diff --git a/doc/observability.md b/doc/observability.md index 3d45b420fd..9d7939a9c6 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -754,18 +754,49 @@ To add a name or an enum value, extend the literal constant in ### Known behavior: aggregate retained body bytes -The HTTP/2 bridge bounds retained body bytes for one route only. Each route -holds up to 8,388,608 bytes (8 MiB) at its own peak (see -`HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS` in `http2-bridge-server.ts`). The host -process admits up to 128 concurrent routes (see -`DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES` in `plugin-worker-manager.ts`). The -process can therefore retain up to 1,073,741,824 bytes (1 GiB) of body data -across every route at the same time. +Each HTTP/2 bridge route holds up to 168,820,736 bytes (161 MiB) at its own +peak (see `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS` in `http2-bridge-server.ts`). +The host process admits up to 128 concurrent routes (see +`DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES` in `plugin-worker-manager.ts`). Those +two figures alone would let the process retain up to 21,609,054,208 bytes +(about 20.1 GiB) of body data across every route at the same time. + +The process does not reach that figure, on two levels. +`HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES` (`http2-bridge-server.ts`) enforces a +real, live ledger: 1,073,741,824 bytes (1 GiB) across every route, not merely +an accepted paper ceiling. Every HTTP/2 stream creates one `BridgeBodyReservation` owner over +its lifetime, and every source-level full-body buffer that stream retains — +its request-body chunk array, the concatenated request body, the +response-body chunk array, and the concatenated response body — reserves +against that one owner before it allocates. A reservation that would pass the +process total is denied before it copies anything, and the host answers 503 +instead of accepting the body. The reservation stays live for the response +body until the HTTP/2 write actually finishes flowing to the peer or the +stream closes, not merely until the write call returns, so a slow or +backpressured peer cannot hold response bytes in memory the ledger no longer +counts. + +`HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES` adds a second, per-route ledger on top of +that process-wide one: each route's own reservations also check a ceiling +scoped to that one route (its own 168,820,736-byte peak from above), so one +busy or malicious route can pass its own ceiling and get denied with a 503, +but it can never spend the whole process-wide total and deny every sibling +route admission. This accounting covers source-level full-body buffers only: +internal Node.js and Undici copies (socket buffers, HTTP/2 frame buffers, +decompression buffers) stay outside it. + +The generated gateway process inside the sandbox (`getSandboxCallbackBridgeServerSource` +in `sandbox-callback-bridge.ts`) enforces its own separate ledger, independent +of the two host-side ledgers above: each side bounds only the memory in its +own process. `readBodyBytes` reserves a request body's chunk bytes as they +arrive, then reserves the concatenated buffer's own byte count before +`Buffer.concat` allocates it, against a ceiling of `maxBodyBytes * 8` (4 +concurrent bodies, each counted twice for its two live copies). A denied +reservation answers 503 with no forward call. Each request handler releases +its own reservation once the whole request settles: a completed response, a +thrown error, a client abort, or a deadline timeout all reach the same +release call. -This is accepted, known behavior. The process tracks no aggregate byte -ledger across routes: a per-route bound stops one busy route from starving -another route's own budget, but the host enforces no smaller ceiling on the -sum across every route. Keep every dimension low-cardinality and free of user content. ### Shared skill preparation diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 0b3c5134af..ce1d9f6faf 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -10,6 +10,13 @@ import { promisify } from "node:util"; import { afterEach, describe, expect, it, vi } from "vitest"; import { getSandboxDuplexGatewayCodecSource } from "./sandbox-callback-bridge.js"; +import { + createBridgeBodyReservation, + getBridgeBodyReservedBytesForTest, + resetBridgeBodyReservationsForTest, + HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS, + HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES, +} from "./http2-bridge-server.js"; import { __duplexReadinessTesting, @@ -3177,10 +3184,10 @@ describe("sandbox adapter execution targets", () => { */ function http2TestRequest( session: http2.ClientHttp2Session, - request: { method: string; path: string; headers?: Record; body?: string }, - ): Promise<{ status: number; headers: Record; body: string }> { + request: { method: string; path: string; headers?: Record; body?: Buffer }, + ): Promise<{ status: number; headers: Record; body: Buffer }> { return new Promise((resolve, reject) => { - const body = Buffer.from(request.body ?? "", "utf8"); + const body = request.body ?? Buffer.alloc(0); const stream = session.request( { ":method": request.method, ":path": request.path, ...request.headers }, { endStream: body.length === 0 }, @@ -3197,13 +3204,35 @@ describe("sandbox adapter execution targets", () => { } }); stream.on("data", (chunk: Buffer) => chunks.push(chunk)); - stream.once("end", () => resolve({ status, headers, body: Buffer.concat(chunks).toString("utf8") })); + stream.once("end", () => resolve({ status, headers, body: Buffer.concat(chunks) })); stream.once("error", (error) => reject(error)); if (body.length > 0) stream.end(body); else if (!stream.writableEnded) stream.end(); }); } + // Fixed binary payload for an attachment-content download. Byte 0x89 opens + // the PNG signature and is not valid UTF-8 on its own, so a round trip + // through a text decode step would corrupt it. + const ATTACHMENT_DOWNLOAD_BYTES = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0xff, 0x00, 0x7f]); + + // Build a minimal multipart/form-data request body with one binary file + // part, plus the matching `content-type` header value. + function buildMultipartAttachmentUpload(fileBytes: Buffer): { body: Buffer; contentType: string } { + const boundary = "paperclip-test-boundary"; + const head = Buffer.from( + `--${boundary}\r\n` + + `Content-Disposition: form-data; name="file"; filename="upload.bin"\r\n` + + `Content-Type: application/octet-stream\r\n\r\n`, + "utf8", + ); + const tail = Buffer.from(`\r\n--${boundary}--\r\n`, "utf8"); + return { + body: Buffer.concat([head, fileBytes, tail]), + contentType: `multipart/form-data; boundary=${boundary}`, + }; + } + // Start a host API server that records each forwarded request, so a test can // assert the real token and the run id reach the host, or that a rejected // request never forwards. @@ -3215,6 +3244,7 @@ describe("sandbox adapter execution targets", () => { auth: string | null; runId: string | null; headers: Record; + body: Buffer; }>; close: () => Promise; }> { @@ -3224,21 +3254,35 @@ describe("sandbox adapter execution targets", () => { auth: string | null; runId: string | null; headers: Record; + body: Buffer; }> = []; const server = createServer((req, res) => { const headers: Record = {}; for (const [key, value] of Object.entries(req.headers)) { if (typeof value === "string") headers[key] = value; } - requests.push({ - method: req.method ?? "GET", - url: req.url ?? "/", - auth: req.headers.authorization ?? null, - runId: typeof req.headers["x-paperclip-run-id"] === "string" ? req.headers["x-paperclip-run-id"] : null, - headers, + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + requests.push({ + method: req.method ?? "GET", + url: req.url ?? "/", + auth: req.headers.authorization ?? null, + runId: typeof req.headers["x-paperclip-run-id"] === "string" ? req.headers["x-paperclip-run-id"] : null, + headers, + body: Buffer.concat(chunks), + }); + // An attachment-content download answers with a binary body, so a + // test can assert the bytes reach the caller unchanged. Every other + // route keeps the fixed JSON acknowledgement. + if (req.method === "GET" && /^\/api\/attachments\/[^/]+\/content$/.test(req.url ?? "")) { + res.writeHead(200, { "content-type": "application/octet-stream" }); + res.end(ATTACHMENT_DOWNLOAD_BYTES); + return; + } + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); }); - res.writeHead(200, { "content-type": "application/json" }); - res.end(JSON.stringify({ ok: true })); }); await new Promise((resolve, reject) => { server.once("error", reject); @@ -3298,6 +3342,23 @@ describe("sandbox adapter execution targets", () => { expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1"); const fallback = counters.find((c) => c.metric === DUPLEX_COUNTER_FALLBACK_TOTAL); expect(fallback?.dimensions.fallback_reason).toBe("channel_open_failed"); + + // The queue fallback keeps the 415 gate for the attachment upload path: + // it never admits a binary body, and it never forwards the request to + // the host. + const uploadResponse = await fetch( + `${bridge!.env.PAPERCLIP_API_URL}/api/companies/co-1/issues/issue-1/attachments`, + { + method: "POST", + headers: { + authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`, + "content-type": "application/octet-stream", + }, + body: Buffer.from([0x50, 0x4b, 0x03, 0x04]), + }, + ); + expect(uploadResponse.status).toBe(415); + expect(api.requests).toHaveLength(0); } finally { await bridge?.stop(); await api.close(); @@ -3762,7 +3823,7 @@ describe("sandbox adapter execution targets", () => { method: "POST", path: "/api/secret-admin-route", headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, - body: JSON.stringify({ escalate: true }), + body: Buffer.from(JSON.stringify({ escalate: true }), "utf8"), }); expect(forbidden.status).toBe(403); expect(api.requests).toHaveLength(0); @@ -3784,6 +3845,39 @@ describe("sandbox adapter execution targets", () => { runId: "run-http2-403", }); expect(api.requests[0].headers["x-not-allowed"]).toBeUndefined(); + + // The two attachment routes are admitted on the http2 path, and a + // binary body reaches the host and returns unchanged in both + // directions: no re-encoding step touches the multipart upload or the + // binary download. + const uploadBytes = Buffer.from([0x00, 0x01, 0xff, 0x7f, 0x80, 0x0d, 0x0a]); + const upload = buildMultipartAttachmentUpload(uploadBytes); + const uploadResponse = await http2TestRequest(sessionRef.current!, { + method: "POST", + path: "/api/companies/co-1/issues/issue-1/attachments", + headers: { + authorization: `Bearer ${bridgeToken}`, + "content-type": upload.contentType, + }, + body: upload.body, + }); + expect(uploadResponse.status).toBe(200); + expect(api.requests).toHaveLength(2); + expect(api.requests[1]).toMatchObject({ + method: "POST", + url: "/api/companies/co-1/issues/issue-1/attachments", + }); + expect(api.requests[1].headers["content-type"]).toBe(upload.contentType); + expect(api.requests[1].body.equals(upload.body)).toBe(true); + + const downloadResponse = await http2TestRequest(sessionRef.current!, { + method: "GET", + path: "/api/attachments/att-1/content", + headers: { authorization: `Bearer ${bridgeToken}` }, + }); + expect(downloadResponse.status).toBe(200); + expect(api.requests).toHaveLength(3); + expect(downloadResponse.body.equals(ATTACHMENT_DOWNLOAD_BYTES)).toBe(true); } finally { sessionRef.current?.close(); await bridge?.stop(); @@ -3791,6 +3885,83 @@ describe("sandbox adapter execution targets", () => { } }, 20000); + it("test_http2_forward_preserves_request_and_response_bytes", async () => { + // Byte 0xC3 opens a two-byte UTF-8 sequence; 0x28 is not a valid + // continuation byte, so this body is not valid UTF-8. The forward path + // must carry these exact bytes on the way in, and the host's own + // response bytes on the way out, with no re-encoding step on either leg. + const malformedBytes = Buffer.from([0x7b, 0x22, 0x61, 0x22, 0x3a, 0xc3, 0x28, 0x7d]); + const receivedRequestBodies: Buffer[] = []; + const echoServer = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + receivedRequestBodies.push(Buffer.concat(chunks)); + res.writeHead(200, { "content-type": "application/octet-stream" }); + res.end(malformedBytes); + }); + }); + await new Promise((resolve, reject) => { + echoServer.once("error", reject); + echoServer.listen(0, "127.0.0.1", () => resolve()); + }); + const echoAddress = echoServer.address(); + if (!echoAddress || typeof echoAddress === "string") { + throw new Error("Expected the echo server to listen on a TCP port."); + } + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-raw-bytes-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-raw-bytes", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: `http://127.0.0.1:${echoAddress.port}`, + enableSandboxDuplexBridge: true, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + + const response = await http2TestRequest(sessionRef.current!, { + method: "POST", + path: "/api/issues/issue-1/comments", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/octet-stream" }, + body: malformedBytes, + }); + + expect(response.status).toBe(200); + expect(receivedRequestBodies).toHaveLength(1); + expect(receivedRequestBodies[0]?.equals(malformedBytes)).toBe(true); + expect(response.body.equals(malformedBytes)).toBe(true); + } finally { + sessionRef.current?.close(); + await bridge?.stop(); + await new Promise((resolve) => echoServer.close(() => resolve())); + } + }, 20000); + // One recording telemetry recorder. It captures every span, counter, and event // the fixed duplex surface produces, so a test asserts the exact names, // dimensions, and values. An optional `failEvery` flag makes every method throw, @@ -4234,7 +4405,7 @@ describe("sandbox adapter execution targets", () => { method: "POST", path: `/api/issues/${ROUTE_SENTINEL}/comments?secret=${QUERY_SENTINEL}`, headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, - body: JSON.stringify({ body: BODY_SENTINEL }), + body: Buffer.from(JSON.stringify({ body: BODY_SENTINEL }), "utf8"), }); expect(response.status).toBe(200); await waitForCondition( @@ -5350,6 +5521,11 @@ describe("sandbox adapter execution targets", () => { // failed, so a retry could double-apply it. The host answers a // non-retryable 504 with the indeterminate marker instead — the same // rule `forwardBridgeRequest` already applies on every transport. + // + // `maxBodyBytes: 1` below now bounds the request body too, since the + // host and the gateway share one resolved ceiling: this request carries + // no body, so only the mock host's own response — comfortably over one + // byte — trips the size check this test exists to force. const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-unsafe-indeterminate-")); cleanupDirs.push(rootDir); const remoteCwd = path.join(rootDir, "workspace"); @@ -5389,7 +5565,6 @@ describe("sandbox adapter execution targets", () => { method: "POST", path: "/api/issues/issue-1/comments", headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, - body: JSON.stringify({ body: "hello" }), }); expect(response.status).toBe(504); expect(response.headers["x-paperclip-bridge-outcome"]).toBe("indeterminate"); @@ -5481,6 +5656,509 @@ describe("sandbox adapter execution targets", () => { } }, 20000); + it("test_a_denied_response_chunk_cancels_the_reader_before_it_copies_the_chunk", async () => { + // Fill the process ledger so only a sliver of headroom remains, then let + // the mock host answer with a response chunk far bigger than that + // sliver. `readBridgeForwardResponseBody` (`execution-target.ts`) must + // deny that chunk's reservation, cancel its reader (closing the + // outbound connection to the mock host), and retain no copy of it. + resetBridgeBodyReservationsForTest(); + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true); + + let resolveHostConnectionClosed: (() => void) | undefined; + const hostConnectionClosed = new Promise((resolve) => { + resolveHostConnectionClosed = resolve; + }); + const api = createServer((_req, res) => { + res.writeHead(200, { "content-type": "application/octet-stream" }); + // Far bigger than the 100 bytes of headroom the filler above left: + // this one chunk alone must pass the process ceiling. + res.write(Buffer.alloc(1_000, "a")); + res.on("close", () => resolveHostConnectionClosed!()); + }); + await new Promise((resolve, reject) => { + api.once("error", reject); + api.listen(0, "127.0.0.1", () => resolve()); + }); + const apiAddress = api.address(); + if (!apiAddress || typeof apiAddress === "string") { + throw new Error("Expected the mock host server to listen on a TCP port."); + } + const apiOrigin = `http://127.0.0.1:${apiAddress.port}`; + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-denied-response-chunk-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-denied-response-chunk", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: apiOrigin, + enableSandboxDuplexBridge: true, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + // A capacity denial reaches the client as the retryable 503 the + // HTTP/2 bridge server's own capacity-denial path answers + // (`forwardBridgeRequest` rethrows `BridgeProcessCapacityError` before + // the method-safety classification runs), not the generic 502 that + // classification would give any other response-body read fault. + const response = await http2TestRequest(sessionRef.current!, { + method: "GET", + path: "/api/agents/me", + headers: { authorization: `Bearer ${bridgeToken}` }, + }); + expect(response.status).toBe(503); + // The reader actually cancelled: the mock host observes its + // connection close, instead of staying open with the chunk + // unacknowledged. + await hostConnectionClosed; + } finally { + sessionRef.current?.close(); + await bridge?.stop(); + await new Promise((resolve) => api.close(() => resolve())); + // The denied chunk was never retained: releasing the filler is the + // only release this test needs to reach zero. If the denied response + // copy had reserved anything despite being denied, this would be + // nonzero. + filler.release(); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } + }, 20000); + + it("test_a_denied_concatenated_response_body_never_allocates_the_copy", async () => { + // Leave room for the one response chunk but not for the second, + // concatenated copy `readBridgeForwardResponseBody` (`execution-target.ts`) + // builds from it: a correct reader checks the reservation before + // `Buffer.concat` allocates the copy, so the denied concatenated copy + // must never call `Buffer.concat` at all. The same denial must also + // cancel the upstream response reader, instead of leaving it open after + // the throw. + resetBridgeBodyReservationsForTest(); + const chunkBytes = 200_000; + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - Math.floor(chunkBytes * 1.5))).toBe(true); + + const api = createServer((_req, res) => { + res.writeHead(200, { "content-type": "application/octet-stream" }); + res.end(Buffer.alloc(chunkBytes, "a")); + }); + await new Promise((resolve, reject) => { + api.once("error", reject); + api.listen(0, "127.0.0.1", () => resolve()); + }); + const apiAddress = api.address(); + if (!apiAddress || typeof apiAddress === "string") { + throw new Error("Expected the mock host server to listen on a TCP port."); + } + const apiOrigin = `http://127.0.0.1:${apiAddress.port}`; + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-denied-response-concat-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-denied-response-concat", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: apiOrigin, + enableSandboxDuplexBridge: true, + }); + const concatSpy = vi.spyOn(Buffer, "concat"); + const readerCancelSpy = vi.spyOn(ReadableStreamDefaultReader.prototype, "cancel"); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + // A capacity denial must reach the client as the retryable 503 the + // HTTP/2 bridge server's own capacity-denial path answers, not the + // generic 502 the method-safety classification would otherwise apply + // to any other response-body read fault (`forwardBridgeRequest` + // rethrows `BridgeProcessCapacityError` before that classification + // runs). This reads the response with a plain string accumulator, not + // `Buffer.concat`, so the spy below counts only the calls the bridge + // code under test makes. + const response = await new Promise<{ status: number; body: string }>((resolve, reject) => { + const stream = sessionRef.current!.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + let status = 0; + let body = ""; + stream.setEncoding("utf8"); + stream.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + stream.on("data", (chunk) => (body += chunk)); + stream.once("end", () => resolve({ status, body })); + stream.once("error", reject); + stream.end(); + }); + expect(response.status).toBe(503); + expect(JSON.parse(response.body)).toEqual({ + error: "The bridge host reached its reserved process body byte ceiling. Retry later.", + }); + // The GET request itself carries no body, so the host's own read of + // that empty request body still calls `Buffer.concat` on an empty + // array — that call is unrelated to this test. No call may carry any + // response byte, since the denied concatenated response copy must + // never allocate. + for (const [chunks] of concatSpy.mock.calls) { + expect((chunks as Buffer[]).reduce((sum, chunk) => sum + chunk.length, 0)).toBe(0); + } + // The denied concatenated-body reservation must cancel the upstream + // response reader before it throws, instead of leaving it open. + expect(readerCancelSpy).toHaveBeenCalled(); + } finally { + readerCancelSpy.mockRestore(); + concatSpy.mockRestore(); + sessionRef.current?.close(); + await bridge?.stop(); + await new Promise((resolve) => api.close(() => resolve())); + // The denied concatenated copy reserved nothing: releasing the filler + // is the only release this test needs to reach zero. + filler.release(); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } + }, 20000); + + it("test_a_denied_response_chunk_on_a_mutating_method_is_indeterminate_not_retryable", async () => { + // A POST may already have committed on the host by the time the + // response-body read hits the process capacity ceiling: the host + // delivered response headers before the read even starts. Unlike the + // safe-method case above, this denial must not reach the client as a + // retryable 503 — a caller that retries would apply the mutation twice. + // It must fall through to the same non-retryable indeterminate 504 any + // other response-body read fault on a mutating method gets. + resetBridgeBodyReservationsForTest(); + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true); + + const api = createServer((_req, res) => { + res.writeHead(200, { "content-type": "application/octet-stream" }); + // Far bigger than the 100 bytes of headroom the filler above left. + res.end(Buffer.alloc(1_000, "a")); + }); + await new Promise((resolve, reject) => { + api.once("error", reject); + api.listen(0, "127.0.0.1", () => resolve()); + }); + const apiAddress = api.address(); + if (!apiAddress || typeof apiAddress === "string") { + throw new Error("Expected the mock host server to listen on a TCP port."); + } + const apiOrigin = `http://127.0.0.1:${apiAddress.port}`; + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-denied-response-mutation-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-denied-response-mutation", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: apiOrigin, + enableSandboxDuplexBridge: true, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + const response = await http2TestRequest(sessionRef.current!, { + method: "POST", + path: "/api/issues/issue-1/comments", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, + }); + expect(response.status).toBe(504); + expect(response.headers["x-paperclip-bridge-outcome"]).toBe("indeterminate"); + expect(JSON.parse(response.body.toString("utf8"))).toMatchObject({ + outcome: "indeterminate", + retryable: false, + }); + } finally { + sessionRef.current?.close(); + await bridge?.stop(); + await new Promise((resolve) => api.close(() => resolve())); + filler.release(); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } + }, 20000); + + it("test_concurrent_request_and_response_bodies_never_pass_the_process_ceiling", async () => { + resetBridgeBodyReservationsForTest(); + const bodyBytes = 2 * 1024 * 1024; + const requestBody = Buffer.alloc(bodyBytes, "a"); + const samples: number[] = []; + const api = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + // Sampled while this stream's own request-body copies are still + // live and at least one sibling stream may also be mid-flight: the + // real, concurrent, multi-stream shape this test exists to prove. + samples.push(getBridgeBodyReservedBytesForTest()); + res.writeHead(200, { "content-type": "application/octet-stream" }); + res.end(Buffer.alloc(bodyBytes, "b")); + }); + }); + await new Promise((resolve, reject) => { + api.once("error", reject); + api.listen(0, "127.0.0.1", () => resolve()); + }); + const apiAddress = api.address(); + if (!apiAddress || typeof apiAddress === "string") { + throw new Error("Expected the mock host server to listen on a TCP port."); + } + const apiOrigin = `http://127.0.0.1:${apiAddress.port}`; + + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-ceiling-aggregate-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-ceiling-aggregate", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: apiOrigin, + enableSandboxDuplexBridge: true, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + + const responses = await Promise.all( + Array.from({ length: HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS }, () => + http2TestRequest(sessionRef.current!, { + method: "POST", + path: "/api/issues/abc/comments", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/octet-stream" }, + body: requestBody, + }), + ), + ); + for (const response of responses) { + expect(response.status).toBe(200); + expect(response.body.byteLength).toBe(bodyBytes); + } + expect(samples).toHaveLength(HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS); + for (const sample of samples) { + expect(sample).toBeGreaterThan(0); + expect(sample).toBeLessThanOrEqual(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES); + } + // Every stream's owner released once its forward settled. + await waitForCondition( + () => getBridgeBodyReservedBytesForTest() === 0, + "the process reservation total to return to zero", + 2_000, + ); + } finally { + sessionRef.current?.close(); + await bridge?.stop(); + await new Promise((resolve) => api.close(() => resolve())); + } + }, 20000); + + it("test_the_host_denies_a_body_over_the_resolved_limit_not_only_the_gateway", async () => { + const api = await startRecordingApiServer(); + const sessionRef: { current: http2.ClientHttp2Session | null } = { current: null }; + let bridgeToken = ""; + const { runner } = makeHttp2SelectionRunner((ctx) => { + bridgeToken = ctx.bridgeToken; + ctx.emitReady(); + sessionRef.current = ctx.connectHttp2(); + }); + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-host-body-limit-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + await mkdir(remoteCwd, { recursive: true }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "daytona", + remoteCwd, + timeoutMs: 30_000, + runner, + effectiveCapabilities: duplexCapabilities(true), + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-host-body-limit", + target, + runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"), + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: api.origin, + enableSandboxDuplexBridge: true, + // A ceiling far under the body this test sends, resolved for this + // run. `createHttp2BridgeServer()` must receive this same resolved + // value, so the host itself enforces it on the raw wire — this test + // talks to the host directly, with no sandbox-side gateway script in + // between to enforce anything on its own. + maxBodyBytes: 100, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1"); + await waitForCondition(() => sessionRef.current !== null, "the http2 client session to open", 4000); + const response = await http2TestRequest(sessionRef.current!, { + method: "POST", + path: "/api/issues/abc/comments", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, + body: Buffer.alloc(1_000, "a"), + }); + // The oversized body never reaches the host API: the resolved ceiling + // rejects it before any forward call runs, so the mock host records + // no request. (The host's own size-violation path resets the stream + // before it can write a status line, so the client sees no ordinary + // response — this test asserts the one thing that channel does prove: + // the forward call itself never ran.) + expect(api.requests).toHaveLength(0); + expect(response.status).not.toBe(200); + } finally { + sessionRef.current?.close(); + await bridge?.stop(); + await api.close(); + } + }, 20000); + + it("test_the_queue_transport_still_forwards_with_no_reservation_owner", async () => { + // The queue transport's `handleRequest` callback + // (`execution-target.ts`'s queue callback) calls `forwardBridgeRequest` + // with no `reservation` option at all. `readBridgeForwardResponseBody` + // must behave exactly as it did before that option existed: it still + // forwards the request and still returns the host's body. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-queue-no-reservation-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex"); + await mkdir(runtimeRootDir, { recursive: true }); + + const apiServer = createServer((req, res) => { + res.writeHead(201, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, echoedMethod: req.method })); + }); + await new Promise((resolve, reject) => { + apiServer.once("error", reject); + apiServer.listen(0, "127.0.0.1", () => resolve()); + }); + const address = apiServer.address(); + if (!address || typeof address === "string") { + throw new Error("Expected the queue-transport test API server to listen on a TCP port."); + } + + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "e2b", + environmentId: "env-1", + leaseId: "lease-1", + remoteCwd, + runner: createLocalSandboxRunner(), + timeoutMs: 30_000, + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-queue-no-reservation", + target, + runtimeRootDir, + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: `http://127.0.0.1:${address.port}`, + }); + try { + expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1"); + const response = await fetch(`${bridge!.env.PAPERCLIP_API_URL}/api/issues/abc/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`, + "content-type": "application/json", + }, + body: JSON.stringify({ body: "hello" }), + }); + expect(response.status).toBe(201); + expect(await response.json()).toEqual({ ok: true, echoedMethod: "POST" }); + } finally { + await bridge?.stop(); + await new Promise((resolve) => apiServer.close(() => resolve())); + } + }); + // --------------------------------------------------------------------------- // Real-PTY replay. // diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index ccc4a9e2ab..dcede3ea14 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -35,6 +35,7 @@ import { createSandboxCallbackBridgeAsset, createSandboxCallbackBridgeToken, DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES, + HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT, SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE, sandboxCallbackBridgeDirectories, @@ -45,6 +46,8 @@ import { } from "./sandbox-callback-bridge.js"; import { createHttp2BridgeServer, + BridgeProcessCapacityError, + type BridgeBodyReservation, type Http2BridgeForwardHandler, } from "./http2-bridge-server.js"; import { @@ -87,12 +90,18 @@ import type { RunnerIngressEndpoint } from "./runner-connectivity.js"; export type { RuntimeProgressSink } from "./runtime-progress.js"; -export function postedIssueCommentLogMarker(method: string, requestPath: string, status: number, body: string) { +export function postedIssueCommentLogMarker( + method: string, + requestPath: string, + status: number, + body: Buffer | string, +) { if (method !== "POST" || !/^\/api\/issues\/[^/]+\/comments$/.test(requestPath) || status < 200 || status >= 300) { return null; } + const bodyText = typeof body === "string" ? body : body.toString("utf8"); try { - const parsed = JSON.parse(body) as { id?: unknown }; + const parsed = JSON.parse(bodyText) as { id?: unknown }; return typeof parsed.id === "string" && parsed.id.length > 0 ? `comment id: ${parsed.id}\n` : null; } catch { return null; @@ -646,12 +655,19 @@ function preferredSandboxShell(target: AdapterSandboxExecutionTarget): "bash" | type AdapterCommandCapableExecutionTarget = AdapterSshExecutionTarget | AdapterSandboxExecutionTarget; +// The Secure Shell command runner's own output buffer. This value used to +// derive from the bridge body limit (`DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES +// * 4`), so a bridge limit rise silently grew it too. It now stands on its +// own local constant, independent of the bridge body limit, so a later +// bridge limit change never resizes this buffer as a side effect. +const SSH_COMMAND_MAX_BUFFER_BYTES = 1024 * 1024; + function adapterExecutionTargetCommandRunner(target: AdapterCommandCapableExecutionTarget): CommandManagedRuntimeRunner { if (target.transport === "ssh") { return createSshCommandManagedRuntimeRunner({ spec: target.spec, defaultCwd: target.remoteCwd, - maxBufferBytes: DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES * 4, + maxBufferBytes: SSH_COMMAND_MAX_BUFFER_BYTES, }); } return requireSandboxRunner(target); @@ -1617,15 +1633,30 @@ function bridgeResponseBodyLimitError(maxBodyBytes: number): Error { } /** - * Read the forward response body into a string. The per-request `maxBodyBytes` - * limit rejects a body larger than the configured per-request ceiling. + * Read the forward response body into a `Buffer`, with no text decoding. The + * per-request `maxBodyBytes` limit rejects a body larger than the configured + * per-request ceiling. * - * This function reserves no process-wide byte budget: it enforces only the - * one request's own ceiling. See the "Known behavior: aggregate retained - * body bytes" section in `doc/observability.md` for the accepted aggregate - * ceiling this leaves across every concurrent route. + * When the caller passes a `reservation`, this reserves each chunk's bytes + * against it immediately after `reader.read()` yields the chunk, and before + * `Buffer.from(value)` copies it — the allocation happens inside that + * expression, so reserving only before the later `chunks.push` would let the + * copy happen first. It also reserves the concatenated buffer's own byte + * count before `Buffer.concat` allocates it: the chunk array and the + * concatenated buffer are two separate live copies. A denied reservation + * cancels the reader and throws {@link BridgeProcessCapacityError}, copying + * no further chunk. This function never releases the reservation; the + * stream owner that created it does, once the whole forward call settles. + * + * With no `reservation`, this function enforces only the one request's own + * ceiling, exactly as it did before this parameter existed — the queue + * transport calls it with no reservation, and its behavior must not change. */ -async function readBridgeForwardResponseBody(response: Response, maxBodyBytes: number): Promise { +async function readBridgeForwardResponseBody( + response: Response, + maxBodyBytes: number, + reservation?: BridgeBodyReservation, +): Promise { const rawContentLength = response.headers.get("content-length"); if (rawContentLength) { const contentLength = Number.parseInt(rawContentLength, 10); @@ -1635,7 +1666,7 @@ async function readBridgeForwardResponseBody(response: Response, maxBodyBytes: n } if (!response.body) { - return ""; + return Buffer.alloc(0); } const reader = response.body.getReader(); @@ -1651,9 +1682,17 @@ async function readBridgeForwardResponseBody(response: Response, maxBodyBytes: n await reader.cancel().catch(() => undefined); throw bridgeResponseBodyLimitError(maxBodyBytes); } + if (reservation && !reservation.reserve(chunkBytes)) { + await reader.cancel().catch(() => undefined); + throw new BridgeProcessCapacityError(); + } chunks.push(Buffer.from(value)); } - return Buffer.concat(chunks, totalBytes).toString("utf8"); + if (reservation && !reservation.reserve(totalBytes)) { + await reader.cancel().catch(() => undefined); + throw new BridgeProcessCapacityError(); + } + return Buffer.concat(chunks, totalBytes); } const PROCESS_SESSION_PROXY_SCRIPT = "paperclip-process-session-proxy.mjs"; @@ -4099,14 +4138,24 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { path: string; query: string; headers: Record; - /** The file bridge passes the whole request body here as one string. */ - body?: string; + /** The file bridge passes the whole request body here as one string. + * The HTTP/2 bridge passes it as the raw `Buffer` it read off the wire. */ + body?: string | Buffer; }, signal?: AbortSignal, options?: { suppressDebugLog?: boolean; + /** + * The caller's stream reservation owner, if it has one. The HTTP/2 + * bridge passes the stream's own owner here, so the response body copy + * reserves against the same ceiling the request body copy already + * reserved against. The queue transport passes no owner, so its + * response-body read enforces only the per-request size ceiling, exactly + * as it did before this option existed. + */ + reservation?: BridgeBodyReservation; }, - ): Promise<{ status: number; headers: Record; body: string }> => { + ): Promise<{ status: number; headers: Record; body: Buffer }> => { const method = request.method.trim().toUpperCase() || "GET"; // The per-request debug log prints the method, the path, and the query. The // duplex path suppresses it, so no route or query rides a log line there. The @@ -4131,14 +4180,19 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { const timeoutSignal = AbortSignal.timeout(forwardTimeoutMs); const forwardSignal = signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal; // Build the request-body init. A GET or a HEAD carries no body. The file - // bridge passes the whole body as one string. + // bridge passes the whole body as one string; the HTTP/2 bridge passes it + // as a raw `Buffer`. Undici accepts a `Buffer` request body directly (a + // `Buffer` is an `ArrayBufferView`), so neither shape needs a conversion. + // The cast below only bridges a `BodyInit` typing gap: the DOM library + // type this project's ambient `RequestInit` resolves to excludes a + // `Buffer`, though Undici accepts one at runtime. const forwardInit: RequestInit = { method, headers, signal: forwardSignal, }; - if (method !== "GET" && method !== "HEAD" && typeof request.body === "string") { - forwardInit.body = request.body; + if (method !== "GET" && method !== "HEAD" && request.body !== undefined) { + forwardInit.body = request.body as BodyInit; } const response = await fetch(buildBridgeForwardUrl(hostApiUrl, request), forwardInit); if (emitDebugLog) { @@ -4159,10 +4213,23 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { // non-retryable 504 and marks the outcome indeterminate, exactly like an // aborted in-flight forward. The in-sandbox server maps the indeterminate 504 // to a non-retryable 409 for both the file bridge and the duplex broker. - let responseBody: string; + let responseBody: Buffer; try { - responseBody = await readBridgeForwardResponseBody(response, maxBodyBytes); + responseBody = await readBridgeForwardResponseBody(response, maxBodyBytes, options?.reservation); } catch (error) { + // A denied reservation is retryable capacity pressure for a safe + // method, not a body-read fault: rethrow it before the method-safety + // classification below runs, so it reaches the HTTP/2 bridge's own + // capacity-denial catch (which answers the retryable 503 and settles + // the stream) instead of this function turning it into a 502. For an + // unsafe (mutating) method, the host has already delivered response + // headers by this point, so it may already have committed the + // mutation. Rethrowing there too would let the retryable 503 reach a + // caller that repeats the request, applying the mutation twice. An + // unsafe method's capacity denial falls through to the same + // non-retryable indeterminate 504 any other response-body read fault + // gets below. + if (error instanceof BridgeProcessCapacityError && isSafeBridgeMethod(method)) throw error; if (isSafeBridgeMethod(method)) { // The method is safe, so a retry cannot double-apply a mutation. Return a // retryable 502 with no indeterminate marker, so the gateway passes it @@ -4170,9 +4237,12 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { return { status: 502, headers: { "content-type": "application/json" }, - body: JSON.stringify({ - error: error instanceof Error ? error.message : String(error), - }), + body: Buffer.from( + JSON.stringify({ + error: error instanceof Error ? error.message : String(error), + }), + "utf8", + ), }; } return { @@ -4181,11 +4251,14 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { "content-type": "application/json", "x-paperclip-bridge-outcome": "indeterminate", }, - body: JSON.stringify({ - error: error instanceof Error ? error.message : String(error), - outcome: "indeterminate", - retryable: false, - }), + body: Buffer.from( + JSON.stringify({ + error: error instanceof Error ? error.message : String(error), + outcome: "indeterminate", + retryable: false, + }), + "utf8", + ), }; } const commentMarker = postedIssueCommentLogMarker(method, request.path, response.status, responseBody); @@ -4387,10 +4460,10 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { path: request.pathname, query: request.query, headers: request.headers, - body: request.body.toString("utf8"), + body: request.body, }, request.signal, - { suppressDebugLog: true }, + { suppressDebugLog: true, reservation: request.reservation }, ); duplexObservability.recordRequest({ latencyMs: Date.now() - dispatchStartMs, outcome: "ok" }); return { status: result.status, headers: result.headers, body: result.body }; @@ -4403,6 +4476,13 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { const http2Server = createHttp2BridgeServer({ bridgeToken, forwardRequest: http2ForwardRequest, + routes: HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, + // The same resolved limit the launch environment hands the + // sandbox-side gateway (`PAPERCLIP_BRIDGE_MAX_BODY_BYTES`, + // below), so the host check and the gateway check enforce one + // value instead of the host silently falling back to the + // package default. + maxBodyBytes, onGoaway: () => recordHttp2Loss("session_goaway"), onSessionError: () => recordHttp2Loss("session_error"), }); @@ -4510,7 +4590,13 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { maxBodyBytes, getRuntimeParentContext: input.getRuntimeParentContext, runtimeSpan: input.runtimeSpan, - handleRequest: async (request, options) => forwardBridgeRequest(request, options?.signal), + // The queue transport writes the response body to a text file, so this + // is the one place the forward path decodes the response `Buffer` to a + // UTF-8 string. The queue's own on-wire behavior does not change. + handleRequest: async (request, options) => { + const result = await forwardBridgeRequest(request, options?.signal); + return { status: result.status, headers: result.headers, body: result.body.toString("utf8") }; + }, }); server = await startSandboxCallbackBridgeServer({ runner, diff --git a/packages/adapter-utils/src/http2-bridge-server.test.ts b/packages/adapter-utils/src/http2-bridge-server.test.ts index 5613620a38..c3acb87aa6 100644 --- a/packages/adapter-utils/src/http2-bridge-server.test.ts +++ b/packages/adapter-utils/src/http2-bridge-server.test.ts @@ -2,14 +2,19 @@ import { duplexPair } from "node:stream"; import type { Duplex } from "node:stream"; import http2 from "node:http2"; -import { describe, expect, it } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { buildHttp2BridgeForwardUrl, classifyStreamAgainstGoaway, + createBridgeBodyReservation, + createBridgeRouteBodyLedger, createHttp2BridgeServer, + getBridgeBodyReservedBytesForTest, parseCanonicalBridgeRequestPath, + resetBridgeBodyReservationsForTest, wrapDuplexChannelAsNodeDuplex, + BridgeProcessCapacityError, DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES, DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS, DEFAULT_HTTP2_BRIDGE_PING_STALL_MS, @@ -19,6 +24,8 @@ import { HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE, HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS, HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE, + HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES, + HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES, HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES, HTTP2_BRIDGE_MAX_SESSION_MEMORY, HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS, @@ -32,6 +39,8 @@ import { import { createSandboxHttp2BridgeGateway, DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES, + HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, + type SandboxCallbackBridgeRouteRule, } from "./sandbox-callback-bridge.js"; import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js"; @@ -82,6 +91,10 @@ interface TestPairOptions { requestBodyTimeoutMs?: number; requestBodyLifetimeCeilingMs?: number; closeGraceMs?: number; + capacityDenialSettleDeadlineMs?: number; + responseWriteSettleDeadlineMs?: number; + maxBodyBytes?: number; + routes?: readonly SandboxCallbackBridgeRouteRule[]; onGoaway?: (record: Http2BridgeGoawayRecord) => void; onSessionError?: (error: Error) => void; onSession?: (session: http2.ServerHttp2Session) => void; @@ -97,7 +110,7 @@ function bindTestServer(options: TestPairOptions = {}) { (async (request: Http2BridgeForwardRequest) => ({ status: 200, headers: { "content-type": "application/json" }, - body: JSON.stringify({ echoedMethod: request.method, echoedPath: request.pathname }), + body: Buffer.from(JSON.stringify({ echoedMethod: request.method, echoedPath: request.pathname }), "utf8"), })); const handle = createHttp2BridgeServer({ bridgeToken, @@ -107,6 +120,10 @@ function bindTestServer(options: TestPairOptions = {}) { requestBodyTimeoutMs: options.requestBodyTimeoutMs, requestBodyLifetimeCeilingMs: options.requestBodyLifetimeCeilingMs, closeGraceMs: options.closeGraceMs, + capacityDenialSettleDeadlineMs: options.capacityDenialSettleDeadlineMs, + responseWriteSettleDeadlineMs: options.responseWriteSettleDeadlineMs, + maxBodyBytes: options.maxBodyBytes, + routes: options.routes, onGoaway: options.onGoaway, onSessionError: options.onSessionError, onSession: options.onSession, @@ -130,9 +147,15 @@ function createTestPair(options: TestPairOptions = {}) { /** Open a raw HTTP/2 client session directly against one side of the pair, * bypassing the sandbox gateway. Some tests need direct stream control (an * explicit RST_STREAM, an explicit GOAWAY) the gateway's `forwardRequest` - * abstraction does not expose. */ -function connectRawClient(clientSide: Duplex): http2.ClientHttp2Session { - return http2.connect("http://bridge.internal", { createConnection: () => clientSide }); + * abstraction does not expose. `settings`, when given, rides the client's + * own initial SETTINGS frame — for example `{ initialWindowSize: 0 }` to + * deny the server any flow-control credit to send response bytes with, a + * deterministic stall independent of the fake transport's own buffering. */ +function connectRawClient( + clientSide: Duplex, + settings?: http2.Settings, +): http2.ClientHttp2Session { + return http2.connect("http://bridge.internal", { createConnection: () => clientSide, settings }); } /** Track whether `forwardRequest` ran, so a test can prove a denied or @@ -175,6 +198,13 @@ async function expectSessionStillServesARequest( } describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { + // Every test in this file shares the module-scope process reservation + // total. A reservation one test leaves unreleased would otherwise lower + // the ceiling every later test sees, so each test starts from zero. + beforeEach(() => { + resetBridgeBodyReservationsForTest(); + }); + it("test_one_session_over_a_fake_channel_forwards_a_request", async () => { const { handle, gateway } = createTestPair(); try { @@ -208,7 +238,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { return { status: 200, headers: {}, - body: JSON.stringify({ echoedPath: request.pathname }), + body: Buffer.from(JSON.stringify({ echoedPath: request.pathname }), "utf8"), }; }, }); @@ -316,7 +346,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { // Hold this one open long enough for the test to RST it. await new Promise((resolve) => setTimeout(resolve, 200)); } - return { status: 200, headers: {}, body: JSON.stringify({ path: request.pathname }) }; + return { status: 200, headers: {}, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") }; }, }); const rawClient = connectRawClient(clientSide); @@ -362,7 +392,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { // Hold the handler open, so the test controls exactly when the // client-side stream close happens relative to the forward. await forwardHeld; - return { status: 200, body: "{}" }; + return { status: 200, body: Buffer.from("{}", "utf8") }; }, }); const rawClient = connectRawClient(clientSide); @@ -414,7 +444,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { } else if (request.pathname === "/api/agents/survivor") { await survivorHeld; } - return { status: 200, body: JSON.stringify({ path: request.pathname }) }; + return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") }; }, }); const rawClient = connectRawClient(clientSide); @@ -491,7 +521,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { } request.signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); }); - return { status: 200, body: JSON.stringify({ path: request.pathname }) }; + return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") }; } finally { liveForwards -= 1; } @@ -579,7 +609,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { requestBodyTimeoutMs: 80, forwardRequest: async (request) => ({ status: 200, - body: JSON.stringify({ bodyLength: request.body.byteLength }), + body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8"), }), }); const rawClient = connectRawClient(clientSide); @@ -687,7 +717,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { requestBodyLifetimeCeilingMs: 5_000, forwardRequest: async (request) => ({ status: 200, - body: JSON.stringify({ bodyLength: request.body.byteLength }), + body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8"), }), }); const rawClient = connectRawClient(clientSide); @@ -728,7 +758,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { requestBodyLifetimeCeilingMs: 5_000, forwardRequest: async (request) => { forwarderTracker.markCalled(); - return { status: 200, body: JSON.stringify({ bodyLength: request.body.byteLength }) }; + return { status: 200, body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8") }; }, }); const rawClient = connectRawClient(clientSide); @@ -845,17 +875,877 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { }); it("test_the_host_body_budget_matches_the_stream_limit", () => { - // The multiplier counts every retained `Buffer` and string copy of one - // live forward's request and response body: four exact `Buffer` rows, - // plus two string rows. Each string row applies two bytes to each UTF-16 - // code unit of the body limit (`sandbox-callback-bridge.ts`), for an - // accounting peak of eight times the body limit for one live forward. + // The multiplier counts every retained `Buffer` copy of one live + // forward's request and response body: four exact `Buffer` rows. The + // forward path decodes no body to a string, so no row applies the + // two-bytes-per-UTF-16-code-unit string overhead any more. // `test_live_forward_work_never_passes_the_stream_limit` proves the // count of live forwards never passes `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS`, // so this multiplier bounds live forwards, not merely open streams. - expect( - HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 8 * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES, - ).toBe(8_388_608); + // `HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES` uses this exact same formula to + // enforce it as a real per-route cap, not merely a derived figure. + const expectedRouteBudget = + HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 4 * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES; + expect(expectedRouteBudget).toBe(168_820_736); + expect(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES).toBe(expectedRouteBudget); + }); + + describe("createBridgeBodyReservation", () => { + it("test_a_reservation_denies_a_chunk_that_passes_the_process_ceiling", () => { + const owner = createBridgeBodyReservation(); + try { + expect(owner.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES)).toBe(true); + expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES); + expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES); + + // One more byte passes the ceiling: denied, and it holds no bytes. + expect(owner.reserve(1)).toBe(false); + expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES); + expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES); + } finally { + owner.release(); + } + }); + + it("test_a_reservation_denies_the_concatenated_copy_that_passes_the_process_ceiling", () => { + const owner = createBridgeBodyReservation(); + try { + // The chunk-array reservation alone passes comfortably. + const chunkArrayBytes = Math.floor(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES * 0.6); + expect(owner.reserve(chunkArrayBytes)).toBe(true); + + // The concatenated `Buffer.concat` copy is a second, separate live + // copy of the same bytes: reserving it on top of the chunk array + // this owner already holds passes the ceiling, even though the + // chunk-array reservation alone did not. + expect(owner.reserve(chunkArrayBytes)).toBe(false); + expect(owner.heldBytes).toBe(chunkArrayBytes); + expect(getBridgeBodyReservedBytesForTest()).toBe(chunkArrayBytes); + } finally { + owner.release(); + } + }); + + it("test_a_reservation_releases_every_held_byte_one_time_only", () => { + const owner = createBridgeBodyReservation(); + expect(owner.reserve(1_000)).toBe(true); + expect(getBridgeBodyReservedBytesForTest()).toBe(1_000); + + owner.release(); + expect(owner.heldBytes).toBe(0); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + + // A second release must not double-subtract: the total must not go + // below zero. + owner.release(); + expect(owner.heldBytes).toBe(0); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + }); + }); + + describe("route isolation", () => { + it("test_a_route_ledger_denies_a_reservation_that_passes_its_own_ceiling_with_the_process_ceiling_still_open", () => { + // The process-wide ceiling has ample room (1 GiB); only this one + // route's own share is tight. A route-scoped owner must still deny + // the second reservation, proving the route ceiling is a real, + // independent check, not merely a reflection of the process total. + const routeLedger = createBridgeRouteBodyLedger(); + const owner = createBridgeBodyReservation(routeLedger); + try { + expect(owner.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true); + expect(routeLedger.reservedBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES); + + expect(owner.reserve(1)).toBe(false); + expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES); + expect(routeLedger.reservedBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES); + // The process-wide total only ever grew by what this owner actually + // holds: the denied byte reserved against neither total. + expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES); + } finally { + owner.release(); + } + }); + + it("test_one_route_at_its_own_ceiling_never_blocks_a_sibling_routes_reservation", () => { + // Two routes, two ledgers. Route A spends its own entire ceiling. + // Route B's reservation, against its own separate ledger, must still + // succeed: the process-wide total (1 GiB) has room for both routes' + // ceilings many times over, so only route isolation — not the shared + // total — could explain a denial here. + const routeLedgerA = createBridgeRouteBodyLedger(); + const routeLedgerB = createBridgeRouteBodyLedger(); + const ownerA = createBridgeBodyReservation(routeLedgerA); + const ownerB = createBridgeBodyReservation(routeLedgerB); + try { + expect(ownerA.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true); + expect(ownerA.reserve(1)).toBe(false); + + expect(ownerB.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true); + } finally { + ownerA.release(); + ownerB.release(); + } + }); + + it("test_two_bridge_server_routes_isolate_their_own_reservations_end_to_end", async () => { + // The end-to-end proof: while route A's own forward holds its own + // entire per-route ceiling in flight, a second, independent route's + // request must still succeed. Each `bindTestServer` call is its own + // route (its own `createHttp2BridgeServer` call, so its own route + // ledger). Route A's forward parks on `releaseRouteAHold` after it + // reserves, so its reservation stays live — not merely reserved and + // immediately released — for the whole time route B's request runs. + let markRouteAReserved: (() => void) | undefined; + const routeAReserved = new Promise((resolve) => { + markRouteAReserved = resolve; + }); + let releaseRouteAHold: (() => void) | undefined; + const routeAHold = new Promise((resolve) => { + releaseRouteAHold = resolve; + }); + const routeA = bindTestServer({ + forwardRequest: async (request) => { + // Reserve this route's own entire ceiling against its own ledger, + // simulating route A at its own documented peak. + if (!request.reservation.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)) { + throw new BridgeProcessCapacityError(); + } + markRouteAReserved!(); + await routeAHold; + return { status: 200 }; + }, + }); + const routeB = bindTestServer({ + forwardRequest: async () => ({ status: 200 }), + }); + const rawClientA = connectRawClient(routeA.clientSide); + const rawClientB = connectRawClient(routeB.clientSide); + try { + const pendingResponseA = expectSessionStillServesARequest(rawClientA, { + method: "GET", + path: "/api/agents/me", + token: routeA.bridgeToken, + }); + + // Wait until route A's forward actually holds its own ceiling — + // not merely until the request was sent — before checking route B. + await routeAReserved; + expect(getBridgeBodyReservedBytesForTest()).toBeGreaterThanOrEqual(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES); + + // Route A now holds its own entire per-route ceiling, live. The + // process-wide total still has headroom (1 GiB minus one 161 MiB + // route), so route B's independent request succeeds only if its own + // ledger is genuinely separate from route A's. + const responseB = await expectSessionStillServesARequest(rawClientB, { + method: "GET", + path: "/api/agents/me", + token: routeB.bridgeToken, + }); + expect(responseB.status).toBe(200); + + releaseRouteAHold!(); + const responseA = await pendingResponseA; + expect(responseA.status).toBe(200); + } finally { + rawClientA.close(); + rawClientB.close(); + await routeA.handle.close(); + await routeB.handle.close(); + } + }); + }); + + it("test_the_stream_handler_releases_its_reservation_on_completion_error_abort_timeout_and_close", async () => { + // Every stream, no matter how it ends, must leave the process-wide + // reservation total at zero: `handleStream`'s `finally` block releases + // exactly one owner exactly one time on every exit path. + + async function runCompletionCase(): Promise { + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async (request) => { + request.reservation.reserve(1_000); + return { status: 200, body: Buffer.from("{}", "utf8") }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(response.status).toBe(200); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + } + + async function runErrorCase(): Promise { + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async (request) => { + request.reservation.reserve(1_000); + throw new Error("forward handler fault"); + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(response.status).toBe(502); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + } + + async function runAbortCase(): Promise { + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async (request) => { + request.reservation.reserve(1_000); + await new Promise((resolve) => { + if (request.signal.aborted) { + resolve(); + return; + } + request.signal.addEventListener("abort", () => resolve(), { once: true }); + }); + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const stream = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + const streamClosed = new Promise((resolve) => { + stream.on("error", () => resolve()); + stream.on("close", () => resolve()); + }); + stream.end(); + await new Promise((resolve) => setTimeout(resolve, 20)); + stream.close(http2.constants.NGHTTP2_CANCEL); + await streamClosed; + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + } + + async function runTimeoutCase(): Promise { + const { handle, bridgeToken, clientSide } = bindTestServer({ + requestBodyTimeoutMs: 30, + forwardRequest: async () => ({ status: 200 }), + }); + const rawClient = connectRawClient(clientSide); + try { + await new Promise((resolve) => { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => resolve()); + req.on("close", () => resolve()); + req.write("partial-body"); + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + } + + async function runCloseCase(): Promise { + const { handle, bridgeToken, clientSide } = bindTestServer({ + requestBodyTimeoutMs: 60_000, + closeGraceMs: 30, + }); + const rawClient = connectRawClient(clientSide); + try { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + req.write("partial-body"); + await new Promise((resolve) => setTimeout(resolve, 20)); + await handle.close(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + } + } + + await runCompletionCase(); + await runErrorCase(); + await runAbortCase(); + await runTimeoutCase(); + await runCloseCase(); + }, 10_000); + + it("test_a_backpressured_response_holds_its_reservation_until_the_write_settles", async () => { + // `stream.end(body)` only queues `body` for asynchronous transmission. + // A client that never reads its response leaves those bytes in process + // memory well after `end()` returns, so the reservation covering them + // must stay held until the write actually settles, not merely until the + // call to `end()` returns. + const responseBytes = 4 * 1024 * 1024; + const responseBody = Buffer.alloc(responseBytes, 0x61); + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async (request) => { + // Mirror the real forward path: the handler reserves the response + // body it read from the sandbox target before returning it. + if (!request.reservation.reserve(responseBytes)) { + throw new BridgeProcessCapacityError(); + } + return { status: 200, body: responseBody }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + // Deliberately paused: no `resume()` and no `data` listener yet, so + // the client never issues the flow-control credit the host needs to + // finish writing a response this size. + const headersReceived = new Promise((resolve) => { + req.once("response", () => resolve()); + }); + req.end(); + await headersReceived; + // Give the host's `stream.end()` call, and its microtask queue, a + // turn to run — the write is now queued but the client still is not + // draining it. + await new Promise((resolve) => setImmediate(resolve)); + expect(getBridgeBodyReservedBytesForTest()).toBeGreaterThan(0); + + const drained = new Promise((resolve) => { + req.on("data", () => { + // Discard: this test only cares that the bytes left the host. + }); + req.once("end", () => resolve()); + }); + req.resume(); + await drained; + + // Give the host's `finally` block a turn to run after the write + // settles. + await new Promise((resolve) => setImmediate(resolve)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_peer_reset_holds_the_reservation_until_the_active_forward_settles", async () => { + let capturedRequest: Http2BridgeForwardRequest | undefined; + let releaseForward: (() => void) | undefined; + const forwardHeld = new Promise((resolve) => { + releaseForward = resolve; + }); + let forwardSettled = false; + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async (request) => { + capturedRequest = request; + // Stand in for a live response-body copy `readBridgeForwardResponseBody` + // (`execution-target.ts`) would reserve against this same owner. + request.reservation.reserve(10_000); + // The forward stays active past the peer's own reset: it settles + // only when the test releases it below, not merely when the abort + // signal fires — the same shape a real outbound `fetch` bound to + // `request.signal` has, since an abort does not settle the fetch + // promise synchronously. + await new Promise((resolve) => { + if (request.signal.aborted) { + resolve(); + return; + } + request.signal.addEventListener("abort", () => resolve(), { once: true }); + }); + await forwardHeld; + forwardSettled = true; + return { status: 200, body: Buffer.from("{}", "utf8") }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const stream = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + const streamClosed = new Promise((resolve) => { + stream.on("error", () => resolve()); + stream.on("close", () => resolve()); + }); + stream.end(); + // Give the request time to reach the server, enter `forwardRequest`, + // and land its reservation. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(capturedRequest).toBeDefined(); + expect(getBridgeBodyReservedBytesForTest()).toBe(10_000); + + // The peer resets the stream. The forward call is still pending. + stream.close(http2.constants.NGHTTP2_CANCEL); + await streamClosed; + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(capturedRequest!.signal.aborted).toBe(true); + // The owner still holds its bytes: the active forward has not + // settled yet, so `handleStream`'s `finally` has not released it. + expect(forwardSettled).toBe(false); + expect(getBridgeBodyReservedBytesForTest()).toBe(10_000); + + // Let the forward settle. The `finally` block releases the owner + // exactly one time, and the process total returns to zero. + releaseForward!(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(forwardSettled).toBe(true); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_stalled_body_releases_its_reservation_while_other_streams_hold_theirs", async () => { + const releaseByPath = new Map void>(); + const { handle, bridgeToken, clientSide } = bindTestServer({ + requestBodyTimeoutMs: 60, + forwardRequest: async (request) => { + // Stand in for a live response-body buffer: reserve, then hold the + // forward open until the test releases this exact path. + request.reservation.reserve(20_000); + await new Promise((resolve) => { + releaseByPath.set(request.pathname, resolve); + }); + return { status: 200, body: Buffer.from("{}", "utf8") }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const streamA = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/a", + authorization: `Bearer ${bridgeToken}`, + }); + const streamB = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/b", + authorization: `Bearer ${bridgeToken}`, + }); + // Drain each response as it arrives: `session.close()` later in this + // test's teardown waits for every stream to fully end on both sides, + // and an unread response can hold a stream open past that point. + streamA.resume(); + streamB.resume(); + streamA.end(); + streamB.end(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(40_000); + + // The third stream's request body stalls past the idle bound. It + // never reaches `forwardRequest`, so it never reserves anything of + // its own; its owner still holds zero bytes at the moment it releases. + const streamC = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + const streamCClosed = new Promise((resolve) => { + streamC.on("error", () => resolve()); + streamC.on("close", () => resolve()); + }); + streamC.write("partial-body"); + await streamCClosed; + await new Promise((resolve) => setTimeout(resolve, 20)); + + // The stalled stream's release changed nothing: the other two streams + // keep their reservations. + expect(getBridgeBodyReservedBytesForTest()).toBe(40_000); + + releaseByPath.get("/api/agents/a")?.(); + releaseByPath.get("/api/agents/b")?.(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_denied_stream_drains_its_body_and_holds_no_bytes", async () => { + const { handle, clientSide } = bindTestServer({ + forwardRequest: async () => ({ status: 200 }), + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await new Promise<{ status: number }>((resolve, reject) => { + // No `authorization` header: the token check denies this stream + // before its body is ever read. + const req = rawClient.request({ ":method": "POST", ":path": "/api/issues/abc/comments" }); + let status = 0; + req.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + req.on("data", () => undefined); + req.on("end", () => resolve({ status })); + req.on("error", reject); + req.write(Buffer.alloc(50_000, "a")); + req.end(); + }); + expect(response.status).toBe(401); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_request_body_reservation_denial_answers_503_on_the_wire", async () => { + // A denied request-body chunk must leave the stream alive long enough + // to answer 503 for real, on the wire — not merely destroy the stream + // and leave the caller with a bare reset. `respondJson` only queues the + // 503 write; the handler must wait for that write to settle before it + // destroys the stream, or the client can see a reset instead of the + // full response body asserted below. + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true); + const forwarderTracker = createForwarderCallTracker(); + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async () => { + forwarderTracker.markCalled(); + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await new Promise<{ status: number; body: string }>((resolve, reject) => { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + let status = 0; + let body = ""; + req.setEncoding("utf8"); + req.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => resolve({ status, body })); + req.on("error", reject); + // Ten times the 100 bytes of headroom the filler above left. + req.end(Buffer.alloc(1_000, "a")); + }); + expect(response.status).toBe(503); + expect(forwarderTracker.called).toBe(false); + // A raw reset delivers no body at all: parsing it here proves the + // full JSON response actually reached the client, not merely the + // status line. + expect(JSON.parse(response.body)).toEqual({ + error: "The bridge host reached its reserved process body byte ceiling. Retry later.", + }); + + // The stream ended with a real response, not a raw reset: the session + // stays healthy, so a second, complete request still succeeds. + const survivingResponse = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(survivingResponse.status).toBe(200); + } finally { + rawClient.close(); + await handle.close(); + filler.release(); + } + }); + + it("test_a_stalled_client_still_frees_the_capacity_denial_reservation_and_slot", async () => { + // The capacity-denial (503) path must not wait forever for its queued + // write to settle: a client that grants no flow-control credit for the + // response would otherwise hold this stream's reservation and + // concurrent-stream slot open indefinitely. A zero initial window + // denies the server any credit to send the 503 body with, at the + // protocol layer, regardless of the fake transport's own buffering — + // the same deterministic stall a real stalled peer produces. + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true); + const forwarderTracker = createForwarderCallTracker(); + const { handle, bridgeToken, clientSide } = bindTestServer({ + capacityDenialSettleDeadlineMs: 30, + forwardRequest: async () => { + forwarderTracker.markCalled(); + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide, { initialWindowSize: 0 }); + try { + const startMs = Date.now(); + const closed = new Promise((resolve) => { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => undefined); + req.on("close", () => resolve()); + // Ten times the 100 bytes of headroom the filler above left: the + // request body itself denies the reservation and enters the 503 + // path. + req.end(Buffer.alloc(1_000, "a")); + }); + await closed; + // The deadline bounded the wait: the stream closed near the deadline + // bound, not after some much longer natural settle that never comes. + expect(Date.now() - startMs).toBeLessThan(5_000); + expect(forwarderTracker.called).toBe(false); + + // The stream's own reservation released even though the client never + // drained the 503 body — only the filler's own bytes remain held. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(filler.heldBytes); + + // The stream's slot freed too. Restore a normal flow-control window + // first: only the denied stream above needs to stall. + await new Promise((resolve) => rawClient.settings({ initialWindowSize: 65_535 }, () => resolve())); + const survivingResponse = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(survivingResponse.status).toBe(200); + } finally { + rawClient.close(); + await handle.close(); + filler.release(); + } + }); + + it("test_a_stalled_normal_response_still_frees_its_reservation_and_slot", async () => { + // The completed-response (200) write path must not wait forever for its + // queued write to settle either: a client that grants no flow-control + // credit for the response would otherwise hold this stream's + // reservation and concurrent-stream slot open indefinitely, the same + // failure mode the capacity-denial path already guards against. A zero + // initial window denies the server any credit to send the response body + // with, at the protocol layer, regardless of the fake transport's own + // buffering. + const forwarderTracker = createForwarderCallTracker(); + const { handle, bridgeToken, clientSide } = bindTestServer({ + responseWriteSettleDeadlineMs: 30, + forwardRequest: async () => { + forwarderTracker.markCalled(); + return { status: 200, headers: {}, body: Buffer.alloc(1_000, "a") }; + }, + }); + const rawClient = connectRawClient(clientSide, { initialWindowSize: 0 }); + try { + const startMs = Date.now(); + const closed = new Promise((resolve) => { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => undefined); + req.on("close", () => resolve()); + req.end(); + }); + await closed; + expect(forwarderTracker.called).toBe(true); + // The deadline bounded the wait: the stream closed near the deadline + // bound, not after some much longer natural settle that never comes. + expect(Date.now() - startMs).toBeLessThan(5_000); + + // This stream's own reservation released even though the client never + // drained the response body. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + + // The stream's slot freed too. Restore a normal flow-control window + // first: only the stalled stream above needs to stall. + await new Promise((resolve) => rawClient.settings({ initialWindowSize: 65_535 }, () => resolve())); + const survivingResponse = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(survivingResponse.status).toBe(200); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_normal_response_that_settles_in_time_is_left_alone", async () => { + // A response that genuinely settles inside the deadline must not be + // force-destroyed: the bounded wait exists only for a stream that never + // settles on its own. + const { handle, bridgeToken, clientSide } = bindTestServer({ + responseWriteSettleDeadlineMs: 5_000, + forwardRequest: async () => ({ + status: 200, + headers: { "content-type": "application/json" }, + body: Buffer.from(JSON.stringify({ ok: true }), "utf8"), + }), + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(response.status).toBe(200); + expect(JSON.parse(response.body)).toEqual({ ok: true }); + expect(getBridgeBodyReservedBytesForTest()).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_response_reservation_denial_answers_503_on_the_wire", async () => { + // The response reader on the sandbox target side + // (`readBridgeForwardResponseBody`) denies its own reservation and + // throws `BridgeProcessCapacityError` from inside `forwardRequest`, + // after the forward handler already ran — a different call site than + // the request-body denial above, but the same 503-before-destroy + // contract must hold: the stream must stay alive long enough to answer + // 503 for real, on the wire, not merely destroy and leave the caller + // with a bare reset. + const forwarderTracker = createForwarderCallTracker(); + // Only the first call denies: the surviving-request check below reuses + // the same session for a second, independent request, which must + // succeed once this first stream's reservation released. + let forwardCalls = 0; + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async () => { + forwarderTracker.markCalled(); + forwardCalls += 1; + if (forwardCalls === 1) { + throw new BridgeProcessCapacityError(); + } + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await new Promise<{ status: number; body: string }>((resolve, reject) => { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + let status = 0; + let body = ""; + req.setEncoding("utf8"); + req.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => resolve({ status, body })); + req.on("error", reject); + req.end(); + }); + expect(response.status).toBe(503); + expect(forwarderTracker.called).toBe(true); + // A raw reset delivers no body at all: parsing it here proves the + // full JSON response actually reached the client, not merely the + // status line. + expect(JSON.parse(response.body)).toEqual({ + error: "The bridge host reached its reserved process body byte ceiling. Retry later.", + }); + + // The stream ended with a real response, not a raw reset: the session + // stays healthy, so a second, complete request still succeeds. + const survivingResponse = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(survivingResponse.status).toBe(200); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("test_a_denied_concatenated_request_body_never_allocates_the_copy", async () => { + // The chunk-array copy and the concatenated copy are two separate live + // buffers, so each must reserve on its own. Leave room for the first but + // not the second: a correct reader checks the reservation before + // `Buffer.concat` allocates the copy, so the denied concatenated copy + // must never call `Buffer.concat` at all. + const chunkBytes = 200_000; + const filler = createBridgeBodyReservation(); + expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - Math.floor(chunkBytes * 1.5))).toBe(true); + const concatSpy = vi.spyOn(Buffer, "concat"); + const forwarderTracker = createForwarderCallTracker(); + const { handle, bridgeToken, clientSide } = bindTestServer({ + forwardRequest: async () => { + forwarderTracker.markCalled(); + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const response = await new Promise<{ status: number; body: string }>((resolve, reject) => { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + let status = 0; + let body = ""; + req.setEncoding("utf8"); + req.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => resolve({ status, body })); + req.on("error", reject); + req.write(Buffer.alloc(chunkBytes, "a"), () => { + // Give the host a turn to receive the one chunk and reserve its + // bytes before this test ends the stream and triggers the + // concatenated-copy reservation attempt. + setTimeout(() => { + // The chunk-array reservation alone must already hold, on top of + // the filler above, exactly the one chunk's bytes: the + // concatenated-copy reservation has not run yet, because the + // stream has not ended. + expect(getBridgeBodyReservedBytesForTest()).toBe(filler.heldBytes + chunkBytes); + req.end(); + }, 50); + }); + }); + expect(response.status).toBe(503); + expect(forwarderTracker.called).toBe(false); + expect(concatSpy).not.toHaveBeenCalled(); + } finally { + concatSpy.mockRestore(); + rawClient.close(); + await handle.close(); + filler.release(); + } }); describe("parseCanonicalBridgeRequestPath", () => { @@ -940,7 +1830,7 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { const { handle, clientSide } = bindTestServer({ forwardRequest: async (request) => { forwarderTracker.markCalled(); - return { status: 200, body: JSON.stringify({ path: request.pathname }) }; + return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") }; }, }); const rawClient = connectRawClient(clientSide); @@ -1045,6 +1935,56 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { } }); + it("test_a_maximum_size_multipart_attachment_upload_fits_through_the_bridge", async () => { + // A file at the exact attachment content ceiling (10 MiB — the same + // figure `MAX_ATTACHMENT_BYTES` in `server/src/attachment-types.ts` + // enforces) still crosses the bridge wrapped in a multipart body: the + // boundary line and the part's own headers add bytes on top of that file + // content. The bridge's own body limit needs headroom for that framing, + // or a valid maximum-size attachment fails here before the server ever + // sees it. + const maxAttachmentBytes = 10 * 1024 * 1024; + const boundary = "----PaperclipTestBoundary1234567890abcdef"; + const preamble = Buffer.from( + `--${boundary}\r\n` + + `Content-Disposition: form-data; name="file"; filename="attachment.bin"\r\n` + + `Content-Type: application/octet-stream\r\n\r\n`, + "utf8", + ); + const fileContent = Buffer.alloc(maxAttachmentBytes, 0x61); + const epilogue = Buffer.from(`\r\n--${boundary}--\r\n`, "utf8"); + const multipartBody = Buffer.concat([preamble, fileContent, epilogue]); + + // The wrapper really does add bytes on top of the file content alone — + // otherwise this test would prove nothing about framing headroom. + expect(multipartBody.byteLength).toBeGreaterThan(maxAttachmentBytes); + expect(multipartBody.byteLength).toBeLessThanOrEqual(DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES); + + let receivedBodyBytes = 0; + const { gateway, handle } = createTestPair({ + routes: HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, + forwardRequest: async (request) => { + receivedBodyBytes = request.body.byteLength; + return { status: 201, body: Buffer.from("{}", "utf8") }; + }, + }); + try { + const response = await gateway.forwardRequest({ + method: "POST", + path: "/api/companies/co-1/issues/issue-1/attachments", + query: "", + headers: { "content-type": `multipart/form-data; boundary=${boundary}` }, + body: multipartBody, + receivedToken: BRIDGE_TOKEN, + }); + expect(response.status).toBe(201); + expect(receivedBodyBytes).toBe(multipartBody.byteLength); + } finally { + await gateway.close(); + await handle.close(); + } + }, 20_000); + it("the sandbox gateway keeps its own token check before it opens a stream", async () => { const forwarderTracker = createForwarderCallTracker(); const { gateway, handle } = createTestPair({ diff --git a/packages/adapter-utils/src/http2-bridge-server.ts b/packages/adapter-utils/src/http2-bridge-server.ts index 0d07fc5ae9..3d96787ff2 100644 --- a/packages/adapter-utils/src/http2-bridge-server.ts +++ b/packages/adapter-utils/src/http2-bridge-server.ts @@ -52,23 +52,32 @@ export const HTTP2_BRIDGE_ENABLE_PUSH = false; * Open streams. The host keeps one forward, its request body, and its * response body alive for the life of a stream, and — before this file binds * each forward to its own stream's abort signal — a forward can outlive its - * stream's own HTTP/2 slot until the forward's own timeout runs out. Counting - * every retained `Buffer` and string copy of one stream's request and - * response body against the {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES} - * body limit (`sandbox-callback-bridge.ts`) gives an accounting peak of eight - * times that limit for one live forward. This bound is the per-route - * in-flight-body budget: `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 8 * - * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES` bytes = 4 * 8 * 262,144 - * bytes = 8,388,608 bytes for one route. + * stream's own HTTP/2 slot until the forward's own timeout runs out. The + * forward path carries a request body and a response body as raw `Buffer` + * values with no string copy. Counting every retained `Buffer` copy of one + * stream's request and response body against the + * {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES} body limit + * (`sandbox-callback-bridge.ts`) gives an accounting peak of four times that + * limit for one live forward. This bound is the per-route in-flight-body + * budget: `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 4 * + * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES` bytes = 4 * 4 * 10,551,296 + * bytes = 168,820,736 bytes (161 MiB) for one route. {@link + * HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES} enforces this figure as a real, live + * cap on every reservation, so one busy route cannot pass it, no matter how + * much of the process-wide ceiling below still sits free. * - * Known aggregate behavior: this budget applies to one route only. The host - * process admits up to `DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES` (128, in - * `plugin-worker-manager.ts`) routes at the same time, and each route holds - * its own 8,388,608-byte peak. The process can therefore retain up to - * 1,073,741,824 bytes (1 GiB) of live body data across every route at once. - * This document accepts that ceiling: the host tracks no process-wide byte - * total, so no single route can starve another route's own budget, but the - * host also enforces no smaller sum across every route. + * Aggregate behavior: the host process admits up to + * `DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES` (128, in `plugin-worker-manager.ts`) + * routes at the same time. The aggregate across every route is bounded too: + * every stream's {@link BridgeBodyReservation} owner also reserves against + * the shared {@link HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES} total + * (1,073,741,824 bytes, 1 GiB), so the process retains no more than that + * many live body bytes no matter how many routes or streams run at once. One + * full-size stream's four retained copies cost `4 * 10,551,296` = + * 42,205,184 bytes of that total, so the process admits at least 25 + * concurrent full-size streams, spread across at least six routes each at + * their own per-route ceiling, before it starts denying the rest with a 503 + * response. */ export const HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS = 4; /** One decompressed header list. The Node default is 65535. */ @@ -112,6 +121,176 @@ export const HTTP2_BRIDGE_SERVER_OPTIONS: http2.ServerOptions = { streamResetBurst: HTTP2_BRIDGE_STREAM_RESET_BURST, }; +// --------------------------------------------------------------------------- +// Process-wide body byte reservation +// --------------------------------------------------------------------------- + +/** + * The most process memory, in bytes, this file lets every route hold in live + * request and response body buffers at the same time. Every + * {@link BridgeBodyReservation} owner reserves against this one shared + * total, so no combination of concurrent streams, across every route, can + * retain more than this many bytes at once. This value keeps the accepted + * process ceiling at 1,073,741,824 bytes (1 GiB) — the same ceiling + * `doc/observability.md` already accepted before the per-body limit rose to + * 10 MiB — now enforced by this reservation instead of left as an unenforced + * document note. See the per-route budget comment above for the full + * accounting. + */ +export const HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES = 1024 * 1024 * 1024; + +/** + * The most memory, in bytes, one route (one {@link createHttp2BridgeServer} + * call, one sandbox run's bridge session) may hold in live request and + * response body buffers at the same time, on top of the shared process-wide + * ceiling above. This is the same per-route figure the budget comment above + * already derives from stream concurrency: naming it here and checking it on + * every reservation stops one busy route from spending the whole + * process-wide ceiling and denying every sibling route admission. See that + * comment for the full accounting. + */ +export const HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES = + HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 4 * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES; + +// The process-wide running total, in bytes, every `BridgeBodyReservation` +// owner reserves against. Module-scope state is correct here: one host +// process runs one bridge, and every route and every stream in that process +// must share the same ceiling. +let reservedProcessBodyBytes = 0; + +/** + * One route's own running total, in bytes, against + * {@link HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES}. `createHttp2BridgeServer` + * creates exactly one ledger per route and every stream that route ever + * handles reserves against it, so one route's own activity can never pass + * its own ceiling, regardless of how much of the process-wide total remains + * free for other routes. + */ +export interface BridgeRouteBodyLedger { + /** + * Reserve `byteCount` more bytes against this route's own ceiling. Returns + * `false`, and reserves nothing, when the new route total would pass + * {@link HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES}. + */ + reserve(byteCount: number): boolean; + /** Release `byteCount` bytes this route previously reserved. */ + release(byteCount: number): void; + /** The bytes this route currently holds. */ + readonly reservedBytes: number; +} + +/** Create one fresh {@link BridgeRouteBodyLedger}, holding zero bytes. One + * `createHttp2BridgeServer` call creates exactly one, before its first + * stream, and every stream that route ever handles shares it. */ +export function createBridgeRouteBodyLedger(): BridgeRouteBodyLedger { + let reservedBytes = 0; + return { + reserve(byteCount: number): boolean { + if (reservedBytes + byteCount > HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES) { + return false; + } + reservedBytes += byteCount; + return true; + }, + release(byteCount: number): void { + reservedBytes -= byteCount; + }, + get reservedBytes(): number { + return reservedBytes; + }, + }; +} + +/** + * One HTTP/2 stream's reservation owner. `handleStream` creates exactly one + * owner per stream and releases it in its existing `finally` block, so every + * live request or response body buffer that stream produces reserves + * against the same owner, and the process reclaims those bytes exactly one + * time when the stream ends. + */ +export interface BridgeBodyReservation { + /** + * Reserve `byteCount` more bytes against the process-wide total, and + * against this owner's route ledger when it has one. Returns `false` and + * reserves nothing against either total when either check fails. A failed + * reservation allocates nothing: the caller must not copy the bytes it + * asked to reserve. + */ + reserve(byteCount: number): boolean; + /** + * Release every byte this owner currently holds. Safe to call more than + * one time: a second call releases nothing. + */ + release(): void; + /** The bytes this owner currently holds. */ + readonly heldBytes: number; +} + +/** + * Create one fresh {@link BridgeBodyReservation} owner, holding zero bytes. + * A caller that passes `routeLedger` also checks and reserves against that + * route's own ceiling on every call, isolating this owner's route from every + * other route sharing the process-wide total. A caller with no route to + * isolate (a test filling only the process-wide total, for example) omits + * it, and this owner checks the process-wide ceiling alone. + */ +export function createBridgeBodyReservation(routeLedger?: BridgeRouteBodyLedger): BridgeBodyReservation { + let heldBytes = 0; + let released = false; + return { + reserve(byteCount: number): boolean { + if (released) return false; + if (reservedProcessBodyBytes + byteCount > HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES) { + return false; + } + if (routeLedger && !routeLedger.reserve(byteCount)) { + return false; + } + reservedProcessBodyBytes += byteCount; + heldBytes += byteCount; + return true; + }, + release(): void { + if (released) return; + released = true; + reservedProcessBodyBytes -= heldBytes; + routeLedger?.release(heldBytes); + heldBytes = 0; + }, + get heldBytes(): number { + return heldBytes; + }, + }; +} + +/** + * A reservation owner denied a request or response body copy because the + * process-wide ceiling would otherwise be passed. The stream handler answers + * 503 for this error, not 413: a 413 tells a caller its own body is too + * large; a 503 tells a caller the host is busy and to retry later. + */ +export class BridgeProcessCapacityError extends Error { + constructor() { + super("The bridge host reached its reserved process body byte ceiling. Retry later."); + this.name = "BridgeProcessCapacityError"; + } +} + +/** + * Test-only. Reset the process-wide reservation total to zero. A test file + * that exercises {@link createBridgeBodyReservation} must call this between + * tests, so a reservation one test left unreleased cannot lower the ceiling + * for a later test. + */ +export function resetBridgeBodyReservationsForTest(): void { + reservedProcessBodyBytes = 0; +} + +/** Test-only. Read the current process-wide reservation total. */ +export function getBridgeBodyReservedBytesForTest(): number { + return reservedProcessBodyBytes; +} + // --------------------------------------------------------------------------- // Duplex channel adapter // --------------------------------------------------------------------------- @@ -463,6 +642,24 @@ export const DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS = 480_000; * session that carries a stalled stream would otherwise hold `close()` open * forever, because `session.close()` waits for every open stream to end. */ export const DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS = 5_000; +/** The default bound the capacity-denial (503) response path waits for its + * queued write to settle before it force-destroys the stream. A normal, + * draining peer settles well inside this bound, so it still receives the + * full 503 body. A stalled peer that never grants the flow-control credit + * the write needs would otherwise hold this stream's reservation and + * {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} slot open forever. */ +export const DEFAULT_HTTP2_BRIDGE_CAPACITY_DENIAL_SETTLE_DEADLINE_MS = 5_000; +/** The default bound the completed-response (normal, non-denial) write path + * waits for its queued write to settle before it force-destroys the stream. + * A normal, draining peer settles well inside this bound. A stalled peer + * that grants no flow-control credit would otherwise hold this stream's + * reservation and {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} slot open + * forever — the same failure mode the capacity-denial path already guards + * against. Set above {@link DEFAULT_HTTP2_BRIDGE_CAPACITY_DENIAL_SETTLE_DEADLINE_MS} + * because a completed response can carry a full-size body (up to the + * configured body-byte ceiling), not just a small JSON error payload, so a + * slow-but-genuine peer needs more room to drain it. */ +export const DEFAULT_HTTP2_BRIDGE_RESPONSE_WRITE_SETTLE_DEADLINE_MS = 30_000; function startHttp2BridgePingWatchdog( session: http2.ServerHttp2Session, @@ -517,7 +714,7 @@ function startHttp2BridgePingWatchdog( export interface Http2BridgeForwardResult { status: number; headers?: Record; - body?: Buffer | string; + body?: Buffer; } /** @@ -538,6 +735,16 @@ export interface Http2BridgeForwardRequest { * never fires for any other stream or for the session. */ signal: AbortSignal; + /** + * This stream's one {@link BridgeBodyReservation} owner. A forward handler + * that itself retains a full response body buffer — `execution-target.ts` + * does, through `forwardBridgeRequest`'s optional `reservation` option — + * reserves against this same owner, so the request body and the response + * body of one stream share one ceiling. `handleStream` releases this owner + * exactly one time, after the forward call settles; the forward handler + * must never release it. + */ + reservation: BridgeBodyReservation; } export type Http2BridgeForwardHandler = ( @@ -587,6 +794,14 @@ export interface CreateHttp2BridgeServerOptions { * session to close on its own before it force-destroys the session. The * default is {@link DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS}. */ closeGraceMs?: number; + /** The bound the capacity-denial (503) response path waits for its queued + * write to settle before it force-destroys the stream. The default is + * {@link DEFAULT_HTTP2_BRIDGE_CAPACITY_DENIAL_SETTLE_DEADLINE_MS}. */ + capacityDenialSettleDeadlineMs?: number; + /** The bound the completed-response (normal) write path waits for its + * queued write to settle before it force-destroys the stalled stream. The + * default is {@link DEFAULT_HTTP2_BRIDGE_RESPONSE_WRITE_SETTLE_DEADLINE_MS}. */ + responseWriteSettleDeadlineMs?: number; /** The cap, in bytes, on data this server holds once a bound `Duplex` * reports its readable side is full (`push()` returns `false`). Past this * cap the server treats the channel as stuck, not merely slow: see @@ -667,13 +882,21 @@ export interface Http2BridgeBodyBounds { * stream ends for any reason at all — a normal end, an error, a timeout- or * shutdown-triggered `destroy()`, or a peer reset — so the promise always * settles and the caller never awaits a stream that already went away. + * + * `onChunk` and `onEnd` decide what the read retains, if anything: + * {@link readHttp2StreamBody} accumulates chunks and reserves against a + * {@link BridgeBodyReservation}; {@link drainHttp2StreamBody} discards every + * chunk and reserves nothing. Either callback may throw to reject the read + * (a denied reservation, for example) — the throw destroys the stream the + * same way a size, idle, or lifetime fault does. */ -function readHttp2StreamBody( +function readOrDrainHttp2StreamBody( stream: http2.ServerHttp2Stream, bounds: Http2BridgeBodyBounds, -): Promise { + onChunk: (chunk: Buffer, totalBytes: number) => void, + onEnd: () => T, +): Promise { return new Promise((resolve, reject) => { - const chunks: Buffer[] = []; let totalBytes = 0; let settled = false; let idleTimer: ReturnType; @@ -709,18 +932,153 @@ function readHttp2StreamBody( stream.destroy(); return; } - chunks.push(chunk); + try { + onChunk(chunk, totalBytes); + } catch (error) { + settle(() => reject(error instanceof Error ? error : new Error(String(error)))); + // A denied reservation answers 503 through the caller's own + // `respondJson` call, after this promise rejects — destroying the + // stream here, before that call runs, would make it a no-op (a + // destroyed stream refuses `.respond()`). The caller destroys the + // stream itself, once its response is actually on the wire. Every + // other body-read fault has no such response to protect, so it + // destroys the stream immediately, exactly as before. + if (!(error instanceof BridgeProcessCapacityError)) { + stream.destroy(); + } + return; + } // The chunk is real progress, so the peer is not stalled: reset the // idle bound. The lifetime ceiling timer above does not reset here. armIdleTimer(); }); - stream.once("end", () => settle(() => resolve(Buffer.concat(chunks)))); + stream.once("end", () => { + let result: T; + try { + result = onEnd(); + } catch (error) { + settle(() => reject(error instanceof Error ? error : new Error(String(error)))); + return; + } + settle(() => resolve(result)); + }); stream.once("error", (error) => settle(() => reject(error instanceof Error ? error : new Error(String(error))))); stream.once("aborted", () => settle(() => reject(new Error("Bridge request stream aborted.")))); stream.once("close", () => settle(() => reject(new Error("Bridge request stream closed before it completed.")))); }); } +/** + * Read one request body into a `Buffer`. When the caller passes a + * `reservation`, this reserves each chunk's bytes against it before the + * chunk joins the retained array, and reserves the concatenated buffer's own + * byte count before `Buffer.concat` allocates it — the chunk array and the + * concatenated buffer are two separate live copies, so both reserve. A + * denied reservation rejects with {@link BridgeProcessCapacityError} and + * destroys the stream, retaining no further chunk. + */ +function readHttp2StreamBody( + stream: http2.ServerHttp2Stream, + bounds: Http2BridgeBodyBounds, + reservation?: BridgeBodyReservation, +): Promise { + const chunks: Buffer[] = []; + let retainedBytes = 0; + return readOrDrainHttp2StreamBody( + stream, + bounds, + (chunk, totalBytes) => { + if (reservation && !reservation.reserve(chunk.byteLength)) { + throw new BridgeProcessCapacityError(); + } + chunks.push(chunk); + retainedBytes = totalBytes; + }, + () => { + if (reservation && !reservation.reserve(retainedBytes)) { + throw new BridgeProcessCapacityError(); + } + return Buffer.concat(chunks); + }, + ); +} + +/** + * Drain and discard one denied stream's request body, under the same size, + * idle, and lifetime bounds an authenticated request gets, but retaining no + * chunk and reserving no bytes. `denyRequest` calls this instead of + * {@link readHttp2StreamBody}, so a stream that never carries a valid bridge + * token cannot retain a full body buffer merely by sending one. + */ +function drainHttp2StreamBody(stream: http2.ServerHttp2Stream, bounds: Http2BridgeBodyBounds): Promise { + return readOrDrainHttp2StreamBody( + stream, + bounds, + () => { + // No `chunks.push`: a denied stream's body content never reaches the + // forward handler, so this reader retains no chunk and reserves + // nothing against the process ceiling. + }, + () => undefined, + ); +} + +/** + * Wait until a stream's queued write actually leaves process memory, or + * until the stream closes for any other reason. `stream.end(body)` only + * queues `body` for asynchronous transmission: Node keeps the bytes in + * memory until HTTP/2 flow control lets them flow, which a slow or + * backpressured peer can delay well past the moment `end()` returns. A + * caller that reserves the response body's bytes must hold that reservation + * across this whole wait, not merely across the call to `end()`, or a + * backpressured response keeps bytes in memory the ledger already believes + * it reclaimed. + * + * `finish` is the normal settle: every queued byte reached the session. A + * `close` or an `error` settle the wait the same way, so a peer reset or a + * destroyed stream cannot leave a caller waiting forever for a `finish` + * that will never come. + * + * `deadlineMs`, when given, bounds the wait itself: past that many + * milliseconds with none of the three events above, this resolves anyway. + * A peer that neither drains the write nor resets nor errors the stream — + * one that simply stalls, granting no flow-control credit — would otherwise + * hold the wait open forever with none of the three settle events ever + * firing. The caller stays responsible for destroying the stream once this + * resolves; a bounded resolve here does not by itself free the stream's + * slot or its reservation. + * + * The returned `settled` flag tells the caller which way this resolved: + * `true` for a genuine `finish`/`close`/`error` event, `false` for the + * deadline. A caller that must free the stream's slot only when the peer + * truly stalled reads this flag instead of destroying an already-finished + * stream unconditionally. + */ +function waitForHttp2StreamWriteToSettle( + stream: http2.ServerHttp2Stream, + deadlineMs?: number, +): Promise<{ settled: boolean }> { + if (stream.writableFinished || stream.destroyed || stream.closed) return Promise.resolve({ settled: true }); + return new Promise((resolve) => { + let deadlineTimer: ReturnType | undefined; + const onSettle = (settled: boolean): void => { + stream.removeListener("finish", onFinish); + stream.removeListener("close", onFinish); + stream.removeListener("error", onFinish); + if (deadlineTimer) clearTimeout(deadlineTimer); + resolve({ settled }); + }; + const onFinish = (): void => onSettle(true); + stream.once("finish", onFinish); + stream.once("close", onFinish); + stream.once("error", onFinish); + if (deadlineMs !== undefined) { + deadlineTimer = setTimeout(() => onSettle(false), deadlineMs); + deadlineTimer.unref?.(); + } + }); +} + function respondJson(stream: http2.ServerHttp2Stream, status: number, body: unknown): void { if (stream.destroyed || stream.closed) return; try { @@ -741,7 +1099,10 @@ function respondJson(stream: http2.ServerHttp2Stream, status: number, body: unkn * peer that leaves the body unfinished keeps the stream open, holding one * of the {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} slots for as long as * it chooses. Nothing awaits the discard; the caller has already answered - * the request and moves on to the next stream. + * the request and moves on to the next stream. The discard retains no + * chunk and reserves no process byte budget: a stream that fails the bridge + * token check gets no reservation, so it cannot retain a full body buffer + * merely by sending one. */ function denyRequest( stream: http2.ServerHttp2Stream, @@ -751,7 +1112,7 @@ function denyRequest( ): void { respondJson(stream, status, body); if (stream.destroyed || stream.closed) return; - readHttp2StreamBody(stream, bounds).catch(() => { + drainHttp2StreamBody(stream, bounds).catch(() => { // The idle or lifetime bound above already destroyed the stream, or the // peer reset it first. Either way the slot is free; the discarded body // content is irrelevant to a denial. @@ -773,6 +1134,10 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) const requestBodyLifetimeCeilingMs = options.requestBodyLifetimeCeilingMs ?? DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS; const closeGraceMs = options.closeGraceMs ?? DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS; + const capacityDenialSettleDeadlineMs = + options.capacityDenialSettleDeadlineMs ?? DEFAULT_HTTP2_BRIDGE_CAPACITY_DENIAL_SETTLE_DEADLINE_MS; + const responseWriteSettleDeadlineMs = + options.responseWriteSettleDeadlineMs ?? DEFAULT_HTTP2_BRIDGE_RESPONSE_WRITE_SETTLE_DEADLINE_MS; const maxBufferedReadBytes = options.maxBufferedReadBytes ?? DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES; const readBackpressureStallMs = options.readBackpressureStallMs ?? DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS; @@ -787,6 +1152,11 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) const server = http2.createServer(HTTP2_BRIDGE_SERVER_OPTIONS); const activeSessions = new Set(); + // One route ledger for this one `createHttp2BridgeServer` call. Every + // stream this route ever handles reserves against it, so this route's own + // activity can never pass its own share of the process-wide ceiling and + // deny a sibling route admission. + const routeLedger = createBridgeRouteBodyLedger(); async function handleStream( stream: http2.ServerHttp2Stream, @@ -804,6 +1174,17 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) stream.once("close", abortForThisStream); stream.once("aborted", abortForThisStream); stream.once("error", abortForThisStream); + // One reservation owner for this one stream's entire life. Every live + // request or response body buffer this stream produces reserves against + // this same owner; the `finally` block below releases it exactly one + // time, on every exit path (a deny, a body-read fault, a forward fault, + // or a completed response), including a peer reset or a timeout, both of + // which route through the abort listeners above into the forward call's + // combined signal, so the owner keeps its bytes reserved until that + // forward call actually settles. It reserves against this route's own + // ledger too, so it can never spend more than this route's own share of + // the process-wide ceiling. + const reservation = createBridgeBodyReservation(routeLedger); try { // Accepted security fix 4: the constant-time bridge-token compare runs // before route processing and before header processing. This host check @@ -838,8 +1219,24 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) let body: Buffer; try { - body = await readHttp2StreamBody(stream, bodyBounds); + body = await readHttp2StreamBody(stream, bodyBounds, reservation); } catch (error) { + if (error instanceof BridgeProcessCapacityError) { + // The reader above left this stream open (undestroyed) exactly so + // this response could reach the wire. `respondJson` only queues + // the 503 write; destroying the stream right after queues a + // RST_STREAM before a backpressured peer drains it, so the client + // can see a reset instead of the 503. Wait for the write to + // settle first, then destroy to free this stream's + // concurrent-stream slot. The wait carries its own deadline: a + // stalled peer that grants no flow-control credit would otherwise + // never settle the write, holding this stream's reservation and + // slot open forever. + respondJson(stream, 503, { error: error.message }); + await waitForHttp2StreamWriteToSettle(stream, capacityDenialSettleDeadlineMs); + if (!stream.destroyed) stream.destroy(); + return; + } respondJson(stream, 413, { error: error instanceof Error ? error.message : String(error) }); return; } @@ -853,8 +1250,23 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) headers: sanitizedHeaders, body, signal: controller.signal, + reservation, }); } catch (error) { + if (error instanceof BridgeProcessCapacityError) { + // The forward handler's own response reader denies its reservation + // and cancels there, but leaves this stream open exactly so this + // 503 can reach the wire — the same contract the request-body + // capacity denial above keeps. `respondJson` only queues the + // write, so wait for it to settle first (bounded, so a stalled + // peer that grants no flow-control credit cannot hold this + // stream's reservation and slot open forever), then destroy to + // free the slot. + respondJson(stream, 503, { error: error.message }); + await waitForHttp2StreamWriteToSettle(stream, capacityDenialSettleDeadlineMs); + if (!stream.destroyed) stream.destroy(); + return; + } respondJson(stream, 502, { error: error instanceof Error ? error.message : String(error) }); return; } @@ -868,6 +1280,21 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) try { stream.respond(responseHeaders); stream.end(result.body); + // `end()` only queues `result.body`; a slow or backpressured peer + // can hold those bytes in process memory well after this call + // returns. Wait for the write to actually settle before the + // `finally` block below releases the reservation those bytes hold. + // The wait carries its own deadline: a peer that grants no + // flow-control credit and never resets or errors the stream would + // otherwise hold this stream's reservation and + // HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS slot open forever, the same + // failure mode the capacity-denial path above already guards + // against. A genuinely stalled stream gets force-destroyed once the + // deadline passes; a stream that settled on its own (`finish`, + // `close`, or `error`) is left alone, since it is already ending or + // ended. + const { settled } = await waitForHttp2StreamWriteToSettle(stream, responseWriteSettleDeadlineMs); + if (!settled && !stream.destroyed) stream.destroy(); } catch { // The peer reset the stream (RST_STREAM) between dispatch and response. // One stream's write fault stays local to that stream. @@ -876,7 +1303,12 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) // Every path above reaches this exactly once: a deny, a body-read // fault, a forward fault, or a completed response. A completed stream // must leak no listener; the session, not this one stream, outlives - // the handler. + // the handler. The reservation release is idempotent, but this is + // still the one place this stream's owner ever releases: releasing + // here, after the response write above has actually settled, keeps + // the owner's bytes reserved for the whole time Node still holds a + // live copy of them, not merely until the write call returns. + reservation.release(); stream.removeListener("close", abortForThisStream); stream.removeListener("aborted", abortForThisStream); stream.removeListener("error", abortForThisStream); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index a967201428..6cc661f5f2 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -14,7 +14,9 @@ import { createFileSystemSandboxCallbackBridgeQueueClient, createSandboxCallbackBridgeAsset, createSandboxCallbackBridgeToken, + getSandboxBridgeProcessBodyLedgerSource, getSandboxCallbackBridgeServerSource, + HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, sandboxCallbackBridgeDirectories, syncRemoteTextFileWithHashSkip, syncSandboxCallbackBridgeEntrypoint, @@ -966,6 +968,24 @@ describe("sandbox callback bridge", () => { await expect(nonJsonResponse.json()).resolves.toEqual({ error: "Bridge only accepts JSON request bodies.", }); + + // The queue transport keeps its 415 gate for the attachment upload path + // too: it carries a string envelope only, so it never admits a binary body. + const attachmentOctetStreamResponse = await fetch( + `${bridge.baseUrl}/api/companies/co-1/issues/issue-1/attachments`, + { + method: "POST", + headers: { + authorization: `Bearer ${bridgeToken}`, + "content-type": "application/octet-stream", + }, + body: Buffer.from([0x50, 0x4b, 0x03, 0x04]), + }, + ); + expect(attachmentOctetStreamResponse.status).toBe(415); + await expect(attachmentOctetStreamResponse.json()).resolves.toEqual({ + error: "Bridge only accepts JSON request bodies.", + }); }); it("returns a 502 when the host response times out", async () => { @@ -1425,6 +1445,44 @@ describe("sandbox callback bridge", () => { `Route not allowed: ${request.method} ${request.path}`, ); } + + // The HTTP/2 route list adds the two binary attachment routes on top of + // the documented heartbeat surface. + const http2Allowed: Array<{ method: string; path: string }> = [ + { method: "POST", path: "/api/companies/co-1/issues/issue-1/attachments" }, + { method: "GET", path: "/api/attachments/att-1/content" }, + ]; + for (const request of http2Allowed) { + expect( + authorizeSandboxCallbackBridgeRequestWithRoutes(request, HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST), + ).toBeNull(); + } + + const http2Denied: Array<{ method: string; path: string }> = [ + // Wrong method for each attachment rule. + { method: "GET", path: "/api/companies/co-1/issues/issue-1/attachments" }, + { method: "POST", path: "/api/attachments/att-1/content" }, + // Extra path segment for each attachment rule. + { method: "POST", path: "/api/companies/co-1/issues/issue-1/attachments/att-1" }, + { method: "GET", path: "/api/attachments/att-1/content/extra" }, + ]; + for (const request of http2Denied) { + expect( + authorizeSandboxCallbackBridgeRequestWithRoutes(request, HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST), + ).toBe(`Route not allowed: ${request.method} ${request.path}`); + } + }); + + it("denies both attachment routes on the default (queue) route list", () => { + const attachmentRequests: Array<{ method: string; path: string }> = [ + { method: "POST", path: "/api/companies/co-1/issues/issue-1/attachments" }, + { method: "GET", path: "/api/attachments/att-1/content" }, + ]; + for (const request of attachmentRequests) { + expect(authorizeSandboxCallbackBridgeRequestWithRoutes(request)).toBe( + `Route not allowed: ${request.method} ${request.path}`, + ); + } }); it("marks command-managed bridge operations with the bridge execution channel", async () => { @@ -3252,7 +3310,11 @@ describe("sandbox callback bridge", () => { bridgeToken, forwardRequest: async (request) => { seenBodies.push(request.body); - return { status: 200, headers: { "content-type": "application/json" }, body: JSON.stringify({ ok: true }) }; + return { + status: 200, + headers: { "content-type": "application/json" }, + body: Buffer.from(JSON.stringify({ ok: true }), "utf8"), + }; }, }); @@ -3278,19 +3340,22 @@ describe("sandbox callback bridge", () => { it("forwards malformed UTF-8 bytes to the HTTP/2 host handler unchanged", async () => { const bridgeToken = createSandboxCallbackBridgeToken(); const seenBodies: Buffer[] = []; + // Byte 0xC3 opens a two-byte UTF-8 sequence; 0x28 is not a valid + // continuation byte, so this body is not valid UTF-8. The gateway does + // not decode or validate the body, so these exact bytes must still + // arrive at the host handler unchanged, and the same bytes must return + // to the caller unchanged. The host answers with a non-JSON content + // type, so the round trip proves the gateway applies no format-specific + // handling on the response leg either. + const malformedBytes = Buffer.from([0x7b, 0x22, 0x61, 0x22, 0x3a, 0xc3, 0x28, 0x7d]); const gateway = await startHttp2GatewayForTest({ bridgeToken, forwardRequest: async (request) => { seenBodies.push(request.body); - return { status: 200, headers: {}, body: "" }; + return { status: 200, headers: { "content-type": "application/octet-stream" }, body: malformedBytes }; }, }); - // Byte 0xC3 opens a two-byte UTF-8 sequence; 0x28 is not a valid - // continuation byte, so this body is not valid UTF-8. The gateway does - // not decode or validate the body, so these exact bytes must still - // arrive at the host handler unchanged. - const malformedBytes = Buffer.from([0x7b, 0x22, 0x61, 0x22, 0x3a, 0xc3, 0x28, 0x7d]); const response = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { method: "POST", headers: { @@ -3302,6 +3367,8 @@ describe("sandbox callback bridge", () => { expect(response.status).toBe(200); expect(seenBodies).toHaveLength(1); expect(seenBodies[0]?.equals(malformedBytes)).toBe(true); + const responseBytes = Buffer.from(await response.arrayBuffer()); + expect(responseBytes.equals(malformedBytes)).toBe(true); }, 15_000); it("rejects a request body over maxBodyBytes on the HTTP/2 path before it forwards a byte", async () => { @@ -3313,7 +3380,7 @@ describe("sandbox callback bridge", () => { maxBodyBytes, forwardRequest: async () => { forwardCalls += 1; - return { status: 200, headers: {}, body: "" }; + return { status: 200, headers: {}, body: Buffer.alloc(0) }; }, }); @@ -3333,6 +3400,133 @@ describe("sandbox callback bridge", () => { expect(forwardCalls).toBe(0); }, 15_000); + /** A resolvable gate a test can await, then release on its own schedule. */ + function createGate(): { reached: Promise; reach: () => void } { + let reach!: () => void; + const reached = new Promise((resolve) => { + reach = resolve; + }); + return { reached, reach }; + } + + it("denies a body once concurrent reservations reach the gateway's own process ledger ceiling, then admits again once they release", async () => { + // The per-body maxBodyBytes cap alone does not bound how many bodies + // this gateway process holds in memory at once. The generated gateway's + // own process ledger gives it a second, independent ceiling: 4 + // concurrent max-size bodies, each counted twice (the retained chunk + // array and the concatenated copy), exactly fills maxBodyBytes * 8. This + // drives 4 concurrent requests to that exact ceiling, held open by a + // forward call that does not return, then proves a 5th is denied and a + // later one succeeds once the 4 held requests release their bytes. + const bridgeToken = createSandboxCallbackBridgeToken(); + const maxBodyBytes = 400; + const holdBody = Buffer.alloc(maxBodyBytes, 0x42); + const HOLD_COUNT = 4; + const reachedGates = Array.from({ length: HOLD_COUNT }, () => createGate()); + const releaseGates = Array.from({ length: HOLD_COUNT }, () => createGate()); + let forwardCalls = 0; + const gateway = await startHttp2GatewayForTest({ + bridgeToken, + maxBodyBytes, + forwardRequest: async () => { + const callIndex = forwardCalls; + forwardCalls += 1; + if (callIndex < HOLD_COUNT) { + reachedGates[callIndex]!.reach(); + await releaseGates[callIndex]!.reached; + } + return { status: 200, headers: {}, body: Buffer.alloc(0) }; + }, + }); + + const postHoldBody = () => + fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, + body: holdBody, + }); + + const holdResponses = Promise.all(Array.from({ length: HOLD_COUNT }, () => postHoldBody())); + // Each held request's own forward call started, which only happens + // after readBodyBytes already reserved that request's bytes, so all 4 + // reservations are live once every gate below resolves. + await Promise.all(reachedGates.map((gate) => gate.reached)); + + const deniedResponse = await postHoldBody(); + expect(deniedResponse.status).toBe(503); + await expect(deniedResponse.json()).resolves.toMatchObject({ + error: "The bridge gateway process reached its reserved body byte ceiling. Retry later.", + }); + // The denial happened during the body read, before this stream's own + // forward call ever ran. + expect(forwardCalls).toBe(HOLD_COUNT); + + releaseGates.forEach((gate) => gate.reach()); + const responses = await holdResponses; + for (const response of responses) { + expect(response.status).toBe(200); + } + + // Every held reservation released once its own forward call settled: a + // fresh request at the same size now succeeds again. + const recoveredResponse = await postHoldBody(); + expect(recoveredResponse.status).toBe(200); + }, 15_000); + + it("maps an indeterminate host outcome to a non-retryable 409 on the HTTP/2 gateway for an unsafe method", async () => { + // The host answers a mutating request's response-body capacity denial with a + // 504 and the indeterminate outcome header, exactly like an aborted in-flight + // forward: the host may have already committed the mutation, so the status + // must not be retryable. The HTTP/2 gateway must map that 504 to a 409, the + // same map the file gateway already applies, so a standard retry policy does + // not repeat the mutation. The outcome header and body must survive the map. + const bridgeToken = createSandboxCallbackBridgeToken(); + const indeterminateBody = JSON.stringify({ error: "response body over limit", outcome: "indeterminate" }); + const gateway = await startHttp2GatewayForTest({ + bridgeToken, + forwardRequest: async () => ({ + status: 504, + headers: { "content-type": "application/json", "x-paperclip-bridge-outcome": "indeterminate" }, + body: Buffer.from(indeterminateBody, "utf8"), + }), + }); + + const response = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { authorization: `Bearer ${bridgeToken}`, "content-type": "application/json" }, + body: JSON.stringify({ note: "test" }), + }); + + expect(response.status).toBe(409); + expect(response.status).toBeLessThan(500); + expect(response.headers.get("x-paperclip-bridge-outcome")).toBe("indeterminate"); + const responseBytes = Buffer.from(await response.arrayBuffer()); + expect(responseBytes.toString("utf8")).toBe(indeterminateBody); + }, 15_000); + + it("passes through a safe method's retryable 503 on the HTTP/2 gateway unchanged", async () => { + // A safe method's response-body capacity denial carries no indeterminate + // outcome header, so the gateway must keep it retryable: the map above must + // not fire on a plain 503. + const bridgeToken = createSandboxCallbackBridgeToken(); + const gateway = await startHttp2GatewayForTest({ + bridgeToken, + forwardRequest: async () => ({ + status: 503, + headers: { "content-type": "application/json" }, + body: Buffer.from(JSON.stringify({ error: "response body over limit" }), "utf8"), + }), + }); + + const response = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { + method: "GET", + headers: { authorization: `Bearer ${bridgeToken}` }, + }); + + expect(response.status).toBe(503); + expect(response.headers.get("x-paperclip-bridge-outcome")).toBeNull(); + }, 15_000); + it("rejects a request body over maxBodyBytes on the queue path before it writes the queue file", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-queue-maxbody-")); cleanupDirs.push(rootDir); @@ -3453,3 +3647,40 @@ describe("sandbox callback bridge", () => { expect(seenRequests[0]?.body).toBe(requestBodyText); }); }); + +interface EmbeddedBridgeProcessBodyLedger { + reserve(byteCount: number): boolean; + release(byteCount: number): void; + readonly reservedBytes: number; +} + +// This describe block covers the zero-dependency process body-byte ledger +// every generated gateway embeds (`BRIDGE_PROCESS_BODY_LEDGER_SOURCE`), +// exercised directly with no spawned process involved, the same way the +// codec source in `execution-target-sandbox.test.ts` gets its own direct +// coverage. +describe("embedded sandbox gateway process body ledger", () => { + it("reserves and releases bytes against its own ceiling, denying only once it is exceeded", () => { + const ledgerFactory = new Function( + `${getSandboxBridgeProcessBodyLedgerSource()}\nreturn createBridgeProcessBodyLedger;`, + ) as unknown as () => (maxBytes: number) => EmbeddedBridgeProcessBodyLedger; + const createBridgeProcessBodyLedger = ledgerFactory(); + const ledger = createBridgeProcessBodyLedger(100); + + expect(ledger.reserve(60)).toBe(true); + expect(ledger.reservedBytes).toBe(60); + // A denied reservation reserves nothing: the total stays exactly what + // the first call reserved. + expect(ledger.reserve(41)).toBe(false); + expect(ledger.reservedBytes).toBe(60); + expect(ledger.reserve(40)).toBe(true); + expect(ledger.reservedBytes).toBe(100); + + ledger.release(60); + expect(ledger.reservedBytes).toBe(40); + // Room freed by the release admits a request the full ceiling would + // have denied. + expect(ledger.reserve(60)).toBe(true); + expect(ledger.reservedBytes).toBe(100); + }); +}); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index de4bdb1c50..8a9ac43b70 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -20,7 +20,23 @@ const DEFAULT_BRIDGE_POLL_INTERVAL_MS = 100; const DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS = 30_000; const DEFAULT_BRIDGE_STOP_TIMEOUT_MS = 2_000; const DEFAULT_BRIDGE_MAX_QUEUE_DEPTH = 64; -const DEFAULT_BRIDGE_MAX_BODY_BYTES = 256 * 1024; +// A `BridgeBodyReservation` owner (`http2-bridge-server.ts`) now bounds the +// process-wide total of live request and response body bytes at +// `HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES` (1 GiB), so this per-body limit can +// rise to the same ceiling `MAX_ATTACHMENT_BYTES` +// (`server/src/attachment-types.ts`) already accepts, with no unbounded +// growth in process memory. +// +// The attachment upload route carries its file inside a multipart body, so a +// file at the exact `MAX_ATTACHMENT_BYTES` ceiling needs more than +// `MAX_ATTACHMENT_BYTES` raw bytes to cross the bridge: the multipart +// boundary line, each part's `Content-Disposition` and `Content-Type` +// headers, and the small JSON metadata field this route also accepts all add +// bytes on top of the file content the server measures. This headroom +// covers that framing with a wide margin, so a valid maximum-size attachment +// never fails at the bridge before the server ever sees it. +const BRIDGE_MULTIPART_FRAMING_HEADROOM_BYTES = 64 * 1024; +const DEFAULT_BRIDGE_MAX_BODY_BYTES = 10 * 1024 * 1024 + BRIDGE_MULTIPART_FRAMING_HEADROOM_BYTES; // 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 // iteration and never false-fires on a slow-but-live call. It is also well @@ -183,6 +199,16 @@ export const DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST: readonly SandboxCa { method: "DELETE", path: /^\/api\/routine-triggers\/[^/]+$/ }, ] as const; +// The HTTP/2 bridge carries the request body as raw bytes, so it can admit +// the two binary attachment routes. The queue transport keeps +// `DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST`, because its envelope +// carries a string body only. +export const HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST: readonly SandboxCallbackBridgeRouteRule[] = [ + ...DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST, + { method: "POST", path: /^\/api\/companies\/[^/]+\/issues\/[^/]+\/attachments$/ }, + { method: "GET", path: /^\/api\/attachments\/[^/]+\/content$/ }, +] as const; + export const DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST = [ "accept", "content-type", @@ -2039,6 +2065,48 @@ export function getSandboxDuplexGatewayCodecSource(): string { return DUPLEX_GATEWAY_CODEC_SOURCE; } +/** + * Zero-dependency source for the gateway's own aggregate body-byte ledger. + * `readBodyBytes` in the generated gateway reserves against one instance of + * this ledger before it retains a chunk and before it concatenates the final + * buffer, so a burst of concurrent requests cannot grow the gateway + * process's own memory without limit. This ledger is separate from, and + * independent of, the ceiling `http2-bridge-server.ts` enforces on the host + * side of the bridge connection: each side bounds only the memory in its own + * process. It uses no global beyond plain JavaScript, so it embeds inside the + * gateway template literal with no escape. A test wraps this source directly + * to exercise `reserve`/`release`, the same way + * {@link getSandboxDuplexGatewayCodecSource} lets a test exercise the codec. + */ +const BRIDGE_PROCESS_BODY_LEDGER_SOURCE = `function createBridgeProcessBodyLedger(maxBytes) { + let reservedBytes = 0; + return { + reserve(byteCount) { + if (reservedBytes + byteCount > maxBytes) { + return false; + } + reservedBytes += byteCount; + return true; + }, + release(byteCount) { + reservedBytes -= byteCount; + }, + get reservedBytes() { + return reservedBytes; + }, + }; +}`; + +/** + * Return the exact zero-dependency ledger source the generated gateway + * embeds. A test wraps this source and calls `createBridgeProcessBodyLedger` + * to prove the embedded copy reserves and releases bytes correctly, with no + * spawned process involved. + */ +export function getSandboxBridgeProcessBodyLedgerSource(): string { + return BRIDGE_PROCESS_BODY_LEDGER_SOURCE; +} + export function getSandboxCallbackBridgeServerSource(): string { return `import { randomUUID, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; @@ -2119,6 +2187,32 @@ process.on("unhandledRejection", (reason) => { // gateway ignores it. ${DUPLEX_GATEWAY_CODEC_SOURCE} +// The embedded zero-dependency process body-byte ledger. Both gateway modes +// use it: readBodyBytes reserves against it, and each mode's request +// handler releases what it reserved once the body is no longer needed. +${BRIDGE_PROCESS_BODY_LEDGER_SOURCE} + +// The multiplier matches HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS (4) in +// http2-bridge-server.ts, doubled because readBodyBytes reserves a body's +// bytes twice: once for the retained chunk array, once for the concatenated +// copy, since both are live buffers at once. This gives the gateway process +// its own aggregate ceiling on live request-body bytes, so a burst of +// concurrent requests cannot grow this process's memory without limit, even +// though every individual body already passes the maxBodyBytes check below. +// This ceiling is independent of, and separate from, the ceiling the host +// enforces on its own side of the bridge connection. +const maxProcessBodyBytes = maxBodyBytes * 8; +const processBodyLedger = createBridgeProcessBodyLedger(maxProcessBodyBytes); + +// A denied process-ledger reservation answers 503: the sandbox client should +// retry once other in-flight bodies finish and release their bytes, the same +// retry contract the host side gives for its own capacity denial. +class BridgeProcessCapacityError extends Error { + constructor() { + super("The bridge gateway process reached its reserved body byte ceiling. Retry later."); + } +} + function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -2136,22 +2230,49 @@ function normalizeHeaders(headers) { return out; } +// Reserves each chunk's bytes against the process ledger before the chunk +// joins the retained array, and reserves the concatenated buffer's own byte +// count before Buffer.concat allocates it, mirroring the order +// readHttp2StreamBody enforces on the host side. Returns the body buffer +// together with a release function: the caller must call release exactly +// once, after the body is no longer needed, so its reserved bytes return to +// the ledger on completion, on an error the caller raises later, on a client +// abort, and on a timeout — every path funnels through the caller's own +// finally block. A read that fails here (the size limit, or a denied +// process reservation) releases its own partial reservation immediately, so +// no caller-side release call is needed for that path. async function readBodyBytes(req) { const chunks = []; 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) { - throw new Error("Bridge request body exceeded the configured size limit."); + let reservedBytes = 0; + try { + for await (const chunk of req) { + const nextChunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + totalBytes += nextChunk.byteLength; + if (totalBytes > maxBodyBytes) { + throw new Error("Bridge request body exceeded the configured size limit."); + } + if (!processBodyLedger.reserve(nextChunk.byteLength)) { + throw new BridgeProcessCapacityError(); + } + reservedBytes += nextChunk.byteLength; + chunks.push(nextChunk); } + if (!processBodyLedger.reserve(totalBytes)) { + throw new BridgeProcessCapacityError(); + } + reservedBytes += totalBytes; + const body = Buffer.concat(chunks); + return { body, release: () => processBodyLedger.release(reservedBytes) }; + } catch (error) { + processBodyLedger.release(reservedBytes); + throw error; } - return Buffer.concat(chunks); } async function readBody(req) { - return (await readBodyBytes(req)).toString("utf8"); + const { body, release } = await readBodyBytes(req); + return { body: body.toString("utf8"), release }; } function tokensMatch(received) { @@ -2211,6 +2332,12 @@ async function runFileGateway() { } const server = createServer(async (req, res) => { + // readBody reserves the body's bytes against the process ledger and + // hands back a release function; this holds it so the finally below + // releases those bytes exactly once no matter how this handler ends — + // its normal completion, a thrown error, a client abort, or a deadline + // timeout all reach the same finally. + let releaseBodyReservation = null; try { const auth = req.headers.authorization || ""; const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : ""; @@ -2236,7 +2363,8 @@ async function runFileGateway() { return; } const requestId = randomUUID(); - const requestBody = await readBody(req); + const { body: requestBody, release } = await readBody(req); + releaseBodyReservation = release; const payload = { id: requestId, method: req.method || "GET", @@ -2284,7 +2412,13 @@ async function runFileGateway() { } res.end(typeof response.body === "string" ? response.body : ""); } catch (error) { - writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); + // A denied process-ledger reservation is retryable: the caller should + // try again once other in-flight bodies release their bytes. Every + // other body-read or handling fault stays a generic 502. + const status = error instanceof BridgeProcessCapacityError ? 503 : 502; + writeJsonResponse(res, status, { error: error instanceof Error ? error.message : String(error) }); + } finally { + releaseBodyReservation?.(); } }); @@ -2465,6 +2599,12 @@ function runHttp2Gateway() { } const server = createServer(async (req, res) => { + // readBodyBytes reserves the body's bytes against the process ledger + // and hands back a release function; this holds it so the finally + // below releases those bytes exactly once no matter how this handler + // ends — its normal completion, a thrown error, a client abort, or a + // deadline timeout all reach the same finally. + let releaseBodyReservation = null; try { const auth = req.headers.authorization || ""; const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : ""; @@ -2477,12 +2617,8 @@ function runHttp2Gateway() { return; } const url = new URL(req.url || "/", "http://127.0.0.1"); - const contentType = typeof req.headers["content-type"] === "string" ? req.headers["content-type"] : ""; - if (req.method && req.method !== "GET" && req.method !== "HEAD" && !/json/i.test(contentType)) { - writeJsonResponse(res, 415, { error: "Bridge only accepts JSON request bodies." }); - return; - } - const requestBodyBuffer = await readBodyBytes(req); + const { body: requestBodyBuffer, release } = await readBodyBytes(req); + releaseBodyReservation = release; let response; try { response = await forwardOverHttp2({ @@ -2496,14 +2632,32 @@ function runHttp2Gateway() { writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); return; } - res.statusCode = typeof response.status === "number" ? response.status : 200; + // The host marks a possibly-committed mutation with an indeterminate outcome. + // The host cannot cancel a host operation that is in flight, so the mutation + // may have committed before the worker aborted the handler. A 5xx status is + // retryable by convention, so a caller that retries 5xx would apply the + // mutation twice. Map the indeterminate outcome to a non-retryable 409, so a + // standard retry policy does not repeat the request. The outcome header and + // body stay, so a caller that reads them still sees the indeterminate result. + const bridgeOutcome = (response.headers || {})["x-paperclip-bridge-outcome"]; + if (bridgeOutcome === "indeterminate") { + res.statusCode = 409; + } else { + res.statusCode = typeof response.status === "number" ? response.status : 200; + } for (const [key, value] of Object.entries(response.headers || {})) { if (typeof value !== "string" || key.toLowerCase() === "content-length") continue; res.setHeader(key, value); } res.end(response.body); } catch (error) { - writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); + // A denied process-ledger reservation is retryable: the caller should + // try again once other in-flight bodies release their bytes. Every + // other body-read or handling fault stays a generic 502. + const status = error instanceof BridgeProcessCapacityError ? 503 : 502; + writeJsonResponse(res, status, { error: error instanceof Error ? error.message : String(error) }); + } finally { + releaseBodyReservation?.(); } }); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index bdcf10cb8b..af6721f3d4 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -3558,11 +3558,18 @@ export interface PluginWorkerManagerOptions { * * Known aggregate behavior: this ceiling bounds route count only, not * retained bytes. Each HTTP/2 bridge route bounds its own retained body - * bytes to 8,388,608 bytes (see `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS` in - * `http2-bridge-server.ts`), so this route ceiling caps the process's - * aggregate retained body bytes at 128 * 8,388,608 = 1,073,741,824 bytes - * (1 GiB). This is accepted, known behavior, not a defect: the process - * tracks no aggregate byte ledger across routes. + * bytes to 168,820,736 bytes (see `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS` in + * `http2-bridge-server.ts`), so this route ceiling alone would let the + * process's aggregate retained body bytes reach 128 * 168,820,736 = + * 21,609,054,208 bytes (about 20.1 GiB) if nothing else bounded it. It does + * not reach that figure: every stream's `BridgeBodyReservation` owner + * (`http2-bridge-server.ts`) reserves against the shared + * `HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES` total (1,073,741,824 bytes, 1 GiB) + * across every route, so that reservation — not this route ceiling — is the + * process's real aggregate-byte bound. `HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES` + * (`http2-bridge-server.ts`) adds a second bound scoped to one route at a + * time, so one busy route cannot spend that whole process-wide total by + * itself and deny every sibling route admission. */ export const DEFAULT_MAX_CONCURRENT_DUPLEX_ROUTES = 128;