fix(mobile): WS heartbeat to reap half-open mobile sockets (#1500)

Co-authored-by: Orca <help@stably.ai>

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": {

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()
+  })
+})

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<typeof setTimeout> | null = null
+  let connectTimer: ReturnType<typeof setTimeout> | null = null
   let handshakeTimer: ReturnType<typeof setTimeout> | null = null
+  let activityProbeTimer: ReturnType<typeof setInterval> | 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

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)
 })

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<typeof setInterval> | 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<WebSocket>()
   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<WebSocket, string>()
 
-  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<void> {
@@ -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,