aboutsummaryrefslogtreecommitdiff
path: root/src/daemon
diff options
context:
space:
mode:
authorUttkarsh <38376852+uttkarsh-26@users.noreply.github.com>2026-08-06 00:20:49 +0530
committerGitHub <noreply@github.com>2026-08-06 00:20:49 +0530
commitd9bc46b508f8c6f7e0f0ff5d54d804340f3e6c80 (patch)
tree0ea1315ccc0feaee1ddf184ed16f16173455b3b4 /src/daemon
parent31c1b27a50bc2b9ba7138ae67a4cfaba6945582e (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/daemon')
-rw-r--r--src/daemon/bridge.ts361
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; },
};