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 +