diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 4296be939..ac9065c05 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -187,6 +187,10 @@ - Fixed interrupted speech audio by ensuring segments queue and drain in order - Fixed speech vocalization starting only after the entire reply was synthesized: ONNX inference blocks the TTS worker's event loop, so per-segment IPC audio chunks queued unflushed and arrived in one burst. Streaming sends now drain the IPC channel before the next segment's inference, cutting time-to-first-audio to ~1.5s regardless of reply length. - Fixed an unhandled `EPIPE: broken pipe, write` rejection at the end of speech playback: the streaming player's `stop()` raced an un-awaited `FileSink.end()` against the backend SIGKILL, and mid-session writes never awaited the flush. Writes now await the flush (so a dead backend is detected and the chunk replays on the next candidate or the per-file path) and `stop()` swallows the expected teardown rejection. +### Fixed + +- Apply WebSocket send backpressure with ordered drain retry to prevent unbounded `bufferedAmount` growth ([#4248](https://github.com/can1357/oh-my-pi/issues/4248)) + ## [16.3.0] - 2026-07-02 diff --git a/packages/coding-agent/src/collab/relay-client.ts b/packages/coding-agent/src/collab/relay-client.ts index 130e92dec..b64e525bc 100644 --- a/packages/coding-agent/src/collab/relay-client.ts +++ b/packages/coding-agent/src/collab/relay-client.ts @@ -21,6 +21,9 @@ const BACKOFF_BASE_MS = 1_000; const BACKOFF_MAX_MS = 30_000; /** Max enveloped frames buffered while a reconnect is pending; overflow is dropped. */ const MAX_PENDING_SENDS = 256; +const WS_BACKPRESSURE_THRESHOLD = 64 * 1024; +const WS_BACKPRESSURE_DRAIN_THRESHOLD = 32 * 1024; +const WS_BACKPRESSURE_DRAIN_RETRY_MS = 25; export interface CollabSocketOptions { /** wss://host[:port]/r/ — no query string. */ @@ -40,6 +43,7 @@ export class CollabSocket { readonly #opts: CollabSocketOptions; #ws: WebSocket | null = null; #retryTimer: NodeJS.Timeout | undefined; + #backpressureDrainTimer: NodeJS.Timeout | undefined; #attempt = 0; /** Terminal state: intentional close or fatal failure. Cleared by connect(). */ #closed = false; @@ -72,28 +76,84 @@ export class CollabSocket { logger.debug("collab: dropping frame, socket closed", { t: frame.t }); return; } + const openWs = this.#ws; + if (openWs && openWs.readyState === WebSocket.OPEN) this.#drainPendingSends(openWs); const sealed = await seal(this.#opts.key, frame); const envelope = packEnvelope(targetPeer, sealed); const ws = this.#ws; if (ws && ws.readyState === WebSocket.OPEN) { + if (this.#pendingSends.length > 0) { + this.#enqueuePendingSend(envelope, frame.t); + if (ws.bufferedAmount < WS_BACKPRESSURE_DRAIN_THRESHOLD) { + this.#drainPendingSends(ws); + } else { + this.#scheduleBackpressureDrain(ws); + } + return; + } + if (ws.bufferedAmount >= WS_BACKPRESSURE_THRESHOLD) { + this.#enqueuePendingSend(envelope, frame.t); + this.#scheduleBackpressureDrain(ws); + return; + } ws.send(envelope); return; } - if (this.#pendingSends.length >= MAX_PENDING_SENDS) { - logger.debug("collab: dropping frame, reconnect buffer full", { t: frame.t }); - return; - } - this.#pendingSends.push(envelope); + this.#enqueuePendingSend(envelope, frame.t); }) .catch((err: unknown) => { logger.debug("collab: send failed", { error: String(err) }); }); } + #enqueuePendingSend(envelope: Uint8Array, frameType: CollabFrame["t"]): void { + if (this.#pendingSends.length >= MAX_PENDING_SENDS) { + logger.debug("collab: dropping frame, reconnect buffer full", { t: frameType }); + return; + } + this.#pendingSends.push(envelope); + } + + #drainPendingSends(ws: WebSocket): void { + while ( + this.#pendingSends.length > 0 && + ws.readyState === WebSocket.OPEN && + ws.bufferedAmount < WS_BACKPRESSURE_DRAIN_THRESHOLD + ) { + const envelope = this.#pendingSends.shift(); + if (!envelope) return; + ws.send(envelope); + } + } + + #scheduleBackpressureDrain(ws: WebSocket): void { + if (this.#backpressureDrainTimer !== undefined) return; + this.#backpressureDrainTimer = setTimeout(() => { + this.#backpressureDrainTimer = undefined; + this.#sendChain = this.#sendChain + .then(async () => { + if (this.#closed || this.#ws !== ws || ws.readyState !== WebSocket.OPEN) return; + this.#drainPendingSends(ws); + if (this.#pendingSends.length > 0) this.#scheduleBackpressureDrain(ws); + }) + .catch((err: unknown) => { + logger.debug("collab: backpressure drain failed", { error: String(err) }); + }); + }, WS_BACKPRESSURE_DRAIN_RETRY_MS); + } + + #clearBackpressureDrain(): void { + if (this.#backpressureDrainTimer !== undefined) { + clearTimeout(this.#backpressureDrainTimer); + this.#backpressureDrainTimer = undefined; + } + } + /** Intentional close: clears any retry timer, suppresses reconnect. A later connect() starts fresh. */ close(): void { const hadActivity = this.#ws !== null || this.#retryTimer !== undefined; this.#clearRetry(); + this.#clearBackpressureDrain(); const wasClosed = this.#closed; this.#closed = true; this.#pendingSends.length = 0; @@ -110,6 +170,7 @@ export class CollabSocket { } #openSocket(): void { + this.#clearBackpressureDrain(); const ws = new WebSocket(`${this.#opts.wsUrl}?role=${this.#opts.role}`); ws.binaryType = "arraybuffer"; this.#ws = ws; @@ -129,6 +190,7 @@ export class CollabSocket { }; ws.onclose = (event: CloseEvent) => { if (this.#ws !== ws) return; + this.#clearBackpressureDrain(); this.#ws = null; this.#handleClose(event.code, event.reason); }; @@ -167,6 +229,7 @@ export class CollabSocket { #handleClose(code: number, reason: string): void { if (this.#closed) return; + this.#clearBackpressureDrain(); const fatalReason = FATAL_CLOSE_REASONS[code]; if (fatalReason !== undefined) { this.#closed = true; @@ -186,6 +249,7 @@ export class CollabSocket { this.#pendingSends.length = 0; const ws = this.#ws; this.#ws = null; + this.#clearBackpressureDrain(); if (ws) { try { ws.close(1000);