diff options
| author | Uttkarsh <38376852+uttkarsh-26@users.noreply.github.com> | 2026-08-06 00:20:49 +0530 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2026-08-06 00:20:49 +0530 |
| commit | d9bc46b508f8c6f7e0f0ff5d54d804340f3e6c80 (patch) | |
| tree | 0ea1315ccc0feaee1ddf184ed16f16173455b3b4 /src | |
| parent | 31c1b27a50bc2b9ba7138ae67a4cfaba6945582e (diff) | |
fix: reconnect to Chrome only on explicit browser demand (#18)
* fix: reconnect to Chrome only on browser demand
* fix(daemon): gate stale CDP bridge reconnect attempts
* Harden bridge reconnect lifecycle against stale socket callbacks
* fix(daemon): handle stale attempt failures without deferred settlement
* test: align waitFor assertions and stale-socket case labels
* fix(daemon): re-check socket after connect wait in handleRequest
Restores the post-wait re-check from the original PR branch (dropped during
the main rebase merge); TypeScript needs the local narrowing before send.
* fix(daemon): reject in-flight requests when an open connection drops
`failCurrentAttempt` delegated callback rejection to `settleAttempt`, but an
attempt that reached `open` had already settled, so the close path returned
early and never answered its in-flight requests. They hung until the 10s
command timeout instead of failing immediately, which the on-demand reconnect
change had promised to preserve.
Reject the attempt's callbacks directly in `failCurrentAttempt`, before the
close handler notifies clients.
Move the connection-lifecycle tests out of test/manual/ — which nothing runs —
into test/daemon/, where `npm test` picks them up, and convert the hand-rolled
pass/fail counters to node:test. Adds a case covering the drop path above; it
fails without the fix.
---------
Co-authored-by: Hermes Subagent <hermes-subagent@example.com>
Co-authored-by: Hermes Agent <agent@example.com>
Co-authored-by: uttkarsh-26 <uttkarsh@users.noreply.github.com>
Co-authored-by: amankumarsingh77 <amankumarsingh7702@gmail.com>
Diffstat (limited to 'src')
| -rw-r--r-- | src/daemon/bridge.ts | 361 |
1 files changed, 262 insertions, 99 deletions
diff --git a/src/daemon/bridge.ts b/src/daemon/bridge.ts index 6503aa4..dc10f2b 100644 --- a/src/daemon/bridge.ts +++ b/src/daemon/bridge.ts @@ -5,6 +5,8 @@ import type { CdpRawMessage } from "../cdp/types"; import type { WireRequest, WireResponse, WireEvent } from "./protocol"; import { CDP_CONNECT_TIMEOUT_MS, CDP_COMMAND_TIMEOUT_MS } from "./protocol"; import { asString, isRecord } from "../util/guards"; +import type { CdpError } from "../cdp/errors"; +import type { Result } from "../util/result"; export type SendToClient = (clientId: string, msg: WireResponse | WireEvent) => void; @@ -117,9 +119,22 @@ export const createEventRouter = (): EventRouter => { }; }; -const CONNECT_WAIT_MS = 15_000; +type CdpBridgeDependencies = { + discoverWsUrl?: () => Promise<Result<string, CdpError>>; + createWebSocket?: (url: string) => WebSocket; + connectTimeoutMs?: number; + onAttemptSettled?: () => void; +}; + +export const createCdpBridge = (options: CdpBridgeDependencies = {}): CdpBridge => { + const { + discoverWsUrl: discoverWsUrlImpl = discoverWsUrl, + createWebSocket: createWebSocketImpl = (url: string): WebSocket => + new WebSocket(url, { perMessageDeflate: false }), + connectTimeoutMs = CDP_CONNECT_TIMEOUT_MS, + onAttemptSettled, + } = options; -export const createCdpBridge = (): CdpBridge => { let ws: WebSocket | null = null; let wsUrl: string | null = null; const mux = createIdMultiplexer(); @@ -127,23 +142,15 @@ export const createCdpBridge = (): CdpBridge => { let eventHandler: EventHandler | null = null; let closeHandler: CloseHandler | null = null; - const connectedWaiters: Array<() => void> = []; - const signalConnected = (): void => { - for (const w of connectedWaiters.splice(0)) w(); - }; + // daemonId -> { generation, clientId }: lets a failed attempt reject only its own + // in-flight requests, and a gone client's entries be dropped without late answers. + const generations = new Map<number, { generation: number; clientId: string }>(); - const waitForConnection = (timeoutMs: number): Promise<void> => - 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); - }); + let stopped = false; + let connecting = false; + let attemptGeneration = 0; + let activeAttempt = 0; + let activeSocket: WebSocket | null = null; const onChromeMessage = (raw: string): void => { let parsed: unknown; @@ -152,6 +159,7 @@ export const createCdpBridge = (): CdpBridge => { const msg: CdpRawMessage = parsed; if (msg.id !== undefined) { + generations.delete(msg.id); const cb = mux.take(msg.id); if (!cb) return; const localId = cb.localId; @@ -195,98 +203,215 @@ export const createCdpBridge = (): CdpBridge => { } }; - let reconnectTimer: ReturnType<typeof setTimeout> | null = null; - let reconnectAttempt = 0; - let stopped = false; + const rejectCallbacksWhere = ( + pred: (info: { generation: number; clientId: string }) => boolean, + ): void => { + for (const [daemonId, info] of generations) { + if (!pred(info)) continue; + generations.delete(daemonId); + const cb = mux.take(daemonId); + if (!cb) continue; + cb.send(cb.clientId, { + type: "response", + id: cb.localId, + error: { code: -32000, message: "Chrome disconnected" }, + }); + } + }; + + const rejectAttemptCallbacks = (generation: number): void => { + rejectCallbacksWhere((info) => info.generation === generation); + }; + + const rejectAllPendingCallbacks = (): void => { + rejectCallbacksWhere(() => true); + }; + + const closeSocket = (socket: WebSocket): void => { + try { + socket.close(); + } catch {} + }; + + const isCurrentAttempt = (generation: number): boolean => + generation === activeAttempt && !stopped; + + const isCurrentAttemptSocket = (generation: number, socket: WebSocket): boolean => + isCurrentAttempt(generation) && activeSocket === socket; + + const clearActiveAttempt = (generation: number): void => { + if (activeAttempt === generation) { + activeAttempt = 0; + activeSocket = null; + } + }; + // On-demand Chrome connection. Attempts happen only from start() and from + // handleRequest() while disconnected; a failed or disconnected bridge stays + // disconnected until the next explicit browser demand. No background retry loop. const tryConnect = (): void => { - if (stopped) return; - if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } + if (stopped || connecting || (ws && ws.readyState === WebSocket.OPEN)) return; - const attempt = async (): Promise<void> => { - if (ws && ws.readyState === WebSocket.OPEN) return; + const generation = ++attemptGeneration; + activeAttempt = generation; + connecting = true; - const cachedUrl = wsUrl; - let url: string; - if (cachedUrl === null || cachedUrl === "" || reconnectAttempt > 0) { - const d = await discoverWsUrl(); - if (!d.success) { scheduleRetry(); return; } - url = d.data; - wsUrl = url; - } else { - url = cachedUrl; - } + const attempt = async (): Promise<void> => { + let settled = false; + let sock: WebSocket | null = null; + let timeout: ReturnType<typeof setTimeout> | null = null; + let settleAttempt: () => void = () => {}; - const settledPromise = new Promise<void>((settle) => { - let settled = false; - const settleOnce = (): void => { + const settledPromise = new Promise<void>((resolve) => { + settleAttempt = (): void => { if (settled) return; settled = true; - settle(); + if (timeout) { + clearTimeout(timeout); + timeout = null; + } + if (generation === activeAttempt && !stopped) { + wsUrl = null; + connecting = false; + } + rejectAttemptCallbacks(generation); + onAttemptSettled?.(); + resolve(); }; + }); - let sock: WebSocket; - try { - sock = new WebSocket(url, { perMessageDeflate: false }); - } catch { - scheduleRetry(); + const clearAttemptTimeout = (): void => { + if (timeout) { + clearTimeout(timeout); + timeout = null; + } + }; + + const failCurrentAttempt = (notifyClose: boolean): void => { + clearAttemptTimeout(); + const isCurrent = isCurrentAttempt(generation); + if (!isCurrent) { + rejectAttemptCallbacks(generation); return; } - const timer = setTimeout(() => { - sock.close(); - settleOnce(); - }, CDP_CONNECT_TIMEOUT_MS); + clearActiveAttempt(generation); + if (ws === sock) ws = null; + wsUrl = null; + connecting = false; + // Not via settleAttempt: a connection that opened already settled its attempt, so + // its requests would otherwise hang to the command timeout instead of failing now. + rejectAttemptCallbacks(generation); + if (notifyClose) closeHandler?.(); + settleAttempt(); + }; - sock.on("open", () => { - clearTimeout(timer); - ws = sock; - reconnectAttempt = 0; - ws.send(JSON.stringify({ id: 0, method: "Target.setDiscoverTargets", params: { discover: true } })); - console.log("[pi-browser-daemon] Connected to Chrome ✓"); - signalConnected(); - settleOnce(); - }); + timeout = setTimeout(() => { + if (!isCurrentAttempt(generation) || stopped) { + failCurrentAttempt(false); + return; + } - sock.on("message", (data: WebSocket.Data) => { - onChromeMessage(typeof data === "string" ? data : data.toString()); - }); + if (sock && sock.readyState !== WebSocket.CLOSED) { + closeSocket(sock); + } + failCurrentAttempt(true); + }, connectTimeoutMs); + + let url = wsUrl; + if (!url) { + const d = await discoverWsUrlImpl(); + if (!isCurrentAttempt(generation) || stopped) { + failCurrentAttempt(false); + return; + } + if (!d.success) { + failCurrentAttempt(false); + return; + } + url = d.data; + } - sock.on("error", () => { - clearTimeout(timer); - settleOnce(); - }); + if (!isCurrentAttempt(generation) || stopped) { + failCurrentAttempt(false); + return; + } - sock.on("close", () => { - clearTimeout(timer); - ws = null; - for (const cb of mux.takeAll()) { - cb.send(cb.clientId, { - type: "response", - id: cb.localId, - error: { code: -32000, message: "Chrome disconnected" }, - }); + try { + sock = createWebSocketImpl(url); + } catch { + failCurrentAttempt(false); + return; + } + + if (!isCurrentAttempt(generation) || stopped) { + closeSocket(sock); + failCurrentAttempt(false); + return; + } + + wsUrl = url; + activeSocket = sock; + + const onOpen = (): void => { + if (!sock || !isCurrentAttemptSocket(generation, sock)) { + clearAttemptTimeout(); + if (sock && sock.readyState !== WebSocket.CLOSED) { + closeSocket(sock); } - closeHandler?.(); - scheduleRetry(); - }); - }); + failCurrentAttempt(false); + return; + } + + clearAttemptTimeout(); + ws = sock; + try { + ws.send(JSON.stringify({ id: 0, method: "Target.setDiscoverTargets", params: { discover: true } })); + } catch { + failCurrentAttempt(true); + if (sock.readyState !== WebSocket.CLOSED) { + closeSocket(sock); + } + return; + } + console.log("[pi-browser-daemon] Connected to Chrome ✓"); + settleAttempt(); + }; + const onMessage = (data: WebSocket.Data): void => { + if (!sock || !isCurrentAttemptSocket(generation, sock)) return; + onChromeMessage(typeof data === "string" ? data : data.toString()); + }; + + const onError = (): void => { + clearAttemptTimeout(); + const notifyClose = !!sock && isCurrentAttemptSocket(generation, sock); + failCurrentAttempt(notifyClose); + if (sock && sock.readyState !== WebSocket.CLOSED) { + closeSocket(sock); + } + }; + + const onClose = (): void => { + clearAttemptTimeout(); + const notifyClose = !!sock && isCurrentAttemptSocket(generation, sock); + failCurrentAttempt(notifyClose); + }; + + sock.on("open", onOpen); + sock.on("message", onMessage); + sock.on("error", onError); + sock.on("close", onClose); + + // The promise must always settle even on constructor errors. await settledPromise; - if (!ws || ws.readyState !== WebSocket.OPEN) { - scheduleRetry(); - } }; - attempt().catch(() => scheduleRetry()); - }; - - const scheduleRetry = (): void => { - if (stopped) return; - const delay = Math.min(1000 * Math.pow(2, Math.min(reconnectAttempt, 6)), 60_000); - reconnectAttempt++; - wsUrl = null; - reconnectTimer = setTimeout(tryConnect, delay); + attempt().catch(() => { + if (isCurrentAttempt(generation)) { + connecting = false; + } + }); }; const start = async (): Promise<void> => { @@ -296,22 +421,51 @@ export const createCdpBridge = (): CdpBridge => { const stop = async (): Promise<void> => { stopped = true; - if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } - mux.takeAll(); - signalConnected(); - if (ws) { try { ws.close(1000, "Shutdown"); } catch {} ws = null; } + attemptGeneration += 1; + activeAttempt = 0; + activeSocket = null; + connecting = false; + rejectAllPendingCallbacks(); + if (ws) { + try { + ws.close(1000, "Shutdown"); + } catch {} + ws = null; + } wsUrl = null; }; + const waitForConnection = async (): Promise<boolean> => { + const deadline = Date.now() + connectTimeoutMs; + while (connecting && Date.now() < deadline) { + if (ws && ws.readyState === WebSocket.OPEN) return true; + await new Promise((r) => setTimeout(r, 5)); + } + return ws !== null && ws.readyState === WebSocket.OPEN; + }; + const handleRequest = async ( req: WireRequest, clientId: string, send: SendToClient, ): Promise<void> => { - // 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); + let activeSocket = ws; + if (!activeSocket || activeSocket.readyState !== WebSocket.OPEN) { + // Connection attempts happen only in response to explicit browser work. + tryConnect(); + const ready = await waitForConnection(); + if (!ready) { + send(clientId, { + type: "response", + id: req.id, + error: { code: -32000, message: "Chrome not connected" }, + }); + return; + } + } - if (!ws || ws.readyState !== WebSocket.OPEN) { + activeSocket = ws; + if (!activeSocket || activeSocket.readyState !== WebSocket.OPEN) { send(clientId, { type: "response", id: req.id, @@ -323,11 +477,14 @@ export const createCdpBridge = (): CdpBridge => { const isDetach = req.method === "Target.detachFromTarget"; if (isDetach && req.sessionId) router.release(req.sessionId); + const isAttach = req.method === "Target.attachToTarget"; + const daemonId = mux.allocate( - { clientId, localId: req.id, send, isAttach: req.method === "Target.attachToTarget" }, + { clientId, localId: req.id, send, isAttach }, (id) => setTimeout(() => { mux.take(id); + generations.delete(id); send(clientId, { type: "response", id: req.id, @@ -335,6 +492,7 @@ export const createCdpBridge = (): CdpBridge => { }); }, CDP_COMMAND_TIMEOUT_MS), ); + generations.set(daemonId, { generation: attemptGeneration, clientId }); const payload: Record<string, unknown> = { id: daemonId, @@ -344,9 +502,10 @@ export const createCdpBridge = (): CdpBridge => { if (req.sessionId) payload["sessionId"] = req.sessionId; try { - ws.send(JSON.stringify(payload)); + activeSocket.send(JSON.stringify(payload)); } catch (e) { mux.take(daemonId); + generations.delete(daemonId); send(clientId, { type: "response", id: req.id, @@ -361,7 +520,11 @@ export const createCdpBridge = (): CdpBridge => { handleRequest, isAlive: () => ws !== null && ws.readyState === WebSocket.OPEN, getSessionOwner: (sid) => router.getOwner(sid), - removeClient: (cid) => { mux.clearClient(cid); router.removeClient(cid); }, + removeClient: (cid) => { + rejectCallbacksWhere((info) => info.clientId === cid); + mux.clearClient(cid); + router.removeClient(cid); + }, onEvent: (h) => { eventHandler = h; }, onClose: (h) => { closeHandler = h; }, }; |
