From 893a9d338ceb5d37aefece1df637c9a5888a120e Mon Sep 17 00:00:00 2001 From: amankumarsingh77 Date: Tue, 28 Jul 2026 21:50:38 +0530 Subject: refactor: one CDP transport, not two MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit createCdpTransport was a second full WebSocket-to-Chrome implementation reachable only through client.ts's `?? createCdpTransport()` fallback, which production never took — every session and every automated test passes the daemon transport. transport.ts now holds only the CdpTransport interface, and the client requires a transport instead of inventing one. The "Session with given id not found" match that both transports carried becomes classifyRemoteError in cdp/errors.ts, next to the error kind it produces. test/manual/profile-concurrency-test.ts was the sole caller of the fallback: it drives its own Chrome on a private port. It now spawns a daemon per scenario with BU_CDP_WS set, and refuses to run when a daemon is already up, since that one is bound to a different browser. Co-Authored-By: Claude Opus 5 (1M context) --- src/cdp/daemon-transport.ts | 14 +--- src/cdp/errors.ts | 4 + src/cdp/transport.ts | 136 +------------------------------- src/client.ts | 5 +- test/manual/profile-concurrency-test.ts | 38 ++++++++- 5 files changed, 50 insertions(+), 147 deletions(-) diff --git a/src/cdp/daemon-transport.ts b/src/cdp/daemon-transport.ts index 14f78c4..eb2bbd5 100644 --- a/src/cdp/daemon-transport.ts +++ b/src/cdp/daemon-transport.ts @@ -1,8 +1,8 @@ import { createConnection, type Socket } from "node:net"; import { createInterface } from "node:readline"; import { type Result, err, ok } from "../util/result"; -import { type CdpError, cdpError } from "../cdp/errors"; -import type { CdpTransport } from "../cdp/transport"; +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 { @@ -91,13 +91,7 @@ export const createDaemonTransport = (clientId: string): CdpTransport => { clearTimeout(p.timer); if (msg.error) { - p.resolve(err(cdpError( - msg.error.message.includes("Session with given id not found") - ? "session_not_found" - : "remote_error", - msg.error.message, - p.method, - ))); + p.resolve(err(cdpError(classifyRemoteError(msg.error.message), msg.error.message, p.method))); } else { p.resolve(ok(msg.result)); } @@ -149,7 +143,7 @@ export const createDaemonTransport = (clientId: string): CdpTransport => { params: Record, opts?: { sessionId?: string | null; timeoutMs?: number }, ): Promise> => { - const timeoutMs = opts?.timeoutMs ?? 15_000; + const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; const sessionId = opts?.sessionId ?? null; const sock = socket; diff --git a/src/cdp/errors.ts b/src/cdp/errors.ts index cf1dab7..bf77bbd 100644 --- a/src/cdp/errors.ts +++ b/src/cdp/errors.ts @@ -12,6 +12,10 @@ export type CdpError = { readonly method?: string; }; +// Chrome reports a dead session as a generic protocol error, so the message text is the only signal that a reattach — not a retry — is what will fix it. +export const classifyRemoteError = (message: string): CdpErrorKind => + message.includes("Session with given id not found") ? "session_not_found" : "remote_error"; + export const cdpError = ( kind: CdpErrorKind, message: string, diff --git a/src/cdp/transport.ts b/src/cdp/transport.ts index 33b9412..4036ff7 100644 --- a/src/cdp/transport.ts +++ b/src/cdp/transport.ts @@ -1,11 +1,8 @@ -import WebSocket from "ws"; -import { type Result, err, ok } from "../util/result"; -import { type CdpError, cdpError } from "./errors"; -import { isCdpRawMessage } from "./types"; -import type { CdpEvent, CdpRawMessage } from "./types"; -import { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue"; +import type { Result } from "../util/result"; +import type { CdpError } from "./errors"; +import type { CdpEvent } from "./types"; -const DEFAULT_TIMEOUT_MS = 15_000; +export const DEFAULT_TIMEOUT_MS = 15_000; export type CdpTransport = { connect(url: string, opts?: { timeoutMs?: number }): Promise>; @@ -19,128 +16,3 @@ export type CdpTransport = { state(): "open" | "closed" | "connecting"; onClose(cb: () => void): () => void; }; - -export const createCdpTransport = (): CdpTransport => { - let ws: WebSocket | null = null; - let nextId = 1; - const pending = new Map(); - const closeListeners = new Set<() => void>(); - - let queue = makeEventQueue(); - - const handleMessage = (raw: string): void => { - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch { - return; - } - if (!isCdpRawMessage(parsed)) return; - const msg: CdpRawMessage = parsed; - if (msg.id === undefined) { - if (msg.method) { - queue.push({ - method: msg.method, - params: msg.params, - ...(msg.sessionId !== undefined ? { sessionId: msg.sessionId } : {}), - }); - } - return; - } - const p = pending.get(msg.id); - if (!p) return; - pending.delete(msg.id); - clearTimeout(p.timer); - if (msg.error) { - const kind = msg.error.message.includes("Session with given id not found") - ? "session_not_found" - : "remote_error"; - p.resolve(err(cdpError(kind, msg.error.message, p.method))); - return; - } - p.resolve(ok(msg.result)); - }; - - const cleanup = (reason: string): void => { - rejectAllPending(pending, reason); - queue.end(); - queue = makeEventQueue(); - for (const cb of closeListeners) cb(); - }; - - return { - connect(url, opts = {}): Promise> { - const timeoutMs = opts.timeoutMs ?? 10_000; - return new Promise((resolve) => { - let settled = false; - const settle = (r: Result): void => { - if (settled) return; - settled = true; - resolve(r); - }; - try { - ws = new WebSocket(url); - } catch (e) { - settle(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e)))); - return; - } - const timer = setTimeout(() => { - ws?.close(); - ws = null; - settle(err(cdpError("timeout", `CDP WebSocket connection timed out after ${timeoutMs}ms`))); - }, timeoutMs); - ws.onopen = () => { - clearTimeout(timer); - settle(ok(undefined)); - }; - ws.onmessage = (ev: WebSocket.MessageEvent) => { - const data = ev.data; - handleMessage(typeof data === "string" ? data : data.toString()); - }; - ws.onerror = () => { - clearTimeout(timer); - settle(err(cdpError("transport_closed", "CDP WebSocket error during connection"))); - }; - ws.onclose = () => { - clearTimeout(timer); - ws = null; - cleanup("WebSocket closed"); - settle(err(cdpError("transport_closed", "CDP WebSocket closed"))); - }; - }); - }, - close(): Promise { - if (ws) { - try { ws.close(1000, "Shutdown"); } catch {} - ws = null; - } - cleanup("close() called"); - return Promise.resolve(); - }, - request(method, params, opts = {}): Promise> { - const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; - const sessionId = opts.sessionId ?? null; - if (!ws || ws.readyState !== WebSocket.OPEN) { - return Promise.resolve(err(cdpError("transport_closed", "Browser not connected. Is Chrome running?", method))); - } - // Bind to a local const so the non-null holds across microtasks between the readyState check and send(). - const sock = ws; - const id = nextId++; - const payload: Record = { id, method, params }; - if (sessionId) payload["sessionId"] = sessionId; - const json = JSON.stringify(payload); - return sendWithTimeout(pending, id, method, timeoutMs, "CDP", () => sock.send(json)); - }, - events(): AsyncIterable { - // Single-consumer: a second for-await loop over this iterable silently steals events from the first. - return queue.iter; - }, - state(): "open" | "closed" | "connecting" { - if (!ws) return "closed"; - if (ws.readyState === WebSocket.CONNECTING) return "connecting"; - if (ws.readyState === WebSocket.OPEN) return "open"; - return "closed"; - }, - onClose: makeOnClose(closeListeners), - }; -}; diff --git a/src/client.ts b/src/client.ts index 5367489..f1d2cb7 100644 --- a/src/client.ts +++ b/src/client.ts @@ -7,7 +7,6 @@ import { type Mutex, createMutex } from "./util/mutex"; import { discoverEndpoint } from "./cdp/discovery"; import { type CdpError, cdpError } from "./cdp/errors"; import type { CdpTransport } from "./cdp/transport"; -import { createCdpTransport } from "./cdp/transport"; import { type CdpSession, createCdpSession } from "./cdp/session"; import { type OwnershipRegistry, createOwnershipRegistry } from "./cdp/ownership"; import { ensureHarnessWindow, openHarnessTab } from "./cdp/target-factory"; @@ -17,7 +16,7 @@ import type { DaemonStatus, DialogInfo, PageInfo, TabInfo } from "./cdp/types"; export type BrowserClientOptions = { readonly namespace: string; - readonly transport?: CdpTransport; + readonly transport: CdpTransport; readonly remote?: { readonly cdpUrl: string; readonly browserId: string }; readonly initialOwnership?: { readonly ownedTargetIds?: ReadonlyArray; @@ -96,7 +95,7 @@ const parsePageInfoPayload = (raw: string): Result => { }; export const createBrowserClient = (opts: BrowserClientOptions): BrowserClient => { - const transport = opts.transport ?? createCdpTransport(); + const transport = opts.transport; const ownershipInit: { ownedTargetIds?: ReadonlyArray; harnessWindowTargetId?: string; harnessWindowId?: number } = {}; if (opts.initialOwnership?.ownedTargetIds !== undefined) { ownershipInit.ownedTargetIds = opts.initialOwnership.ownedTargetIds; diff --git a/test/manual/profile-concurrency-test.ts b/test/manual/profile-concurrency-test.ts index 1a5771d..43cb880 100644 --- a/test/manual/profile-concurrency-test.ts +++ b/test/manual/profile-concurrency-test.ts @@ -3,6 +3,8 @@ 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 { isDaemonRunning, spawnDaemon } from "../../src/daemon/spawn"; import { closeIsolatedTab, openIsolatedTab } from "../../src/domains/isolated-tab"; const PORT = Number(process.env["PROFILE_CONC_PORT"] ?? 9488); @@ -56,6 +58,25 @@ const killTree = (pid: number | undefined): void => { } }; +// The daemon owns the only socket to Chrome and reads BU_CDP_WS once, at spawn — so each scenario needs its own daemon pointed at its own browser. +const startDaemonFor = async (endpoint: string): Promise<{ readonly pid: number | undefined } | null> => { + process.env["BU_CDP_WS"] = endpoint; + const child = spawnDaemon(); + if (!child) return null; + for (let i = 0; i < 50; i++) { + if (await isDaemonRunning()) return { pid: child.pid }; + await sleep(200); + } + killTree(child.pid); + return null; +}; + +const stopDaemon = async (pid: number | undefined): Promise => { + killTree(pid); + for (let i = 0; i < 50 && (await isDaemonRunning()); i++) await sleep(200); + delete process.env["BU_CDP_WS"]; +}; + async function scenario(exePath: string, label: string, pinned: boolean): Promise { const udd = mkdtempSync(join(tmpdir(), "pi-profile-conc-")); const base = [ @@ -72,6 +93,7 @@ async function scenario(exePath: string, label: string, pinned: boolean): Promis }); browser.unref(); + let daemonPid: number | undefined; try { let endpoint: string | undefined; for (let i = 0; i < 60 && !endpoint; i++) { @@ -94,9 +116,15 @@ async function scenario(exePath: string, label: string, pinned: boolean): Promis await sleep(6_000); } - process.env["BU_CDP_WS"] = endpoint; + const daemon = await startDaemonFor(endpoint); + check(daemon !== null, `${label}: daemon started against the test browser`); + if (!daemon) return; + daemonPid = daemon.pid; + + const namespace = `conc-${pinned ? "pinned" : "plain"}`; const client = createBrowserClient({ - namespace: `conc-${pinned ? "pinned" : "plain"}`, + namespace, + transport: createDaemonTransport(namespace), ...(pinned ? { profilePin: { userDataDir: udd, profileDir: "Profile 2", label: "pinned", savedAt: "" } } : {}), @@ -138,6 +166,7 @@ async function scenario(exePath: string, label: string, pinned: boolean): Promis await client.closeOwnedTabs(); await client.stop(); } finally { + await stopDaemon(daemonPid); killTree(browser.pid); try { rmSync(udd, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }); @@ -151,6 +180,11 @@ async function main(): Promise { console.error("No Chrome/Chromium found — set CHROME_BIN. Skipping."); process.exit(process.env["CI"] ? 1 : 0); } + // A daemon that is already up is bound to a different Chrome, and the socket path is fixed — this test would silently drive the wrong browser. + if (await isDaemonRunning()) { + console.error("A pi browser daemon is already running. Stop it before running this test."); + process.exit(1); + } console.log(`browser: ${exePath}\n`); await scenario(exePath, "UNPINNED", false); -- cgit v1.2.3