From 07d10c8e8d4883026c0c4529262c8b7f5b7e3908 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Wed, 6 May 2026 02:05:38 -0700 Subject: [PATCH] fix(mobile): WS heartbeat to reap half-open mobile sockets (#1500) Co-authored-by: Orca --- mobile/app.json | 2 +- mobile/src/transport/rpc-client.test.ts | 111 ++++++++++++++++++ mobile/src/transport/rpc-client.ts | 133 +++++++++++++++++++--- src/main/runtime/rpc/ws-transport.test.ts | 56 +++++++++ src/main/runtime/rpc/ws-transport.ts | 81 ++++++++++++- 5 files changed, 367 insertions(+), 16 deletions(-) create mode 100644 mobile/src/transport/rpc-client.test.ts diff --git a/mobile/app.json b/mobile/app.json index 6ad232bec..5e03ae105 100644 --- a/mobile/app.json +++ b/mobile/app.json @@ -16,7 +16,7 @@ "ios": { "supportsTablet": false, "bundleIdentifier": "com.stably.orca.mobile", - "buildNumber": "9", + "buildNumber": "10", "infoPlist": { "NSLocalNetworkUsageDescription": "Orca connects to the desktop app on your local network.", "NSAppTransportSecurity": { diff --git a/mobile/src/transport/rpc-client.test.ts b/mobile/src/transport/rpc-client.test.ts new file mode 100644 index 000000000..505a5722b --- /dev/null +++ b/mobile/src/transport/rpc-client.test.ts @@ -0,0 +1,111 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { connect } from './rpc-client' + +vi.mock('./e2ee', () => ({ + generateKeyPair: () => ({ + publicKey: new Uint8Array(32), + secretKey: new Uint8Array(32) + }), + deriveSharedKey: () => new Uint8Array(32), + publicKeyFromBase64: () => new Uint8Array(32), + publicKeyToBase64: () => 'client-public-key', + encrypt: (plaintext: string) => `encrypted:${plaintext}`, + decrypt: (raw: string) => raw.replace(/^encrypted:/, '') +})) + +class MockWebSocket { + static CONNECTING = 0 + static OPEN = 1 + static CLOSING = 2 + static CLOSED = 3 + + readonly CONNECTING = MockWebSocket.CONNECTING + readonly OPEN = MockWebSocket.OPEN + readonly CLOSING = MockWebSocket.CLOSING + readonly CLOSED = MockWebSocket.CLOSED + + readyState = MockWebSocket.CONNECTING + onopen: (() => void) | null = null + onclose: (() => void) | null = null + onmessage: ((event: { data: string }) => void) | null = null + onerror: (() => void) | null = null + emitCloseOnClose = true + sent: string[] = [] + close = vi.fn(() => { + if (this.readyState === MockWebSocket.CLOSED) return + this.readyState = MockWebSocket.CLOSED + if (this.emitCloseOnClose) { + this.onclose?.() + } + }) + + constructor(readonly endpoint: string) { + mockSockets.push(this) + } + + send(payload: string): void { + this.sent.push(payload) + } + + open(): void { + this.readyState = MockWebSocket.OPEN + this.onopen?.() + } + + receive(payload: string): void { + this.onmessage?.({ data: payload }) + } +} + +const mockSockets: MockWebSocket[] = [] +const originalWebSocket = globalThis.WebSocket + +describe('mobile rpc-client connection timeout', () => { + beforeEach(() => { + vi.useFakeTimers() + mockSockets.length = 0 + globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket + }) + + afterEach(() => { + vi.useRealTimers() + globalThis.WebSocket = originalWebSocket + }) + + it('closes a socket that never opens so reconnect can run', () => { + const states: string[] = [] + const client = connect('ws://desktop.invalid', 'token', 'server-key', (state) => { + states.push(state) + }) + + expect(client.getState()).toBe('connecting') + expect(mockSockets).toHaveLength(1) + mockSockets[0]!.emitCloseOnClose = false + + vi.advanceTimersByTime(12_000) + + expect(mockSockets[0]!.close).toHaveBeenCalledTimes(1) + expect(client.getState()).toBe('reconnecting') + expect(states).toContain('reconnecting') + + client.close() + }) + + it('clears the open timeout once the socket opens and authenticates', () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + + socket.open() + socket.receive(JSON.stringify({ type: 'e2ee_ready' })) + socket.receive('encrypted:{"type":"e2ee_authenticated"}') + + expect(client.getState()).toBe('connected') + + vi.advanceTimersByTime(12_000) + + expect(socket.close).not.toHaveBeenCalled() + expect(client.getState()).toBe('connected') + + client.close() + }) +}) diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 7ada7965e..d12690003 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -41,7 +41,24 @@ export type RpcClient = { // "magic". Shorter backoff makes the auto-recovery path feel as fast. const RECONNECT_DELAYS = [500, 1000, 2000, 4000] const REQUEST_TIMEOUT_MS = 30_000 +const CONNECT_TIMEOUT_MS = 12_000 const HANDSHAKE_TIMEOUT_MS = 5_000 +// Why: RN's WebSocket implementation may not expose static readyState +// constants, but the protocol value for CONNECTING is stable across runtimes. +const WEBSOCKET_CONNECTING_STATE = 0 + +// Why: app-level liveness probe. The server runs its own ping/pong sweep +// at 15s, but RN's WebSocket runtime auto-pongs at the native layer +// without surfacing anything to JS — so the mobile side can't *see* that +// the server thinks the link is fine. To detect a half-open socket from +// the mobile direction (e.g. server crashed, phone moved between wifi +// and cellular without TCP RST) we periodically round-trip a tiny RPC. +// If two consecutive probes time out we force-close the WS, which fires +// the existing reconnect path. 20s cadence + the 30s request timeout = +// worst-case ~50s before mobile decides the link is dead and kicks +// reconnect, which is still inside the user's perceived "responsive" +// window and well below iOS's typical background-disconnect window. +const ACTIVITY_PROBE_INTERVAL_MS = 20_000 export function connect( endpoint: string, @@ -54,7 +71,9 @@ export function connect( let requestCounter = 0 let reconnectAttempt = 0 let reconnectTimer: ReturnType | null = null + let connectTimer: ReturnType | null = null let handshakeTimer: ReturnType | null = null + let activityProbeTimer: ReturnType | null = null let intentionallyClosed = false // Why: fresh ephemeral keypair per connection provides forward secrecy. @@ -105,8 +124,23 @@ export function connect( sharedKey = null ws = new WebSocket(endpoint) + const openingWs = ws + + // Why: React Native can leave TCP/WebSocket opens pending indefinitely on + // flaky network handoffs. Force the existing onclose reconnect path if + // onopen never arrives, instead of leaving the UI stuck at "Connecting...". + connectTimer = setTimeout(() => { + connectTimer = null + if (ws === openingWs && openingWs.readyState === WEBSOCKET_CONNECTING_STATE) { + openingWs.close() + if (ws === openingWs) { + handleSocketClosed(openingWs) + } + } + }, CONNECT_TIMEOUT_MS) ws.onopen = () => { + clearConnectTimer() reconnectAttempt = 0 setState('handshaking') @@ -161,6 +195,7 @@ export function connect( handshakeTimer = null } setState('connected') + startActivityProbe() for (const [id, stream] of streamListeners) { sendEncrypted({ id, deviceToken, method: stream.method, params: stream.params }) } @@ -243,20 +278,7 @@ export function connect( } ws.onclose = () => { - ws = null - sharedKey = null - if (handshakeTimer) { - clearTimeout(handshakeTimer) - handshakeTimer = null - } - if (intentionallyClosed) { - setState('disconnected') - rejectAllPending('Connection closed') - return - } - rejectAllPending('Connection interrupted') - setState('reconnecting') - scheduleReconnect() + handleSocketClosed(openingWs) } ws.onerror = () => { @@ -264,6 +286,28 @@ export function connect( } } + function handleSocketClosed(closedWs: WebSocket) { + if (ws !== closedWs) { + return + } + clearConnectTimer() + ws = null + sharedKey = null + if (handshakeTimer) { + clearTimeout(handshakeTimer) + handshakeTimer = null + } + stopActivityProbe() + if (intentionallyClosed) { + setState('disconnected') + rejectAllPending('Connection closed') + return + } + rejectAllPending('Connection interrupted') + setState('reconnecting') + scheduleReconnect() + } + function scheduleReconnect() { const delay = RECONNECT_DELAYS[Math.min(reconnectAttempt, RECONNECT_DELAYS.length - 1)]! reconnectAttempt++ @@ -273,6 +317,65 @@ export function connect( }, delay) } + function clearConnectTimer() { + if (connectTimer) { + clearTimeout(connectTimer) + connectTimer = null + } + } + + // Why: app-level liveness probe — see ACTIVITY_PROBE_INTERVAL_MS comment + // at the top of the file. Fires while the channel is in 'connected' + // state, sends a tiny status.get, and force-closes the WS if the probe + // fails (which the existing onclose path then turns into a reconnect). + function startActivityProbe() { + stopActivityProbe() + activityProbeTimer = setInterval(() => { + // Why: only probe while the channel is actually in 'connected'. The + // sendRequest path itself waits for connected, but a probe scheduled + // during a reconnect would just stack up timeouts and confuse logs. + if (state !== 'connected' || !ws) return + const probeWs = ws + // Why: short timeout (8s) — server's heartbeat is 15s, so if we + // don't see *anything* back within 8s the link is almost certainly + // half-open. Using REQUEST_TIMEOUT_MS (30s) here would make the + // user wait nearly a minute before reconnect kicks in. + const id = nextId() + let timedOut = false + const timeout = setTimeout(() => { + timedOut = true + pending.delete(id) + // Why: only force-close if this is still the same socket the + // probe was sent on; a normal close that already swapped `ws` + // shouldn't trigger a redundant terminate. + if (probeWs === ws && probeWs.readyState === WebSocket.OPEN) { + probeWs.close() + } + }, 8_000) + pending.set(id, { + resolve: () => { + if (timedOut) return + clearTimeout(timeout) + }, + reject: () => { + if (timedOut) return + clearTimeout(timeout) + } + }) + if (!sendEncrypted({ id, deviceToken, method: 'status.get' })) { + clearTimeout(timeout) + pending.delete(id) + } + }, ACTIVITY_PROBE_INTERVAL_MS) + } + + function stopActivityProbe() { + if (activityProbeTimer) { + clearInterval(activityProbeTimer) + activityProbeTimer = null + } + } + function rejectAllPending(reason: string) { const error = new Error(reason) for (const [id, req] of pending) { @@ -385,10 +488,12 @@ export function connect( clearTimeout(reconnectTimer) reconnectTimer = null } + clearConnectTimer() if (handshakeTimer) { clearTimeout(handshakeTimer) handshakeTimer = null } + stopActivityProbe() if (ws) { ws.close() ws = null diff --git a/src/main/runtime/rpc/ws-transport.test.ts b/src/main/runtime/rpc/ws-transport.test.ts index ea3acd486..8bd80ca53 100644 --- a/src/main/runtime/rpc/ws-transport.test.ts +++ b/src/main/runtime/rpc/ws-transport.test.ts @@ -253,4 +253,60 @@ describe('WebSocketTransport', () => { const ws = await connectWs(second.resolvedPort) ws.close() }) + + it('reaps a half-open client that stops responding to pings', async () => { + // Why: regression cover for the half-open-socket leak that would + // strand mobile clients in the connection pool until OS TCP keepalive + // (~2 hours) reaped them. With the heartbeat, two consecutive ping + // ticks without a pong should cause terminate() to fire and free the + // slot. Verifying via the server's connection-close handler, which + // is what frees up the MAX_WS_CONNECTIONS budget in production. + const tls = makeTls() + const port = findFreePort() + const transport = new WebSocketTransport({ + host: '127.0.0.1', + port, + tlsCert: tls.cert, + tlsKey: tls.key, + heartbeatIntervalMs: 50 + }) + transport.onMessage(() => {}) + transports.push(transport) + + let serverClosed = false + transport.onConnectionClose(() => { + serverClosed = true + }) + + // Why: setClientId is what registers the ws → clientId mapping that + // onConnectionClose fires off. Hook the connection event before + // start so we can stamp every accepted ws with a token. + await transport.start() + + const ws = await connectWs(port) + // Why: pausing the underlying TCP socket halts both read (ping in) + // and write (pong out) at the kernel level, so the `ws` library's + // auto-pong can't actually be flushed back. From the server's + // perspective the client looks half-open — exactly the production + // failure mode iOS produces when it suspends a backgrounded socket. + const underlying = (ws as unknown as { _socket: { pause: () => void } })._socket + underlying.pause() + + // Why: we need a clientId on the ws so onConnectionClose actually + // fires. The transport sets it lazily via setClientId in production + // (after auth); in this test we don't run auth, so reach in. + const wss = (transport as unknown as { wss: { clients: Set<{ readyState: number }> } }).wss + for (const c of wss.clients) { + transport.setClientId(c as never, 'test-client') + } + + // Wait long enough for two heartbeat ticks (50ms each) plus slack. + const start = Date.now() + while (!serverClosed && Date.now() - start < 2_000) { + await new Promise((r) => setTimeout(r, 25)) + } + + expect(serverClosed).toBe(true) + expect(wss.clients.size).toBe(0) + }, 5_000) }) diff --git a/src/main/runtime/rpc/ws-transport.ts b/src/main/runtime/rpc/ws-transport.ts index 2d5881324..5a97642a3 100644 --- a/src/main/runtime/rpc/ws-transport.ts +++ b/src/main/runtime/rpc/ws-transport.ts @@ -11,11 +11,25 @@ import type { RpcTransport } from './transport' const MAX_WS_MESSAGE_BYTES = 1024 * 1024 const MAX_WS_CONNECTIONS = 32 +// Why: mobile clients (iOS/Android) regularly background-suspend their +// sockets without the OS sending a TCP FIN/RST, leaving the server with +// half-open WebSockets that count toward MAX_WS_CONNECTIONS. Without this +// heartbeat the only thing that ever reaps them is the OS's TCP keepalive +// (default macOS: ~2 hours idle + 11 min of probes), which is the +// "connection randomly turns green again after a long delay" symptom. +// Pinging every 15s and terminating any client that hasn't pong'd by the +// next tick collapses that worst case to ~30s. RN/browser WebSocket +// runtimes auto-respond to server pings with pongs at the protocol layer, +// so this works for any client that speaks RFC 6455. +const HEARTBEAT_INTERVAL_MS = 15_000 + export type WebSocketTransportOptions = { host: string port: number tlsCert?: string tlsKey?: string + // Why: test-only override. Production uses HEARTBEAT_INTERVAL_MS. + heartbeatIntervalMs?: number } export class WebSocketTransport implements RpcTransport { @@ -23,8 +37,14 @@ export class WebSocketTransport implements RpcTransport { private readonly port: number private readonly tlsCert: string | undefined private readonly tlsKey: string | undefined + private readonly heartbeatIntervalMs: number private httpServer: HttpsServer | HttpServer | null = null private wss: WebSocketServer | null = null + private heartbeatTimer: ReturnType | null = null + // Why: tracks whether each socket has pong'd since the last heartbeat + // sweep. A socket missing from the set when the next sweep fires is + // assumed dead and terminated. + private wsAlive = new WeakSet() private messageHandler: | ((msg: string, reply: (response: string) => void, ws: WebSocket) => void) | null = null @@ -35,11 +55,12 @@ export class WebSocketTransport implements RpcTransport { // so ws.on('close') can notify the runtime which mobile client disconnected. private wsClientIds = new Map() - constructor({ host, port, tlsCert, tlsKey }: WebSocketTransportOptions) { + constructor({ host, port, tlsCert, tlsKey, heartbeatIntervalMs }: WebSocketTransportOptions) { this.host = host this.port = port this.tlsCert = tlsCert this.tlsKey = tlsKey + this.heartbeatIntervalMs = heartbeatIntervalMs ?? HEARTBEAT_INTERVAL_MS } onMessage( @@ -133,6 +154,49 @@ export class WebSocketTransport implements RpcTransport { this.httpServer = httpServer this.wss = wss + this.startHeartbeat() + } + + // Why: ping every live socket on a fixed cadence and terminate any that + // didn't pong since the previous tick. This is the only thing that + // reliably reaps half-open mobile sockets stranded by background + // suspension without a TCP FIN. See HEARTBEAT_INTERVAL_MS comment. + private startHeartbeat(): void { + if (this.heartbeatTimer) { + return + } + this.heartbeatTimer = setInterval(() => { + const wss = this.wss + if (!wss) { + return + } + for (const ws of wss.clients) { + if (!this.wsAlive.has(ws)) { + // Why: terminate() (vs close()) skips the close handshake and + // immediately fires the 'close' event, freeing the slot. close() + // on an already-dead socket can hang for the OS-level TCP timeout. + ws.terminate() + continue + } + this.wsAlive.delete(ws) + try { + ws.ping() + } catch { + // Why: ping() can throw on a socket that's mid-tear-down; the + // close handler will run regardless, so swallow the throw. + } + } + }, this.heartbeatIntervalMs) + if (typeof this.heartbeatTimer.unref === 'function') { + this.heartbeatTimer.unref() + } + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer) + this.heartbeatTimer = null + } } async stop(): Promise { @@ -140,6 +204,7 @@ export class WebSocketTransport implements RpcTransport { const httpServer = this.httpServer this.wss = null this.httpServer = null + this.stopHeartbeat() if (wss) { for (const client of wss.clients) { @@ -166,7 +231,21 @@ export class WebSocketTransport implements RpcTransport { // connection via the RPC `id` field. The transport delegates all auth // and dispatch logic to the message handler set by OrcaRuntimeRpcServer. private handleConnection(ws: WebSocket): void { + // Why: seed alive=true so the first heartbeat tick after connect doesn't + // treat a fresh socket as dead. Subsequent pongs (or any inbound traffic) + // re-arm it. + this.wsAlive.add(ws) + + ws.on('pong', () => { + this.wsAlive.add(ws) + }) + ws.on('message', (data) => { + // Why: any inbound traffic counts as proof of life, not just pongs. + // RN's WebSocket runtime auto-pongs server pings transparently, but + // app-level frames also count toward liveness so an actively-talking + // client doesn't get terminated mid-request. + this.wsAlive.add(ws) const msg = typeof data === 'string' ? data : data.toString('utf-8') this.messageHandler?.( msg,