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 547fefc1abe..c26350ed848 100644 --- a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts +++ b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts @@ -781,13 +781,20 @@ describe("duplex bridge local end-to-end harness", () => { expect(receivedBodyLength).toBe(apiLimitBytes); }, 30000); - it("answers 413 request_too_large, not 502, when the body passes the bridge cap", async () => { + it("answers 413 from the declared length alone, before it reserves or reads any body byte", async () => { // Regression test: the body-size overflow path threw a generic Error that // unwound to the 502 catch. A 502 is retryable and leaked the internal // message. The overflow must answer a clean 413 request_too_large. Assert the // status code specifically, because the wrong 502 stood for a long time. const maxBodyBytes = 4096; - const harness = await createHarness({ maxBodyBytes }); + const laterBodySize = "small".length; + // Just enough admission budget for the later small read below (its + // declared size plus the fixed in-flight frame allowance), with no + // margin. If the gateway reserved the oversize request's declared length + // before it rejected the request, this budget would stay exhausted and + // the later request would be shed with a 503 instead of admitted. + const maxInflightBodyBytes = laterBodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES; + const harness = await createHarness({ maxBodyBytes, maxInflightBodyBytes }); harnesses.push(harness); await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); @@ -812,18 +819,126 @@ describe("duplex bridge local end-to-end harness", () => { // The oversize body never reached the API, and the response leaked no // internal message. expect(apiWasCalled).toBe(false); + + // The rejected request reserved nothing, so a small request right after + // is still admitted -- proving the gateway answered the 413 from the + // declared length header alone, before it ever reserved or read a byte. + const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(later.status).toBe(200); + expect(apiWasCalled).toBe(true); + }, 20000); + + it("reserves the worst case and answers 413 for a chunked request with no declared length that overruns the cap", async () => { + // A request with no Content-Length header -- for example a chunked + // request -- gives the gateway no declared size ahead of time. The + // gateway must fall back to the worst case (maxBodyBytes) for its + // reserve, and it must still answer a clean 413 once the body it + // actually reads exceeds the cap. + const maxBodyBytes = 4096; + const harness = await createHarness({ maxBodyBytes }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + let apiWasCalled = false; + harness.api.setResponder((_req, res) => { + apiWasCalled = true; + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + + const oversizeBody = "a".repeat(maxBodyBytes + 1024); + // A ReadableStream body sends with chunked transfer-encoding and no + // Content-Length header, so the gateway cannot know the real size ahead + // of time. + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(oversizeBody)); + controller.close(); + }, + }); + const response = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: stream, + duplex: "half", + } as RequestInit & { duplex: "half" }); + expect(response.status).toBe(413); + await expect(response.json()).resolves.toEqual({ error: "request_too_large" }); + expect(apiWasCalled).toBe(false); + + // The rejected read released its reserve, so a later request is admitted + // again and reaches the API. + const later = await fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body: "small", + }); + expect(later.status).toBe(200); + }, 20000); + + it("admits many small concurrent requests that the old flat 10 MiB charge would have shed with 503", async () => { + // Production constants. Under the old flat charge every active read + // reserved maxBodyBytes (about 10 MiB) regardless of the real body size, + // so the 64 MiB inflight bound admitted at most + // floor(64 MiB / (10 MiB + 1)) = 6 concurrent reads. Charging the + // declared body size instead means a control-plane-sized body reserves + // only its own few hundred bytes plus the fixed frame allowance, so the + // same bound admits far more than 6 concurrent small reads at once. + const maxBodyBytes = 10 * 1024 * 1024 + 1; + const maxInflightBodyBytes = 64 * 1024 * 1024; + const concurrency = 20; + const harness = await createHarness({ maxBodyBytes, maxInflightBodyBytes }); + 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 })); + }); + + const body = JSON.stringify({ body: "a small status comment" }); + const post = () => + fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${harness.bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + const responses = await Promise.all(Array.from({ length: concurrency }, () => post())); + expect(responses.map((response) => response.status)).toEqual( + Array.from({ length: concurrency }, () => 200), + ); }, 20000); it("sheds a concurrent body read past the admission bound, then releases the reserve", async () => { - // The gateway reserves one worst-case body per active read against the local - // admission bound. With the bound set to one body, the first read is admitted - // and the second read is shed with a 503. The gateway releases the reserve on - // the response timeout, so a later request is admitted again. + // The gateway reserves the declared body size plus the fixed in-flight + // frame allowance for each active read against the local admission + // bound. With the bound set to exactly one such reserve, the first read + // is admitted and the second read is shed with a 503. The gateway + // releases the reserve on the response timeout, so a later request is + // admitted again. const maxBodyBytes = 4096; + const bodySize = 1024; + const reservePerRead = bodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES; const harness = await createHarness({ maxBodyBytes, - // The bound admits exactly one concurrent body read. - maxInflightBodyBytes: maxBodyBytes, + // The bound admits exactly one concurrent body read at bodySize. + maxInflightBodyBytes: reservePerRead, // A short response deadline releases the first (hung) read fast. responseTimeoutMs: 700, }); @@ -833,7 +948,7 @@ describe("duplex bridge local end-to-end harness", () => { // The first admitted request hangs on the forward until it times out. The // second request cannot reserve, so the gateway sheds it with a 503. harness.setForwardMode("hang"); - const body = "a".repeat(1024); + const body = "a".repeat(bodySize); const post = () => fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { method: "POST", @@ -923,16 +1038,21 @@ describe("duplex bridge local end-to-end harness", () => { }, 30000); it("caps concurrent body reads at the retained-byte model, then releases the reserve", async () => { - // Production-size accounting. Admission charges maxBodyBytes per read against - // the inflight bound, so the bound admits floor(bound / maxBodyBytes) reads. - // With the production constants that is floor(64 MiB / (10 MiB + 1)) = 6. The - // reservation is the worst-case body per read, so a small body still charges - // the full maxBodyBytes. So the test drives the model with small bodies and - // asserts the admitted count matches the model. + // Production-size accounting. The gateway charges each active read the + // declared body size plus the fixed in-flight frame allowance, not the + // flat body cap, so the bound admits + // floor(inflightBound / (declaredBytes + frameAllowance)) reads. The + // declared body here is well under the cap, so this model admits far + // more than the old flat-charge model's 6 -- but the test still keeps + // the count under the gateway's separate maxQueueDepth bound (64), so + // that unrelated queue-depth gate does not also shed a request and + // confound the assertions below. const maxBodyBytes = 10 * 1024 * 1024 + 1; const maxInflightBodyBytes = 64 * 1024 * 1024; - const admitCount = Math.floor(maxInflightBodyBytes / maxBodyBytes); - expect(admitCount).toBe(6); + const bodySize = 1024 * 1024; + const reservePerRead = bodySize + DEFAULT_MAX_DUPLEX_FRAME_BYTES; + const admitCount = Math.floor(maxInflightBodyBytes / reservePerRead); + expect(admitCount).toBe(32); const harness = await createHarness({ maxBodyBytes, maxInflightBodyBytes, @@ -945,7 +1065,7 @@ describe("duplex bridge local end-to-end harness", () => { // Each admitted read hangs on the forward until it times out, so all admitted // reads hold their reserve at once and the next read cannot reserve. harness.setForwardMode("hang"); - const body = "a".repeat(1024); + const body = "a".repeat(bodySize); const post = () => fetch(`${harness.baseUrl}/api/issues/issue-1/comments`, { method: "POST", @@ -992,7 +1112,9 @@ describe("duplex bridge local end-to-end harness", () => { const concurrency = 4; const harness = await createHarness({ maxBodyBytes, - maxInflightBodyBytes: maxBodyBytes * concurrency, + // Each concurrent read charges its declared body size plus the fixed + // in-flight frame allowance, so the bound must admit all four at once. + maxInflightBodyBytes: (maxBodyBytes + DEFAULT_MAX_DUPLEX_FRAME_BYTES) * concurrency, responseTimeoutMs: 20000, }); harnesses.push(harness); @@ -1098,8 +1220,9 @@ describe("duplex bridge local end-to-end harness", () => { const maxBodyBytes = 2 * 1024 * 1024; const harness = await createHarness({ maxBodyBytes, - // Admits exactly one concurrent body read. - maxInflightBodyBytes: 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: 15000, }); harnesses.push(harness); @@ -1166,8 +1289,9 @@ describe("duplex bridge local end-to-end harness", () => { const maxBodyBytes = 2 * 1024 * 1024; const harness = await createHarness({ maxBodyBytes, - // Admits exactly one concurrent body read. - maxInflightBodyBytes: 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: 1200, }); harnesses.push(harness); diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index fcf05759bc1..6ed28a1d8ab 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -5854,6 +5854,37 @@ describe("sandbox duplex gateway", () => { expect(stream.writes).toEqual(["frame-1\n"]); }); + it("settles every write queued in the same tick as a stream error, not only the queue head", async () => { + // Regression test. The writer's top-of-loop error check rejected only the + // queue head and discarded the rest of the queue with no reject call, so + // a second write queued in the same synchronous tick as the error never + // settled. Separately, the code path that resumed after a stalled write + // shifted the queue head unconditionally, so a write queued during that + // stall -- after the stream had already errored, but before the writer's + // pending wait had a chance to resume -- was removed and never settled + // either. Queue two more writes in the same tick as the error, so both + // failure modes are on the table, and prove every one of the three + // pending promises rejects. + const { createStdoutFrameWriter } = loadEmbeddedStdoutFrameWriter(); + const stream = new FakeOutboundStream({ accepting: false }); + const writeLine = createStdoutFrameWriter(stream); + + const first = writeLine("frame-1\n"); + await Promise.resolve(); + expect(stream.writes).toEqual(["frame-1\n"]); + + const boom = new Error("EPIPE"); + // Queue two more lines in the same synchronous tick as the error, before + // the writer's drain-or-error wait for frame-1 resumes. + stream.errorOut(boom); + const second = writeLine("frame-2\n"); + const third = writeLine("frame-3\n"); + + await expect(first).rejects.toBe(boom); + await expect(second).rejects.toBe(boom); + await expect(third).rejects.toBe(boom); + }); + 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.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index b0300d6d7d1..59029ec819b 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -885,6 +885,166 @@ describe("sandbox callback bridge", () => { }); }); + it("reserves the declared body size, not the flat body cap, against the file gateway's admission bound", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-body-reserve-")); + cleanupDirs.push(rootDir); + + const localWorkspaceDir = path.join(rootDir, "local-workspace"); + const remoteWorkspaceDir = path.join(rootDir, "remote-workspace"); + await mkdir(localWorkspaceDir, { recursive: true }); + await mkdir(remoteWorkspaceDir, { recursive: true }); + await writeFile(path.join(localWorkspaceDir, "README.md"), "bridge body reserve test\n", "utf8"); + + const runner = createExecRunner(); + const bridgeAsset = await createSandboxCallbackBridgeAsset(); + cleanupFns.push(bridgeAsset.cleanup); + const prepared = await prepareCommandManagedRuntime({ + runner, + spec: { + remoteCwd: remoteWorkspaceDir, + timeoutMs: 30_000, + }, + adapterKey: "codex", + workspaceLocalDir: localWorkspaceDir, + assets: [{ key: "bridge", localDir: bridgeAsset.localDir }], + }); + + const queueDir = path.posix.join(prepared.runtimeRootDir, "paperclip-bridge"); + const bridgeToken = createSandboxCallbackBridgeToken(); + + const maxBodyBytes = 4096; + const bodySize = 1024; + // The file gateway charges 15x the declared body size for its reserve: + // 1x for the raw buffer, up to 2x for the decoded string, and up to 6x + // each for the JSON-escaped string and its UTF-8 write buffer. See the + // comment on FILE_GATEWAY_BODY_RESERVE_FACTOR in sandbox-callback-bridge.ts. + const reservePerRead = bodySize * 15; + const bridge = await startSandboxCallbackBridgeServer({ + runner, + remoteCwd: remoteWorkspaceDir, + assetRemoteDir: prepared.assetDirs.bridge, + queueDir, + bridgeToken, + timeoutMs: 30_000, + pollIntervalMs: 10, + // No worker answers in this test, so an admitted read hangs until this + // deadline and times out with a 502. A shed read never reaches that + // far; it answers a 503 at once. + responseTimeoutMs: 300, + maxBodyBytes, + // Admits exactly one concurrent body read at bodySize. + maxInflightBodyBytes: reservePerRead, + }); + cleanupFns.push(async () => { + await bridge.stop(); + }); + + const body = "a".repeat(bodySize); + const post = () => + fetch(`${bridge.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + const [first, second] = await Promise.all([post(), post()]); + const statuses = [first.status, second.status].sort((a, b) => a - b); + // One request was admitted (it later times out with a 502, because no + // worker answers). One request was shed with a 503 before it could + // buffer a second body. + expect(statuses).toEqual([502, 503]); + const shed = first.status === 503 ? first : second; + await expect(shed.json()).resolves.toEqual({ error: "bridge_busy" }); + + // The timed-out read released its reserve, so a later request is + // admitted again -- it also times out with a 502, not a 503. + const later = await post(); + expect(later.status).toBe(502); + }, 20000); + + it("admits many small concurrent file-gateway requests that the old flat 10 MiB charge would have shed with 503", async () => { + // Production constants. Under the old flat charge every active read + // reserved maxBodyBytes (about 10 MiB) regardless of the real body size, + // so the 64 MiB inflight bound admitted at most + // floor(64 MiB / (10 MiB + 1)) = 6 concurrent reads. Charging the + // declared body size instead (times the file gateway's factor) means a + // control-plane-sized body reserves only a small multiple of its own + // size, so the same bound admits far more than 6 concurrent small reads + // at once. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-small-concurrent-")); + cleanupDirs.push(rootDir); + + const localWorkspaceDir = path.join(rootDir, "local-workspace"); + const remoteWorkspaceDir = path.join(rootDir, "remote-workspace"); + await mkdir(localWorkspaceDir, { recursive: true }); + await mkdir(remoteWorkspaceDir, { recursive: true }); + await writeFile(path.join(localWorkspaceDir, "README.md"), "bridge small concurrent test\n", "utf8"); + + const runner = createExecRunner(); + const bridgeAsset = await createSandboxCallbackBridgeAsset(); + cleanupFns.push(bridgeAsset.cleanup); + const prepared = await prepareCommandManagedRuntime({ + runner, + spec: { + remoteCwd: remoteWorkspaceDir, + timeoutMs: 30_000, + }, + adapterKey: "codex", + workspaceLocalDir: localWorkspaceDir, + assets: [{ key: "bridge", localDir: bridgeAsset.localDir }], + }); + + const queueDir = path.posix.join(prepared.runtimeRootDir, "paperclip-bridge"); + const bridgeToken = createSandboxCallbackBridgeToken(); + + const worker = await startSandboxCallbackBridgeWorker({ + client: createFileSystemSandboxCallbackBridgeQueueClient(), + queueDir, + authorizeRequest: async () => null, + handleRequest: async (request) => ({ + status: 200, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ok: true, id: request.id }), + }), + }); + cleanupFns.push(async () => { + await worker.stop(); + }); + + const bridge = await startSandboxCallbackBridgeServer({ + runner, + remoteCwd: remoteWorkspaceDir, + assetRemoteDir: prepared.assetDirs.bridge, + queueDir, + bridgeToken, + timeoutMs: 30_000, + pollIntervalMs: 10, + maxBodyBytes: 10 * 1024 * 1024 + 1, + // maxInflightBodyBytes stays at the production default (64 MiB). + }); + cleanupFns.push(async () => { + await bridge.stop(); + }); + + const concurrency = 20; + const body = JSON.stringify({ body: "a small status comment" }); + const post = () => + fetch(`${bridge.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${bridgeToken}`, + "content-type": "application/json", + }, + body, + }); + const responses = await Promise.all(Array.from({ length: concurrency }, () => post())); + expect(responses.map((response) => response.status)).toEqual( + Array.from({ length: concurrency }, () => 200), + ); + }, 20000); + it("returns a 502 when the host response times out", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-timeout-")); cleanupDirs.push(rootDir); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index a0743d69813..21f38c1c016 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -44,11 +44,12 @@ const DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES = 8 * 1024 * 1024; // operating-system process inside the sandbox, so it cannot share the host // aggregate byte ledger. At a 10 MiB body cap, the 64-deep pending queue could // let many concurrent reads each retain a full body and exhaust the sandbox -// process memory before any host limit applies. The gateway reserves one -// worst-case body per active read against this bound, and sheds a new read with -// a 503 when the bound is full. The host passes the value through -// PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES. This bound protects the sandbox -// process; it is separate from the host aggregate byte ledger. +// process memory before any host limit applies. The gateway reserves bytes for +// each active read against this bound -- the declared body size times a +// per-gateway representation factor, not a flat per-request charge -- and sheds +// a new read with a 503 when the bound is full. The host passes the value +// through PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES. This bound protects the +// sandbox process; it is separate from the host aggregate byte ledger. const DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES = 64 * 1024 * 1024; // Per-iteration timeout for one poll-loop client call. A healthy control-plane // round trip finishes in well under one second, so 10s is far above a normal @@ -476,6 +477,7 @@ export function buildSandboxCallbackBridgeEnv(input: { responseTimeoutMs?: number | null; maxQueueDepth?: number | null; maxBodyBytes?: number | null; + maxInflightBodyBytes?: number | null; }): Record { return { PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_FILE_MODE, @@ -495,6 +497,9 @@ export function buildSandboxCallbackBridgeEnv(input: { PAPERCLIP_BRIDGE_MAX_BODY_BYTES: String( normalizeTimeoutMs(input.maxBodyBytes, DEFAULT_BRIDGE_MAX_BODY_BYTES), ), + PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES: String( + normalizeTimeoutMs(input.maxInflightBodyBytes, DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES), + ), }; } @@ -1643,6 +1648,7 @@ export async function startSandboxCallbackBridgeServer(input: { shellCommand?: "bash" | "sh" | null; maxQueueDepth?: number | null; maxBodyBytes?: number | null; + maxInflightBodyBytes?: number | null; }): Promise { const timeoutMs = normalizeTimeoutMs(input.timeoutMs, DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS); const shellCommand = preferredShellForSandbox(input.shellCommand); @@ -1668,6 +1674,7 @@ export async function startSandboxCallbackBridgeServer(input: { responseTimeoutMs: input.responseTimeoutMs, maxQueueDepth: input.maxQueueDepth, maxBodyBytes: input.maxBodyBytes, + maxInflightBodyBytes: input.maxInflightBodyBytes, }); const nodeCommand = input.nodeCommand?.trim() || "node"; const startResult = await input.runner.execute({ @@ -2023,22 +2030,29 @@ const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(st 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, + // not only the one pump happened to be holding. + function strandQueue(error) { + const stranded = queue.splice(0, queue.length); + for (const entry of stranded) entry.reject(error); + } + 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); + strandQueue(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); + strandQueue(streamError); break; } + const entry = queue[0]; let wroteOk = false; try { wroteOk = stream.write(entry.line); @@ -2064,12 +2078,18 @@ const DUPLEX_GATEWAY_STDOUT_WRITER_SOURCE = `function createStdoutFrameWriter(st stream.once("error", onStreamErrorWhileWaiting); }); } - queue.shift(); if (streamError) { - entry.reject(streamError); - } else { - entry.resolve(); + // 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. + continue; } + queue.shift(); + entry.resolve(); } pumping = false; } @@ -2140,9 +2160,10 @@ const maxDuplexDecoderBytes = Number( process.env.PAPERCLIP_BRIDGE_MAX_DUPLEX_DECODER_BYTES || "${DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES}", ); // The host passes the local in-sandbox body-read admission bound here. The -// gateway reserves one worst-case body (maxBodyBytes) per active read and sheds -// a new read with a 503 when the reserve is full. This bound protects the -// sandbox process memory; it never shares the host aggregate byte ledger. +// gateway reserves declaredBytes times a per-gateway factor for each active +// read and sheds a new read with a 503 when the reserve is full. This bound +// protects the sandbox process memory; it never shares the host aggregate +// byte ledger. const maxInflightBodyBytes = Number( process.env.PAPERCLIP_BRIDGE_MAX_INFLIGHT_BODY_BYTES || "${DEFAULT_BRIDGE_MAX_INFLIGHT_BODY_BYTES}", ); @@ -2193,9 +2214,10 @@ function normalizeHeaders(headers) { return out; } -// Typed marker for a request body that exceeds maxBodyBytes. readBody throws it, -// so each gateway answers a clean 413 request_too_large instead of the generic -// 502. A 502 is retryable by convention and would leak the internal message. +// Typed marker for a request body that exceeds its declared length or the +// bridge cap. readBody throws it, so each gateway answers a clean 413 +// request_too_large instead of the generic 502. A 502 is retryable by +// convention and would leak the internal message. class BridgeBodyTooLargeError extends Error { constructor() { super("request_too_large"); @@ -2203,50 +2225,98 @@ class BridgeBodyTooLargeError extends Error { } } +// Read the declared body size from the Content-Length request header. Node's +// HTTP server validates the header's syntax before a request reaches a +// handler, so this only needs to guard the two cases that leave the gateway +// with no declared size: the header is absent, or its value is not a plain +// non-negative integer (a chunked request carries no Content-Length header +// at all). Both cases return maxBodyBytes, the prior worst-case charge, so a +// request that declares nothing keeps the old admission behavior. +function declaredBodyBytes(req) { + const raw = req.headers["content-length"]; + if (typeof raw !== "string" || !/^[0-9]+$/.test(raw)) { + return maxBodyBytes; + } + const parsed = Number(raw); + return Number.isSafeInteger(parsed) ? parsed : maxBodyBytes; +} + +// The file gateway holds four representations of one request body at once +// after it reads the raw bytes -- see the comment at the writeFile call in +// runFileGateway. Each one's size relative to the declared body size D: +// raw body buffer 1x D (readBody allocates exactly D) +// decodeUtf8Body string 2x D (worst case: V8 stores the string two bytes +// per character) +// JSON.stringify string 6x D (worst case: every raw byte is a JSON +// control character, so it escapes to +// "\\u00XX", six characters, for one input +// byte) +// fs.writeFile Buffer 6x D (the escaped string above is pure ASCII, so +// its UTF-8 encoding is one byte per +// character, matching the escaped string's +// six-to-one ratio) +// 1 + 2 + 6 + 6 = 15. The factor charges the sum of every representation's +// own worst case, not the true instantaneous peak, so it stays a safe upper +// bound even if a future change makes two representations overlap that do +// not overlap today. +const FILE_GATEWAY_BODY_RESERVE_FACTOR = 15; + +// The duplex gateway holds one body buffer plus at most one in-flight +// base64-encoded frame while it sends that buffer to the host -- see the +// retained-peak note in runDuplexGateway. The frame never exceeds +// DEFAULT_MAX_DUPLEX_FRAME_BYTES, and that bound does not grow with the body +// (the gateway always sends the body as a series of fixed-size frames), so +// the reserve adds it as a fixed term instead of folding it into a +// multiplier. +const DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES = DEFAULT_MAX_DUPLEX_FRAME_BYTES; + // The running total of reserved body-read bytes across all active reads. let reservedBodyBytes = 0; const noopRelease = () => {}; -// Reserve capacity for one body read against the local admission bound. Return a -// release function on success, or null when the bound is full. Each caller -// releases on every terminal path: success, size rejection, other error, -// timeout, and connection close. -function reserveBodyRead() { - if (reservedBodyBytes + maxBodyBytes > maxInflightBodyBytes) { +// Reserve reserveBytes against the local admission bound for one active read. +// Return a release function on success, or null when the bound is full. Each +// caller releases on every terminal path: success, size rejection, other +// error, timeout, and connection close. +// +// A caller passes the bytes the read will actually retain -- the declared +// body size times the calling gateway's representation factor -- not a flat +// per-request charge. +function reserveBodyRead(reserveBytes) { + if (reservedBodyBytes + reserveBytes > maxInflightBodyBytes) { return null; } - reservedBodyBytes += maxBodyBytes; + reservedBodyBytes += reserveBytes; let released = false; return () => { if (released) return; released = true; - reservedBodyBytes -= maxBodyBytes; + reservedBodyBytes -= reserveBytes; }; } -// Read one request body into a single buffer sized to the admission charge. +// Read one request body into a single buffer sized to the declared body +// length. // // 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. +// full body at once, so the transient peak was 2x the raw body -- a +// dishonest bound. This version preallocates one buffer of declaredBytes 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 for the raw body is exactly one declaredBytes +// buffer, matching the reserve for it. // -// 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 out = Buffer.allocUnsafe(maxBodyBytes); +// Buffer.allocUnsafe is safe here because the returned view never exposes a +// byte the loop did not write: out.subarray(0, totalBytes) stops at +// totalBytes, and totalBytes only ever advances by the byte count a copy +// call actually wrote. +async function readBody(req, declaredBytes) { + const out = Buffer.allocUnsafe(declaredBytes); let totalBytes = 0; for await (const chunk of req) { const nextChunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - if (totalBytes + nextChunk.byteLength > maxBodyBytes) { + if (totalBytes + nextChunk.byteLength > declaredBytes) { throw new BridgeBodyTooLargeError(); } nextChunk.copy(out, totalBytes); @@ -2347,13 +2417,22 @@ async function runFileGateway() { } const requestId = randomUUID(); const hasRequestBody = req.method && req.method !== "GET" && req.method !== "HEAD"; - const releaseBodyRead = hasRequestBody ? reserveBodyRead() : noopRelease; + const declaredBytes = hasRequestBody ? declaredBodyBytes(req) : 0; + if (hasRequestBody && declaredBytes > maxBodyBytes) { + // Reject an oversize body from its declared length alone, before the + // gateway reserves or reads one byte of it. + writeJsonResponse(res, 413, { error: "request_too_large" }); + return; + } + const releaseBodyRead = hasRequestBody + ? reserveBodyRead(declaredBytes * FILE_GATEWAY_BODY_RESERVE_FACTOR) + : noopRelease; if (!releaseBodyRead) { writeJsonResponse(res, 503, { error: "bridge_busy" }); return; } try { - const requestBody = decodeUtf8Body(await readBody(req)); + const requestBody = decodeUtf8Body(await readBody(req, declaredBytes)); const payload = { id: requestId, method: req.method || "GET", @@ -2365,6 +2444,11 @@ async function runFileGateway() { }; const requestPath = path.posix.join(requestsDir, \`\${requestId}.json\`); const tempPath = \`\${requestPath}.tmp\`; + // JSON.stringify builds a third representation of the body (the escaped + // copy inside payload), and fs.writeFile's "utf8" encoding builds a + // fourth (the write Buffer). FILE_GATEWAY_BODY_RESERVE_FACTOR charges + // for all four representations at once; see its comment for the + // worst-case size of each. await fs.writeFile(tempPath, \`\${JSON.stringify(payload)}\\n\`, "utf8"); await fs.rename(tempPath, requestPath); @@ -2679,7 +2763,16 @@ function runDuplexGateway() { } const requestId = randomUUID(); const hasRequestBody = req.method && req.method !== "GET" && req.method !== "HEAD"; - const releaseBodyRead = hasRequestBody ? reserveBodyRead() : noopRelease; + const declaredBytes = hasRequestBody ? declaredBodyBytes(req) : 0; + if (hasRequestBody && declaredBytes > maxBodyBytes) { + // Reject an oversize body from its declared length alone, before the + // gateway reserves or reads one byte of it. + writeJsonResponse(res, 413, { error: "request_too_large" }); + return; + } + const releaseBodyRead = hasRequestBody + ? reserveBodyRead(declaredBytes + DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES) + : noopRelease; if (!releaseBodyRead) { writeJsonResponse(res, 503, { error: "bridge_busy" }); return; @@ -2690,7 +2783,7 @@ function runDuplexGateway() { // every path that never commits to a send. let handedOffToSend = false; try { - const requestBodyBuffer = await readBody(req); + const requestBodyBuffer = await readBody(req, declaredBytes); if (unavailable) { writeJsonResponse(res, 503, { error: "bridge_unavailable" }); return; @@ -2730,7 +2823,7 @@ function runDuplexGateway() { return; } // Retained-peak note. readBody preallocates one buffer sized to - // maxBodyBytes and returns a view over it, so no second (concatenated) + // declaredBytes 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 @@ -2742,20 +2835,18 @@ function runDuplexGateway() { // 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 body buffer <= declaredBytes // 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. + // and admission charges exactly that sum + // (declaredBytes + DUPLEX_GATEWAY_BODY_READ_RESERVE_EXTRA_BYTES) per + // read against the 64 MiB inflight bound, so a small request no longer + // charges the full 10 MiB body cap and a large 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. handedOffToSend = true; let sendDone = false; let responseDone = false;