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) --- src/cdp/daemon-transport.ts | 178 -------------------------------------------- src/cdp/event-queue.ts | 47 ------------ src/cdp/pending-requests.ts | 47 ++++++++++++ 3 files changed, 47 insertions(+), 225 deletions(-) delete mode 100644 src/cdp/daemon-transport.ts create mode 100644 src/cdp/pending-requests.ts (limited to 'src/cdp') 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); + }; -- cgit v1.2.3