From 630f13bfe623c6b288951c61c8423291d8bdb007 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 5 Aug 2026 18:46:34 -0700 Subject: [PATCH 1/2] refactor(mobile): separate native chat controller contract (#12826) --- .../mobile-native-chat-controller-contract.ts | 64 +++++++++++++++ .../use-mobile-native-chat-controller.ts | 82 ++----------------- 2 files changed, 69 insertions(+), 77 deletions(-) create mode 100644 mobile/src/session/mobile-native-chat-controller-contract.ts diff --git a/mobile/src/session/mobile-native-chat-controller-contract.ts b/mobile/src/session/mobile-native-chat-controller-contract.ts new file mode 100644 index 00000000000..a817b47b3ee --- /dev/null +++ b/mobile/src/session/mobile-native-chat-controller-contract.ts @@ -0,0 +1,64 @@ +import type { Dispatch, MutableRefObject, SetStateAction } from 'react' +import type { detectAgentPermission } from './mobile-native-chat-permission' +import type { parseAgentQuestion } from './mobile-native-chat-question' +import type { AskAnswerSelection, AskPrompt, parseAskFromStatus } from './mobile-native-chat-ask' +import type { MobileNativeChatSendOutcome } from './mobile-native-chat-send' +import type { MobileNativeChatPendingMessage } from './use-mobile-native-chat-drafts' +import type { useMobileNativeChatSession } from './use-mobile-native-chat-session' +import type { MobileNativeChatSessionOptionPickersProps } from './MobileNativeChatSessionOptionPickers' + +export type MobileNativeChatController = { + /** Whether a tab's effective view is chat (per-tab override, else the default). */ + isTabChatView: (tabId: string) => boolean + toggleTabChatView: (tabId: string) => void + showNativeChat: boolean + showNativeChatRef: MutableRefObject + /** Resolved agent for the active chat tab (names the empty-state copy). */ + nativeChatAgent: string | null + chatComposerText: string + setChatComposerText: Dispatch> + chatPending: MobileNativeChatPendingMessage[] + chatImagePreviewsByMessageId: Record + nativeChatSession: ReturnType + nativeChatAgentWorking: boolean + nativeChatStreamingText?: string + /** Agent mid-turn, regardless of whether chat is the visible view. */ + nativeChatStreamLive: boolean + /** Host/workspace/tab/session scope for stateful streaming suppression. */ + nativeChatStreamScopeKey: string + nativeChatPermission: ReturnType + nativeChatQuestion: ReturnType + /** The pending ask, already null while dismissed (dismissal lives here so it + * survives the chat-view subtree unmounting on a view toggle). */ + nativeChatAsk: ReturnType + /** Stable key for the current ask card (keys the card component). */ + nativeChatAskKey: string | null + /** Hide the current ask until a genuinely different question arrives. */ + dismissNativeChatAsk: () => void + handleNativeChatAnswerAsk: ( + prompt: AskPrompt, + selections: AskAnswerSelection[] + ) => Promise + handleNativeChatCancelAsk: () => Promise + handleNativeChatRespondPermission: (text: string) => Promise + handleNativeChatStop: () => void + nativeChatFilePaths: string[] + loadNativeChatFiles: (query: string) => void + handleNativeChatQuestionAnswer: (text: string) => Promise + handleNativeChatSend: (text: string, images?: string[]) => Promise + /** Outcome-preserving send: callers that pasted terminal input beforehand + * (image sends) must see 'unknown' to heal a possibly-orphaned paste. Such a + * caller passes its own `deadline` so the paste it already spent and this text + * body share one budget instead of holding the composer for two. */ + handleNativeChatSendWithOutcome: ( + text: string, + images?: string[], + deadline?: number + ) => Promise + /** Launch-context text still parked on the agent's TUI input line, or null. + * Image sends read it to size their leading clear (one Ctrl+U per line). */ + readSeededLaunchDraft: () => string | null + /** Model/session-option pickers for the composer, or null when the active + * agent has no session-option catalog. */ + nativeChatSessionOptions: MobileNativeChatSessionOptionPickersProps | null +} diff --git a/mobile/src/session/use-mobile-native-chat-controller.ts b/mobile/src/session/use-mobile-native-chat-controller.ts index eaa36fc5051..46012dadbe4 100644 --- a/mobile/src/session/use-mobile-native-chat-controller.ts +++ b/mobile/src/session/use-mobile-native-chat-controller.ts @@ -1,101 +1,29 @@ -import { - useCallback, - useLayoutEffect, - useRef, - type Dispatch, - type MutableRefObject, - type SetStateAction -} from 'react' +import { useCallback, useLayoutEffect, useRef, type MutableRefObject } from 'react' import { encodeNativeChatTranscriptIdentity } from '../../../src/shared/native-chat-transcript-retention' import { useMobileSessionViewMode } from './use-mobile-session-view-mode' import type { RpcClient } from '../transport/rpc-client' import type { ConnectionState } from '../transport/types' -import { - parseAskFromStatus, - type AskAnswerSelection, - type AskPrompt -} from './mobile-native-chat-ask' import { type MobileNativeChatTab, resolveMobileNativeChat } from './mobile-native-chat-eligibility' -import { detectAgentPermission } from './mobile-native-chat-permission' -import { parseAgentQuestion } from './mobile-native-chat-question' import { useMobileNativeChatPermissionSend } from './mobile-native-chat-permission-send' -import type { MobileNativeChatSendOutcome } from './mobile-native-chat-send' import { useMobileNativeChatAnswerSend } from './use-mobile-native-chat-answer-send' import { useMobileNativeChatAskDismiss } from './use-mobile-native-chat-ask-dismiss' import { useMobileNativeChatCancelAsk } from './use-mobile-native-chat-cancel-ask' -import { - useMobileNativeChatDrafts, - type MobileNativeChatPendingMessage -} from './use-mobile-native-chat-drafts' +import { useMobileNativeChatDrafts } from './use-mobile-native-chat-drafts' import { useMobileNativeChatFileSearch } from './use-mobile-native-chat-file-search' import { useMobileNativeChatMessageSend } from './use-mobile-native-chat-message-send' import { mobileNativeChatScopeKey } from './mobile-native-chat-scope-key' import { useMobileNativeChatSession } from './use-mobile-native-chat-session' import { useMobileNativeChatSessionOptions } from './use-mobile-native-chat-session-options' -import type { MobileNativeChatSessionOptionPickersProps } from './MobileNativeChatSessionOptionPickers' import { useMobileNativeChatPrompts } from './use-mobile-native-chat-prompts' import { useMobileNativeChatStop } from './use-mobile-native-chat-stop' import { useNativeChatAcceptedAction } from './use-native-chat-action-outcomes' import { useThrottledLatestValue } from './use-throttled-latest-value' +import type { MobileNativeChatController } from './mobile-native-chat-controller-contract' + +export type { MobileNativeChatController } from './mobile-native-chat-controller-contract' const NATIVE_CHAT_STREAM_THROTTLE_MS = 50 -export type MobileNativeChatController = { - /** Whether a tab's effective view is chat (per-tab override, else the default). */ - isTabChatView: (tabId: string) => boolean - toggleTabChatView: (tabId: string) => void - showNativeChat: boolean - showNativeChatRef: MutableRefObject - /** Resolved agent for the active chat tab (names the empty-state copy). */ - nativeChatAgent: string | null - chatComposerText: string - setChatComposerText: Dispatch> - chatPending: MobileNativeChatPendingMessage[] - chatImagePreviewsByMessageId: Record - nativeChatSession: ReturnType - nativeChatAgentWorking: boolean - nativeChatStreamingText?: string - /** Agent mid-turn, regardless of whether chat is the visible view. */ - nativeChatStreamLive: boolean - /** Host/workspace/tab/session scope for stateful streaming suppression. */ - nativeChatStreamScopeKey: string - nativeChatPermission: ReturnType - nativeChatQuestion: ReturnType - /** The pending ask, already null while dismissed (dismissal lives here so it - * survives the chat-view subtree unmounting on a view toggle). */ - nativeChatAsk: ReturnType - /** Stable key for the current ask card (keys the card component). */ - nativeChatAskKey: string | null - /** Hide the current ask until a genuinely different question arrives. */ - dismissNativeChatAsk: () => void - handleNativeChatAnswerAsk: ( - prompt: AskPrompt, - selections: AskAnswerSelection[] - ) => Promise - handleNativeChatCancelAsk: () => Promise - handleNativeChatRespondPermission: (text: string) => Promise - handleNativeChatStop: () => void - nativeChatFilePaths: string[] - loadNativeChatFiles: (query: string) => void - handleNativeChatQuestionAnswer: (text: string) => Promise - handleNativeChatSend: (text: string, images?: string[]) => Promise - /** Outcome-preserving send: callers that pasted terminal input beforehand - * (image sends) must see 'unknown' to heal a possibly-orphaned paste. Such a - * caller passes its own `deadline` so the paste it already spent and this text - * body share one budget instead of holding the composer for two. */ - handleNativeChatSendWithOutcome: ( - text: string, - images?: string[], - deadline?: number - ) => Promise - /** Launch-context text still parked on the agent's TUI input line, or null. - * Image sends read it to size their leading clear (one Ctrl+U per line). */ - readSeededLaunchDraft: () => string | null - /** Model/session-option pickers for the composer, or null when the active - * agent has no session-option catalog. */ - nativeChatSessionOptions: MobileNativeChatSessionOptionPickersProps | null -} - /** Owns mobile native-chat state and teardown outside the already dense session * route. The route remains responsible only for choosing and rendering the view. */ export function useMobileNativeChatController(args: { From 211b2d1a35200cabc5eccf792685f703640b0c58 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Wed, 5 Aug 2026 18:59:20 -0700 Subject: [PATCH 2/2] fix(runtime): require consecutive missed probes before reaping a paired socket (#12790) The paired-runtime WS heartbeat terminated a client after a single unanswered 15s ping. One missed pong is UNKNOWN, not proof the peer is gone: a cellular or Tailscale blackhole, or a stalled TCP retransmit, routinely swallows one pong from a peer that is still there. Users on flaky paths saw constant drops, each costing a full redial plus E2EE re-handshake and subscription replay. Reap now needs MISSED_PROBE_LIMIT (3) consecutive unanswered probes, counted per socket rather than timed. Any proof of life -- pong or any inbound frame -- clears the count, as does a resume from a server-loop pause, since a gap the client was never given a chance to answer must not top up its budget. Missed sweeps still re-probe, so a recovered path proves itself on the next tick. Three matches the liveness budgets already in the product: the web client gives 45s (25s idle + 20s probe grace) and the relay control gives 75s. The paired transport's single miss was the outlier. Also gives the web client's redial the one-sided jitter the shared-control path already had, so a fleet dropped by one shared blip does not re-dial in lockstep; the helper is extracted to src/shared/reconnect-jitter.ts and shared by both. STA-3320, #12327 Co-authored-by: Jinwoo-H --- ...r-heartbeat-missed-probe-tolerance.test.ts | 139 ++++++++++++++++++ .../remote-runtime-server-heartbeat.test.ts | 20 ++- .../rpc/remote-runtime-server-heartbeat.ts | 36 ++++- .../rpc/ws-transport-accept-order.test.ts | 19 ++- ...ws-transport-transient-packet-loss.test.ts | 83 +++++++++++ src/main/runtime/rpc/ws-transport.ts | 2 +- .../src/web/web-runtime-client.test.ts | 3 + src/renderer/src/web/web-runtime-client.ts | 4 +- src/shared/reconnect-jitter.test.ts | 17 +++ src/shared/reconnect-jitter.ts | 7 + .../remote-runtime-shared-control-state.ts | 7 +- 11 files changed, 310 insertions(+), 27 deletions(-) create mode 100644 src/main/runtime/rpc/remote-runtime-server-heartbeat-missed-probe-tolerance.test.ts create mode 100644 src/main/runtime/rpc/ws-transport-transient-packet-loss.test.ts create mode 100644 src/shared/reconnect-jitter.test.ts create mode 100644 src/shared/reconnect-jitter.ts diff --git a/src/main/runtime/rpc/remote-runtime-server-heartbeat-missed-probe-tolerance.test.ts b/src/main/runtime/rpc/remote-runtime-server-heartbeat-missed-probe-tolerance.test.ts new file mode 100644 index 00000000000..2e3a3427175 --- /dev/null +++ b/src/main/runtime/rpc/remote-runtime-server-heartbeat-missed-probe-tolerance.test.ts @@ -0,0 +1,139 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { WebSocket } from 'ws' +import { RemoteRuntimeServerHeartbeat } from './remote-runtime-server-heartbeat' + +afterEach(() => { + vi.useRealTimers() +}) + +const INTERVAL_MS = 100 + +function makeSocket(): WebSocket { + return { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket +} + +describe('RemoteRuntimeServerHeartbeat missed-probe tolerance', () => { + it('keeps a client whose pong is merely delayed past one probe window', async () => { + vi.useFakeTimers() + let now = 1_000 + const socket = makeSocket() + const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now) + heartbeat.noteAlive(socket) + heartbeat.start(() => [socket]) // probe #1 + heartbeat.noteAlive(socket) // pongs probe #1 + + // A transient blackhole: probe #2 goes out and its pong is stuck in the network. + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + expect(socket.ping).toHaveBeenCalledTimes(2) + expect(socket.terminate).not.toHaveBeenCalled() + + // The path recovers and the delayed pong lands right after the next sweep would have run. + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + heartbeat.noteAlive(socket) + + // One unanswered probe is not evidence the peer is gone: it must still be connected, + // and it must have been re-probed rather than reaped. + expect(socket.terminate).not.toHaveBeenCalled() + expect(socket.ping).toHaveBeenCalledTimes(3) + heartbeat.stop() + }) + + it('reaps only after consecutive probes go unanswered, counted not timed', async () => { + vi.useFakeTimers() + let now = 1_000 + const socket = makeSocket() + const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now) + heartbeat.noteAlive(socket) + heartbeat.start(() => [socket]) + heartbeat.noteAlive(socket) + + const missesBeforeReap: number[] = [] + for (let probe = 0; probe < 6; probe += 1) { + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + missesBeforeReap.push((socket.terminate as unknown as ReturnType).mock.calls.length) + } + + // Silence must be tolerated for more than a single probe, and must eventually end the socket. + expect(missesBeforeReap[0]).toBe(0) + expect(missesBeforeReap[1]).toBe(0) + expect(missesBeforeReap.at(-1)).toBeGreaterThan(0) + heartbeat.stop() + }) + + it('lets any proof of life reset accumulated misses', async () => { + vi.useFakeTimers() + let now = 1_000 + const socket = makeSocket() + const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now) + heartbeat.noteAlive(socket) + heartbeat.start(() => [socket]) + heartbeat.noteAlive(socket) + + // Two silent probes, then a single inbound frame, repeated well past any fixed budget. + for (let cycle = 0; cycle < 5; cycle += 1) { + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + heartbeat.noteAlive(socket) + } + + expect(socket.terminate).not.toHaveBeenCalled() + heartbeat.stop() + }) + + it('clears accumulated misses when the server event loop resumes from a pause', async () => { + vi.useFakeTimers() + let now = 1_000 + const socket = makeSocket() + const heartbeat = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now) + heartbeat.noteAlive(socket) + heartbeat.start(() => [socket]) + heartbeat.noteAlive(socket) + + const sweep = async (): Promise => { + now += INTERVAL_MS + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + } + const reaped = (candidate: WebSocket): boolean => + (candidate.terminate as unknown as ReturnType).mock.calls.length > 0 + + // Discover the budget rather than hardcoding it: the first sweep consumes the pong above, so + // every sweep after it is an unanswered probe. + await sweep() + let toleratedMisses = 0 + // Bounded so a heartbeat that never reaps fails here instead of hanging out to the test timeout. + while (!reaped(socket) && toleratedMisses < 20) { + await sweep() + toleratedMisses += 1 + } + expect(reaped(socket)).toBe(true) + expect(toleratedMisses).toBeGreaterThan(1) + heartbeat.stop() + + // Replay on a fresh socket: bank misses to one short of the limit, suspend, then bank the same + // number again. The pause must write the earlier misses off, not top them up to the limit. + const resumed = new RemoteRuntimeServerHeartbeat(INTERVAL_MS, () => now) + const resumedSocket = makeSocket() + resumed.noteAlive(resumedSocket) + resumed.start(() => [resumedSocket]) + resumed.noteAlive(resumedSocket) + await sweep() + for (let miss = 0; miss < toleratedMisses - 1; miss += 1) { + await sweep() + } + expect(reaped(resumedSocket)).toBe(false) + + now += 3_600_000 + await vi.advanceTimersByTimeAsync(INTERVAL_MS) + for (let miss = 0; miss < toleratedMisses - 1; miss += 1) { + await sweep() + } + + expect(reaped(resumedSocket)).toBe(false) + resumed.stop() + }) +}) diff --git a/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts b/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts index a34b43c3c8b..e1a58d23587 100644 --- a/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts +++ b/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts @@ -7,12 +7,13 @@ afterEach(() => { }) describe('RemoteRuntimeServerHeartbeat', () => { - it('still reaps a client that misses a probe while another remains alive', async () => { + it('still reaps a persistently silent client while another remains alive', async () => { vi.useFakeTimers() let now = 1_000 const responsiveSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket const deadSocket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket - const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now) + // Limit of 2 keeps the sweep count readable; production uses the module default. + const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now, 128, 2) heartbeat.noteAlive(responsiveSocket) heartbeat.noteAlive(deadSocket) // start() probes immediately: both are pinged now (probe #1) and cleared to await a pong. @@ -20,12 +21,20 @@ describe('RemoteRuntimeServerHeartbeat', () => { // Only the responsive socket pongs the immediate probe. heartbeat.noteAlive(responsiveSocket) + // Miss #1: not yet evidence, so the silent socket is re-probed rather than reaped. + now += 100 + await vi.advanceTimersByTimeAsync(100) + heartbeat.noteAlive(responsiveSocket) + expect(deadSocket.ping).toHaveBeenCalledTimes(2) + expect(deadSocket.terminate).not.toHaveBeenCalled() + + // Miss #2 reaches the limit: consecutive silence is evidence. now += 100 await vi.advanceTimersByTimeAsync(100) - expect(responsiveSocket.ping).toHaveBeenCalledTimes(2) + expect(responsiveSocket.ping).toHaveBeenCalledTimes(3) expect(responsiveSocket.terminate).not.toHaveBeenCalled() - expect(deadSocket.ping).toHaveBeenCalledTimes(1) + expect(deadSocket.ping).toHaveBeenCalledTimes(2) expect(deadSocket.terminate).toHaveBeenCalledTimes(1) heartbeat.stop() }) @@ -34,7 +43,8 @@ describe('RemoteRuntimeServerHeartbeat', () => { vi.useFakeTimers() let now = 1_000 const socket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket - const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now) + // Limit of 1 isolates the resume grant: without it the very next sweep would reap. + const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now, 128, 1) heartbeat.noteAlive(socket) // start() probes immediately (ping #1); the socket pongs it. heartbeat.start(() => [socket]) diff --git a/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts b/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts index 3b5bb9760df..3913e9d9b6b 100644 --- a/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts +++ b/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts @@ -1,16 +1,27 @@ import type { WebSocket } from 'ws' +// Why: one unanswered probe is UNKNOWN, not death — a cellular/Tailscale blackhole or a stalled TCP +// retransmit routinely swallows a single pong from a peer that is still there (STA-3320). Only a run of +// consecutive unanswered probes is evidence. Three matches the web client's own 45s liveness budget +// (25s idle + 20s probe grace), so both ends of a paired session give up on roughly the same evidence. +const MISSED_PROBE_LIMIT = 3 + export class RemoteRuntimeServerHeartbeat { private timer: ReturnType | null = null private lastTickAt: number | null = null private readonly alive = new WeakSet() + // Why: consecutive probes a socket has left unanswered. Absent = zero, so live sockets cost nothing. + private readonly missedProbes = new WeakMap() constructor( private readonly intervalMs: number, private readonly now: () => number = Date.now, - private readonly warningClientCount = 128 + private readonly warningClientCount = 128, + private readonly missedProbeLimit = MISSED_PROBE_LIMIT ) {} + // Why: hot path — every inbound frame calls this. The sweep clears the miss count when it observes + // the flag, so proof of life stays a single WeakSet write. noteAlive(socket: WebSocket): void { this.alive.add(socket) } @@ -47,16 +58,27 @@ export class RemoteRuntimeServerHeartbeat { for (const socket of clients) { clientCount += 1 if (resumedFromPause) { - // Why: a delayed server tick cannot infer that clients missed a probe they had no chance to answer. + // Why: a delayed server tick cannot infer that clients missed a probe they had no chance to + // answer. Seeding alive also clears any misses banked before the pause, so a suspend can never + // top up a budget the client was never given a chance to defend. this.alive.add(socket) } - if (!this.alive.has(socket)) { - socket.terminate() - reaped += 1 - continue + if (this.alive.has(socket)) { + this.alive.delete(socket) + this.missedProbes.delete(socket) + } else { + const missed = (this.missedProbes.get(socket) ?? 0) + 1 + if (missed >= this.missedProbeLimit) { + this.missedProbes.delete(socket) + socket.terminate() + reaped += 1 + continue + } + this.missedProbes.set(socket, missed) } - this.alive.delete(socket) try { + // Why: re-probe every sweep, including missed ones, so a path that just recovered can prove + // itself on the next tick instead of waiting out the rest of the budget. socket.ping() } catch { // Why: a mid-teardown socket is finalized by its close/error listener. diff --git a/src/main/runtime/rpc/ws-transport-accept-order.test.ts b/src/main/runtime/rpc/ws-transport-accept-order.test.ts index 4f56a2da53e..20ae28f0f0e 100644 --- a/src/main/runtime/rpc/ws-transport-accept-order.test.ts +++ b/src/main/runtime/rpc/ws-transport-accept-order.test.ts @@ -83,7 +83,7 @@ describe('WebSocketTransport accepted socket ordering', () => { ).toEqual([0, 0, 0, 0]) }) - it('reaps an unresponsive accepted socket by the first interval deadline', async () => { + it('reaps an unresponsive accepted socket once consecutive probes go unanswered', async () => { vi.useFakeTimers() const events: string[] = [] let socket: WebSocket @@ -99,19 +99,22 @@ describe('WebSocketTransport accepted socket ordering', () => { socket.once('open', () => events.push('open')) socket.emit('open') lifecycle.handleConnection(socket) - await vi.advanceTimersByTimeAsync(99) + // A single silent interval is not evidence: the socket is re-probed, not reaped. + await vi.advanceTimersByTimeAsync(100) + expect(socket.terminate).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(199) expect(socket.terminate).not.toHaveBeenCalled() await vi.advanceTimersByTimeAsync(1) expect(socket.terminate).toHaveBeenCalledTimes(1) - expect(events).toEqual(['open', 'ping', 'close']) + expect(events).toEqual(['open', 'ping', 'ping', 'ping', 'close']) expect(lifecycle.heartbeatConnections.size).toBe(0) expect(lifecycle.heartbeat.timer).toBeNull() expect(vi.getTimerCount()).toBe(0) }) - it('keeps later sockets on the shared cadence and reaps them within two intervals', async () => { + it('keeps later sockets on the shared cadence and reaps them on the shared budget', async () => { vi.useFakeTimers() vi.setSystemTime(0) const firstPingTimes: number[] = [] @@ -150,13 +153,15 @@ describe('WebSocketTransport accepted socket ordering', () => { expect(laterPingTimes).toEqual([100]) expect(laterSocket.terminate).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(99) + // Re-probed on every sweep while its misses bank, so a recovered path can answer immediately. + await vi.advanceTimersByTimeAsync(299) + expect(laterPingTimes).toEqual([100, 200, 300]) expect(laterSocket.terminate).not.toHaveBeenCalled() await vi.advanceTimersByTimeAsync(1) expect(laterSocket.terminate).toHaveBeenCalledTimes(1) - expect(laterReapTimes).toEqual([200]) - expect(firstPingTimes).toEqual([0, 100, 200]) + expect(laterReapTimes).toEqual([400]) + expect(firstPingTimes).toEqual([0, 100, 200, 300, 400]) expect( ['pong', 'message', 'close', 'error'].map((event) => laterSocket.listenerCount(event)) ).toEqual([0, 0, 0, 0]) diff --git a/src/main/runtime/rpc/ws-transport-transient-packet-loss.test.ts b/src/main/runtime/rpc/ws-transport-transient-packet-loss.test.ts new file mode 100644 index 00000000000..a8a5eb9c4de --- /dev/null +++ b/src/main/runtime/rpc/ws-transport-transient-packet-loss.test.ts @@ -0,0 +1,83 @@ +// Real ws server + real ws client over loopback: a live, well-behaved peer whose pong is swallowed by +// a transient blackhole must not be terminated (STA-3320). +import { afterEach, describe, expect, it, vi } from 'vitest' +import WebSocket from 'ws' +import { WebSocketTransport } from './ws-transport' + +const PROBE_INTERVAL_MS = 150 + +// Why: the heartbeat calls now() exactly once per sweep, so a per-call step makes elapsed exactly one +// interval every sweep — real scheduler jitter can never be mistaken for a suspended loop, and no +// assertion in this file depends on wall time. +function createStepClock(stepMs: number): () => number { + let calls = 0 + return () => 1_000_000 + calls++ * stepMs +} + +describe('WebSocketTransport under transient packet loss', () => { + const transports: WebSocketTransport[] = [] + const sockets: WebSocket[] = [] + + afterEach(async () => { + for (const socket of sockets) { + socket.removeAllListeners() + socket.terminate() + } + sockets.length = 0 + await Promise.all(transports.map((t) => t.stop().catch(() => {}))) + transports.length = 0 + }) + + async function startTransport(): Promise { + const transport = new WebSocketTransport({ + host: '127.0.0.1', + port: 0, + heartbeatIntervalMs: PROBE_INTERVAL_MS, + heartbeatNow: createStepClock(PROBE_INTERVAL_MS), + // Why: this test is about liveness, not the auth handshake; keep the pre-auth reaper out of it. + preAuthTimeoutMs: 600_000 + }) + transports.push(transport) + await transport.start() + return transport + } + + it('keeps a live client whose pong is swallowed for a single probe', async () => { + const transport = await startTransport() + // autoPong:false hands us the protocol-level pong so we can drop exactly one, the way a + // cellular/Tailscale blackhole drops exactly the frames in flight during the stall. + const client = new WebSocket(`ws://127.0.0.1:${transport.resolvedPort}`, { autoPong: false }) + sockets.push(client) + + let probesReceived = 0 + const pongedProbes: number[] = [] + const swallowedProbe = 2 + let closed = false + client.on('ping', () => { + probesReceived += 1 + if (probesReceived === swallowedProbe) { + return + } + pongedProbes.push(probesReceived) + client.pong() + }) + client.on('close', () => { + closed = true + }) + await new Promise((resolve, reject) => { + client.once('open', resolve) + client.once('error', reject) + }) + + // Bounded by counted probe events, never by elapsed time. + await vi.waitFor(() => expect(closed || probesReceived >= swallowedProbe + 2).toBe(true), { + timeout: 15_000, + interval: 10 + }) + + expect({ closed, probesReceived }).toEqual({ closed: false, probesReceived: 4 }) + // The peer was demonstrably alive across the blackhole: it answered the probes on both sides of it. + expect(pongedProbes).toEqual([1, 3, 4]) + expect(client.readyState).toBe(WebSocket.OPEN) + }) +}) diff --git a/src/main/runtime/rpc/ws-transport.ts b/src/main/runtime/rpc/ws-transport.ts index 1444813fd65..ec58f8697ae 100644 --- a/src/main/runtime/rpc/ws-transport.ts +++ b/src/main/runtime/rpc/ws-transport.ts @@ -21,7 +21,7 @@ type WebSocketMessageHandler = { ): void }['bivarianceHack'] -// Why: mobile clients background-suspend sockets with no TCP FIN, leaving half-opens that otherwise only the OS keepalive (~2h) reaps; a 15s ping/pong sweep bounds that to ~30s (clients auto-pong per RFC 6455). +// Why: mobile clients background-suspend sockets with no TCP FIN, leaving half-opens that otherwise only the OS keepalive (~2h) reaps; a 15s ping/pong sweep bounds that to ~60s (clients auto-pong per RFC 6455), since a reap needs consecutive unanswered probes rather than one (STA-3320). const HEARTBEAT_INTERVAL_MS = 15_000 export type WebSocketTransportOptions = { diff --git a/src/renderer/src/web/web-runtime-client.test.ts b/src/renderer/src/web/web-runtime-client.test.ts index 85a5575e3bf..e9d510a6234 100644 --- a/src/renderer/src/web/web-runtime-client.test.ts +++ b/src/renderer/src/web/web-runtime-client.test.ts @@ -210,6 +210,8 @@ describe('WebRuntimeClient', () => { } timerWindow.setTimeout = setTimeout timerWindow.clearTimeout = clearTimeout + // Pin the one-sided reconnect jitter to zero so the redial lands on its exact backoff step. + const random = vi.spyOn(Math, 'random').mockReturnValue(0) const client = new WebRuntimeClient({ v: 2, endpoint: 'ws://127.0.0.1:6768', @@ -235,6 +237,7 @@ describe('WebRuntimeClient', () => { expect(replacementSocket.send).toHaveBeenCalledTimes(1) } finally { client.close() + random.mockRestore() vi.useRealTimers() } }) diff --git a/src/renderer/src/web/web-runtime-client.ts b/src/renderer/src/web/web-runtime-client.ts index 12507f6225f..3ed33e18887 100644 --- a/src/renderer/src/web/web-runtime-client.ts +++ b/src/renderer/src/web/web-runtime-client.ts @@ -16,6 +16,7 @@ import { } from './web-e2ee' import { SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' import { createWebRuntimeUnauthorizedError } from './web-runtime-client-error' +import { withReconnectJitter } from '../../../shared/reconnect-jitter' type WebRuntimeConnectionState = | 'disconnected' @@ -641,8 +642,9 @@ export class WebRuntimeClient { if (this.reconnectTimer || this.intentionallyClosed) { return } - const delay = + const delay = withReconnectJitter( RECONNECT_DELAYS_MS[Math.min(this.reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)] + ) this.reconnectAttempt += 1 this.reconnectTimer = window.setTimeout(() => { this.reconnectTimer = null diff --git a/src/shared/reconnect-jitter.test.ts b/src/shared/reconnect-jitter.test.ts new file mode 100644 index 00000000000..81b0bd448dd --- /dev/null +++ b/src/shared/reconnect-jitter.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, it } from 'vitest' +import { withReconnectJitter } from './reconnect-jitter' + +describe('withReconnectJitter', () => { + it('never returns less than the backoff floor', () => { + expect(withReconnectJitter(500, () => 0)).toBe(500) + }) + + it('spreads a fleet that was dropped by one shared blip', () => { + const fleet = Array.from({ length: 32 }, (_, index) => + withReconnectJitter(500, () => index / 32) + ) + expect(new Set(fleet).size).toBeGreaterThan(1) + expect(Math.min(...fleet)).toBeGreaterThanOrEqual(500) + expect(Math.max(...fleet)).toBeLessThanOrEqual(600) + }) +}) diff --git a/src/shared/reconnect-jitter.ts b/src/shared/reconnect-jitter.ts new file mode 100644 index 00000000000..576db7f0270 --- /dev/null +++ b/src/shared/reconnect-jitter.ts @@ -0,0 +1,7 @@ +// Why: a blip on a shared path (Tailscale, cellular, host restart) drops every socket on that path in +// the same instant. A jitterless backoff ladder then re-dials all of them at the same millisecond, +// which is exactly the thundering herd the host is least able to absorb while it is still recovering. +// One-sided so a delay is never shortened below its backoff floor. +export function withReconnectJitter(delayMs: number, random: () => number = Math.random): number { + return delayMs + Math.floor(delayMs * 0.2 * random()) +} diff --git a/src/shared/remote-runtime-shared-control-state.ts b/src/shared/remote-runtime-shared-control-state.ts index c74c50916c9..eba1d4a1be8 100644 --- a/src/shared/remote-runtime-shared-control-state.ts +++ b/src/shared/remote-runtime-shared-control-state.ts @@ -10,6 +10,7 @@ import type { SharedControlReadyWaiter } from './remote-runtime-shared-control-types' import { getSubscriptionId, isEndResult } from './remote-runtime-shared-control-protocol' +import { withReconnectJitter } from './reconnect-jitter' import { tagRuntimeSubscriptionReplayResponse } from './runtime-subscription-replay' export function buildSharedControlDiagnostics(args: { @@ -208,9 +209,3 @@ export function scheduleSharedControlReconnect(args: { return { timer, reconnectAttempt: args.reconnectAttempt + 1 } } -function withReconnectJitter(delayMs: number): number { - // Why: when a remote host restarts, all passive subscriptions reconnect - // together. A small one-sided jitter avoids synchronized retry spikes. - const jitterMs = Math.floor(delayMs * 0.2 * Math.random()) - return delayMs + jitterMs -}