mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 08:03:22 +00:00
fix(mobile): make the :6768 WebSocket accept loop leak-proof and self-healing
The desktop runtime's mobile WebSocket server on :6768 could wedge with the socket in LISTEN but the TCP accept loop stalled — new connections piling up in SYN_RCVD with zero ESTABLISHED, even from localhost, with no self-heal. Since Orca mobile has no APNs (the persistent WS doubles as the push channel), a wedged accept loop silently drops every notification while the app still shows cached UI. Root cause: connections were only capped at the WebSocket-upgrade layer (MAX_WS_CONNECTIONS), after the socket is already accepted. TCP-level sockets were unbounded, so leaked/half-open sockets (backgrounded phones, flaky relay, reconnect storms) could grow until the process ran out of file descriptors and accept() started failing with EMFILE. - Bound TCP sockets via httpServer.maxConnections (2x the WS cap). At the cap Node accepts-then-closes, so the accept loop always keeps draining and the EMFILE/SYN_RCVD wedge is structurally impossible on this listener. - Handle accept-level errors: ws forwards httpServer 'error' onto the WebSocketServer; with no listener Node rethrows it as an uncaught exception. Swallow + log and keep listening. - Force-terminate over-capacity sockets: a bare ws.close(1013) left half-open phones lingering at >cap forever, rejecting everyone after them. Add an error handler + a 1s terminate fallback so the descriptor is always freed. - Log heartbeat reaping / near-cap live counts so the leak is observable.
This commit is contained in:
@@ -327,6 +327,75 @@ describe('WebSocketTransport', () => {
|
||||
liveClient.close()
|
||||
})
|
||||
|
||||
it('bounds TCP-level sockets so a leak cannot wedge the accept loop', async () => {
|
||||
// Why: the :6768 accept-wedge — unbounded sockets exhausted the FD budget,
|
||||
// accept() started failing with EMFILE, and new SYNs stalled in SYN_RCVD
|
||||
// with zero ESTABLISHED and no self-heal. A TCP-level maxConnections makes
|
||||
// Node accept-then-close past the cap so the listener never stops draining.
|
||||
const { transport } = await createTransport()
|
||||
await transport.start()
|
||||
|
||||
const httpServer = (transport as unknown as { httpServer: { maxConnections: number } })
|
||||
.httpServer
|
||||
expect(httpServer.maxConnections).toBeGreaterThan(128)
|
||||
expect(Number.isFinite(httpServer.maxConnections)).toBe(true)
|
||||
})
|
||||
|
||||
it('keeps listening after an accept-level error and still accepts new clients', async () => {
|
||||
// Why: an EMFILE/ENFILE from accept() surfaces as an 'error' on the http
|
||||
// server, which `ws` forwards onto the WebSocketServer. Without a listener
|
||||
// Node rethrows it as an uncaught exception that can silently kill the
|
||||
// transport. The handler must swallow it and leave the listener able to
|
||||
// accept the next connection.
|
||||
const { transport } = await createTransport((msg, reply) => {
|
||||
const request = JSON.parse(msg)
|
||||
reply(JSON.stringify({ id: request.id, ok: true }))
|
||||
})
|
||||
await transport.start()
|
||||
|
||||
const httpServer = (
|
||||
transport as unknown as { httpServer: { emit: (event: string, err: Error) => void } }
|
||||
).httpServer
|
||||
// Why: emit on the http server (not wss) to exercise the real forwarding
|
||||
// path — libuv raises accept errors on the underlying net server.
|
||||
expect(() =>
|
||||
httpServer.emit('error', Object.assign(new Error('accept EMFILE'), { code: 'EMFILE' }))
|
||||
).not.toThrow()
|
||||
|
||||
const ws = await connectWs(transport)
|
||||
const response = await sendAndReceive(ws, JSON.stringify({ id: 'after-error', method: 'test' }))
|
||||
expect(JSON.parse(response)).toMatchObject({ id: 'after-error', ok: true })
|
||||
ws.close()
|
||||
})
|
||||
|
||||
it('force-terminates an over-capacity socket that ignores the close frame', async () => {
|
||||
// Why: a backgrounded/half-open phone may never ack the 1013 close, so a
|
||||
// bare ws.close() would leave the socket in wss.clients at >cap forever and
|
||||
// permanently reject everyone after it. rejectOverCapacity must hard-close.
|
||||
const { transport } = await createTransport()
|
||||
await transport.start()
|
||||
|
||||
const ws = await connectWs(transport)
|
||||
const wss = (transport as unknown as { wss: { clients: Set<WebSocket> } }).wss
|
||||
const serverSocket = Array.from(wss.clients)[0]
|
||||
expect(serverSocket).toBeDefined()
|
||||
const terminateSpy = vi.spyOn(serverSocket!, 'terminate')
|
||||
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
;(transport as unknown as { rejectOverCapacity(ws: WebSocket): void }).rejectOverCapacity(
|
||||
serverSocket!
|
||||
)
|
||||
vi.advanceTimersByTime(1_000)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
|
||||
expect(terminateSpy).toHaveBeenCalled()
|
||||
terminateSpy.mockRestore()
|
||||
ws.close()
|
||||
})
|
||||
|
||||
it('is idempotent on double start', async () => {
|
||||
const { transport } = await createTransport()
|
||||
|
||||
|
||||
@@ -14,6 +14,15 @@ const MAX_WS_MESSAGE_BYTES = 1024 * 1024
|
||||
// streams (session tabs, terminals, file watches, browser streams). Keep the
|
||||
// cap high enough that leaked/stale streams do not starve short control RPCs.
|
||||
const MAX_WS_CONNECTIONS = 128
|
||||
// Why: hard bound on TCP-level sockets, set ABOVE the WS-upgrade cap so it only
|
||||
// bites under a genuine leak/abuse. When the http server is at this cap Node
|
||||
// accepts-then-immediately-closes each new socket, so the accept loop KEEPS
|
||||
// DRAINING — the listener can never stall with SYNs stuck in SYN_RCVD the way
|
||||
// an unbounded server does once leaked/half-open sockets exhaust the process
|
||||
// file-descriptor budget (EMFILE). The WS-upgrade cap alone runs only after a
|
||||
// socket is already accepted, so it cannot prevent that FD wedge. See the
|
||||
// :6768 accept-wedge post-mortem.
|
||||
const MAX_TCP_CONNECTIONS = MAX_WS_CONNECTIONS * 2
|
||||
const PRE_AUTH_TIMEOUT_MS = 10_000
|
||||
type WebSocketMessagePayload = string | Uint8Array<ArrayBufferLike>
|
||||
type WebSocketMessageHandler = {
|
||||
@@ -224,14 +233,28 @@ export class WebSocketTransport implements RpcTransport {
|
||||
})
|
||||
})
|
||||
|
||||
// Why: bound TCP-level sockets so a leak/flood cannot exhaust the process
|
||||
// FD budget and stall accept() with EMFILE (the :6768 accept-wedge: SYNs
|
||||
// pile up in SYN_RCVD, zero ESTABLISHED, even localhost times out, no
|
||||
// self-heal). At this cap Node accepts-then-closes new sockets, keeping the
|
||||
// accept loop draining instead of silently wedging.
|
||||
httpServer.maxConnections = MAX_TCP_CONNECTIONS
|
||||
|
||||
const wss = new WebSocketServer({
|
||||
server: httpServer,
|
||||
maxPayload: MAX_WS_MESSAGE_BYTES
|
||||
})
|
||||
|
||||
// Why: an accept()-level failure (EMFILE/ENFILE) is emitted as 'error' on
|
||||
// the http server, which `ws` forwards onto the WebSocketServer. With no
|
||||
// listener Node rethrows it as an uncaught exception that can silently tear
|
||||
// the transport down. Log and keep listening so the accept path is
|
||||
// observable and recovers as descriptors free up.
|
||||
wss.on('error', (error) => console.warn('[ws-transport] listener error on accept path:', error))
|
||||
|
||||
wss.on('connection', (ws) => {
|
||||
if (wss.clients.size > MAX_WS_CONNECTIONS) {
|
||||
ws.close(1013, 'Maximum connections reached')
|
||||
this.rejectOverCapacity(ws)
|
||||
return
|
||||
}
|
||||
this.handleConnection(ws)
|
||||
@@ -242,6 +265,22 @@ export class WebSocketTransport implements RpcTransport {
|
||||
this.startHeartbeat()
|
||||
}
|
||||
|
||||
// Why: over the WS-upgrade cap, request a graceful 1013 close but force the
|
||||
// socket down shortly after. A backgrounded/half-open phone may never ack the
|
||||
// close frame, so a bare ws.close() would leave the socket in wss.clients (and
|
||||
// holding a descriptor) at >cap forever, permanently rejecting every client
|
||||
// after it. The 'error' listener keeps an ECONNRESET-while-closing from
|
||||
// surfacing as an unhandled 'error' event (which would throw).
|
||||
private rejectOverCapacity(ws: WebSocket): void {
|
||||
ws.on('error', () => {})
|
||||
ws.close(1013, 'Maximum connections reached')
|
||||
// Why: give the 1013 close a brief window, then hard-terminate so the
|
||||
// descriptor is freed even if the client never acks the close frame.
|
||||
const terminateTimer = setTimeout(() => ws.terminate(), 1_000)
|
||||
terminateTimer.unref?.()
|
||||
ws.once('close', () => clearTimeout(terminateTimer))
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -255,12 +294,14 @@ export class WebSocketTransport implements RpcTransport {
|
||||
if (!wss) {
|
||||
return
|
||||
}
|
||||
let reaped = 0
|
||||
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()
|
||||
reaped++
|
||||
continue
|
||||
}
|
||||
this.wsAlive.delete(ws)
|
||||
@@ -271,6 +312,12 @@ export class WebSocketTransport implements RpcTransport {
|
||||
// close handler will run regardless, so swallow the throw.
|
||||
}
|
||||
}
|
||||
// Why: make the accept path observable. Steady reaping or a client count
|
||||
// riding the cap are the early signs of the half-open-socket leak that
|
||||
// used to wedge :6768 silently — surface them instead of failing dark.
|
||||
if (reaped > 0 || wss.clients.size >= MAX_WS_CONNECTIONS) {
|
||||
console.warn(`[ws-transport] heartbeat reaped ${reaped}; ${wss.clients.size} live sockets`)
|
||||
}
|
||||
}, this.heartbeatIntervalMs)
|
||||
if (typeof this.heartbeatTimer.unref === 'function') {
|
||||
this.heartbeatTimer.unref()
|
||||
|
||||
Reference in New Issue
Block a user