fix(browser): bound CDP output for stalled clients (#20949)

* fix(browser): bound CDP output for stalled clients

* fix(browser): log CDP outbound overflow before terminating the client

The outbound queue terminated the automation client silently on overflow, so
the client saw a socket close indistinguishable from a crash. Surface the cap
that tripped and the backlog held when it did.

The queue dropped its backlog before invoking onOverflow, so the counters were
already zero at the callback. Snapshot them first and pass them through.

---------

Co-authored-by: m4air <m4air@Mac.localdomain>
Co-authored-by: Neil <neil@stably.ai>

src/main/browser/cdp-client-response-writer.test.ts

@@ -0,0 +1,123 @@
+import { once } from 'node:events'
+import { setImmediate as nextTurn } from 'node:timers/promises'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { WebSocket, WebSocketServer } from 'ws'
+import { CdpClientResponseWriter } from './cdp-client-response-writer'
+
+const MiB = 1024 * 1024
+
+describe('CDP outbound backpressure', () => {
+  let server: WebSocketServer
+  let peer: WebSocket
+  let socket: WebSocket
+  let writer: CdpClientResponseWriter
+
+  beforeEach(async () => {
+    server = new WebSocketServer({ host: '127.0.0.1', port: 0 })
+    await once(server, 'listening')
+    const address = server.address()
+    if (!address || typeof address === 'string') {
+      throw new Error('Expected TCP address')
+    }
+    const accepted = new Promise<WebSocket>((resolve) => server.once('connection', resolve))
+    peer = new WebSocket(`ws://127.0.0.1:${address.port}`)
+    await once(peer, 'open')
+    socket = await accepted
+    writer = new CdpClientResponseWriter(() => socket)
+  })
+
+  afterEach(async () => {
+    writer.forgetClient(socket)
+    vi.useRealTimers()
+    vi.restoreAllMocks()
+    socket.terminate()
+    peer.terminate()
+    await new Promise<void>((resolve) => server.close(() => resolve()))
+  })
+
+  it('drains replies and events in order with their original session correlation', () => {
+    vi.useFakeTimers()
+    const buffered = vi.spyOn(socket, 'bufferedAmount', 'get').mockReturnValue(9 * MiB)
+    const send = vi.spyOn(socket, 'send').mockImplementation(() => {})
+    writer.recordRequestSessionId(socket, 7, { sessionId: 'session-one' })
+    writer.sendResult(7, { ok: true })
+    writer.send({ method: 'Page.loadEventFired', sessionId: 'session-one' })
+    writer.recordRequestSessionId(socket, 8, { sessionId: 'session-two' })
+    writer.sendError(8, 'failed')
+    expect(send).not.toHaveBeenCalled()
+
+    buffered.mockReturnValue(0)
+    vi.advanceTimersByTime(25)
+    expect(send.mock.calls.map(([frame]) => JSON.parse(String(frame)))).toEqual([
+      { id: 7, result: { ok: true }, sessionId: 'session-one' },
+      { method: 'Page.loadEventFired', sessionId: 'session-one' },
+      { id: 8, error: { code: -32000, message: 'failed' }, sessionId: 'session-two' }
+    ])
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it.each(['forget', 'close'] as const)('releases a parked queue on %s', (boundary) => {
+    vi.useFakeTimers()
+    const initialCloseListeners = socket.listenerCount('close')
+    const buffered = vi.spyOn(socket, 'bufferedAmount', 'get').mockReturnValue(9 * MiB)
+    const send = vi.spyOn(socket, 'send').mockImplementation(() => {})
+    writer.sendResult(1, { value: 'abandoned' })
+    expect(vi.getTimerCount()).toBe(1)
+    if (boundary === 'forget') {
+      writer.forgetClient(socket)
+    } else {
+      socket.emit('close', 1000, Buffer.alloc(0))
+    }
+    buffered.mockReturnValue(0)
+    vi.advanceTimersByTime(100)
+    expect(send).not.toHaveBeenCalled()
+    expect(vi.getTimerCount()).toBe(0)
+    expect(socket.listenerCount('close')).toBeLessThanOrEqual(initialCloseListeners)
+  })
+
+  it('preserves a single large reply on a clear connection', () => {
+    const send = vi.spyOn(socket, 'send').mockImplementation(() => {})
+    writer.sendResult(1, { data: 'x'.repeat(65 * MiB) })
+    expect(send).toHaveBeenCalledOnce()
+    expect(socket.readyState).toBe(WebSocket.OPEN)
+  })
+
+  it('terminates a real stalled reader before native buffering follows all produced bytes', async () => {
+    peer.pause()
+    const payload = 'x'.repeat(64 * 1024)
+    let peakBuffered = 0
+    for (let index = 0; index < 2048; index++) {
+      writer.send({ method: 'Runtime.consoleAPICalled', params: { data: payload } })
+      peakBuffered = Math.max(peakBuffered, socket.bufferedAmount)
+      if (index % 16 === 0) {
+        await nextTurn()
+      }
+      if (socket.readyState !== WebSocket.OPEN) {
+        break
+      }
+    }
+    expect(peakBuffered).toBeLessThan(9 * MiB)
+    expect(socket.readyState).not.toBe(WebSocket.OPEN)
+  })
+
+  it('delivers a transient burst completely to a reading client', async () => {
+    const frames: string[] = []
+    const allReceived = new Promise<void>((resolve) => {
+      peer.on('message', (frame) => {
+        frames.push(frame.toString())
+        if (frames.length === 192) {
+          resolve()
+        }
+      })
+    })
+    const payload = 'y'.repeat(64 * 1024)
+    for (let index = 0; index < 192; index++) {
+      writer.sendResult(index, { data: payload })
+    }
+    await allReceived
+    expect(frames).toEqual(
+      Array.from({ length: 192 }, (_, id) => JSON.stringify({ id, result: { data: payload } }))
+    )
+    expect(socket.readyState).toBe(WebSocket.OPEN)
+  })
+})

src/main/browser/cdp-client-response-writer.ts

@@ -1,4 +1,8 @@
 import { WebSocket } from 'ws'
+import {
+  createWsOutboundBackpressureQueue,
+  type WsOutboundBackpressureQueue
+} from '../../shared/ws-outbound-backpressure-queue'
 
 /**
  * Serializes CDP replies to the connected websocket client and echoes the
@@ -6,16 +10,51 @@ import { WebSocket } from 'ws'
  */
 export class CdpClientResponseWriter {
   private readonly responseSessionIdsByClient = new WeakMap<WebSocket, Map<number, string>>()
+  private readonly outboundByClient = new WeakMap<
+    WebSocket,
+    { queue: WsOutboundBackpressureQueue<string>; onClose: () => void }
+  >()
 
   constructor(private readonly getClient: () => WebSocket | null) {}
 
   send(payload: unknown, client = this.getClient()): void {
     const responsePayload = client ? this.addResponseSessionId(payload, client) : payload
     if (client?.readyState === WebSocket.OPEN) {
-      client.send(JSON.stringify(responsePayload))
+      this.outboundQueue(client).enqueue(JSON.stringify(responsePayload))
     }
   }
 
+  private outboundQueue(client: WebSocket): WsOutboundBackpressureQueue<string> {
+    const existing = this.outboundByClient.get(client)
+    if (existing) {
+      return existing.queue
+    }
+    const queue = createWsOutboundBackpressureQueue<string>({
+      send: (frame) => client.send(frame),
+      byteLengthOf: (frame) => Buffer.byteLength(frame),
+      getBufferedAmount: () => client.bufferedAmount,
+      isWritable: () => client.readyState === WebSocket.OPEN,
+      onOverflow: (evidence) => {
+        // The client only sees an abrupt socket close, so name the cap here.
+        console.warn('[cdp] outbound queue overflow; terminating automation client:', {
+          cap: evidence.cap,
+          queuedBytes: evidence.queuedBytes,
+          queuedFrames: evidence.queuedFrames,
+          maxQueuedBytes: evidence.maxQueuedBytes,
+          maxQueuedFrames: evidence.maxQueuedFrames
+        })
+        client.terminate()
+      },
+      // Preserve large PDF/screenshot replies on a draining connection; queued bursts stay capped.
+      maxFrameBytes: Number.POSITIVE_INFINITY,
+      maxDrainFramesPerTurn: 128
+    })
+    const onClose = (): void => this.forgetClient(client)
+    this.outboundByClient.set(client, { queue, onClose })
+    client.once('close', onClose)
+    return queue
+  }
+
   private addResponseSessionId(payload: unknown, client: WebSocket): unknown {
     if (typeof payload !== 'object' || payload === null) {
       return payload
@@ -54,5 +93,11 @@ export class CdpClientResponseWriter {
 
   forgetClient(client: WebSocket): void {
     this.responseSessionIdsByClient.delete(client)
+    const outbound = this.outboundByClient.get(client)
+    if (outbound) {
+      this.outboundByClient.delete(client)
+      client.off('close', outbound.onClose)
+      outbound.queue.dispose()
+    }
   }
 }

src/main/browser/cdp-debugger-channel.ts

@@ -53,7 +53,7 @@ export class CdpDebuggerChannel {
       // agent-browser filters events by the sessionId from Target.attachToTarget.
       const msg: Record<string, unknown> = { method, params }
       msg.sessionId = sessionId || this.sessions.primarySessionId
-      client.send(JSON.stringify(msg))
+      this.responder.send(msg, client)
     }
     this.debuggerDetachHandler = () => {
       this.attached = false

src/shared/ws-outbound-backpressure-queue.ts

@@ -10,6 +10,16 @@
 // Generic over the frame type so it serves both the text reply path (encrypted
 // base64 strings) and the binary send path (Uint8Array frames).
 
+/** Which hard bound tripped, plus the backlog held when it did. */
+export type WsOutboundOverflowEvidence = {
+  cap: 'maxQueuedBytes' | 'maxQueuedFrames' | 'maxFrameBytes' | 'claimQueuedBytes' | 'sendFailed'
+  queuedBytes: number
+  queuedFrames: number
+  maxQueuedBytes: number
+  maxQueuedFrames: number
+  maxFrameBytes: number
+}
+
 export type WsOutboundBackpressureQueueOptions<TFrame> = {
   /** Send a frame on the wire. Called only when under the soft cap. */
   send: (frame: TFrame) => void
@@ -25,8 +35,9 @@ export type WsOutboundBackpressureQueueOptions<TFrame> = {
    * Called once when queued bytes exceed maxQueuedBytes — the link is wedged.
    * The caller should tear the connection down so a fresh subscription can
    * replay an authoritative snapshot. The queue drops its backlog afterward.
+   * Receives the backlog measured before that drop, so callers can report it.
    */
-  onOverflow: () => void
+  onOverflow: (evidence: WsOutboundOverflowEvidence) => void
   /** Soft cap: stop draining onto the wire while bufferedAmount is above this. */
   softCapBytes?: number
   /** Hard cap on bytes held in this queue before onOverflow fires. */
@@ -135,13 +146,22 @@ export function createWsOutboundBackpressureQueue<TFrame>(
     stopTimer()
   }
 
-  const failOverflow = (): void => {
+  const failOverflow = (cap: WsOutboundOverflowEvidence['cap']): void => {
     if (disposed || overflowed) {
       return
     }
     overflowed = true
+    // Snapshot before dropBacklog() zeroes the counters.
+    const evidence: WsOutboundOverflowEvidence = {
+      cap,
+      queuedBytes: queued,
+      queuedFrames,
+      maxQueuedBytes,
+      maxQueuedFrames,
+      maxFrameBytes
+    }
     dropBacklog()
-    options.onOverflow()
+    options.onOverflow(evidence)
   }
 
   const sendFrame = (frame: TFrame): boolean => {
@@ -149,7 +169,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
       options.send(frame)
       return true
     } catch {
-      failOverflow()
+      failOverflow('sendFailed')
       return false
     }
   }
@@ -243,7 +263,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
     }
     const bytes = options.byteLengthOf(frame)
     if (!Number.isFinite(bytes) || bytes < 0 || bytes > maxFrameBytes) {
-      failOverflow()
+      failOverflow('maxFrameBytes')
       return { accepted: false, queued: false, cancel: () => false }
     }
     // Fast path: nothing parked and the wire is under the cap — send directly.
@@ -261,7 +281,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
     }
     const queuedBytesClaim = options.claimQueuedBytes?.(bytes)
     if (options.claimQueuedBytes && !queuedBytesClaim) {
-      failOverflow()
+      failOverflow('claimQueuedBytes')
       return { accepted: false, queued: false, cancel: () => false }
     }
     const entry: QueueEntry = {
@@ -274,7 +294,7 @@ export function createWsOutboundBackpressureQueue<TFrame>(
     queued += bytes
     queuedFrames += 1
     if (queued > maxQueuedBytes || queuedFrames > maxQueuedFrames) {
-      failOverflow()
+      failOverflow(queued > maxQueuedBytes ? 'maxQueuedBytes' : 'maxQueuedFrames')
       return { accepted: false, queued: false, cancel: () => false }
     }
     if (timer === null) {