diff --git a/docs/browser-observation.md b/docs/browser-observation.md index 0bfbb2789f..355d7f98ea 100644 --- a/docs/browser-observation.md +++ b/docs/browser-observation.md @@ -343,8 +343,10 @@ an invalid supplied size remains unknown with incomplete resource coverage while the resource's other facts are retained. Zero and nonnegative fractional sizes are preserved. -CDP discovery, connection, and commands wait for the peer or caller cancellation; -a stalled peer can leave an operation waiting. WebSocket messages remain subject +CDP discovery has a 10-second deadline. Each WebSocket handshake and command +response has its own 20-second deadline, separate from the caller-selected +observation window. A stalled handshake or command reports `provider_timeout`; +caller cancellation still reports cancellation. WebSocket messages remain subject to the [64 MiB limit](#request-boundary). REA does not impose a message-count or pending-command quota. On cancellation, it closes only its own WebSocket, waiting up to one second before terminating it. Malformed unsolicited events diff --git a/docs/javascript-runtime-observation.md b/docs/javascript-runtime-observation.md index 439572eba0..6d8404b1ed 100644 --- a/docs/javascript-runtime-observation.md +++ b/docs/javascript-runtime-observation.md @@ -95,6 +95,11 @@ pauses or resumes execution, calls `Runtime.runIfWaitingForDebugger`, launches the target, instruments JavaScript, or invokes Electron APIs. Closing a call closes only REA's WebSocket. +Each WebSocket handshake and command response has its own 20-second deadline, +separate from the caller-selected observation window. A stalled handshake or +command reports `provider_timeout` and retains observations already received; +caller cancellation still reports cancellation. + The direct observations are: - `Debugger.scriptParsed`, establishing that a bounded script location was diff --git a/src/browser/CdpConnection.ts b/src/browser/CdpConnection.ts index e970056686..57319c1d16 100644 --- a/src/browser/CdpConnection.ts +++ b/src/browser/CdpConnection.ts @@ -23,7 +23,7 @@ interface PendingCommand { readonly sessionId?: string; readonly resolve: (value: unknown) => void; readonly reject: (error: AnalysisError) => void; - readonly removeAbort: () => void; + readonly dispose: () => void; } /** Correlated JSON-RPC transport for one CDP target operation. */ @@ -79,6 +79,8 @@ export class CdpConnection { `CDP maxPayloadBytes must be a positive safe integer no greater than ${String(MAX_WEBSOCKET_PAYLOAD_BYTES)}.`, ); const socket = new WebSocket(url, { + // waitForOpen owns the deadline, cancellation, cleanup, and typed errors; + // keep ws's timer disabled so handshake failures have one owner. handshakeTimeout: 0, maxPayload: maxPayloadBytes, perMessageDeflate: false, @@ -115,7 +117,7 @@ export class CdpConnection { return () => this.#protocolFailureListeners.delete(listener); } - /** Execute one command, optionally within a flat target session. */ + /** Bound each response separately from the caller-selected observation window. */ async send( method: string, params: Readonly> = {}, @@ -130,41 +132,51 @@ export class CdpConnection { throw new AnalysisCancelledError(this.operation); const id = this.#nextId; this.#nextId += 1; + const message = JSON.stringify({ + id, + method, + params, + ...(sessionId === undefined ? {} : { sessionId }), + }); return await new Promise((resolve, reject) => { const onAbort = (): void => { const pending = this.#pending.get(id); if (pending === undefined) return; - pending.removeAbort(); - this.#pending.delete(id); + this.#complete(id, pending); reject(new AnalysisCancelledError(this.operation)); }; + const timer = setTimeout(() => { + const pending = this.#pending.get(id); + if (pending === undefined) return; + this.#complete(id, pending); + reject( + new BrowserObservationError(this.operation, "timeout", { + detail: `CDP command ${method} exceeded its ${String(WEB_RUNTIME_LIMITS.commandTimeoutMs)} ms response deadline.`, + }), + ); + }, WEB_RUNTIME_LIMITS.commandTimeoutMs); signal?.addEventListener("abort", onAbort, { once: true }); this.#pending.set(id, { method, ...(sessionId === undefined ? {} : { sessionId }), resolve, reject, - removeAbort: () => signal?.removeEventListener("abort", onAbort), - }); - this.socket.send( - JSON.stringify({ - id, - method, - params, - ...(sessionId === undefined ? {} : { sessionId }), - }), - (error) => { - if (error === undefined || error === null) return; - const pending = this.#pending.get(id); - if (pending === undefined) return; - this.#complete(id, pending); - reject( - new BrowserObservationError(this.operation, "disconnected", { - cause: error, - }), - ); + dispose: () => { + clearTimeout(timer); + signal?.removeEventListener("abort", onAbort); }, - ); + }); + this.socket.send(message, (error) => { + if (error === undefined || error === null) return; + const pending = this.#pending.get(id); + if (pending === undefined) return; + this.#complete(id, pending); + reject( + new BrowserObservationError(this.operation, "disconnected", { + cause: error, + }), + ); + }); }); } @@ -282,7 +294,7 @@ export class CdpConnection { } #complete(id: number, pending: PendingCommand): void { - pending.removeAbort(); + pending.dispose(); this.#pending.delete(id); } @@ -370,11 +382,21 @@ const waitForOpen = async ( onFailure(new AnalysisCancelledError(operation)); }; const cleanup = (): void => { + clearTimeout(timer); socket.off("open", onOpen); socket.off("error", onFailure); socket.off("close", onClose); signal?.removeEventListener("abort", onAbort); }; + const timer = setTimeout(() => { + terminateSocket(socket); + cleanup(); + reject( + new BrowserObservationError(operation, "timeout", { + detail: `CDP WebSocket handshake exceeded its ${String(WEB_RUNTIME_LIMITS.commandTimeoutMs)} ms deadline.`, + }), + ); + }, WEB_RUNTIME_LIMITS.commandTimeoutMs); socket.once("open", onOpen); socket.once("error", onFailure); socket.once("close", onClose); diff --git a/tests/boundary/browser/cdpTargetDeadlines.test.ts b/tests/boundary/browser/cdpTargetDeadlines.test.ts new file mode 100644 index 0000000000..965c867fff --- /dev/null +++ b/tests/boundary/browser/cdpTargetDeadlines.test.ts @@ -0,0 +1,277 @@ +import { createServer, type Socket } from "node:net"; + +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +import { CdpBrowserProvider } from "../../../src/browser/CdpBrowserProvider.js"; +import { CdpConnection } from "../../../src/browser/CdpConnection.js"; +import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js"; +import { inspectWebPageInputSchema } from "../../../src/domain/browserObservation.js"; +import { WEB_RUNTIME_LIMITS } from "../../../src/domain/webRuntime.js"; +import { V8InspectorProvider } from "../../../src/inspector/V8InspectorProvider.js"; +import { startFakeCdpBrowser } from "../../fixtures/fakeCdpBrowser.js"; +import { startFakeV8Inspector } from "../../fixtures/inspector/fakeV8Inspector.js"; + +// Keep real socket scheduling; expire only a timer whose owning phase is ready. +beforeEach(() => { + vi.spyOn(globalThis, "setTimeout"); + vi.spyOn(globalThis, "clearTimeout"); +}); + +afterEach(() => vi.restoreAllMocks()); + +const activeTimers = (milliseconds: number) => { + const scheduled = vi.mocked(globalThis.setTimeout); + const cleared = vi.mocked(globalThis.clearTimeout); + return scheduled.mock.calls.flatMap(([callback, delay], index) => { + const result = scheduled.mock.results[index]; + if ( + delay !== milliseconds || + result?.type !== "return" || + cleared.mock.calls.some(([timer]) => timer === result.value) + ) + return []; + return [{ callback, timer: result.value }]; + }); +}; + +const expireTimer = (milliseconds: number): void => { + const timers = activeTimers(milliseconds); + expect(timers).toHaveLength(1); + const timer = timers[0]; + if (timer === undefined) throw new Error("Expected one active timer"); + try { + timer.callback(); + if (milliseconds === WEB_RUNTIME_LIMITS.commandTimeoutMs) + expect(globalThis.clearTimeout).toHaveBeenCalledWith(timer.timer); + } finally { + // The observation callback resolves naturally without cancelling its timer. + clearTimeout(timer.timer); + } +}; + +it("bounds a stalled WebSocket handshake and releases its socket", async () => { + const sockets = new Set(); + const server = createServer((socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + socket.resume(); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (address === null || typeof address === "string") + throw new Error("Expected a TCP address"); + try { + const rejection = expect( + CdpConnection.connect( + `ws://127.0.0.1:${String(address.port)}/target`, + "inspect_web_page", + AbortSignal.timeout(10_000), + ), + ).rejects.toMatchObject({ reason: "timeout" }); + await vi.waitFor(() => expect(sockets.size).toBe(1), { timeout: 5_000 }); + expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs); + await rejection; + await vi.waitFor(() => expect(sockets.size).toBe(0)); + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + } +}); + +it.each(["Target.attachToTarget", "Page.enable", "Runtime.enable"])( + "times out %s after HTTP discovery and closes the target session", + async (hangOnMethod) => { + const browser = await startFakeCdpBrowser({ hangOnMethod }); + try { + const pending = new CdpBrowserProvider().inspectPage( + inspectWebPageInputSchema.parse({ + cdp_endpoint: browser.endpoint, + target_id: "allowed-page", + observation_ms: 0, + }), + { signal: AbortSignal.timeout(10_000) }, + ); + await vi.waitFor( + () => + expect(browser.commands.map(({ method }) => method)).toContain( + hangOnMethod, + ), + { timeout: 5_000 }, + ); + expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs); + const result = await pending; + if (result.ok) throw new Error("Expected a target command timeout"); + expect(projectAnalysisError(result.error)).toMatchObject({ + code: "provider_timeout", + category: "timeout", + }); + expect(browser.commands.map(({ method }) => method)).toContain( + hangOnMethod, + ); + await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0)); + } finally { + await browser.close(); + } + }, +); + +it("ignores a late timed-out reply and keeps following commands correlated", async () => { + const browser = await startFakeCdpBrowser({ hangOnMethod: "Page.enable" }); + const connection = await CdpConnection.connect( + browser.browserWebSocketUrl, + "inspect_web_page", + ); + try { + const rejection = expect( + connection.send( + "Page.enable", + {}, + undefined, + AbortSignal.timeout(10_000), + ), + ).rejects.toMatchObject({ reason: "timeout" }); + await vi.waitFor(() => expect(browser.commands).toHaveLength(1), { + timeout: 5_000, + }); + expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs); + await rejection; + const command = browser.commands[0]; + if (command === undefined) + throw new Error("Expected the timed-out command"); + browser.emitRawMessage(JSON.stringify({ id: command.id, result: {} })); + await expect(connection.send("Runtime.enable")).resolves.toEqual({}); + } finally { + await connection.close(); + await browser.close(); + } +}); + +it.each(["success", "rejection", "abort", "disconnect"])( + "releases handshake and command timers on %s", + async (outcome) => { + const cleared = vi.mocked(globalThis.clearTimeout); + const browser = await startFakeCdpBrowser({ + ...(outcome === "abort" || outcome === "success" + ? { hangOnMethod: "Runtime.enable" } + : {}), + ...(outcome === "disconnect" ? { closeOnMethod: "Runtime.enable" } : {}), + }); + const controller = new AbortController(); + const removeAbort = vi.spyOn(controller.signal, "removeEventListener"); + const connection = await CdpConnection.connect( + browser.browserWebSocketUrl, + "inspect_web_page", + controller.signal, + ); + try { + const pending = connection.send( + outcome === "rejection" ? "Fixture.unmodeledMethod" : "Runtime.enable", + {}, + undefined, + controller.signal, + ); + const assertion = + outcome === "success" + ? expect(pending).resolves.toEqual({}) + : expect(pending).rejects.toMatchObject({ + _tag: + outcome === "abort" + ? "AnalysisCancelledError" + : "BrowserObservationError", + }); + if (outcome === "success") { + await vi.waitFor(() => expect(browser.commands).toHaveLength(1), { + timeout: 5_000, + }); + // A healthy reply may take longer than the former 100 ms test budget. + await new Promise((resolve) => setTimeout(resolve, 250)); + const command = browser.commands[0]; + if (command === undefined) + throw new Error("Expected a pending command"); + browser.emitRawMessage(JSON.stringify({ id: command.id, result: {} })); + } + if (outcome === "abort") controller.abort(); + await assertion; + const scheduled = vi.mocked(globalThis.setTimeout); + const deadlines = scheduled.mock.calls.flatMap((call, index) => + call[1] === WEB_RUNTIME_LIMITS.commandTimeoutMs + ? [scheduled.mock.results[index]?.value] + : [], + ); + expect(deadlines).toHaveLength(2); + for (const timer of deadlines) + expect(cleared).toHaveBeenCalledWith(timer); + expect(removeAbort).toHaveBeenCalledTimes(2); + } finally { + await connection.close(); + await browser.close(); + } + }, +); + +it.each(["Runtime.enable", "Debugger.enable"])( + "times out Inspector %s and preserves observations received before expiry", + async (hangOnMethod) => { + const inspector = await startFakeV8Inspector({ + targetUrl: "https://fixture.example/app.js", + hangOnMethod, + }); + try { + const pending = new V8InspectorProvider().observe( + { + inspector_endpoint: inspector.endpoint, + target_id: inspector.targetId, + observation_ms: 0, + }, + { signal: AbortSignal.timeout(10_000) }, + ); + await vi.waitFor( + () => + expect(inspector.commands.map(({ method }) => method)).toContain( + hangOnMethod, + ), + { timeout: 5_000 }, + ); + expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs); + const result = await pending; + if (result.ok) throw new Error("Expected an Inspector command timeout"); + expect(projectAnalysisError(result.error)).toMatchObject({ + code: "provider_timeout", + category: "timeout", + }); + expect(result.error.partialObservation).toMatchObject({ + capture: { + events_observed: hangOnMethod === "Debugger.enable" ? 1 : 0, + }, + }); + } finally { + await inspector.close(); + } + }, +); + +it("keeps the chosen observation window independent of command deadlines", async () => { + const browser = await startFakeCdpBrowser(); + const observationMs = WEB_RUNTIME_LIMITS.commandTimeoutMs + 250; + try { + const pending = new CdpBrowserProvider().inspectPage( + inspectWebPageInputSchema.parse({ + cdp_endpoint: browser.endpoint, + target_id: "allowed-page", + observation_ms: observationMs, + }), + ); + await vi.waitFor( + () => expect(activeTimers(observationMs)).toHaveLength(1), + { + timeout: 5_000, + }, + ); + expect(activeTimers(WEB_RUNTIME_LIMITS.commandTimeoutMs)).toHaveLength(0); + expireTimer(observationMs); + const result = await pending; + expect(result.ok).toBe(true); + } finally { + await browser.close(); + } +}); diff --git a/tests/fixtures/inspector/fakeV8Inspector.ts b/tests/fixtures/inspector/fakeV8Inspector.ts index 14ae1ed4d7..c9e3f69150 100644 --- a/tests/fixtures/inspector/fakeV8Inspector.ts +++ b/tests/fixtures/inspector/fakeV8Inspector.ts @@ -28,6 +28,7 @@ interface FakeV8InspectorOptions { readonly additionalTargetUrl?: string; readonly additionalTargetCount?: number; readonly closeOnMethod?: string; + readonly hangOnMethod?: string; } export interface FakeV8Inspector { @@ -147,6 +148,7 @@ const registerInspectorSocket = ( socket.close(); return; } + if (options.hangOnMethod === command.method) return; respondToInspectorCommand(socket, command, options); socket.send(JSON.stringify({ id: command.id, result: {} })); });