From 1646bb072bae9c6c94e50045beecdd883a7e7249 Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Sun, 23 Aug 2026 22:33:44 +0000 Subject: [PATCH] fix(adapter-utils): end a stalled duplex stdout write on a bounded stall timeout A backpressured stdout write in the duplex gateway waited for drain or a stream error, with no other end condition. A pipe that stays open but stops draining -- for example a reader that stops consuming without closing the pipe -- never fires either event, so the wait, the send, and the admission reserve for that read stayed open forever. Later requests then answered 503 bridge_busy for good. Add a stall-timeout end condition to the embedded stdout frame writer. It reuses the existing response timeout as its bound: a write that stays backpressured that long already means no response can arrive in time, so ending the wait costs nothing beyond what the response side already gives up on. Once the bound elapses, the writer treats the wait exactly like a stream error -- it strands every queued line and refuses further writes -- so the same finally block that already releases the reserve on a stream error now also runs. Add a unit test that proves a stalled write past its bound rejects and strands the queue, and a companion test that proves a write which drains before the bound clears its timer and settles normally. Add an end-to-end test that pauses the real child's stdout pipe and never resumes or destroys it, proving a later request is admitted again once the stall bound elapses. Confirmed each new test fails against the prior commit and passes after this change. Co-authored-by: Paperclip --- .../src/duplex-bridge-local-e2e.test.ts | 74 ++++++++++++ .../src/execution-target-sandbox.test.ts | 85 ++++++++++++++ .../src/sandbox-callback-bridge.ts | 107 ++++++++++++------ 3 files changed, 234 insertions(+), 32 deletions(-) diff --git a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts index 203d8a9e8fe..2a3cfdf5874 100644 --- a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts +++ b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts @@ -1462,4 +1462,78 @@ describe("duplex bridge local end-to-end harness", () => { }); expect(later.status).toBe(502); }, 20000); + + it("ends a stalled send on its own stall bound when the outbound pipe never drains and never breaks, and releases the reserve for a later request", async () => { + // Regression test for the stalled-write finding. The prior two tests + // release the reserve because the pipe eventually drains, or because it + // breaks. This proves the reserve also releases when neither happens: + // the reader on the other end just stops consuming, and the pipe itself + // stays open. Never resume or destroy the child's stdout in this test -- + // a release here can only come from the writer's own stall bound. + const maxBodyBytes = 2 * 1024 * 1024; + const responseTimeoutMs = 900; + const harness = await createHarness({ + maxBodyBytes, + // Admits exactly one concurrent body read at maxBodyBytes: the declared + // body size plus the fixed in-flight frame allowance. + maxInflightBodyBytes: maxBodyBytes + DEFAULT_MAX_DUPLEX_FRAME_BYTES, + responseTimeoutMs, + }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + harness.setForwardMode("hang"); + + // Stop reading the child's stdout, so the first request's send stalls + // mid-flight. Nothing in this test ever resumes or destroys it. + harness.child.stdout.pause(); + + const body = "a".repeat(maxBodyBytes); + const first = fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + void first.catch(() => undefined); + + // Give the gateway time to read the body and drive its send into + // backpressure. + await new Promise((resolve) => setTimeout(resolve, 400)); + + // Confirm the reserve is held while the send is stalled: a second + // request cannot reserve, so the gateway sheds it with a clean 503. + const shedWhileStalled = await fetch(`${harness.baseUrl}/api/issues/issue-2/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(shedWhileStalled.status).toBe(503); + + const firstResponse = await first; + // No response frame can ever arrive: the pipe never drains, so the + // gateway's own response timeout ends the request with a local 502. + expect(firstResponse.status).toBe(502); + + // Give the writer's stall bound (== responseTimeoutMs) room to fire and + // release the reserve, well past the point where it must have struck. + await new Promise((resolve) => setTimeout(resolve, responseTimeoutMs + 300)); + + // The reserve released once the stalled send ended on its own stall + // bound and the timed-out response both finished. A later request is + // admitted again, even though the pipe still never drained. + const later = await fetch(`${harness.baseUrl}/api/issues/issue-3/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(later.status).toBe(502); + }, 20000); }); diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 6ed28a1d8ab..437c14f07a6 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -5784,6 +5784,7 @@ describe("sandbox duplex gateway", () => { createStdoutFrameWriter: ( stream: FakeOutboundStream, onBackpressure?: (writableLength: number) => void, + stallTimeoutMs?: number, ) => (line: string) => Promise; } @@ -5885,6 +5886,90 @@ describe("sandbox duplex gateway", () => { await expect(third).rejects.toBe(boom); }); + it("ends a stalled outbound write on its own stall bound when the pipe never drains and never errors", async () => { + // Regression test for the stalled-write finding. A backpressured write + // that meets a pipe which stays open forever -- no drain, no "error" + // event -- used to wait without end. That left the write's caller + // waiting forever too, so an admission reserve tied to that write's + // completion never released. Pass a stall bound, and prove a wait that + // outlasts it ends the same way a stream error does: the queued write + // and every line still queued reject, and the writer never issues + // another write afterward. + vi.useFakeTimers(); + try { + const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter(); + const stream = new FakeOutboundStream({ accepting: false }); + const writeLine = createStdoutFrameWriter(stream, undefined, 1000); + + const first = writeLine("frame-1\n"); + const second = writeLine("frame-2\n"); + // Attach a throwaway handler to each promise right away. Both reject + // once the bound below elapses; a bare `.rejects` assertion later + // still sees the same rejection, but this stops the interpreter from + // reporting an unhandled rejection in the window before that assertion + // runs. + let firstSettled = false; + void first.catch(() => undefined).finally(() => { + firstSettled = true; + }); + void second.catch(() => undefined); + await vi.advanceTimersByTimeAsync(0); + expect(stream.writes).toEqual(["frame-1\n"]); + + // The pipe never drains and never errors. Advance to just before the + // bound: the writer is still waiting, nothing has settled. + await vi.advanceTimersByTimeAsync(999); + expect(firstSettled).toBe(false); + + // Cross the bound. The stall timer fires, ends the wait, and strands + // both queued lines. + await vi.advanceTimersByTimeAsync(1); + await expect(first).rejects.toThrow(/did not drain within 1000ms/); + await expect(second).rejects.toThrow(/did not drain within 1000ms/); + + // The writer treats the stall exactly like a stream error: it never + // issues another write once one has struck. + const third = writeLine("frame-3\n"); + await expect(third).rejects.toThrow(/did not drain within 1000ms/); + expect(stream.writes).toEqual(["frame-1\n"]); + } finally { + vi.useRealTimers(); + } + }); + + it("does not start a stall timer once the pipe drains before the bound elapses", async () => { + // The stall bound only ever fires against a write still waiting when it + // elapses. A write that drains first must settle normally and must leave + // no stray timer that could later reject an unrelated later write. + vi.useFakeTimers(); + try { + const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter(); + const stream = new FakeOutboundStream({ accepting: false }); + const writeLine = createStdoutFrameWriter(stream, undefined, 1000); + + const first = writeLine("frame-1\n"); + await vi.advanceTimersByTimeAsync(0); + expect(stream.writes).toEqual(["frame-1\n"]); + + // Drain well before the bound elapses. The write settles normally. + stream.drain(); + await first; + + // Advance well past the bound the drained write would have hit. No + // stray timer fires and rejects an unrelated later write, because the + // writer cleared frame-1's timer the moment it drained. + await vi.advanceTimersByTimeAsync(5000); + + const second = writeLine("frame-2\n"); + await vi.advanceTimersByTimeAsync(0); + stream.drain(); + await second; + expect(stream.writes).toEqual(["frame-1\n", "frame-2\n"]); + } finally { + vi.useRealTimers(); + } + }); + it("returns the same HTTP response as the file gateway for a forwarded request", async () => { const token = "duplex-token-forward"; const gateway = await startDuplexGateway({ PAPERCLIP_BRIDGE_TOKEN: token }); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index f96ac7da43e..bf5be6eb501 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -2023,17 +2023,22 @@ class DuplexFrameDecoder { // body_chunk frame. The writer issues at most one `stream.write()` call at a // time and waits for the stream to drain before it starts the next queued // line, so a slow reader never lets the gateway queue a whole request's -// frames at once. A stream "error" event ends an in-flight wait the same way -// "drain" does, and it fails every line still queued, because a broken pipe -// can never drain. -const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(stream, onBackpressure) { +// frames at once. Three conditions end an in-flight wait, all the same way: +// the stream drains, the stream emits an "error" event, or the wait outlasts +// `stallTimeoutMs`. The last one covers a pipe that stays open but never +// drains -- for example a reader that stopped consuming without closing the +// pipe -- which the "error" event never fires for on its own. Each of the +// last two fails every line still queued, because neither a broken pipe nor +// one stalled past its bound can ever be trusted to drain. +const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(stream, onBackpressure, stallTimeoutMs) { const queue = []; let pumping = false; let streamError = null; - // Reject every entry still queued and empty the queue. Both the "error" - // listener below and the top-of-loop check inside pump call this, so a - // stream error always settles every entry that was queued when it struck, + // Reject every entry still queued and empty the queue. The "error" + // listener below, the stall timer inside pump's drain wait, and the + // top-of-loop check inside pump all call this, so a stream error or a + // stalled drain always settles every entry that was queued when it struck, // not only the one pump happened to be holding. function strandQueue(error) { const stranded = queue.splice(0, queue.length); @@ -2067,26 +2072,49 @@ const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(st onBackpressure(stream.writableLength); } await new Promise((resolve) => { + let stallTimer = null; + const clearStallTimer = () => { + if (stallTimer !== null) clearTimeout(stallTimer); + }; const onDrain = () => { + clearStallTimer(); stream.off("error", onStreamErrorWhileWaiting); resolve(); }; const onStreamErrorWhileWaiting = () => { + clearStallTimer(); stream.off("drain", onDrain); resolve(); }; stream.once("drain", onDrain); stream.once("error", onStreamErrorWhileWaiting); + if (typeof stallTimeoutMs === "number" && stallTimeoutMs > 0) { + // The pipe accepted the write but stayed above its high-water + // mark past this bound with no drain and no error. Treat that + // exactly like a stream error: a wait this long already means + // the reader on the other end is not consuming, so holding the + // admission reserve for this read any longer serves no request + // that can still complete. + stallTimer = setTimeout(() => { + stream.off("drain", onDrain); + stream.off("error", onStreamErrorWhileWaiting); + streamError = new Error( + "stdout write did not drain within " + stallTimeoutMs + "ms", + ); + strandQueue(streamError); + resolve(); + }, stallTimeoutMs); + } }); } if (streamError) { - // The "error" listener already stranded and rejected every entry - // that was in the queue when the stream failed, including this one - // if it was still queued at that moment. Do not shift here: the - // queue head may now be a line a caller pushed after the stream - // failed but before this wait resumed, and shifting it here would - // discard that entry without ever settling its promise. The - // top-of-loop check above strands it on the next turn instead. + // The "error" listener or the stall timer above already stranded and + // rejected every entry that was in the queue when it struck, + // including this one if it was still queued at that moment. Do not + // shift here: the queue head may now be a line a caller pushed after + // the stream failed but before this wait resumed, and shifting it + // here would discard that entry without ever settling its promise. + // The top-of-loop check above strands it on the next turn instead. continue; } queue.shift(); @@ -2245,9 +2273,9 @@ export function getSandboxDuplexGatewayCodecSource(): string { * Return the exact zero-dependency outbound-writer source the generated duplex * gateway embeds. A test wraps this source with a fake stream, so it proves * the writer serializes every write, waits for drain, and ends a stalled write - * on a stream error, all without a real child process. The source declares - * `createStdoutFrameWriter` but exports it from no module; a caller wraps it - * to read the name. + * on a stream error or on a bounded stall timeout, all without a real child + * process. The source declares `createStdoutFrameWriter` but exports it from + * no module; a caller wraps it to read the name. */ export function getSandboxDuplexGatewayStdoutWriterSource(): string { return DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE; @@ -2784,10 +2812,19 @@ function runDuplexGateway() { // The one serialized, backpressure-aware writer for this process. Every // outbound frame -- the heartbeat, the READY frame, one request envelope, - // and each body_chunk -- writes through it, in call order. - const writeStdoutFrameLine = createStdoutFrameWriter(process.stdout, (writableLength) => { - diag("stdout frame write backpressured: writableLength=" + writableLength); - }); + // and each body_chunk -- writes through it, in call order. Its stall bound + // reuses responseTimeoutMs: a write that stays backpressured that long + // already means no response can arrive over this pipe in time, so ending + // the wait here costs nothing beyond what the response side already gives + // up on, and it lets a stalled read's admission reserve release instead of + // holding it for the rest of the process's life. + const writeStdoutFrameLine = createStdoutFrameWriter( + process.stdout, + (writableLength) => { + diag("stdout frame write backpressured: writableLength=" + writableLength); + }, + responseTimeoutMs, + ); // Write a control frame (heartbeat or READY). This is fire-and-forget: a // dropped heartbeat is not fatal, because the heartbeat repeats on its own @@ -3102,10 +3139,12 @@ function runDuplexGateway() { // request charges no less than its true retained peak. A malformed // body cannot expand, because its raw bytes ride base64 frames // unchanged. The reserve for one read releases only once both its send - // finished (drained fully or ended by a stream error) and its response - // settled -- see releaseReserveWhenBothDone below -- so a slow host - // cannot make the gateway release a reserve while it still retains the - // bytes. + // finished (drained fully, ended by a stream error, or ended by the + // writer's own stall-timeout bound) and its response settled -- see + // releaseReserveWhenBothDone below -- so a slow host cannot make the + // gateway release a reserve while it still retains the bytes, and a + // host that stops reading without ever closing the pipe cannot make + // the gateway hold that reserve forever either. handedOffToSend = true; let sendDone = false; let responseDone = false; @@ -3146,10 +3185,13 @@ function runDuplexGateway() { // already registered, so the response side and the send side start // from the same point and settle independently. There is no cancel // frame in the protocol: once this send writes the envelope, it must - // run to completion or the stdout stream must error, because a - // stopped send would leave a partial body only channel teardown - // reclaims. A response timeout above does not stop this send; it only - // resolves the HTTP caller early while the send keeps running. + // run to completion, or the stdout stream must error, or the writer's + // own stall bound must end it, because a stopped send would leave a + // partial body only channel teardown reclaims. A response timeout + // above does not stop this send; it only resolves the HTTP caller + // early while the send keeps running -- but the writer's own bound + // stops a send whose pipe never drains and never errors, so this + // read's admission reserve still releases even then. void (async () => { try { await writeStdoutFrameLine(encodedRequest.line); @@ -3159,9 +3201,10 @@ function runDuplexGateway() { await writeStdoutFrameLine(encoded.line); } } catch (error) { - // The writer's only two end conditions for a stalled write are a - // drain and a stream error. This catches the stream-error case: it - // ends the send here instead of leaving it stalled forever. + // The writer's three end conditions for a stalled write are a + // drain, a stream error, and its own stall-timeout bound. This + // catches the latter two: it ends the send here instead of + // leaving it stalled forever. diag( "the outbound send for request " + requestId +