diff --git a/apps/web/src/components/WebSocketConnectionSurface.logic.test.ts b/apps/web/src/components/WebSocketConnectionSurface.logic.test.ts index 2e371b7d7644f2cfe27a5b55cd1033741b0321c6..aed81ab01c73dd4aeb443e015b6093224e663bed 100644 --- a/apps/web/src/components/WebSocketConnectionSurface.logic.test.ts +++ b/apps/web/src/components/WebSocketConnectionSurface.logic.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "vitest"; import type { WsConnectionStatus } from "../rpc/wsConnectionState"; -import { shouldAutoReconnect } from "./WebSocketConnectionSurface"; +import { shouldAutoReconnect, shouldRestartStalledReconnect } from "./WebSocketConnectionSurface"; function makeStatus(overrides: Partial = {}): WsConnectionStatus { return { @@ -80,4 +80,34 @@ describe("WebSocketConnectionSurface.logic", () => { ), ).toBe(true); }); + + it("restarts a stalled reconnect window after the scheduled retry time passes", () => { + expect( + shouldRestartStalledReconnect( + makeStatus({ + hasConnected: true, + nextRetryAt: "2026-04-03T20:00:01.000Z", + online: true, + phase: "disconnected", + reconnectAttemptCount: 3, + reconnectPhase: "waiting", + }), + "2026-04-03T20:00:01.000Z", + ), + ).toBe(true); + + expect( + shouldRestartStalledReconnect( + makeStatus({ + hasConnected: true, + nextRetryAt: "2026-04-03T20:00:01.000Z", + online: true, + phase: "disconnected", + reconnectAttemptCount: 3, + reconnectPhase: "attempting", + }), + "2026-04-03T20:00:01.000Z", + ), + ).toBe(false); + }); }); diff --git a/apps/web/src/components/WebSocketConnectionSurface.tsx b/apps/web/src/components/WebSocketConnectionSurface.tsx index 17d514ae705ff6980f53238784a665f8ee3493ff..7b350e98496dc41bdf1d1c933fb7b779472a55a9 100644 --- a/apps/web/src/components/WebSocketConnectionSurface.tsx +++ b/apps/web/src/components/WebSocketConnectionSurface.tsx @@ -2,7 +2,6 @@ import { type ReactNode, useEffect, useEffectEvent, useRef, useState } from "rea import { type SlowRpcAckRequest, useSlowRpcAckRequests } from "../rpc/requestLatencyState"; import { - exhaustWsReconnectIfStillWaiting, getWsConnectionStatus, getWsConnectionUiState, setBrowserOnlineStatus, @@ -105,6 +104,18 @@ export function shouldAutoReconnect( ); } +export function shouldRestartStalledReconnect( + status: WsConnectionStatus, + expectedNextRetryAt: string, +): boolean { + return ( + status.reconnectPhase === "waiting" && + status.nextRetryAt === expectedNextRetryAt && + status.online && + status.hasConnected + ); +} + export function WebSocketConnectionCoordinator() { const status = useWsConnectionStatus(); const [nowMs, setNowMs] = useState(() => Date.now()); @@ -205,7 +216,12 @@ export function WebSocketConnectionCoordinator() { const nextRetryAt = status.nextRetryAt; const timeoutMs = Math.max(0, new Date(nextRetryAt).getTime() - Date.now()) + 1_500; const timeoutId = window.setTimeout(() => { - exhaustWsReconnectIfStillWaiting(nextRetryAt); + const currentStatus = getWsConnectionStatus(); + if (!shouldRestartStalledReconnect(currentStatus, nextRetryAt)) { + return; + } + + runReconnect(false); }, timeoutMs); return () => { diff --git a/apps/web/src/environments/runtime/connection.test.ts b/apps/web/src/environments/runtime/connection.test.ts index 38611014918590f0e83fa4636d2570cf54d78e97..2644fb83990c11ff82be95d677e77e0c5b01bdfc 100644 --- a/apps/web/src/environments/runtime/connection.test.ts +++ b/apps/web/src/environments/runtime/connection.test.ts @@ -100,6 +100,7 @@ function createTestClient(options?: { return { client, getSnapshot, + replayEvents, emitWelcome: (environmentId: EnvironmentId) => { for (const listener of lifecycleListeners) { listener({ @@ -227,6 +228,64 @@ describe("createEnvironmentConnection", () => { await connection.dispose(); }); + it("retries replay recovery after transport disconnects during resubscribe", async () => { + const environmentId = EnvironmentId.makeUnsafe("env-1"); + let replayAttempts = 0; + const applyEventBatch = vi.fn(); + const { client, replayEvents, triggerDomainResubscribe } = createTestClient({ + replayEvents: async () => { + replayAttempts += 1; + if (replayAttempts === 1) { + throw new Error("SocketCloseError: 1006"); + } + + return [ + { + sequence: 2, + type: "thread.created", + payload: {}, + }, + ]; + }, + }); + + const connection = createEnvironmentConnection({ + kind: "saved", + knownEnvironment: { + id: "env-1", + label: "Remote env", + source: "manual", + target: { + httpBaseUrl: "http://example.test", + wsBaseUrl: "ws://example.test", + }, + environmentId, + }, + client, + applyEventBatch, + syncSnapshot: vi.fn(), + applyTerminalEvent: vi.fn(), + }); + + await Promise.resolve(); + await Promise.resolve(); + + triggerDomainResubscribe(); + + await vi.waitFor(() => { + expect(replayEvents).toHaveBeenCalledTimes(2); + expect(applyEventBatch).toHaveBeenCalledWith( + [ + expect.objectContaining({ + sequence: 2, + }), + ], + environmentId, + ); + }); + + await connection.dispose(); + }); it("swallows replay recovery failures triggered by resubscribe", async () => { const environmentId = EnvironmentId.makeUnsafe("env-1"); const snapshotError = new Error("snapshot failed"); diff --git a/apps/web/src/environments/runtime/connection.ts b/apps/web/src/environments/runtime/connection.ts index aa171d4e86d8ed7c765c45d22efb2113971b142e..ef79d3fce6c0ec48263515a8d974193b650e74c4 100644 --- a/apps/web/src/environments/runtime/connection.ts +++ b/apps/web/src/environments/runtime/connection.ts @@ -16,10 +16,13 @@ import { createOrchestrationRecoveryCoordinator, type ReplayRetryTracker, } from "../../orchestrationRecovery"; +import { isTransportConnectionErrorMessage } from "~/rpc/transportError"; import type { WsRpcClient } from "~/rpc/wsRpcClient"; const REPLAY_RECOVERY_RETRY_DELAY_MS = 100; const MAX_NO_PROGRESS_REPLAY_RETRIES = 3; +const RECOVERY_TRANSPORT_RETRY_DELAY_MS = 250; +const MAX_RECOVERY_TRANSPORT_RETRIES = 20; export interface EnvironmentConnection { readonly kind: "primary" | "saved"; @@ -128,6 +131,30 @@ export function createEnvironmentConnection( queueMicrotask(flushPendingDomainEvents); }; + const retryTransportRecoveryOperation = async (operation: () => Promise): Promise => { + for (let attempt = 0; ; attempt += 1) { + try { + return await operation(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if ( + disposed || + !isTransportConnectionErrorMessage(message) || + attempt >= MAX_RECOVERY_TRANSPORT_RETRIES - 1 + ) { + throw error; + } + + await new Promise((resolve) => { + setTimeout(resolve, RECOVERY_TRANSPORT_RETRY_DELAY_MS); + }); + + if (disposed) { + throw error; + } + } + } + }; const scheduleReplayRecovery = (reason: "sequence-gap" | "resubscribe") => { void runReplayRecovery(reason).catch(() => undefined); }; @@ -139,7 +166,9 @@ export function createEnvironmentConnection( const fromSequenceExclusive = recovery.getState().latestSequence; try { - const events = await input.client.orchestration.replayEvents({ fromSequenceExclusive }); + const events = await retryTransportRecoveryOperation(() => + input.client.orchestration.replayEvents({ fromSequenceExclusive }), + ); if (!disposed) { const nextEvents = recovery.markEventBatchApplied(events); if (nextEvents.length > 0) { @@ -149,6 +178,9 @@ export function createEnvironmentConnection( } catch { replayRetryTracker = null; recovery.failReplayRecovery(); + if (disposed) { + return; + } await snapshotBootstrap.ensureSnapshotRecovery("replay-failed"); return; } @@ -198,7 +230,9 @@ export function createEnvironmentConnection( } try { - const snapshot = await input.client.orchestration.getSnapshot(); + const snapshot = await retryTransportRecoveryOperation(() => + input.client.orchestration.getSnapshot(), + ); if (!disposed) { input.syncSnapshot(snapshot, environmentId); if (recovery.completeSnapshotRecovery(snapshot.snapshotSequence)) { diff --git a/apps/web/src/rpc/protocol.ts b/apps/web/src/rpc/protocol.ts index 9dc7696235d3f6a18813a6586cd1f2942b76fcb8..48681b4a27d5032f2ede24377a5b6b010b6b4f26 100644 --- a/apps/web/src/rpc/protocol.ts +++ b/apps/web/src/rpc/protocol.ts @@ -62,14 +62,36 @@ function defaultLifecycleHandlers(): Required { }; } +function composeLifecycleHandlers( + handlers?: WsProtocolLifecycleHandlers, +): Required { + const defaults = defaultLifecycleHandlers(); + + return { + onAttempt: (socketUrl) => { + defaults.onAttempt(socketUrl); + handlers?.onAttempt?.(socketUrl); + }, + onOpen: () => { + defaults.onOpen(); + handlers?.onOpen?.(); + }, + onError: (message) => { + defaults.onError(message); + handlers?.onError?.(message); + }, + onClose: (details) => { + defaults.onClose(details); + handlers?.onClose?.(details); + }, + }; +} + export function createWsRpcProtocolLayer( url: WsRpcProtocolSocketUrlProvider, handlers?: WsProtocolLifecycleHandlers, ) { - const lifecycle = { - ...defaultLifecycleHandlers(), - ...handlers, - }; + const lifecycle = composeLifecycleHandlers(handlers); const resolvedUrl = typeof url === "function" ? Effect.promise(() => url()).pipe( diff --git a/apps/web/src/rpc/wsConnectionState.test.ts b/apps/web/src/rpc/wsConnectionState.test.ts index f4e8689a58cde9aaa50ed9cf5a8bb0439e5342c6..fb0b9d85aeff56950e85e96242556ee599802f10 100644 --- a/apps/web/src/rpc/wsConnectionState.test.ts +++ b/apps/web/src/rpc/wsConnectionState.test.ts @@ -1,7 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { - exhaustWsReconnectIfStillWaiting, getWsConnectionStatus, getWsReconnectDelayMsForRetry, getWsConnectionUiState, @@ -73,40 +72,4 @@ describe("wsConnectionState", () => { reconnectPhase: "exhausted", }); }); - - it("can exhaust a stalled final retry window when no new attempt starts", () => { - recordWsConnectionAttempt("ws://localhost:3020/ws"); - recordWsConnectionOpened(); - - for (let attempt = 0; attempt < WS_RECONNECT_MAX_ATTEMPTS - 1; attempt += 1) { - recordWsConnectionAttempt("ws://localhost:3020/ws"); - recordWsConnectionErrored("Unable to connect to the T3 server WebSocket."); - } - - const finalRetryDelayMs = getWsReconnectDelayMsForRetry(WS_RECONNECT_MAX_ATTEMPTS - 2); - if (finalRetryDelayMs === null) { - throw new Error("Expected a final retry delay."); - } - - const statusBeforeExhaust = getWsConnectionStatus(); - expect(statusBeforeExhaust).toMatchObject({ - nextRetryAt: new Date(Date.now() + finalRetryDelayMs).toISOString(), - reconnectAttemptCount: 7, - reconnectPhase: "waiting", - }); - - const nextRetryAt = statusBeforeExhaust.nextRetryAt; - if (!nextRetryAt) { - throw new Error("Expected a scheduled retry."); - } - - vi.setSystemTime(new Date(Date.now() + finalRetryDelayMs + 1_000)); - exhaustWsReconnectIfStillWaiting(nextRetryAt); - - expect(getWsConnectionStatus()).toMatchObject({ - nextRetryAt: null, - reconnectAttemptCount: WS_RECONNECT_MAX_ATTEMPTS, - reconnectPhase: "exhausted", - }); - }); }); diff --git a/apps/web/src/rpc/wsConnectionState.ts b/apps/web/src/rpc/wsConnectionState.ts index 5020a1c453e425f079702af959e790fa8cdbdee9..d3050a2b4c9b01456efecbabb1beea90685a1686 100644 --- a/apps/web/src/rpc/wsConnectionState.ts +++ b/apps/web/src/rpc/wsConnectionState.ts @@ -149,26 +149,6 @@ export function resetWsReconnectBackoff(): WsConnectionStatus { })); } -export function exhaustWsReconnectIfStillWaiting(expectedNextRetryAt: string): WsConnectionStatus { - return updateWsConnectionStatus((current) => { - if ( - current.reconnectPhase !== "waiting" || - current.nextRetryAt !== expectedNextRetryAt || - !current.online || - !current.hasConnected - ) { - return current; - } - - return { - ...current, - nextRetryAt: null, - reconnectAttemptCount: current.reconnectMaxAttempts, - reconnectPhase: "exhausted", - }; - }); -} - export function resetWsConnectionStateForTests(): void { appAtomRegistry.set(wsConnectionStatusAtom, INITIAL_WS_CONNECTION_STATUS); } diff --git a/apps/web/src/rpc/wsTransport.test.ts b/apps/web/src/rpc/wsTransport.test.ts index fea4454cd6c7e3a64aabc026cfad048a12e05c1e..2863bb0225a94d51b45f890af571eb714b19d2c5 100644 --- a/apps/web/src/rpc/wsTransport.test.ts +++ b/apps/web/src/rpc/wsTransport.test.ts @@ -83,6 +83,7 @@ class MockWebSocket { const originalWebSocket = globalThis.WebSocket; const originalFetch = globalThis.fetch; +const transports: WsTransport[] = []; function getSocket(): MockWebSocket { const socket = sockets.at(-1); @@ -107,9 +108,16 @@ async function waitFor(assertion: () => void, timeoutMs = 1_000): Promise } } +function createTransport(...args: ConstructorParameters): WsTransport { + const transport = new WsTransport(...args); + transports.push(transport); + return transport; +} + beforeEach(() => { vi.useRealTimers(); sockets.length = 0; + transports.length = 0; resetRequestLatencyStateForTests(); resetWsConnectionStateForTests(); @@ -134,6 +142,8 @@ beforeEach(() => { }); afterEach(async () => { + await Promise.allSettled(transports.map((transport) => transport.dispose())); + transports.length = 0; globalThis.WebSocket = originalWebSocket; globalThis.fetch = originalFetch; resetRequestLatencyStateForTests(); @@ -144,7 +154,7 @@ afterEach(async () => { describe("WsTransport", () => { it("normalizes root websocket urls to /ws and preserves query params", async () => { - const transport = new WsTransport("ws://localhost:3020/?token=secret-token"); + const transport = createTransport("ws://localhost:3020/?token=secret-token"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -155,7 +165,7 @@ describe("WsTransport", () => { }); it("uses an explicit secure websocket base url", async () => { - const transport = new WsTransport("wss://app.example.com"); + const transport = createTransport("wss://app.example.com"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -166,7 +176,7 @@ describe("WsTransport", () => { }); it("uses an explicit insecure websocket base url for remote backends", async () => { - const transport = new WsTransport("ws://192.168.1.44:3773"); + const transport = createTransport("ws://192.168.1.44:3773"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -177,7 +187,7 @@ describe("WsTransport", () => { }); it("supports async websocket url providers", async () => { - const transport = new WsTransport(async () => "wss://remote.example.com/?wsToken=dynamic"); + const transport = createTransport(async () => "wss://remote.example.com/?wsToken=dynamic"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -188,7 +198,7 @@ describe("WsTransport", () => { }); it("tracks initial connection failures for the app error state", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -219,7 +229,7 @@ describe("WsTransport", () => { }); it("surfaces reconnecting state after a live socket disconnects", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -248,8 +258,48 @@ describe("WsTransport", () => { await transport.dispose(); }); + it("composes custom lifecycle handlers with default websocket state tracking", async () => { + const onOpen = vi.fn(); + const onClose = vi.fn(); + const transport = createTransport("ws://localhost:3020", { + onOpen, + onClose, + }); + + await waitFor(() => { + expect(sockets).toHaveLength(1); + }); + + const socket = getSocket(); + socket.open(); + + await waitFor(() => { + expect(onOpen).toHaveBeenCalledOnce(); + expect(getWsConnectionStatus()).toMatchObject({ + hasConnected: true, + phase: "connected", + }); + }); + + socket.close(1012, "service restart"); + + await waitFor(() => { + expect(onClose).toHaveBeenCalledWith({ + code: 1012, + reason: "service restart", + }); + expect(getWsConnectionStatus()).toMatchObject({ + attemptCount: 2, + closeReason: "service restart", + phase: "connecting", + }); + }, 2_000); + + await transport.dispose(); + }); + it("reconnects the websocket session without disposing the transport", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); await waitFor(() => { expect(sockets).toHaveLength(1); @@ -314,7 +364,7 @@ describe("WsTransport", () => { it("marks unary requests as slow until the first server ack arrives", async () => { const slowAckThresholdMs = 25; setSlowRpcAckThresholdMsForTests(slowAckThresholdMs); - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const requestPromise = transport.request((client) => client[WS_METHODS.serverUpsertKeybinding]({ @@ -368,7 +418,7 @@ describe("WsTransport", () => { }, 5_000); it("sends unary RPC requests and resolves successful exits", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const requestPromise = transport.request((client) => client[WS_METHODS.serverUpsertKeybinding]({ @@ -426,7 +476,7 @@ describe("WsTransport", () => { }); it("delivers stream chunks to subscribers", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const listener = vi.fn(); const unsubscribe = transport.subscribe( @@ -481,7 +531,7 @@ describe("WsTransport", () => { }); it("re-subscribes stream listeners after the stream exits", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const listener = vi.fn(); const onResubscribe = vi.fn(); @@ -590,7 +640,7 @@ describe("WsTransport", () => { }); it("does not fire onResubscribe when the first stream attempt exits before any value", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const listener = vi.fn(); const onResubscribe = vi.fn(); @@ -636,7 +686,7 @@ describe("WsTransport", () => { }); it("does not retry stream subscriptions after application-level failures", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); let attempts = 0; @@ -675,7 +725,7 @@ describe("WsTransport", () => { }); it("keeps retrying stream subscriptions after transport failures", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); let attempts = 0; @@ -708,7 +758,7 @@ describe("WsTransport", () => { }); it("logs a transport disconnect once even when multiple subscriptions fail together", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined); const unsubscribeA = transport.subscribe( @@ -741,7 +791,7 @@ describe("WsTransport", () => { }); it("streams finite request events without re-subscribing", async () => { - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const listener = vi.fn(); await waitFor(() => { @@ -857,7 +907,7 @@ describe("WsTransport", () => { exportIntervalMs: 10, }); - const transport = new WsTransport("ws://localhost:3020"); + const transport = createTransport("ws://localhost:3020"); const requestPromise = transport.request((client) => client[WS_METHODS.serverGetSettings]({})); await waitFor(() => {