diff --git a/src/main/runtime/rpc/ws-transport.test.ts b/src/main/runtime/rpc/ws-transport.test.ts index ca34340ccb1..a32cdcc7067 100644 --- a/src/main/runtime/rpc/ws-transport.test.ts +++ b/src/main/runtime/rpc/ws-transport.test.ts @@ -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 } }).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() diff --git a/src/main/runtime/rpc/ws-transport.ts b/src/main/runtime/rpc/ws-transport.ts index 0d5403c5991..ebc10037899 100644 --- a/src/main/runtime/rpc/ws-transport.ts +++ b/src/main/runtime/rpc/ws-transport.ts @@ -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 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()