From 4f7c423cd3de48171d6ada283670967a931aff54 Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Wed, 26 Aug 2026 22:46:29 +0000 Subject: [PATCH] fix(adapter-utils): check the http2 bridge wrapper's two read buffers as one shared total wrapDuplexChannelAsNodeDuplex checked the Node readable buffer and its own pending queue against the byte cap as two separate totals. A channel could hold up to twice the cap across the two buffers together. Replace both checks with one: retainedReadBytes() adds duplex.readableLength and the pending queue, and channel.onData tests that shared total against the cap before either the direct-push or the queue path runs. The two existing failure messages stay, chosen by whether the new chunk alone exceeds the cap. Co-authored-by: Paperclip --- .../src/http2-bridge-server.test.ts | 78 +++++++++++++++++++ .../adapter-utils/src/http2-bridge-server.ts | 67 +++++++++------- 2 files changed, 115 insertions(+), 30 deletions(-) diff --git a/packages/adapter-utils/src/http2-bridge-server.test.ts b/packages/adapter-utils/src/http2-bridge-server.test.ts index d7f2e32261c..3b30a5aed9c 100644 --- a/packages/adapter-utils/src/http2-bridge-server.test.ts +++ b/packages/adapter-utils/src/http2-bridge-server.test.ts @@ -1085,6 +1085,84 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { expect(duplex.destroyed).toBe(false); }); + it("the wrapper holds no more than the byte cap across the readable buffer and the pending queue", async () => { + let dataListener: ((chunk: Uint8Array) => void) | undefined; + let stopped = false; + const channel: CommandManagedDuplexChannel = { + write: () => undefined, + onData: (listener) => { + dataListener = listener; + }, + onExit: () => undefined, + stop: () => { + stopped = true; + }, + close: async () => undefined, + }; + const cap = 100_000; + const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes: cap }); + const errored = new Promise((resolve) => duplex.on("error", resolve)); + + // Fill the readable side: this chunk passes the default 65,536-byte high + // water mark, so `push()` reports the readable side full, and every later + // chunk queues in the wrapper's own pending queue instead. + dataListener?.(Buffer.alloc(70_000, "a")); + expect(duplex.readableLength).toBe(70_000); + + // Queue a chunk that brings the combined total to 99,000 of the + // 100,000-byte cap. The readable buffer and the pending queue are two + // separate buffers, so this proves their sum, not each one alone, stays + // under the cap. + dataListener?.(Buffer.alloc(29_000, "b")); + expect(duplex.destroyed).toBe(false); + expect(duplex.readableLength + 29_000).toBeLessThanOrEqual(cap); + + // One more chunk brings the combined total to 100,001 bytes, one byte + // past the cap: the wrapper must fail closed here, not let the sum grow + // past what the cap allows. + dataListener?.(Buffer.alloc(1_001, "c")); + const error = await errored; + expect(error.message).toMatch(/backpressure/i); + expect(stopped).toBe(true); + }); + + it("a chunk that fits alone but passes the cap with the readable buffer fails closed", async () => { + let dataListener: ((chunk: Uint8Array) => void) | undefined; + let stopped = false; + const channel: CommandManagedDuplexChannel = { + write: () => undefined, + onData: (listener) => { + dataListener = listener; + }, + onExit: () => undefined, + stop: () => { + stopped = true; + }, + close: async () => undefined, + }; + const cap = 80_000; + const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes: cap }); + const errored = new Promise((resolve) => duplex.on("error", resolve)); + + // This chunk is under the cap on its own, and it passes the default + // high-water mark, so it pushes directly and fills the readable side. + dataListener?.(Buffer.alloc(70_000, "a")); + expect(duplex.readableLength).toBe(70_000); + + // This chunk is also under the cap on its own (20,000 < 80,000), so a + // check that only bounds one chunk's own size, or only the queue's own + // total, would let it through. Combined with the 70,000 bytes the + // readable side already holds, the total is 90,000 bytes: over the cap. + // The wrapper must fail closed with the reader message, not the + // oversized-chunk message, because no single chunk here passes the cap + // on its own. + dataListener?.(Buffer.alloc(20_000, "b")); + const error = await errored; + expect(error.message).toMatch(/reader could not keep up/i); + expect(error.message).not.toMatch(/delivered one chunk larger/i); + expect(stopped).toBe(true); + }); + describe("the admission gate", () => { it("the response drain timeout defaults to thirty seconds, and the buffered-read byte cap shares one home with the admission module's session budget", () => { expect(DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS).toBe(30_000); diff --git a/packages/adapter-utils/src/http2-bridge-server.ts b/packages/adapter-utils/src/http2-bridge-server.ts index b3e83f9cea9..89bc3eb15dd 100644 --- a/packages/adapter-utils/src/http2-bridge-server.ts +++ b/packages/adapter-utils/src/http2-bridge-server.ts @@ -100,17 +100,17 @@ export const HTTP2_BRIDGE_SERVER_OPTIONS: http2.ServerOptions = { // Duplex channel adapter // --------------------------------------------------------------------------- -/** The default cap, in bytes, on the read-side queue {@link wrapDuplexChannelAsNodeDuplex} - * holds once `Duplex.push()` reports the readable side is full (a `false` - * return). A sandbox-controlled channel has no upstream pause: `onData` below - * keeps delivering bytes whether or not the HTTP/2 session keeps up with - * them. Past this cap the wrapper treats the channel as stuck, not merely - * slow, and fails closed: it stops the channel and destroys the `Duplex`, so - * a producer that keeps outpacing its reader cannot grow host memory without - * bound. This cap also bounds one single chunk: the wrapper checks a chunk's - * own size against it before `push()` ever runs, so one oversized chunk - * cannot cross the cap on its first delivery, before the queue holds - * anything to compare it against. +/** The default cap, in bytes, on the total {@link wrapDuplexChannelAsNodeDuplex} + * retains for one channel across both of its read buffers: the Node readable + * buffer (`Duplex.readableLength`) and the wrapper's own pending queue. A + * sandbox-controlled channel has no upstream pause: `onData` below keeps + * delivering bytes whether or not the HTTP/2 session keeps up with them. + * Past this cap the wrapper treats the channel as stuck, not merely slow, + * and fails closed: it stops the channel and destroys the `Duplex`, so a + * producer that keeps outpacing its reader cannot grow host memory without + * bound. One bound cap bounds the sum of both buffers, so one live session + * never retains more than this many bytes in total, not this many bytes in + * each buffer. * * This value reads {@link HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES} from the * admission module, so the wrapper cap and the admission gate's live-session @@ -145,11 +145,11 @@ export const DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS = 30_000; * wrapper honors that signal instead: while `push()` reports room, it pushes * directly; once `push()` reports the readable side is full, it queues each * later chunk instead of pushing past that signal, and drains the queue from - * `read()`, which Node calls again only once the consumer wants more. Every - * chunk, on either path, first checks against + * `read()`, which Node calls again only once the consumer wants more. Before + * either path runs, every chunk checks against * {@link DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES} (or the caller's - * `maxBufferedReadBytes`) on its own size, and the queue checks against the - * same cap on its cumulative size: past either check the wrapper fails + * `maxBufferedReadBytes`) as one bound on the total the Node readable buffer + * and the pending queue hold together: past that bound the wrapper fails * closed instead of buffering further, because the channel has no pause to * fall back on. A second, independent bound — * {@link DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS} (or the caller's @@ -208,6 +208,14 @@ export function wrapDuplexChannelAsNodeDuplex( duplex.destroy(new Error(message)); } + /** The total bytes this channel retains right now, across both read + * buffers: the Node readable buffer and the wrapper's own pending queue. + * `channel.onData` checks this total, not each buffer alone, against + * `maxBufferedReadBytes`, so the cap bounds the sum the channel can hold. */ + function retainedReadBytes(): number { + return duplex.readableLength + pendingReadBytes; + } + function endReadableIfDrained(): void { if (!channelExited || pendingReads.length > 0 || duplex.destroyed) return; duplex.push(null); @@ -264,14 +272,18 @@ export function wrapDuplexChannelAsNodeDuplex( channel.onData((chunk) => { if (duplex.destroyed) return; const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - // Check one chunk's own size against the cap before either path below - // runs. `push()` never refuses a call on its size, so a single chunk - // larger than the whole cap would otherwise reach Node's internal - // buffer unbounded on the direct-push path, before the queue this - // wrapper owns ever holds anything to compare a later chunk against. - if (bytes.byteLength > maxBufferedReadBytes) { + // Check this chunk against the cap as one bound on the total the Node + // readable buffer and the wrapper's own pending queue hold together, + // before either path below runs. `push()` never refuses a call on its + // size, so a single large chunk would otherwise reach Node's internal + // buffer unbounded on the direct-push path, and a check that bounds only + // one buffer at a time would let the two buffers together hold more + // than the cap allows. + if (retainedReadBytes() + bytes.byteLength > maxBufferedReadBytes) { failClosed( - "Sandbox HTTP/2 channel delivered one chunk larger than the bounded read backpressure buffer.", + bytes.byteLength > maxBufferedReadBytes + ? "Sandbox HTTP/2 channel delivered one chunk larger than the bounded read backpressure buffer." + : "Sandbox HTTP/2 channel exceeded the bounded read backpressure buffer; the reader could not keep up.", ); return; } @@ -281,8 +293,9 @@ export function wrapDuplexChannelAsNodeDuplex( } // The readable side already reported it is full, and the channel has no // pause to slow the sandbox side down: queue this chunk instead of - // pushing past that signal. Bound the queue, so a producer that keeps - // outpacing its reader cannot grow it without limit. + // pushing past that signal. The check above already bounds the queue as + // part of the shared total, so a producer that keeps outpacing its + // reader still cannot grow it without limit. if (pendingReads.length === 0) { // The queue was empty until this chunk: arm the stall bound, so a // consumer that never resumes reading still ends the channel, even @@ -290,12 +303,6 @@ export function wrapDuplexChannelAsNodeDuplex( armBackpressureStallTimer(); } pendingReadBytes += bytes.byteLength; - if (pendingReadBytes > maxBufferedReadBytes) { - failClosed( - "Sandbox HTTP/2 channel exceeded the bounded read backpressure buffer; the reader could not keep up.", - ); - return; - } pendingReads.push(bytes); }); channel.onExit(() => {