RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(adapter-utils): carry binary bodies and attachment routes over the HTTP/2 sandbox bridge (#12923) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Paperclip runs agents in local and remote sandboxes through adapter
utilities
> - The HTTP/2 sandbox bridge decoded every body as UTF-8 text and
rejected non-JSON content
> - This stopped agents from uploading or downloading issue attachments
through that bridge
> - This pull request carries raw bytes, permits the two attachment
routes, and enforces a shared body limit
> - The benefit is correct attachment transfer with a process-wide
memory guard

## Linked Issues or Issue Description

**What existing behavior does this improve?**

The HTTP/2 sandbox bridge forwards request bodies between an agent
sandbox and the Paperclip host. It now supports binary bodies and the
issue attachment routes.

**Current behavior**

The bridge decodes each body as UTF-8 text. It returns HTTP 415 for
content types outside the JSON route list. An agent cannot upload or
download an issue attachment through this transport.

**Proposed behavior**

The bridge carries raw bytes through the forward path. It permits the
attachment upload and content routes. The queue transport and file
gateway keep their existing route behavior. A shared 10 MiB body limit
and process-wide byte reservation protect memory use.

**Reason and benefit**

Attachment clients need byte-preserving transfer. The shared limit keeps
the gateway and host aligned. The reservation prevents concurrent
streams from exceeding the accepted process memory ceiling.

**Breaking changes**

The HTTP/2 bridge accepts two attachment routes and permits binary
content. The queue transport and file gateway keep their previous route
lists and HTTP 415 behavior. No schema or external endpoint changes.

## What Changed

- Carry request and response bodies as raw bytes through the HTTP/2
bridge.
- Permit attachment upload and attachment content routes on the HTTP/2
bridge only.
- Raise the resolved per-body limit to 10 MiB and share it between the
gateway and host.
- Reserve body bytes before allocation and release each stream
reservation on every terminal path.
- Document the body limit, process ceiling, and reservation behavior.

## Verification

- Run `pnpm exec vitest run
packages/adapter-utils/src/http2-bridge-server.test.ts
packages/adapter-utils/src/execution-target-sandbox.test.ts
packages/adapter-utils/src/sandbox-callback-bridge.test.ts`; 226 tests
pass.
- Run `pnpm --filter @paperclipai/adapter-utils typecheck`; it passes.
- Run the direct server TypeScript check with `tsc --noEmit` in
`server/`; it passes with zero errors.
- Verify multipart upload and binary download round trips over HTTP/2
without corruption.
- Verify the queue transport and file gateway return HTTP 415 for the
same routes.
- Verify the host rejects bodies over the resolved limit.
- Verify a denied reservation returns HTTP 503 and allocates no copy.
- Verify stream cleanup releases reservations after completion, error,
abort, timeout, and close.

## Risks

The bridge now accepts larger bodies and binary content. The
process-wide reservation limits total live body bytes to 1 GiB. Route
behavior changes only for the HTTP/2 bridge. The security review found
no blocking issue for this commit range.

## Model Used

OpenAI Codex, GPT-5. The runtime used tool calls and code execution. The
runtime did not expose the context window size. No model-generated code
changes were made for this pull request.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach 2026-09-08 14:18:12 -07:00 committed by GitHub
parent e095b84dab
commit 6019e2bd6e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 2696 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
doc/observability.md
View File

@ -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

708
packages/adapter-utils/src/execution-target-sandbox.test.ts
View File

@ -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<string, string>; body?: string },
): Promise<{ status: number; headers: Record<string, string>; body: string }> {
request: { method: string; path: string; headers?: Record<string, string>; body?: Buffer },
): Promise<{ status: number; headers: Record<string, string>; 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<string, string>;
body: Buffer;
}>;
close: () => Promise<void>;
}> {
@ -3224,21 +3254,35 @@ describe("sandbox adapter execution targets", () => {
auth: string | null;
runId: string | null;
headers: Record<string, string>;
body: Buffer;
}> = [];
const server = createServer((req, res) => {
const headers: Record<string, string> = {};
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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((resolve) => apiServer.close(() => resolve()));
}
});
// ---------------------------------------------------------------------------
// Real-PTY replay.
//

148
packages/adapter-utils/src/execution-target.ts
View File

@ -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<string> {
async function readBridgeForwardResponseBody(
response: Response,
maxBodyBytes: number,
reservation?: BridgeBodyReservation,
): Promise<Buffer> {
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<string, string>;
/** 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<string, string>; body: string }> => {
): Promise<{ status: number; headers: Record<string, string>; 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,

984
packages/adapter-utils/src/http2-bridge-server.test.ts
View File

File diff suppressed because it is too large Load Diff

484
packages/adapter-utils/src/http2-bridge-server.ts
View File

@ -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<string, string>;
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<T>(
stream: http2.ServerHttp2Stream,
bounds: Http2BridgeBodyBounds,
): Promise<Buffer> {
onChunk: (chunk: Buffer, totalBytes: number) => void,
onEnd: () => T,
): Promise<T> {
return new Promise((resolve, reject) => {
const chunks: Buffer[] = [];
let totalBytes = 0;
let settled = false;
let idleTimer: ReturnType<typeof setTimeout>;
@ -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<Buffer> {
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<void> {
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<typeof setTimeout> | 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<http2.ServerHttp2Session>();
// 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);

247
packages/adapter-utils/src/sandbox-callback-bridge.test.ts
View File

@ -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<void>; reach: () => void } {
let reach!: () => void;
const reached = new Promise<void>((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);
});
});

192
packages/adapter-utils/src/sandbox-callback-bridge.ts
View File

@ -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?.();
}
});

17
server/src/services/plugin-worker-manager.ts
View File

@ -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;