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 6d9b0aa6555..547fefc1abe 100644 --- a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts +++ b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts @@ -15,6 +15,7 @@ import { } from "./sandbox-callback-bridge.js"; import { DuplexFrameDecoder, + DEFAULT_MAX_DUPLEX_FRAME_BYTES, DUPLEX_BODY_CHUNK_RAW_BYTES, DUPLEX_FRAME_VERSION, encodeDuplexFrame, @@ -978,4 +979,252 @@ describe("duplex bridge local end-to-end harness", () => { const later = await post(); expect(later.status).toBe(200); }, 30000); + + it("bounds the outbound writer to about one frame per active send while the pipe backpressures", async () => { + // Regression test for the backpressure block. The gateway used to call + // stdout.write() for every frame in a tight loop and ignore the return + // value, so a slow reader let one admitted read queue its whole body's + // worth of encoded frames. Pause the pipe read side, so every later write + // meets a full pipe, then post several concurrent cap-sized bodies at + // once and prove the writer never reports more than about one bounded + // frame stuck behind the full pipe. + const maxBodyBytes = 4 * 1024 * 1024; + const concurrency = 4; + const harness = await createHarness({ + maxBodyBytes, + maxInflightBodyBytes: maxBodyBytes * concurrency, + responseTimeoutMs: 20000, + }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // Stop reading the child's stdout, so the OS pipe fills and the gateway's + // writes meet real backpressure, the same as a slow host reader would + // cause in production. + harness.child.stdout.pause(); + + const body = "a".repeat(maxBodyBytes); + const posts = Array.from({ length: concurrency }, (_unused, index) => + fetch(`${harness.baseUrl}/api/issues/issue-${index}/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body, + }).catch(() => undefined), + ); + + // Give the gateway time to read every body and drive its sends into + // backpressure. + await new Promise((resolve) => setTimeout(resolve, 1500)); + + const backpressureLines = harness + .stderr() + .split("\n") + .filter((line) => line.includes("stdout frame write backpressured")); + expect(backpressureLines.length).toBeGreaterThan(0); + const reportedLengths = backpressureLines.map((line) => { + const match = /writableLength=(\d+)/.exec(line); + if (!match) throw new Error("backpressure diagnostic is missing writableLength: " + line); + return Number(match[1]); + }); + // One buffered frame is at most DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB). An + // unbounded queue would report a writableLength close to one whole body's + // worth of encoded frames -- several megabytes for a 4 MiB body. A + // generous multiple of one frame proves the writer never queued anywhere + // near that, regardless of how long the pipe stays full. + const oneFrameBound = DEFAULT_MAX_DUPLEX_FRAME_BYTES * 2; + for (const length of reportedLengths) { + expect(length).toBeLessThan(oneFrameBound); + } + + // Resume so the harness can read the outstanding frames and tear down + // cleanly. + harness.child.stdout.resume(); + await Promise.allSettled(posts); + }, 20000); + + it("keeps one request's frames in sequence while a concurrent request interleaves its own", async () => { + const harness = await createHarness({ + maxBodyBytes: 2 * 1024 * 1024, + maxInflightBodyBytes: 8 * 1024 * 1024, + }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + + // Each body is large enough to split into several body_chunk frames, so a + // correct sequence is not a coincidence of a one-frame body. + const bodyA = "a".repeat(1_500_000); + const bodyB = "b".repeat(1_500_000); + const post = (body: string) => + fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + + const [responseA, responseB] = await Promise.all([post(bodyA), post(bodyB)]); + expect(responseA.status).toBe(200); + expect(responseB.status).toBe(200); + + const requestFrames = harness.observedFrames.filter( + (frame): frame is DuplexRequestFrame => frame.type === "request", + ); + expect(requestFrames).toHaveLength(2); + for (const requestFrame of requestFrames) { + const chunkSeqs = harness.observedFrames + .filter( + (frame): frame is DuplexBodyChunkFrame => + frame.type === "body_chunk" && frame.id === requestFrame.id, + ) + .map((frame) => frame.seq); + expect(chunkSeqs.length).toBeGreaterThan(1); + // This request's own chunks arrive in increasing sequence order, even + // though the other request's frames may interleave between them. + expect(chunkSeqs).toEqual(Array.from({ length: chunkSeqs.length }, (_unused, index) => index)); + } + }, 20000); + + it("holds the admission reserve while a send is stalled behind a full pipe, then releases it once the send drains", async () => { + const maxBodyBytes = 2 * 1024 * 1024; + const harness = await createHarness({ + maxBodyBytes, + // Admits exactly one concurrent body read. + maxInflightBodyBytes: maxBodyBytes, + responseTimeoutMs: 15000, + }); + 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, well before it ever reaches the host. + 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, 500)); + + // The reserve is still held: a second request cannot reserve, so the + // gateway sheds it with a clean 503. + const shed = await fetch(`${harness.baseUrl}/api/issues/issue-2/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(shed.status).toBe(503); + await expect(shed.json()).resolves.toEqual({ error: "bridge_busy" }); + + // Resume the pipe and let the host answer. The stalled send drains and + // finishes, and the first request's response settles. + harness.child.stdout.resume(); + harness.setForwardMode("proxy"); + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + const firstResponse = await first; + expect(firstResponse.status).toBe(200); + + // The reserve released once the send finished. A later request is + // admitted again. + 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(200); + }, 20000); + + it("ends a stalled send when the outbound pipe breaks, and releases the reserve for a later request", async () => { + const maxBodyBytes = 2 * 1024 * 1024; + const harness = await createHarness({ + maxBodyBytes, + // Admits exactly one concurrent body read. + maxInflightBodyBytes: maxBodyBytes, + responseTimeoutMs: 1200, + }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // Stop reading the child's stdout, so the first request's send is + // genuinely in flight -- stalled behind a full pipe -- when the pipe + // breaks. + 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 start writing frames into + // the full pipe. + await new Promise((resolve) => setTimeout(resolve, 400)); + + // Confirm the reserve is held before the pipe breaks: a second request + // cannot reserve, so the gateway sheds it with a clean 503. + const shedBeforeBreak = await fetch(`${harness.baseUrl}/api/issues/issue-2/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(shedBeforeBreak.status).toBe(503); + + // Break the pipe. The stalled write meets a stream error, which ends the + // wait for drain -- the only other end condition the writer accepts. + harness.child.stdout.destroy(); + + const firstResponse = await first; + // No response frame can ever arrive over a broken pipe. The gateway's own + // wait budget ends the request with its local 502 timeout. + expect(firstResponse.status).toBe(502); + + // The reserve released once the failed send and the timed-out response + // both finished. A later request is admitted again -- it is not shed with + // 503, even though it can never complete over the broken pipe either. + 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 aecce32ec1a..fcf05759bc1 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -1,3 +1,4 @@ +import { EventEmitter } from "node:events"; import { createServer } from "node:http"; import net from "node:net"; import { execFile, spawn } from "node:child_process"; @@ -11,6 +12,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { getSandboxCallbackBridgeServerSource, getSandboxDuplexGatewayCodecSource, + getSandboxDuplexGatewayStdoutWriterSource, } from "./sandbox-callback-bridge.js"; import { @@ -5746,6 +5748,112 @@ describe("sandbox duplex gateway", () => { expect(ledger.liveTokenCount).toBe(0); }); + /** + * A minimal fake outbound stream for the embedded stdout-writer tests below. + * It lets a test control exactly when one write "drains" and when the + * stream "errors", with no real pipe and no child process. + */ + class FakeOutboundStream extends EventEmitter { + writes: string[] = []; + writableLength = 0; + private accepting: boolean; + + constructor(options: { accepting?: boolean } = {}) { + super(); + this.accepting = options.accepting ?? true; + } + + write(line: string): boolean { + this.writes.push(line); + if (this.accepting) return true; + this.writableLength += Buffer.byteLength(line, "utf8"); + return false; + } + + drain(): void { + this.writableLength = 0; + this.emit("drain"); + } + + errorOut(error: Error): void { + this.emit("error", error); + } + } + + interface EmbeddedStdoutWriter { + createStdoutFrameWriter: ( + stream: FakeOutboundStream, + onBackpressure?: (writableLength: number) => void, + ) => (line: string) => Promise; + } + + function loadEmbeddedStdoutFrameWriter(): EmbeddedStdoutWriter { + const factory = new Function( + `${getSandboxDuplexGatewayStdoutWriterSource()}\nreturn { createStdoutFrameWriter };`, + ) as () => EmbeddedStdoutWriter; + return factory(); + } + + it("issues one outbound write at a time and waits for drain before the next", async () => { + // Regression test for the backpressure block. The old gateway called + // `stream.write()` for every frame in a tight loop and ignored the return + // value, so a slow reader let it queue a whole request's frames at once. + // The writer must issue at most one write, then wait for the stream to + // drain before it starts the next queued line. + const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter(); + const stream = new FakeOutboundStream({ accepting: false }); + const backpressureLengths: number[] = []; + const writeLine = createStdoutFrameWriter(stream, (length) => backpressureLengths.push(length)); + + const first = writeLine("frame-1\n"); + const second = writeLine("frame-2\n"); + const third = writeLine("frame-3\n"); + + // Only the first line reached the stream. The writer holds the rest back + // in its own queue; it never hands every queued line to the stream at + // once. + await Promise.resolve(); + expect(stream.writes).toEqual(["frame-1\n"]); + expect(backpressureLengths).toEqual([Buffer.byteLength("frame-1\n", "utf8")]); + + stream.drain(); + await first; + expect(stream.writes).toEqual(["frame-1\n", "frame-2\n"]); + + stream.drain(); + await second; + expect(stream.writes).toEqual(["frame-1\n", "frame-2\n", "frame-3\n"]); + + stream.drain(); + await third; + }); + + it("ends a stalled outbound write and fails every queued line when the stream errors", async () => { + // The writer accepts exactly two end conditions for a stalled write: the + // stream drains, or the stream errors. This proves the second one: a + // stream error ends the wait at once and fails every line still queued, + // because a broken pipe can never drain. + const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter(); + const stream = new FakeOutboundStream({ accepting: false }); + const writeLine = createStdoutFrameWriter(stream); + + const first = writeLine("frame-1\n"); + const second = writeLine("frame-2\n"); + await Promise.resolve(); + expect(stream.writes).toEqual(["frame-1\n"]); + + const boom = new Error("EPIPE"); + stream.errorOut(boom); + + await expect(first).rejects.toBe(boom); + await expect(second).rejects.toBe(boom); + + // The writer never issues another write once the stream has errored. + const third = writeLine("frame-3\n"); + await expect(third).rejects.toBe(boom); + expect(stream.writes).toEqual(["frame-1\n"]); + }); + 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 09930cad077..a0743d69813 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -2009,6 +2009,79 @@ class DuplexFrameDecoder { } }`; +// The embedded zero-dependency outbound frame writer. The duplex gateway +// routes every stdout frame write through the one function this factory +// returns: the heartbeat, the READY frame, one request envelope, and each +// 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) { + const queue = []; + let pumping = false; + let streamError = null; + + stream.on("error", (error) => { + streamError = error instanceof Error ? error : new Error(String(error)); + const stranded = queue.splice(0, queue.length); + for (const entry of stranded) entry.reject(streamError); + }); + + async function pump() { + if (pumping) return; + pumping = true; + while (queue.length > 0) { + const entry = queue[0]; + if (streamError) { + queue.length = 0; + entry.reject(streamError); + break; + } + let wroteOk = false; + try { + wroteOk = stream.write(entry.line); + } catch (error) { + queue.shift(); + entry.reject(error instanceof Error ? error : new Error(String(error))); + continue; + } + if (!wroteOk && !streamError) { + if (typeof onBackpressure === "function") { + onBackpressure(stream.writableLength); + } + await new Promise((resolve) => { + const onDrain = () => { + stream.off("error", onStreamErrorWhileWaiting); + resolve(); + }; + const onStreamErrorWhileWaiting = () => { + stream.off("drain", onDrain); + resolve(); + }; + stream.once("drain", onDrain); + stream.once("error", onStreamErrorWhileWaiting); + }); + } + queue.shift(); + if (streamError) { + entry.reject(streamError); + } else { + entry.resolve(); + } + } + pumping = false; + } + + return function writeStdoutFrameLine(line) { + return new Promise((resolve, reject) => { + queue.push({ line: line, resolve: resolve, reject: reject }); + void pump(); + }); + }; +}`; + /** * Return the exact zero-dependency codec source the generated duplex gateway * embeds. A test runs every fixture vector against this source, so it proves the @@ -2022,6 +2095,18 @@ export function getSandboxDuplexGatewayCodecSource(): string { return DUPLEX_GATEWAY_CODEC_SOURCE; } +/** + * 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. + */ +export function getSandboxDuplexGatewayStdoutWriterSource(): string { + return DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE; +} + export function getSandboxCallbackBridgeServerSource(): string { return `import { randomUUID, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; @@ -2087,6 +2172,10 @@ if (bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" && !queueDir) { // gateway ignores it. ${DUPLEX_GATEWAY_CODEC_SOURCE} +// The embedded zero-dependency outbound frame writer. The duplex gateway uses +// it; the file gateway ignores it. +${DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE} + function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -2135,22 +2224,40 @@ function reserveBodyRead() { }; } +// Read one request body into a single buffer sized to the admission charge. +// +// Accounting choice: the old code pushed each chunk onto a chunks array, then +// called Buffer.concat on it. Both the array and the concat result held the +// full body at once, so the transient peak was 2x the raw body while +// admission charged 1x -- a dishonest bound. This version preallocates one +// buffer of maxBodyBytes and copies each incoming chunk into it in place, then +// returns a view over the written bytes. A view shares the same backing +// memory; it adds no second copy. So the retained peak per active read is one +// maxBodyBytes buffer, matching the charge exactly. +// +// The tradeoff: a small body now retains the full maxBodyBytes allocation for +// the life of the request, where the old code retained only the small body's +// own bytes. reserveBodyRead already charges maxBodyBytes for every active +// read regardless of the real size, so this tradeoff raises no read past what +// admission already assumed was possible; it only makes the real memory shape +// match the existing worst-case charge instead of usually running under it. async function readBody(req) { - const chunks = []; + const out = Buffer.allocUnsafe(maxBodyBytes); let totalBytes = 0; for await (const chunk of req) { const nextChunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - chunks.push(nextChunk); - totalBytes += nextChunk.byteLength; - if (totalBytes > maxBodyBytes) { + if (totalBytes + nextChunk.byteLength > maxBodyBytes) { throw new BridgeBodyTooLargeError(); } + nextChunk.copy(out, totalBytes); + totalBytes += nextChunk.byteLength; } - // Return the raw request bytes. The duplex gateway base64-encodes these bytes - // unchanged, so a malformed body cannot expand during transport and cannot - // exceed the bytes that admission charged. The file gateway decodes a valid - // UTF-8 body with decodeUtf8Body and rejects an invalid one. - return Buffer.concat(chunks); + // Return the raw request bytes as a view, not a copy. The duplex gateway + // base64-encodes these bytes unchanged, so a malformed body cannot expand + // during transport and cannot exceed the bytes that admission charged. The + // file gateway decodes a valid UTF-8 body with decodeUtf8Body and rejects an + // invalid one. + return out.subarray(0, totalBytes); } // Strict UTF-8 decoder for the file gateway. It throws on any invalid byte, so @@ -2368,8 +2475,27 @@ function runDuplexGateway() { process.stderr.write("[paperclip-bridge] " + message + "\\n"); } + // 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); + }); + + // 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 + // interval, and a lost READY frame surfaces as a broker wait timeout. A + // request send awaits its own writes instead -- see the request handler + // below -- because it must run to completion once it starts. function writeFrame(frame) { - process.stdout.write(encodeDuplexFrame(frame)); + writeStdoutFrameLine(encodeDuplexFrame(frame)).catch((error) => { + diag( + "failed to write a " + + frame.type + + " frame: " + + (error && error.message ? error.message : String(error)), + ); + }); } function stopTimers() { @@ -2558,6 +2684,11 @@ function runDuplexGateway() { writeJsonResponse(res, 503, { error: "bridge_busy" }); return; } + // Once the send below starts, releaseReserveWhenBothDone owns the + // reserve release. This flag stays false only while a return above the + // send can still happen, so the outer finally releases directly for + // every path that never commits to a send. + let handedOffToSend = false; try { const requestBodyBuffer = await readBody(req); if (unavailable) { @@ -2598,37 +2729,100 @@ function runDuplexGateway() { writeJsonResponse(res, 413, { error: "request_too_large" }); return; } - // Retained-peak note. The gateway preserves the raw request bytes end to - // end and streams the body as body_chunk frames, one frame at a time. It - // does not hold the raw body plus the full base64 frame array plus the - // full encoded-line array (about 3.7x the raw body). The retained peak per - // read is now the raw body plus one encoded frame: - // raw body <= maxBodyBytes (about 10 MiB) - // one body_chunk <= DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB) - // so the peak is about maxBodyBytes + 1 MB per read. Admission charges - // maxBodyBytes per read against the 64 MiB inflight bound. The bound admits - // at most floor(64 MiB / maxBodyBytes) = 6 reads, so the true retained peak - // stays near 6 * (10 MiB + 1 MB) ~= 66 MiB. That is a fixed small margin - // over the admission threshold, not a multiple of the body. A malformed - // body cannot expand, because its raw bytes ride base64 frames unchanged. + // Retained-peak note. readBody preallocates one buffer sized to + // maxBodyBytes and returns a view over it, so no second (concatenated) + // copy of the body ever coexists with it -- see the comment on readBody. + // The send below builds and encodes one body_chunk frame at a time from + // that same view; it holds no per-request buffer of its own. Every + // frame -- this envelope, each body_chunk, and every heartbeat and READY + // frame from the rest of the process -- writes through the one + // serialized writer above. That writer issues at most one + // written-but-not-yet-drained line at a time, bounded by + // DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB), plus whatever the stream itself + // still buffers up to its own high-water mark; it never queues a whole + // request's frames at once behind a slow reader. So the retained peak + // per active read is about: + // one body buffer <= maxBodyBytes (about 10 MiB) + // one in-flight frame <= DEFAULT_MAX_DUPLEX_FRAME_BYTES (1 MB) + // and admission charges maxBodyBytes per read against the 64 MiB + // inflight bound, so the bound admits at most + // floor(64 MiB / maxBodyBytes) = 6 reads. The true retained peak across + // every admitted read stays near 6 * (10 MiB + 1 MB) ~= 66 MiB, the same + // fixed small margin over the admission threshold as before -- not a + // multiple of the body, and not made worse by a slow host. 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. + handedOffToSend = true; + let sendDone = false; + let responseDone = false; + function releaseReserveWhenBothDone() { + if (sendDone && responseDone) { + // releaseBodyRead is itself idempotent, so a second call here (for + // example if both flags flip true in the same tick) is harmless. + releaseBodyRead(); + } + } + const response = await new Promise((resolve) => { const timer = setTimeout(() => { responseAssembly.delete(requestId); if (pending.delete(requestId)) { - resolve({ + settleResponse({ status: 502, headers: { "content-type": "application/json" }, body: JSON.stringify({ error: "Timed out waiting for host bridge response." }), }); } }, responseTimeoutMs); - pending.set(requestId, { resolve: resolve, timer: timer }); - process.stdout.write(encodedRequest.line); - for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) { - const encoded = encodeDuplexFrameChecked(frame); - // Pass 1 proved every frame fits, so encoded.ok is always true here. - process.stdout.write(encoded.line); + + // Every path that settles this request's response -- a real response, + // this timeout, a local 502 from failRequest, or the 409 triggerLoss + // sends on channel loss -- resolves through entry.resolve, which is + // this function. So responseDone flips true exactly once, on + // whichever of those settles first. + function settleResponse(value) { + clearTimeout(timer); + responseDone = true; + releaseReserveWhenBothDone(); + resolve(value); } + pending.set(requestId, { resolve: settleResponse, timer: timer }); + + // Run the serialized send after the pending entry and the timer are + // 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. + void (async () => { + try { + await writeStdoutFrameLine(encodedRequest.line); + for (const frame of duplexBodyChunkFrames(requestId, requestBodyBuffer)) { + const encoded = encodeDuplexFrameChecked(frame); + // Pass 1 proved every frame fits, so encoded.ok is always true here. + 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. + diag( + "the outbound send for request " + + requestId + + " ended: " + + (error && error.message ? error.message : String(error)), + ); + } finally { + sendDone = true; + releaseReserveWhenBothDone(); + } + })(); }); res.statusCode = typeof response.status === "number" ? response.status : 200; for (const [key, value] of Object.entries(response.headers || {})) { @@ -2637,7 +2831,9 @@ function runDuplexGateway() { } res.end(typeof response.body === "string" ? response.body : ""); } finally { - releaseBodyRead(); + if (!handedOffToSend) { + releaseBodyRead(); + } } } catch (error) { if (error && error.code === "request_too_large") {