From 02f796660ecbe4df46c72ac915f65dac2aa76e6d Mon Sep 17 00:00:00 2001 From: "(CJ) Chukwudi Nwobodo" <142016413+chuks-qua@users.noreply.github.com> Date: Tue, 22 Sep 2026 09:45:27 +0100 Subject: [PATCH] fix(web): settle RPCs parked on reconnect readiness An RPC awaiting the `ready` promise during a failed CONNECTING attempt, a close(), or a socket that dies between readiness and send could sit in `pending` forever: close() resolved `ready` so waiters resumed onto a dead socket where send() silently drops, and a reconnect callback already awaiting URL discovery could still create a fresh socket and pending `ready` after dispose. close() now rejects `ready` so parked and late RPCs fail fast, rpc() and rpcBinary() reject when the socket is no longer OPEN after `await ready`, the in-flight reconnect callback bails if closed during discovery, and waitForConnection propagates the rejection. Fixes #1733 --- apps/web/src/__tests__/connection.test.ts | 197 ++++++++++++++++++ .../thread-reconnect-refresh.test.ts | 1 + .../__tests__/agent-commands.test.ts | 1 + apps/web/src/transport/ws-transport.ts | 40 +++- 4 files changed, 233 insertions(+), 6 deletions(-) diff --git a/apps/web/src/__tests__/connection.test.ts b/apps/web/src/__tests__/connection.test.ts index ac14e0e5c..5f8a69d02 100644 --- a/apps/web/src/__tests__/connection.test.ts +++ b/apps/web/src/__tests__/connection.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import { createWsTransport } from "../transport/ws-transport"; class MockWebSocket { + static readonly OPEN = 1; onopen: (() => void) | null = null; onclose: ((event: { code: number; reason: string }) => void) | null = null; onerror: (() => void) | null = null; @@ -232,3 +233,199 @@ describe("4001 auth failure handling", () => { vi.useRealTimers(); }); }); + +/** + * Browser-faithful socket: send() throws while CONNECTING and silently + * discards while CLOSING/CLOSED, matching the WebSocket spec. The silent + * drop is what orphans an RPC that resumes onto a dead socket. + */ +class LifecycleSocket { + static instances: LifecycleSocket[] = []; + static readonly OPEN = 1; + onopen: (() => void) | null = null; + onclose: ((event: { code: number; reason: string }) => void) | null = null; + onerror: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + readyState = 0; + binaryType = ""; + + constructor() { + LifecycleSocket.instances.push(this); + } + + send(raw: string | ArrayBuffer): void { + if (this.readyState === 0) throw new Error("InvalidStateError: socket is CONNECTING"); + if (this.readyState !== LifecycleSocket.OPEN) return; + if (typeof raw !== "string") return; + const request = JSON.parse(raw) as { id: string; method?: string }; + const result = request.method === "workspace.list" ? [{ id: "fixture" }] : null; + queueMicrotask(() => + this.onmessage?.({ data: JSON.stringify({ id: request.id, result }) }), + ); + } + + open(): void { + this.readyState = 1; + this.onopen?.(); + } + + fail(): void { + this.readyState = 3; + this.onclose?.({ code: 1006, reason: "fixture failure" }); + } + + close(): void { + this.readyState = 3; + this.onclose?.({ code: 1000, reason: "" }); + } +} + +describe("RPC readiness across reconnects (issue #1733)", () => { + let transport: ReturnType; + + function settleSpy(promise: Promise): { state: string } { + const spy = { state: "pending" }; + void promise.then( + () => { spy.state = "resolved"; }, + () => { spy.state = "rejected"; }, + ); + return spy; + } + + beforeEach(() => { + vi.useFakeTimers(); + LifecycleSocket.instances = []; + vi.stubGlobal("WebSocket", LifecycleSocket); + transport = createWsTransport("ws://fixture.invalid"); + }); + + afterEach(() => { + transport.close(); + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it("settles an RPC waiting during a failed connecting attempt after the next socket opens", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + await expect(transport.listWorkspaces()).resolves.toEqual([{ id: "fixture" }]); + + LifecycleSocket.instances[0]!.fail(); + await vi.advanceTimersByTimeAsync(1100); + expect(LifecycleSocket.instances).toHaveLength(2); + + const original = settleSpy(transport.listWorkspaces()); + LifecycleSocket.instances[1]!.fail(); + await vi.advanceTimersByTimeAsync(2100); + expect(LifecycleSocket.instances).toHaveLength(3); + LifecycleSocket.instances[2]!.open(); + await vi.advanceTimersByTimeAsync(0); + + await expect(transport.listWorkspaces()).resolves.toEqual([{ id: "fixture" }]); + await vi.advanceTimersByTimeAsync(60_000); + expect(original.state).not.toBe("pending"); + }); + + it("keeps a waiting RPC alive across consecutive failed reconnect attempts", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + LifecycleSocket.instances[0]!.fail(); + await vi.advanceTimersByTimeAsync(1100); + expect(LifecycleSocket.instances).toHaveLength(2); + + const waiting = settleSpy(transport.listWorkspaces()); + LifecycleSocket.instances[1]!.fail(); + await vi.advanceTimersByTimeAsync(2100); + LifecycleSocket.instances[2]!.fail(); + await vi.advanceTimersByTimeAsync(4100); + expect(LifecycleSocket.instances).toHaveLength(4); + + LifecycleSocket.instances[3]!.open(); + await vi.advanceTimersByTimeAsync(60_000); + expect(waiting.state).toBe("resolved"); + }); + + it("rejects an RPC waiting for readiness when the transport is closed", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + LifecycleSocket.instances[0]!.fail(); + await vi.advanceTimersByTimeAsync(1100); + expect(LifecycleSocket.instances).toHaveLength(2); + + const waiting = settleSpy(transport.listWorkspaces()); + transport.close(); + await vi.advanceTimersByTimeAsync(60_000); + expect(waiting.state).toBe("rejected"); + }); + + it("rejects RPCs issued after close instead of queueing them forever", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + transport.close(); + + const late = settleSpy(transport.listWorkspaces()); + await vi.advanceTimersByTimeAsync(60_000); + expect(late.state).toBe("rejected"); + }); + + it("settles a binary RPC waiting during a failed connecting attempt after the next socket opens", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + LifecycleSocket.instances[0]!.fail(); + await vi.advanceTimersByTimeAsync(1100); + expect(LifecycleSocket.instances).toHaveLength(2); + + const waiting = settleSpy( + transport.saveClipboardFile(new ArrayBuffer(4), "image/png", "clip.png"), + ); + LifecycleSocket.instances[1]!.fail(); + await vi.advanceTimersByTimeAsync(2100); + LifecycleSocket.instances[2]!.open(); + await vi.advanceTimersByTimeAsync(60_000); + expect(waiting.state).toBe("resolved"); + }); + + it("does not open a new socket or re-arm readiness when close() lands mid-reconnect", async () => { + transport.close(); + let releaseDiscovery: (() => void) | null = null; + transport = createWsTransport("ws://fixture.invalid", { + discoverServerUrl: () => + new Promise((resolve) => { + releaseDiscovery = () => resolve("ws://fixture.invalid"); + }), + }); + + // instances[0] belongs to the beforeEach transport closed above. + LifecycleSocket.instances[1]!.open(); + await vi.advanceTimersByTimeAsync(0); + LifecycleSocket.instances[1]!.fail(); + // Reconnect callback fires and suspends inside discoverServerUrl. + await vi.advanceTimersByTimeAsync(1100); + + transport.close(); + releaseDiscovery!(); + await vi.advanceTimersByTimeAsync(60_000); + + // No post-close connect() may run: no new socket, and `ready` stays + // rejected so late RPCs fail fast instead of parking forever. + expect(LifecycleSocket.instances).toHaveLength(2); + const late = settleSpy(transport.listWorkspaces()); + await vi.advanceTimersByTimeAsync(0); + expect(late.state).toBe("rejected"); + }); + + it("rejects a binary RPC waiting for readiness when the transport is closed", async () => { + LifecycleSocket.instances[0]!.open(); + await vi.advanceTimersByTimeAsync(0); + LifecycleSocket.instances[0]!.fail(); + await vi.advanceTimersByTimeAsync(1100); + + const waiting = settleSpy( + transport.saveClipboardFile(new ArrayBuffer(4), "image/png", "clip.png"), + ); + transport.close(); + await vi.advanceTimersByTimeAsync(60_000); + expect(waiting.state).toBe("rejected"); + }); +}); + diff --git a/apps/web/src/__tests__/thread-reconnect-refresh.test.ts b/apps/web/src/__tests__/thread-reconnect-refresh.test.ts index 1e2b21201..c0c9bd060 100644 --- a/apps/web/src/__tests__/thread-reconnect-refresh.test.ts +++ b/apps/web/src/__tests__/thread-reconnect-refresh.test.ts @@ -16,6 +16,7 @@ vi.mock("@/transport", async () => ({ })); class ReconnectWebSocket { + static readonly OPEN = 1; onopen: (() => void) | null = null; onclose: ((event: { code: number; reason: string }) => void) | null = null; onerror: (() => void) | null = null; diff --git a/apps/web/src/transport/__tests__/agent-commands.test.ts b/apps/web/src/transport/__tests__/agent-commands.test.ts index 49ac719d5..ae7d97a4e 100644 --- a/apps/web/src/transport/__tests__/agent-commands.test.ts +++ b/apps/web/src/transport/__tests__/agent-commands.test.ts @@ -4,6 +4,7 @@ import { useSettingsStore } from "@/stores/settingsStore"; import { createWsTransport } from "../ws-transport"; class MockWebSocket { + static readonly OPEN = 1; onopen: (() => void) | null = null; onclose: ((event: { code: number; reason: string }) => void) | null = null; onerror: (() => void) | null = null; diff --git a/apps/web/src/transport/ws-transport.ts b/apps/web/src/transport/ws-transport.ts index c86e39827..aee53c0f2 100644 --- a/apps/web/src/transport/ws-transport.ts +++ b/apps/web/src/transport/ws-transport.ts @@ -340,6 +340,7 @@ export function createWsTransport( /** Resolves when the current WebSocket connection is open. */ let ready: Promise; let resolveReady: () => void = () => undefined; + let rejectReady: (err: Error) => void = () => undefined; let readyPending = false; function resetReady() { @@ -347,12 +348,19 @@ export function createWsTransport( // would strand them forever. if (readyPending) return; readyPending = true; - ready = new Promise((resolve) => { + ready = new Promise((resolve, reject) => { resolveReady = () => { readyPending = false; resolve(); }; + rejectReady = (err) => { + readyPending = false; + reject(err); + }; }); + // Rejection is consumed by rpc() awaiters; this swallows it when close() + // settles `ready` with no one parked on it. + ready.catch(() => {}); } function emitTerminalDataFrame(view: Uint8Array): void { @@ -564,6 +572,9 @@ export function createWsTransport( // Discovery failed, retry with current URL } } + // close() cannot cancel a callback already awaiting discovery; bail so a + // new socket (and a fresh `ready` promise) is not created after dispose. + if (closed) return; connect(); }, delay); } @@ -572,6 +583,13 @@ export function createWsTransport( async function rpc(method: string, params: Record): Promise { await ready; return new Promise((resolve, reject) => { + // `ready` may have resolved on a socket that closed before this + // continuation ran; send() on a dead socket is silently dropped and the + // request would sit in `pending` forever. + if (ws.readyState !== WebSocket.OPEN) { + reject(new Error("WebSocket disconnected")); + return; + } const id = `req_${++idCounter}`; pending.set(id, { resolve: resolve as (v: unknown) => void, @@ -635,6 +653,10 @@ export function createWsTransport( ): Promise { await ready; return new Promise((resolve, reject) => { + if (ws.readyState !== WebSocket.OPEN) { + reject(new Error("WebSocket disconnected")); + return; + } const id = `req_${++idCounter}`; pending.set(id, { resolve: resolve as (v: unknown) => void, @@ -663,10 +685,16 @@ export function createWsTransport( reject(new Error(`Could not connect to server at ${displayUrl}`)); }, timeoutMs); - ready.then(() => { - clearTimeout(timer); - resolve(); - }); + ready.then( + () => { + clearTimeout(timer); + resolve(); + }, + (err) => { + clearTimeout(timer); + reject(err); + }, + ); }); } @@ -1258,7 +1286,7 @@ export function createWsTransport( clearTimeout(reconnectTimer); reconnectTimer = null; } - resolveReady(); + rejectReady(new Error("Transport closed")); rejectPending("Transport closed"); ws.close(); },