From b913f2d4af218ae61de7073778feb56da89a15de Mon Sep 17 00:00:00 2001 From: amankumarsingh77 Date: Tue, 28 Jul 2026 22:08:16 +0530 Subject: refactor: daemon internals, and fix a leak the merge exposed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IdMultiplexer and EventRouter were the only classes in src/; they are now createIdMultiplexer/createEventRouter closures like every other factory here. The multiplexer kept {clientId, localId} per daemon id while the bridge kept {send, clientId, localId, timer, isAttach} under the same key — two maps of overlapping data, and mux.resolve()'s return value was discarded at three of four call sites. One map now holds the union. That merge fixes a real leak: removeClient cleared the multiplexer but never the callbacks map, so a disconnected client's timers stayed armed and later fired a response at a socket that was gone. clearClient now cancels them, and a test pins it. handleRequest polled every 250ms for up to 15s per request when Chrome was down; N queued requests now await one connect signal. server.ts uses its own protocol's serialize/deserialize instead of hand-rolled JSON, and broadcast no longer routes through `this` — the only `this` in the codebase, and the one thing that breaks when the server object is destructured. cdp/event-queue.ts held two unrelated concerns; the request/timeout half moves to cdp/pending-requests.ts. cdp/daemon-transport.ts is the client half of the daemon protocol, so it moves to daemon/transport.ts. test/daemon/ needs adding to the test glob to run at all — there was no daemon test directory before because none of this was reachable without live Chrome. Co-Authored-By: Claude Opus 5 (1M context) --- package.json | 2 +- src/cdp/daemon-transport.ts | 178 --------------------- src/cdp/event-queue.ts | 47 ------ src/cdp/pending-requests.ts | 47 ++++++ src/daemon/bridge.ts | 265 +++++++++++++++++--------------- src/daemon/server.ts | 37 ++--- src/daemon/transport.ts | 179 +++++++++++++++++++++ src/index.ts | 2 +- test/cdp/event-queue-test.ts | 3 +- test/daemon/bridge-routing-test.ts | 131 ++++++++++++++++ test/manual/daemon-transport-test.ts | 2 +- test/manual/e2e-test.ts | 2 +- test/manual/forms-test.ts | 2 +- test/manual/live-daemon-test.ts | 2 +- test/manual/profile-concurrency-test.ts | 2 +- 15 files changed, 519 insertions(+), 382 deletions(-) delete mode 100644 src/cdp/daemon-transport.ts create mode 100644 src/cdp/pending-requests.ts create mode 100644 src/daemon/transport.ts create mode 100644 test/daemon/bridge-routing-test.ts diff --git a/package.json b/package.json index 1ea453e..effbedf 100644 --- a/package.json +++ b/package.json @@ -58,7 +58,7 @@ }, "scripts": { "typecheck": "tsc --noEmit && tsc -p tsconfig.test.json", - "test": "node --conditions=import --import tsx --test \"test/{profile,domains,cdp,util}/**/*-test.ts\"", + "test": "node --conditions=import --import tsx --test \"test/{profile,domains,cdp,daemon,util}/**/*-test.ts\"", "boundaries": "tsx scripts/check-boundaries.ts", "check": "npm run typecheck && npm run boundaries && npm test", "prepublishOnly": "tsc --noEmit" diff --git a/src/cdp/daemon-transport.ts b/src/cdp/daemon-transport.ts deleted file mode 100644 index eb2bbd5..0000000 --- a/src/cdp/daemon-transport.ts +++ /dev/null @@ -1,178 +0,0 @@ -import { createConnection, type Socket } from "node:net"; -import { createInterface } from "node:readline"; -import { type Result, err, ok } from "../util/result"; -import { type CdpError, cdpError, classifyRemoteError } from "../cdp/errors"; -import { DEFAULT_TIMEOUT_MS, type CdpTransport } from "../cdp/transport"; -import type { CdpEvent } from "../cdp/types"; -import { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue"; -import { - DAEMON_SOCKET_PATH, - type WireRequest, - type WireControl, - deserialize, - serialize, -} from "../daemon/protocol"; - -export const createDaemonTransport = (clientId: string): CdpTransport => { - let socket: Socket | null = null; - let rl: ReturnType | null = null; - let queue = makeEventQueue(); - const closeListeners = new Set<() => void>(); - const pending = new Map(); - let registered = false; - let nextRequestId = 1; - - const cleanup = (reason: string): void => { - rejectAllPending(pending, reason); - queue.end(); - queue = makeEventQueue(); - registered = false; - - if (rl) { rl.close(); rl = null; } - if (socket) { try { socket.destroy(); } catch {} socket = null; } - - for (const cb of closeListeners) cb(); - }; - - const connect = (_url: string, opts?: { timeoutMs?: number }): Promise> => { - const timeoutMs = opts?.timeoutMs ?? 10_000; - - if (socket && !socket.destroyed && registered) { - return Promise.resolve(ok(undefined)); - } - - return new Promise((resolve) => { - let settled = false; - const settle = (r: Result): void => { - if (settled) return; - settled = true; - resolve(r); - }; - - let sock: Socket; - try { - sock = createConnection(DAEMON_SOCKET_PATH); - } catch (e) { - settle(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e)))); - return; - } - socket = sock; - - const connectTimer = setTimeout(() => { - cleanup("Connection timeout"); - settle(err(cdpError("timeout", `Daemon connection timed out after ${timeoutMs}ms`))); - }, timeoutMs); - - sock.on("connect", () => { - clearTimeout(connectTimer); - rl = createInterface({ input: sock, crlfDelay: Infinity }); - - rl.on("line", (line: string) => { - const msg = deserialize(line); - if (!msg) return; - - if (msg.type === "control" && msg.action === "registered" && msg.clientId === clientId) { - registered = true; - settle(ok(undefined)); - return; - } - - if (msg.type === "control" && msg.action === "shutdown") { - cleanup(`Daemon shutting down: ${msg.reason ?? "unknown"}`); - return; - } - - if (!registered) return; - - if (msg.type === "response") { - const p = pending.get(msg.id); - if (!p) return; - pending.delete(msg.id); - clearTimeout(p.timer); - - if (msg.error) { - p.resolve(err(cdpError(classifyRemoteError(msg.error.message), msg.error.message, p.method))); - } else { - p.resolve(ok(msg.result)); - } - return; - } - - if (msg.type === "event") { - queue.push({ - method: msg.method, - params: msg.params, - ...(msg.sessionId !== undefined ? { sessionId: msg.sessionId } : {}), - }); - return; - } - }); - - sock.on("error", () => {}); - - sock.on("close", () => { - clearTimeout(connectTimer); - cleanup("Daemon socket closed"); - settle(err(cdpError("transport_closed", "Daemon socket closed before registration"))); - }); - - const regMsg: WireControl = { type: "control", action: "register", clientId }; - sock.write(serialize(regMsg) + "\n"); - }); - - sock.on("error", (e: NodeJS.ErrnoException) => { - clearTimeout(connectTimer); - const msg = e.code === "ENOENT" - ? "Daemon not running — socket not found" - : e.message; - settle(err(cdpError("transport_closed", msg))); - }); - }); - }; - - const close = async (): Promise => { - if (registered && socket && !socket.destroyed) { - const dereg: WireControl = { type: "control", action: "deregister", clientId }; - try { socket.write(serialize(dereg) + "\n"); } catch {} - } - cleanup("close() called"); - }; - - const request = ( - method: string, - params: Record, - opts?: { sessionId?: string | null; timeoutMs?: number }, - ): Promise> => { - const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; - const sessionId = opts?.sessionId ?? null; - - const sock = socket; - if (sock === null || sock.destroyed || !registered) { - return Promise.resolve(err(cdpError("transport_closed", "Daemon not connected", method))); - } - - const id = nextRequestId++; - const req: WireRequest = { - type: "request", - id, - method, - params, - ...(sessionId !== null && sessionId !== undefined ? { sessionId } : {}), - }; - - return sendWithTimeout(pending, id, method, timeoutMs, "Daemon", () => sock.write(serialize(req) + "\n")); - }; - - const events = (): AsyncIterable => queue.iter; - - const state = (): "open" | "closed" | "connecting" => { - if (!socket) return "closed"; - if (!registered) return "connecting"; - if (socket.destroyed) return "closed"; - return "open"; - }; - - const onClose = makeOnClose(closeListeners); - - return { connect, close, request, events, state, onClose }; -}; diff --git a/src/cdp/event-queue.ts b/src/cdp/event-queue.ts index 330bdb8..08024b0 100644 --- a/src/cdp/event-queue.ts +++ b/src/cdp/event-queue.ts @@ -1,9 +1,5 @@ -import { type Result, err } from "../util/result"; -import { type CdpError, cdpError } from "./errors"; import type { CdpEvent } from "./types"; -export type CdpResult = Result; - export type EventQueue = { readonly push: (e: CdpEvent) => void; readonly end: () => void; @@ -40,46 +36,3 @@ export const makeEventQueue = (): EventQueue => { }, }; }; - -export type Pending = { - readonly resolve: (v: Result) => void; - readonly timer: ReturnType; - readonly method: string; -}; - -export const rejectAllPending = (pending: Map, reason: string): void => { - for (const [, p] of pending) { - clearTimeout(p.timer); - p.resolve(err(cdpError("transport_closed", reason, p.method))); - } - pending.clear(); -}; - -export const sendWithTimeout = ( - pending: Map, - id: number, - method: string, - timeoutMs: number, - timeoutLabel: string, - send: () => void, -): Promise => - new Promise((resolve) => { - const timer = setTimeout(() => { - pending.delete(id); - resolve(err(cdpError("timeout", `${timeoutLabel} timeout after ${timeoutMs}ms: ${method}`, method))); - }, timeoutMs); - pending.set(id, { resolve, timer, method }); - try { - send(); - } catch (e) { - clearTimeout(timer); - pending.delete(id); - resolve(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e), method))); - } - }); - -export const makeOnClose = (closeListeners: Set<() => void>) => - (cb: () => void): (() => void) => { - closeListeners.add(cb); - return () => closeListeners.delete(cb); - }; diff --git a/src/cdp/pending-requests.ts b/src/cdp/pending-requests.ts new file mode 100644 index 0000000..03b19cd --- /dev/null +++ b/src/cdp/pending-requests.ts @@ -0,0 +1,47 @@ +import { type Result, err } from "../util/result"; +import { type CdpError, cdpError } from "./errors"; + +export type CdpResult = Result; + +export type Pending = { + readonly resolve: (v: Result) => void; + readonly timer: ReturnType; + readonly method: string; +}; + +export const rejectAllPending = (pending: Map, reason: string): void => { + for (const [, p] of pending) { + clearTimeout(p.timer); + p.resolve(err(cdpError("transport_closed", reason, p.method))); + } + pending.clear(); +}; + +export const sendWithTimeout = ( + pending: Map, + id: number, + method: string, + timeoutMs: number, + timeoutLabel: string, + send: () => void, +): Promise => + new Promise((resolve) => { + const timer = setTimeout(() => { + pending.delete(id); + resolve(err(cdpError("timeout", `${timeoutLabel} timeout after ${timeoutMs}ms: ${method}`, method))); + }, timeoutMs); + pending.set(id, { resolve, timer, method }); + try { + send(); + } catch (e) { + clearTimeout(timer); + pending.delete(id); + resolve(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e), method))); + } + }); + +export const makeOnClose = (closeListeners: Set<() => void>) => + (cb: () => void): (() => void) => { + closeListeners.add(cb); + return () => closeListeners.delete(cb); + }; diff --git a/src/daemon/bridge.ts b/src/daemon/bridge.ts index d9a54ed..6503aa4 100644 --- a/src/daemon/bridge.ts +++ b/src/daemon/bridge.ts @@ -23,96 +23,127 @@ export type CdpBridge = { onClose(handler: CloseHandler): void; }; -interface PendingEntry { - clientId: string; - localId: number; -} - -class IdMultiplexer { - private nextId = 1; - private localToRemote = new Map>(); - private pending = new Map(); - - allocate(clientId: string, localId: number): number { - const daemonId = this.nextId++; - let m = this.localToRemote.get(clientId); - if (!m) { m = new Map(); this.localToRemote.set(clientId, m); } - m.set(localId, daemonId); - this.pending.set(daemonId, { clientId, localId }); - return daemonId; - } - - resolve(daemonId: number): PendingEntry | null { - const e = this.pending.get(daemonId); - if (!e) return null; - this.pending.delete(daemonId); - const m = this.localToRemote.get(e.clientId); - if (m) m.delete(e.localId); - return e; - } - - clearClient(clientId: string): void { - const m = this.localToRemote.get(clientId); - if (m) { for (const [, did] of m) this.pending.delete(did); } - this.localToRemote.delete(clientId); - } -} - -class EventRouter { - private owners = new Map(); - private clientSessions = new Map>(); - - record(clientId: string, sessionId: string): void { - const prev = this.owners.get(sessionId); - if (prev && prev !== clientId) { - this.clientSessions.get(prev)?.delete(sessionId); - } - this.owners.set(sessionId, clientId); - let s = this.clientSessions.get(clientId); - if (!s) { s = new Set(); this.clientSessions.set(clientId, s); } - s.add(sessionId); - } - - release(sessionId: string): void { - const owner = this.owners.get(sessionId); - if (owner) this.clientSessions.get(owner)?.delete(sessionId); - this.owners.delete(sessionId); - } - - getOwner(sessionId: string): string | undefined { - return this.owners.get(sessionId); - } - - removeClient(clientId: string): void { - const sessions = this.clientSessions.get(clientId); - if (sessions) { for (const sid of sessions) this.owners.delete(sid); } - this.clientSessions.delete(clientId); - } - - route(event: WireEvent): string[] { - if (event.sessionId) { - const owner = this.owners.get(event.sessionId); - return owner ? [owner] : []; - } - return [...this.clientSessions.keys()]; - } -} +export type InFlight = { + readonly clientId: string; + readonly localId: number; + readonly send: SendToClient; + readonly timer: ReturnType; + readonly isAttach: boolean; +}; + +// One map, keyed by the daemon-side id: the client that asked, the id it used, and everything needed to answer it. +export type IdMultiplexer = { + allocate(entry: Omit, arm: (daemonId: number) => InFlight["timer"]): number; + take(daemonId: number): InFlight | undefined; + takeAll(): ReadonlyArray; + clearClient(clientId: string): void; +}; + +export const createIdMultiplexer = (): IdMultiplexer => { + let nextId = 1; + const inFlight = new Map(); + + return { + allocate(entry, arm) { + const daemonId = nextId++; + inFlight.set(daemonId, { ...entry, timer: arm(daemonId) }); + return daemonId; + }, + take(daemonId) { + const e = inFlight.get(daemonId); + if (!e) return undefined; + inFlight.delete(daemonId); + clearTimeout(e.timer); + return e; + }, + takeAll() { + const all = [...inFlight.values()]; + inFlight.clear(); + for (const e of all) clearTimeout(e.timer); + return all; + }, + // A disconnected client must not be answered later: dropping its entries also cancels their timeouts. + clearClient(clientId) { + for (const [daemonId, e] of inFlight) { + if (e.clientId !== clientId) continue; + clearTimeout(e.timer); + inFlight.delete(daemonId); + } + }, + }; +}; + +export type EventRouter = { + record(clientId: string, sessionId: string): void; + release(sessionId: string): void; + getOwner(sessionId: string): string | undefined; + removeClient(clientId: string): void; + route(event: WireEvent): string[]; +}; + +export const createEventRouter = (): EventRouter => { + const owners = new Map(); + const clientSessions = new Map>(); + + return { + record(clientId, sessionId) { + const prev = owners.get(sessionId); + if (prev && prev !== clientId) clientSessions.get(prev)?.delete(sessionId); + owners.set(sessionId, clientId); + let s = clientSessions.get(clientId); + if (!s) { s = new Set(); clientSessions.set(clientId, s); } + s.add(sessionId); + }, + release(sessionId) { + const owner = owners.get(sessionId); + if (owner) clientSessions.get(owner)?.delete(sessionId); + owners.delete(sessionId); + }, + getOwner(sessionId) { + return owners.get(sessionId); + }, + removeClient(clientId) { + const sessions = clientSessions.get(clientId); + if (sessions) { for (const sid of sessions) owners.delete(sid); } + clientSessions.delete(clientId); + }, + route(event) { + if (event.sessionId) { + const owner = owners.get(event.sessionId); + return owner ? [owner] : []; + } + return [...clientSessions.keys()]; + }, + }; +}; + +const CONNECT_WAIT_MS = 15_000; export const createCdpBridge = (): CdpBridge => { let ws: WebSocket | null = null; let wsUrl: string | null = null; - const mux = new IdMultiplexer(); - const router = new EventRouter(); + const mux = createIdMultiplexer(); + const router = createEventRouter(); let eventHandler: EventHandler | null = null; let closeHandler: CloseHandler | null = null; - const callbacks = new Map; - isAttach: boolean; - }>(); + const connectedWaiters: Array<() => void> = []; + const signalConnected = (): void => { + for (const w of connectedWaiters.splice(0)) w(); + }; + + const waitForConnection = (timeoutMs: number): Promise => + new Promise((resolve) => { + let settled = false; + const finish = (): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(); + }; + const timer = setTimeout(finish, timeoutMs); + connectedWaiters.push(finish); + }); const onChromeMessage = (raw: string): void => { let parsed: unknown; @@ -121,14 +152,8 @@ export const createCdpBridge = (): CdpBridge => { const msg: CdpRawMessage = parsed; if (msg.id !== undefined) { - const cb = callbacks.get(msg.id); - if (!cb) { - mux.resolve(msg.id); - return; - } - clearTimeout(cb.timer); - callbacks.delete(msg.id); - + const cb = mux.take(msg.id); + if (!cb) return; const localId = cb.localId; if (msg.error) { @@ -146,8 +171,6 @@ export const createCdpBridge = (): CdpBridge => { } cb.send(cb.clientId, { type: "response", id: localId, result: msg.result }); } - - mux.resolve(msg.id); return; } @@ -221,6 +244,7 @@ export const createCdpBridge = (): CdpBridge => { reconnectAttempt = 0; ws.send(JSON.stringify({ id: 0, method: "Target.setDiscoverTargets", params: { discover: true } })); console.log("[pi-browser-daemon] Connected to Chrome ✓"); + signalConnected(); settleOnce(); }); @@ -236,15 +260,12 @@ export const createCdpBridge = (): CdpBridge => { sock.on("close", () => { clearTimeout(timer); ws = null; - for (const [daemonId, cb] of callbacks) { - clearTimeout(cb.timer); + for (const cb of mux.takeAll()) { cb.send(cb.clientId, { type: "response", id: cb.localId, error: { code: -32000, message: "Chrome disconnected" }, }); - callbacks.delete(daemonId); - mux.resolve(daemonId); } closeHandler?.(); scheduleRetry(); @@ -276,8 +297,8 @@ export const createCdpBridge = (): CdpBridge => { const stop = async (): Promise => { stopped = true; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } - for (const [, cb] of callbacks) clearTimeout(cb.timer); - callbacks.clear(); + mux.takeAll(); + signalConnected(); if (ws) { try { ws.close(1000, "Shutdown"); } catch {} ws = null; } wsUrl = null; }; @@ -287,13 +308,8 @@ export const createCdpBridge = (): CdpBridge => { clientId: string, send: SendToClient, ): Promise => { - if (!ws || ws.readyState !== WebSocket.OPEN) { - const deadline = Date.now() + 15_000; - while (Date.now() < deadline) { - await new Promise((r) => setTimeout(r, 250)); - if (ws && ws.readyState === WebSocket.OPEN) break; - } - } + // Every queued request awaits the same connect signal rather than each spinning its own poll loop. + if (!ws || ws.readyState !== WebSocket.OPEN) await waitForConnection(CONNECT_WAIT_MS); if (!ws || ws.readyState !== WebSocket.OPEN) { send(clientId, { @@ -304,26 +320,21 @@ export const createCdpBridge = (): CdpBridge => { return; } - const daemonId = mux.allocate(clientId, req.id); - - const timer = setTimeout(() => { - callbacks.delete(daemonId); - mux.resolve(daemonId); - send(clientId, { - type: "response", - id: req.id, - error: { code: -32000, message: `Timeout after ${CDP_COMMAND_TIMEOUT_MS}ms: ${req.method}` }, - }); - }, CDP_COMMAND_TIMEOUT_MS); - - const isAttach = req.method === "Target.attachToTarget"; const isDetach = req.method === "Target.detachFromTarget"; - - if (isDetach && req.sessionId) { - router.release(req.sessionId); - } - - callbacks.set(daemonId, { send, clientId, localId: req.id, timer, isAttach }); + if (isDetach && req.sessionId) router.release(req.sessionId); + + const daemonId = mux.allocate( + { clientId, localId: req.id, send, isAttach: req.method === "Target.attachToTarget" }, + (id) => + setTimeout(() => { + mux.take(id); + send(clientId, { + type: "response", + id: req.id, + error: { code: -32000, message: `Timeout after ${CDP_COMMAND_TIMEOUT_MS}ms: ${req.method}` }, + }); + }, CDP_COMMAND_TIMEOUT_MS), + ); const payload: Record = { id: daemonId, @@ -335,9 +346,7 @@ export const createCdpBridge = (): CdpBridge => { try { ws.send(JSON.stringify(payload)); } catch (e) { - clearTimeout(timer); - callbacks.delete(daemonId); - mux.resolve(daemonId); + mux.take(daemonId); send(clientId, { type: "response", id: req.id, diff --git a/src/daemon/server.ts b/src/daemon/server.ts index 142bddb..b98f37e 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -4,7 +4,8 @@ import { unlinkSync } from "node:fs"; import { type WireMessage, type WireControl, - isWireMessage, + deserialize, + serialize, DAEMON_SOCKET_PATH, DAEMON_MAX_CLIENTS, } from "./protocol"; @@ -59,14 +60,8 @@ export const createIpcServer = (): IpcServer => { sockets.set(tempId, currentClient); rl.on("line", (line: string) => { - let msg: WireMessage | null; - try { - const parsed: unknown = JSON.parse(line); - if (!isWireMessage(parsed)) return; - msg = parsed; - } catch { - return; - } + const msg = deserialize(line); + if (!msg) return; if (msg.type === "control" && msg.action === "register" && !currentClient.registered) { const clientId = msg.clientId; @@ -78,7 +73,7 @@ export const createIpcServer = (): IpcServer => { action: "shutdown", reason: `clientId ${clientId} is already connected`, }; - socket.write(JSON.stringify(reply) + "\n"); + socket.write(serialize(reply) + "\n"); socket.destroy(); return; } @@ -90,7 +85,7 @@ export const createIpcServer = (): IpcServer => { clients.set(clientId, currentClient); const ack: WireControl = { type: "control", action: "registered", clientId }; - socket.write(JSON.stringify(ack) + "\n"); + socket.write(serialize(ack) + "\n"); connectHandler?.(currentClient); return; @@ -146,6 +141,14 @@ export const createIpcServer = (): IpcServer => { } }); + const send = (clientId: string, msg: WireMessage): void => { + const client = clients.get(clientId); + if (!client) return; + try { + client.socket.write(serialize(msg) + "\n"); + } catch {} + }; + return { start, stop, @@ -158,17 +161,9 @@ export const createIpcServer = (): IpcServer => { onDisconnect(handler) { disconnectHandler = handler; }, - send(clientId, msg) { - const client = clients.get(clientId); - if (!client) return; - try { - client.socket.write(JSON.stringify(msg) + "\n"); - } catch {} - }, + send, broadcast(msg) { - for (const [clientId] of clients) { - this.send(clientId, msg); - } + for (const [clientId] of clients) send(clientId, msg); }, clientCount: (): number => clients.size, }; diff --git a/src/daemon/transport.ts b/src/daemon/transport.ts new file mode 100644 index 0000000..4e75215 --- /dev/null +++ b/src/daemon/transport.ts @@ -0,0 +1,179 @@ +import { createConnection, type Socket } from "node:net"; +import { createInterface } from "node:readline"; +import { type Result, err, ok } from "../util/result"; +import { type CdpError, cdpError, classifyRemoteError } from "../cdp/errors"; +import { DEFAULT_TIMEOUT_MS, type CdpTransport } from "../cdp/transport"; +import type { CdpEvent } from "../cdp/types"; +import { makeEventQueue } from "../cdp/event-queue"; +import { type Pending, makeOnClose, rejectAllPending, sendWithTimeout } from "../cdp/pending-requests"; +import { + DAEMON_SOCKET_PATH, + type WireRequest, + type WireControl, + deserialize, + serialize, +} from "./protocol"; + +export const createDaemonTransport = (clientId: string): CdpTransport => { + let socket: Socket | null = null; + let rl: ReturnType | null = null; + let queue = makeEventQueue(); + const closeListeners = new Set<() => void>(); + const pending = new Map(); + let registered = false; + let nextRequestId = 1; + + const cleanup = (reason: string): void => { + rejectAllPending(pending, reason); + queue.end(); + queue = makeEventQueue(); + registered = false; + + if (rl) { rl.close(); rl = null; } + if (socket) { try { socket.destroy(); } catch {} socket = null; } + + for (const cb of closeListeners) cb(); + }; + + const connect = (_url: string, opts?: { timeoutMs?: number }): Promise> => { + const timeoutMs = opts?.timeoutMs ?? 10_000; + + if (socket && !socket.destroyed && registered) { + return Promise.resolve(ok(undefined)); + } + + return new Promise((resolve) => { + let settled = false; + const settle = (r: Result): void => { + if (settled) return; + settled = true; + resolve(r); + }; + + let sock: Socket; + try { + sock = createConnection(DAEMON_SOCKET_PATH); + } catch (e) { + settle(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e)))); + return; + } + socket = sock; + + const connectTimer = setTimeout(() => { + cleanup("Connection timeout"); + settle(err(cdpError("timeout", `Daemon connection timed out after ${timeoutMs}ms`))); + }, timeoutMs); + + sock.on("connect", () => { + clearTimeout(connectTimer); + rl = createInterface({ input: sock, crlfDelay: Infinity }); + + rl.on("line", (line: string) => { + const msg = deserialize(line); + if (!msg) return; + + if (msg.type === "control" && msg.action === "registered" && msg.clientId === clientId) { + registered = true; + settle(ok(undefined)); + return; + } + + if (msg.type === "control" && msg.action === "shutdown") { + cleanup(`Daemon shutting down: ${msg.reason ?? "unknown"}`); + return; + } + + if (!registered) return; + + if (msg.type === "response") { + const p = pending.get(msg.id); + if (!p) return; + pending.delete(msg.id); + clearTimeout(p.timer); + + if (msg.error) { + p.resolve(err(cdpError(classifyRemoteError(msg.error.message), msg.error.message, p.method))); + } else { + p.resolve(ok(msg.result)); + } + return; + } + + if (msg.type === "event") { + queue.push({ + method: msg.method, + params: msg.params, + ...(msg.sessionId !== undefined ? { sessionId: msg.sessionId } : {}), + }); + return; + } + }); + + sock.on("error", () => {}); + + sock.on("close", () => { + clearTimeout(connectTimer); + cleanup("Daemon socket closed"); + settle(err(cdpError("transport_closed", "Daemon socket closed before registration"))); + }); + + const regMsg: WireControl = { type: "control", action: "register", clientId }; + sock.write(serialize(regMsg) + "\n"); + }); + + sock.on("error", (e: NodeJS.ErrnoException) => { + clearTimeout(connectTimer); + const msg = e.code === "ENOENT" + ? "Daemon not running — socket not found" + : e.message; + settle(err(cdpError("transport_closed", msg))); + }); + }); + }; + + const close = async (): Promise => { + if (registered && socket && !socket.destroyed) { + const dereg: WireControl = { type: "control", action: "deregister", clientId }; + try { socket.write(serialize(dereg) + "\n"); } catch {} + } + cleanup("close() called"); + }; + + const request = ( + method: string, + params: Record, + opts?: { sessionId?: string | null; timeoutMs?: number }, + ): Promise> => { + const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const sessionId = opts?.sessionId ?? null; + + const sock = socket; + if (sock === null || sock.destroyed || !registered) { + return Promise.resolve(err(cdpError("transport_closed", "Daemon not connected", method))); + } + + const id = nextRequestId++; + const req: WireRequest = { + type: "request", + id, + method, + params, + ...(sessionId !== null && sessionId !== undefined ? { sessionId } : {}), + }; + + return sendWithTimeout(pending, id, method, timeoutMs, "Daemon", () => sock.write(serialize(req) + "\n")); + }; + + const events = (): AsyncIterable => queue.iter; + + const state = (): "open" | "closed" | "connecting" => { + if (!socket) return "closed"; + if (!registered) return "connecting"; + if (socket.destroyed) return "closed"; + return "open"; + }; + + const onClose = makeOnClose(closeListeners); + + return { connect, close, request, events, state, onClose }; +}; diff --git a/src/index.ts b/src/index.ts index 6097881..0133f65 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,7 +7,7 @@ import { readPin } from "./profile/store"; import { type BrowserState, defaultState, persistState, restoreState } from "./state"; import { registerAllTools } from "./registry"; import { cleanupTempDirs } from "./util/truncate"; -import { createDaemonTransport } from "./cdp/daemon-transport"; +import { createDaemonTransport } from "./daemon/transport"; import { setDebugClicks } from "./util/debug"; import { asBoolean, asString } from "./util/guards"; diff --git a/test/cdp/event-queue-test.ts b/test/cdp/event-queue-test.ts index cae8143..771ce8e 100644 --- a/test/cdp/event-queue-test.ts +++ b/test/cdp/event-queue-test.ts @@ -1,6 +1,7 @@ import { test, describe } from "node:test"; import assert from "node:assert/strict"; -import { makeEventQueue, rejectAllPending, sendWithTimeout, type Pending } from "../../src/cdp/event-queue"; +import { makeEventQueue } from "../../src/cdp/event-queue"; +import { rejectAllPending, sendWithTimeout, type Pending } from "../../src/cdp/pending-requests"; describe("event queue iteration", () => { test("yields pushed events in order", async () => { diff --git a/test/daemon/bridge-routing-test.ts b/test/daemon/bridge-routing-test.ts new file mode 100644 index 0000000..7c96bad --- /dev/null +++ b/test/daemon/bridge-routing-test.ts @@ -0,0 +1,131 @@ +import { test, describe } from "node:test"; +import assert from "node:assert/strict"; +import { createEventRouter, createIdMultiplexer, type InFlight } from "../../src/daemon/bridge"; +import type { SendToClient } from "../../src/daemon/bridge"; +import type { WireEvent } from "../../src/daemon/protocol"; + +const noSend: SendToClient = () => {}; +const entry = (clientId: string, localId: number): Omit => ({ + clientId, + localId, + send: noSend, + isAttach: false, +}); + +const inertTimer = (): ReturnType => setTimeout(() => {}, 60_000); + +describe("daemon id multiplexer", () => { + test("two clients using the same local id get distinct daemon ids", () => { + const mux = createIdMultiplexer(); + const a = mux.allocate(entry("client-a", 1), inertTimer); + const b = mux.allocate(entry("client-b", 1), inertTimer); + assert.notEqual(a, b); + }); + + test("a response is routed back to the client that asked, with its own id", () => { + const mux = createIdMultiplexer(); + const a = mux.allocate(entry("client-a", 7), inertTimer); + const b = mux.allocate(entry("client-b", 7), inertTimer); + assert.equal(mux.take(b)?.clientId, "client-b"); + const takenA = mux.take(a); + assert.equal(takenA?.clientId, "client-a"); + assert.equal(takenA?.localId, 7); + }); + + test("taking the same daemon id twice yields nothing the second time", () => { + const mux = createIdMultiplexer(); + const id = mux.allocate(entry("client-a", 1), inertTimer); + assert.ok(mux.take(id)); + assert.equal(mux.take(id), undefined); + }); + + test("an unknown daemon id is not mistaken for an in-flight request", () => { + const mux = createIdMultiplexer(); + assert.equal(mux.take(999), undefined); + }); + + test("clearClient drops only that client's in-flight requests", () => { + const mux = createIdMultiplexer(); + const a = mux.allocate(entry("client-a", 1), inertTimer); + const b = mux.allocate(entry("client-b", 1), inertTimer); + mux.clearClient("client-a"); + assert.equal(mux.take(a), undefined); + assert.equal(mux.take(b)?.clientId, "client-b"); + }); + + test("clearClient cancels the pending timeout so a gone client is never answered", () => { + const mux = createIdMultiplexer(); + let fired = false; + mux.allocate(entry("client-a", 1), () => setTimeout(() => { fired = true; }, 1)); + mux.clearClient("client-a"); + return new Promise((resolve) => setTimeout(() => { + assert.equal(fired, false); + resolve(); + }, 30)); + }); + + test("takeAll drains every in-flight request exactly once", () => { + const mux = createIdMultiplexer(); + mux.allocate(entry("client-a", 1), inertTimer); + mux.allocate(entry("client-b", 2), inertTimer); + assert.equal(mux.takeAll().length, 2); + assert.equal(mux.takeAll().length, 0); + }); +}); + +const ev = (sessionId?: string): WireEvent => ({ + type: "event", + method: "Page.loadEventFired", + ...(sessionId !== undefined ? { sessionId } : {}), +}); + +describe("daemon event router", () => { + test("a session-scoped event goes only to the client owning that session", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + router.record("client-b", "S2"); + assert.deepEqual(router.route(ev("S1")), ["client-a"]); + assert.deepEqual(router.route(ev("S2")), ["client-b"]); + }); + + test("an event for an unknown session goes nowhere rather than to everyone", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + assert.deepEqual(router.route(ev("S9")), []); + }); + + test("a session-less event goes to every client that holds a session", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + router.record("client-b", "S2"); + assert.deepEqual(router.route(ev()).sort(), ["client-a", "client-b"]); + }); + + test("re-recording a session moves ownership and does not leave it with the old client", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + router.record("client-b", "S1"); + assert.equal(router.getOwner("S1"), "client-b"); + assert.deepEqual(router.route(ev("S1")), ["client-b"]); + }); + + test("releasing a session leaves its events unrouted", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + router.release("S1"); + assert.equal(router.getOwner("S1"), undefined); + assert.deepEqual(router.route(ev("S1")), []); + }); + + test("removing a client releases every session it owned", () => { + const router = createEventRouter(); + router.record("client-a", "S1"); + router.record("client-a", "S2"); + router.record("client-b", "S3"); + router.removeClient("client-a"); + assert.equal(router.getOwner("S1"), undefined); + assert.equal(router.getOwner("S2"), undefined); + assert.equal(router.getOwner("S3"), "client-b"); + assert.deepEqual(router.route(ev()), ["client-b"]); + }); +}); diff --git a/test/manual/daemon-transport-test.ts b/test/manual/daemon-transport-test.ts index 5738e23..30f5a29 100644 --- a/test/manual/daemon-transport-test.ts +++ b/test/manual/daemon-transport-test.ts @@ -1,5 +1,5 @@ import { createIpcServer } from "../../src/daemon/server"; -import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { createDaemonTransport } from "../../src/daemon/transport"; import { DAEMON_SOCKET_PATH } from "../../src/daemon/protocol"; import type { WireMessage } from "../../src/daemon/protocol"; import { unlinkSync } from "node:fs"; diff --git a/test/manual/e2e-test.ts b/test/manual/e2e-test.ts index 81cb4df..02e04a4 100644 --- a/test/manual/e2e-test.ts +++ b/test/manual/e2e-test.ts @@ -1,6 +1,6 @@ import { createIpcServer } from "../../src/daemon/server"; import { createCdpBridge, type SendToClient } from "../../src/daemon/bridge"; -import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { createDaemonTransport } from "../../src/daemon/transport"; import { DAEMON_SOCKET_PATH } from "../../src/daemon/protocol"; import type { WireRequest } from "../../src/daemon/protocol"; import { unlinkSync } from "node:fs"; diff --git a/test/manual/forms-test.ts b/test/manual/forms-test.ts index 43b2863..274f039 100644 --- a/test/manual/forms-test.ts +++ b/test/manual/forms-test.ts @@ -1,4 +1,4 @@ -import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { createDaemonTransport } from "../../src/daemon/transport"; import { createBrowserClient } from "../../src/client"; import { fillTool, selectOptionTool, setCheckedTool } from "../../src/domains/form"; import { fillFormTool } from "../../src/domains/form-batch"; diff --git a/test/manual/live-daemon-test.ts b/test/manual/live-daemon-test.ts index 28ec7cd..5334cbc 100644 --- a/test/manual/live-daemon-test.ts +++ b/test/manual/live-daemon-test.ts @@ -1,4 +1,4 @@ -import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { createDaemonTransport } from "../../src/daemon/transport"; async function main() { const t = createDaemonTransport("pi-live-test"); diff --git a/test/manual/profile-concurrency-test.ts b/test/manual/profile-concurrency-test.ts index 43cb880..83b233d 100644 --- a/test/manual/profile-concurrency-test.ts +++ b/test/manual/profile-concurrency-test.ts @@ -3,7 +3,7 @@ import { existsSync, mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { createBrowserClient } from "../../src/client"; -import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { createDaemonTransport } from "../../src/daemon/transport"; import { isDaemonRunning, spawnDaemon } from "../../src/daemon/spawn"; import { closeIsolatedTab, openIsolatedTab } from "../../src/domains/isolated-tab"; -- cgit v1.2.3