diff --git a/packages/adapter-utils/src/http2-bridge-server.test.ts b/packages/adapter-utils/src/http2-bridge-server.test.ts index 113aa212e49..1761947b7f6 100644 --- a/packages/adapter-utils/src/http2-bridge-server.test.ts +++ b/packages/adapter-utils/src/http2-bridge-server.test.ts @@ -10,8 +10,10 @@ import { createHttp2BridgeServer, parseCanonicalBridgeRequestPath, wrapDuplexChannelAsNodeDuplex, + DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES, DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS, DEFAULT_HTTP2_BRIDGE_PING_STALL_MS, + DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS, HTTP2_BRIDGE_ENABLE_PUSH, HTTP2_BRIDGE_HEADER_TABLE_SIZE, HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS, @@ -30,6 +32,10 @@ import { } from "./http2-bridge-server.js"; import { createSandboxHttp2BridgeGateway } from "./sandbox-callback-bridge.js"; import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js"; +import { + Http2BridgeAdmissionGate, + HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES, +} from "./http2-bridge-admission.js"; /** * Unit harness for the host HTTP/2 server and the sandbox HTTP/2 client @@ -78,6 +84,9 @@ interface TestPairOptions { requestBodyTimeoutMs?: number; requestBodyLifetimeCeilingMs?: number; closeGraceMs?: number; + maxBodyBytes?: number; + admissionGate?: Http2BridgeAdmissionGate; + responseDrainTimeoutMs?: number; onGoaway?: (record: Http2BridgeGoawayRecord) => void; onSessionError?: (error: Error) => void; onSession?: (session: http2.ServerHttp2Session) => void; @@ -103,6 +112,9 @@ function bindTestServer(options: TestPairOptions = {}) { requestBodyTimeoutMs: options.requestBodyTimeoutMs, requestBodyLifetimeCeilingMs: options.requestBodyLifetimeCeilingMs, closeGraceMs: options.closeGraceMs, + maxBodyBytes: options.maxBodyBytes, + admissionGate: options.admissionGate, + responseDrainTimeoutMs: options.responseDrainTimeoutMs, onGoaway: options.onGoaway, onSessionError: options.onSessionError, onSession: options.onSession, @@ -1072,4 +1084,417 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { await new Promise((resolve) => setTimeout(resolve, 60)); expect(duplex.destroyed).toBe(false); }); + + describe("the admission gate", () => { + it("the response drain timeout defaults to thirty seconds, and the buffered-read byte cap shares one home with the admission module's session budget", () => { + expect(DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS).toBe(30_000); + expect(DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES).toBe(HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES); + }); + + it("one gate caps parallel streams across two bridge server instances", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 2, maxSessions: 4 }); + let concurrent = 0; + let peakConcurrent = 0; + const forwardRequest = async (): Promise => { + concurrent += 1; + peakConcurrent = Math.max(peakConcurrent, concurrent); + await new Promise((resolve) => setTimeout(resolve, 30)); + concurrent -= 1; + return { status: 200 }; + }; + const serverA = bindTestServer({ admissionGate, forwardRequest }); + const serverB = bindTestServer({ admissionGate, forwardRequest }); + const rawClientA = connectRawClient(serverA.clientSide); + const rawClientB = connectRawClient(serverB.clientSide); + try { + await Promise.all([ + expectSessionStillServesARequest(rawClientA, { + method: "GET", + path: "/api/agents/me", + token: serverA.bridgeToken, + }), + expectSessionStillServesARequest(rawClientA, { + method: "GET", + path: "/api/companies/co1", + token: serverA.bridgeToken, + }), + expectSessionStillServesARequest(rawClientB, { + method: "GET", + path: "/api/agents/me", + token: serverB.bridgeToken, + }), + expectSessionStillServesARequest(rawClientB, { + method: "GET", + path: "/api/companies/co1", + token: serverB.bridgeToken, + }), + ]); + // The one shared gate never let more than two streams run at once, + // across both server instances. + expect(peakConcurrent).toBeLessThanOrEqual(2); + // Two of the four streams did run at the same time: this proves the + // cap is real overlap, not accidental full serialization to one. + expect(peakConcurrent).toBe(2); + } finally { + rawClientA.close(); + rawClientB.close(); + await serverA.handle.close(); + await serverB.handle.close(); + } + }); + + it("a queued stream starts its body read only after an earlier stream releases", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 }); + const HOLD_MS = 150; + const IDLE_MS = 80; + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + requestBodyTimeoutMs: IDLE_MS, + forwardRequest: async (request) => { + if (request.pathname === "/api/agents/me") { + // Hold the only slot for HOLD_MS, so the second stream below + // stays queued for that whole span. + await new Promise((resolve) => setTimeout(resolve, HOLD_MS)); + } + return { status: 200, body: JSON.stringify({ bodyLength: request.body.byteLength }) }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const firstReq = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + firstReq.resume(); + firstReq.end(); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(admissionGate.activeCount).toBe(1); + expect(admissionGate.queuedCount).toBe(0); + + const secondResponse = new Promise<{ status: number; body: string }>((resolve, reject) => { + const secondReq = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + let status = 0; + let body = ""; + secondReq.setEncoding("utf8"); + secondReq.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + secondReq.on("data", (chunk) => (body += chunk)); + secondReq.on("end", () => resolve({ status, body })); + secondReq.on("error", reject); + // Wait almost the full hold span, sending no data, before this + // stream's one chunk. If the idle bound had started counting + // while this stream still waited in the queue, this silence alone + // would already exceed IDLE_MS and stall the stream. + setTimeout(() => { + secondReq.write("chunk"); + secondReq.end(); + }, HOLD_MS - 10); + }); + + // The second stream queues while the first holds the only slot. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(admissionGate.queuedCount).toBe(1); + + const response = await secondResponse; + expect(response.status).toBe(200); + expect(JSON.parse(response.body)).toEqual({ bodyLength: "chunk".length }); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("a token, path, or route denial takes no gate slot", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 }); + const { handle, bridgeToken, clientSide } = bindTestServer({ admissionGate }); + const rawClient = connectRawClient(clientSide); + + function requestAndAwaitStatus(input: { method: string; path: string; token?: string }): Promise { + return new Promise((resolve, reject) => { + const headers: http2.OutgoingHttpHeaders = { ":method": input.method, ":path": input.path }; + if (input.token !== undefined) headers.authorization = `Bearer ${input.token}`; + const req = rawClient.request(headers); + let status = 0; + req.on("response", (responseHeaders) => { + status = Number(responseHeaders[":status"]) || 0; + }); + req.on("end", () => resolve(status)); + req.on("error", reject); + req.resume(); + req.end(); + }); + } + + try { + // An invalid token is denied before route or body processing. + expect(await requestAndAwaitStatus({ method: "GET", path: "/api/agents/me" })).toBe(401); + expect(admissionGate.activeCount).toBe(0); + expect(admissionGate.queuedCount).toBe(0); + + // An invalid path is denied before route processing. + expect( + await requestAndAwaitStatus({ method: "GET", path: "/api/../secrets", token: bridgeToken }), + ).toBe(400); + expect(admissionGate.activeCount).toBe(0); + expect(admissionGate.queuedCount).toBe(0); + + // A route the allowlist does not carry is denied. + expect( + await requestAndAwaitStatus({ method: "DELETE", path: "/api/agents/me", token: bridgeToken }), + ).toBe(403); + expect(admissionGate.activeCount).toBe(0); + expect(admissionGate.queuedCount).toBe(0); + + // None of the three denials above spent the one available slot: a + // legitimate request right after them still succeeds at once. + const survivingResponse = await expectSessionStillServesARequest(rawClient, { + method: "GET", + path: "/api/agents/me", + token: bridgeToken, + }); + expect(survivingResponse.status).toBe(200); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("the slot stays held until the response stream closes", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 }); + // A body past the default HTTP/2 flow-control window, so the server + // cannot finish sending it, and the stream cannot close, until the + // client actually reads it. + const bigBody = Buffer.alloc(200_000, "a"); + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + forwardRequest: async () => ({ status: 200, body: bigBody }), + }); + const rawClient = connectRawClient(clientSide); + try { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => undefined); + req.end(); + await new Promise((resolve) => req.once("response", () => resolve())); + // The response headers arrived, but this test reads no body: the + // slot must still be held. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(admissionGate.activeCount).toBe(1); + + const drained = new Promise((resolve) => req.once("end", resolve)); + req.resume(); + await drained; + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(admissionGate.activeCount).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("a peer that never reads the response releases the slot at the drain bound", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 }); + const bigBody = Buffer.alloc(200_000, "a"); + // Long enough that the client reliably observes the response headers + // before the server's own drain bound destroys the stream; short + // enough to keep the test fast. + const RESPONSE_DRAIN_TIMEOUT_MS = 150; + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + responseDrainTimeoutMs: RESPONSE_DRAIN_TIMEOUT_MS, + // The drain bound below force-destroys the stream while the peer + // never read it: a short grace bound keeps this test's own + // `handle.close()` call from waiting on the default five-second one. + closeGraceMs: 30, + forwardRequest: async () => ({ status: 200, body: bigBody }), + }); + const rawClient = connectRawClient(clientSide); + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + try { + req.on("error", () => undefined); + req.end(); + await new Promise((resolve) => req.once("response", () => resolve())); + // Never read the response body: the drain bound must still free the + // slot, instead of holding it forever. + await new Promise((resolve) => setTimeout(resolve, RESPONSE_DRAIN_TIMEOUT_MS + 200)); + expect(admissionGate.activeCount).toBe(0); + } finally { + // The server destroyed its side already, at the drain bound; drop + // the client's own unread stream too, so the still-buffered (and + // now pointless) response bytes do not sit in the transport and + // block the session-level close below. + req.destroy(); + rawClient.close(); + await handle.close(); + } + }); + + it("the gate slot returns after a forward-handler error", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 }); + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + forwardRequest: async () => { + throw new Error("forward handler failed"); + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const status = await new Promise((resolve, reject) => { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("response", (headers) => resolve(Number(headers[":status"]) || 0)); + req.on("error", reject); + req.resume(); + req.end(); + }); + expect(status).toBe(502); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(admissionGate.activeCount).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("the gate slot returns after a body size rejection", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 }); + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + maxBodyBytes: 10, + }); + const rawClient = connectRawClient(clientSide); + try { + // A body past the size limit destroys the stream at once, so the + // client sees an `error` or a `close`, not a clean response — the + // same settling pattern the idle- and lifetime-bound tests above + // exercise. This test's own concern is the admission slot below. + await new Promise((resolve) => { + const req = rawClient.request({ + ":method": "POST", + ":path": "/api/issues/abc/comments", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => resolve()); + req.on("close", () => resolve()); + req.resume(); + req.write(Buffer.alloc(1_000, "a")); + req.end(); + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(admissionGate.activeCount).toBe(0); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("a stream that closes while it waits leaves the queue", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 }); + let releaseFirst: (() => void) | undefined; + const firstHeld = new Promise((resolve) => { + releaseFirst = resolve; + }); + let forwarderCallCount = 0; + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + forwardRequest: async () => { + forwarderCallCount += 1; + if (forwarderCallCount === 1) { + await firstHeld; + } + return { status: 200 }; + }, + }); + const rawClient = connectRawClient(clientSide); + try { + const firstReq = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + firstReq.on("error", () => undefined); + firstReq.resume(); + firstReq.end(); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(admissionGate.activeCount).toBe(1); + + const secondReq = rawClient.request({ + ":method": "GET", + ":path": "/api/companies/co1", + authorization: `Bearer ${bridgeToken}`, + }); + secondReq.on("error", () => undefined); + secondReq.end(); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(admissionGate.queuedCount).toBe(1); + + const secondClosed = new Promise((resolve) => secondReq.once("close", resolve)); + secondReq.close(http2.constants.NGHTTP2_CANCEL); + await secondClosed; + await new Promise((resolve) => setTimeout(resolve, 30)); + // The cancelled stream left the queue, and never took the slot: the + // first stream still holds the only one. + expect(admissionGate.queuedCount).toBe(0); + expect(admissionGate.activeCount).toBe(1); + + const firstClosed = new Promise((resolve) => firstReq.once("close", resolve)); + releaseFirst?.(); + await firstClosed; + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(admissionGate.activeCount).toBe(0); + // The cancelled stream never reached the forward handler. + expect(forwarderCallCount).toBe(1); + } finally { + rawClient.close(); + await handle.close(); + } + }); + + it("server close returns every held gate slot", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 }); + const bigBody = Buffer.alloc(200_000, "a"); + const { handle, bridgeToken, clientSide } = bindTestServer({ + admissionGate, + // Long enough that only the server close below bounds this wait. + responseDrainTimeoutMs: 60_000, + closeGraceMs: 30, + forwardRequest: async () => ({ status: 200, body: bigBody }), + }); + const rawClient = connectRawClient(clientSide); + try { + const req = rawClient.request({ + ":method": "GET", + ":path": "/api/agents/me", + authorization: `Bearer ${bridgeToken}`, + }); + req.on("error", () => undefined); + req.end(); + await new Promise((resolve) => req.once("response", () => resolve())); + // Never read the response: the stream cannot close on its own. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(admissionGate.activeCount).toBe(1); + + await handle.close(); + expect(admissionGate.activeCount).toBe(0); + } finally { + rawClient.close(); + } + }); + }); }); diff --git a/packages/adapter-utils/src/http2-bridge-server.ts b/packages/adapter-utils/src/http2-bridge-server.ts index a42afa17031..e6af4e2e306 100644 --- a/packages/adapter-utils/src/http2-bridge-server.ts +++ b/packages/adapter-utils/src/http2-bridge-server.ts @@ -30,6 +30,11 @@ import { Duplex } from "node:stream"; import http2 from "node:http2"; import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js"; +import { + getHttp2BridgeAdmissionGate, + HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES, + type Http2BridgeAdmissionGate, +} from "./http2-bridge-admission.js"; import { authorizeSandboxCallbackBridgeRequestWithRoutes, compareBridgeTokensConstantTime, @@ -105,8 +110,12 @@ export const HTTP2_BRIDGE_SERVER_OPTIONS: http2.ServerOptions = { * bound. This cap also bounds one single chunk: the wrapper checks a chunk's * own size against it before `push()` ever runs, so one oversized chunk * cannot cross the cap on its first delivery, before the queue holds - * anything to compare it against. */ -export const DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES = HTTP2_BRIDGE_MAX_SESSION_MEMORY * 1024 * 1024; + * anything to compare it against. + * + * This value reads {@link HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES} from the + * admission module, so the wrapper cap and the admission gate's live-session + * reservation resolve through one shared constant and cannot drift apart. */ +export const DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES = HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES; /** The default bound, in milliseconds, on how long the read-side queue * {@link wrapDuplexChannelAsNodeDuplex} holds can stay non-empty with no * chunk draining from it. The byte cap above bounds how much memory a stuck @@ -438,6 +447,13 @@ 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 {@link waitForStreamDrain} waits for a response stream + * to reach its `close` event before the server frees the admission slot the + * stream holds. `stream.end()` returns before the peer has read the + * response, so a wait with no bound would let a peer that never reads a + * response hold the slot open forever. Set to the same value as + * {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS}. */ +export const DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS = 30_000; function startHttp2BridgePingWatchdog( session: http2.ServerHttp2Session, @@ -565,6 +581,16 @@ export interface CreateHttp2BridgeServerOptions { * chunk draining from it. The default is * {@link DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS}. */ readBackpressureStallMs?: number; + /** The process-wide admission gate. The server takes one slot from this + * gate for each admitted stream, after the route check and before the + * request body read, and holds the slot until the stream closes (or the + * response drain bound fires). The default is the process-wide instance + * from {@link getHttp2BridgeAdmissionGate}. A test injects its own gate. */ + admissionGate?: Http2BridgeAdmissionGate; + /** The bound {@link waitForStreamDrain} waits for a response stream to + * close before the server frees its admission slot. The default is + * {@link DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS}. */ + responseDrainTimeoutMs?: number; /** The sink for a GOAWAY the server observed on a session (one the sandbox * side sent to the host). */ onGoaway?: (record: Http2BridgeGoawayRecord) => void; @@ -688,6 +714,141 @@ function readHttp2StreamBody( }); } +/** One settled body-capture result: either the concatenated body, or the + * error the capture settled with. */ +type Http2BridgeBodyCaptureOutcome = { ok: true; body: Buffer } | { ok: false; error: Error }; + +/** + * Begin capturing one request body, ahead of the idle and lifetime timers + * that bound how long the capture can run. A caller that waits on the + * admission gate before it is ready to enforce those timers must still + * attach the underlying stream listeners at once, in the same synchronous + * turn as the `stream` event: attaching them one turn later can lose the + * peer's `end` (Node does not replay a body the JS layer was not yet + * listening for), stalling a request that already fully arrived. + * + * `awaitBody` arms the idle and lifetime timers and returns the promise the + * caller awaits. Call it once, when the caller is ready to enforce the + * bounds (after the admission wait, in this server). If the body already + * finished, or the stream already errored, closed, or aborted, before + * `awaitBody` runs, it delivers that outcome at once and arms no timer. + */ +function beginHttp2BridgeStreamBodyCapture( + stream: http2.ServerHttp2Stream, + maxBodyBytes: number, +): { awaitBody(idleTimeoutMs: number, lifetimeCeilingMs: number): Promise } { + const chunks: Buffer[] = []; + let totalBytes = 0; + let settled = false; + let outcome: Http2BridgeBodyCaptureOutcome | undefined; + let onSettled: ((outcome: Http2BridgeBodyCaptureOutcome) => void) | undefined; + let idleTimer: ReturnType | undefined; + let lifetimeCeilingTimer: ReturnType | undefined; + let armedIdleTimeoutMs: number | undefined; + + const settle = (result: Http2BridgeBodyCaptureOutcome) => { + if (settled) return; + settled = true; + if (idleTimer) clearTimeout(idleTimer); + if (lifetimeCeilingTimer) clearTimeout(lifetimeCeilingTimer); + outcome = result; + onSettled?.(result); + }; + + const armIdleTimer = () => { + // Before `awaitBody` runs, `armedIdleTimeoutMs` is unset: a chunk that + // arrives while the caller still waits on admission must not start (or + // restart) the idle bound, so a queue wait alone can never trip it. + if (armedIdleTimeoutMs === undefined) return; + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => { + settle({ ok: false, error: new Error("Bridge request body stalled before it completed.") }); + stream.destroy(); + }, armedIdleTimeoutMs); + idleTimer.unref?.(); + }; + + stream.on("data", (chunk: Buffer) => { + totalBytes += chunk.byteLength; + if (totalBytes > maxBodyBytes) { + settle({ ok: false, error: new Error("Bridge request body exceeded the configured size limit.") }); + stream.destroy(); + return; + } + chunks.push(chunk); + armIdleTimer(); + }); + stream.once("end", () => settle({ ok: true, body: Buffer.concat(chunks) })); + stream.once("error", (error) => + settle({ ok: false, error: error instanceof Error ? error : new Error(String(error)) }), + ); + stream.once("aborted", () => settle({ ok: false, error: new Error("Bridge request stream aborted.") })); + stream.once("close", () => + settle({ ok: false, error: new Error("Bridge request stream closed before it completed.") }), + ); + + return { + awaitBody(idleTimeoutMs: number, lifetimeCeilingMs: number): Promise { + return new Promise((resolve, reject) => { + const deliver = (result: Http2BridgeBodyCaptureOutcome) => { + if (result.ok) resolve(result.body); + else reject(result.error); + }; + if (outcome) { + // The body already finished (or the stream already went away) + // while this capture waited on admission: deliver it at once, and + // arm no timer for a wait that is already over. + deliver(outcome); + return; + } + onSettled = deliver; + armedIdleTimeoutMs = idleTimeoutMs; + armIdleTimer(); + // Arms one time, from here, and never rearms on a chunk: see + // DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS for why the + // ceiling must stay independent of progress. + lifetimeCeilingTimer = setTimeout(() => { + settle({ ok: false, error: new Error("Bridge request body passed the total lifetime ceiling.") }); + stream.destroy(); + }, lifetimeCeilingMs); + lifetimeCeilingTimer.unref?.(); + }); + }, + }; +} + +/** + * Wait for `stream` to reach its `close` event, or destroy it once + * `timeoutMs` passes with no close. `stream.end()` returns before the peer + * has read the response: the response bytes, and the admission slot the + * stream holds, stay in host memory until the peer actually drains them. + * This bounds that wait, so a peer that never reads a response cannot hold + * the slot open forever. + */ +function waitForStreamDrain(stream: http2.ServerHttp2Stream, timeoutMs: number): Promise { + if (stream.closed || stream.destroyed) { + return Promise.resolve(); + } + return new Promise((resolve) => { + let settled = false; + const onClose = () => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(); + }; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + stream.removeListener("close", onClose); + if (!stream.destroyed) stream.destroy(); + resolve(); + }, timeoutMs); + timer.unref?.(); + stream.once("close", onClose); + }); +} + function respondJson(stream: http2.ServerHttp2Stream, status: number, body: unknown): void { if (stream.destroyed || stream.closed) return; try { @@ -743,6 +904,8 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) const maxBufferedReadBytes = options.maxBufferedReadBytes ?? DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES; const readBackpressureStallMs = options.readBackpressureStallMs ?? DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS; + const admissionGate = options.admissionGate ?? getHttp2BridgeAdmissionGate(); + const responseDrainTimeoutMs = options.responseDrainTimeoutMs ?? DEFAULT_HTTP2_BRIDGE_RESPONSE_DRAIN_TIMEOUT_MS; // Built one time and passed to every readHttp2StreamBody() and // denyRequest() call below, so every stream on this server enforces the // same bounds. @@ -790,40 +953,78 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) headerAllowlist, ); - let body: Buffer; - try { - body = await readHttp2StreamBody(stream, bodyBounds); - } catch (error) { - respondJson(stream, 413, { error: error instanceof Error ? error.message : String(error) }); - return; - } + // Begin capturing the request body at once, in this same synchronous + // turn: the admission wait below can take an arbitrary time, and Node + // does not replay a body this handler was not yet listening for. The + // capture arms no timer until `awaitBody` runs, after admission. + const bodyCapture = beginHttp2BridgeStreamBodyCapture(stream, maxBodyBytes); - let result: Http2BridgeForwardResult; + // The admission gate bounds host memory across every bridge server and + // every session: take one slot before the body read completes, and hold + // it until the response leaves the host. A denial above takes no slot; + // only a routed, authenticated request reaches this point. + // + // Abort the queue wait if the stream closes while it still waits, so a + // cancelled request leaves the queue instead of taking a later slot. + const admissionAbortController = new AbortController(); + const abortAdmissionWaitOnClose = () => admissionAbortController.abort(); + stream.once("close", abortAdmissionWaitOnClose); + let releaseAdmissionSlot: () => void; try { - result = await options.forwardRequest({ - method, - pathname: parsedPath.value.pathname, - query: parsedPath.value.query, - headers: sanitizedHeaders, - body, - }); - } catch (error) { - respondJson(stream, 502, { error: error instanceof Error ? error.message : String(error) }); - return; - } - - if (stream.destroyed || stream.closed) return; - const responseHeaders: http2.OutgoingHttpHeaders = { ":status": result.status }; - for (const [key, value] of Object.entries(result.headers ?? {})) { - if (key.toLowerCase() === "content-length") continue; - responseHeaders[key] = value; - } - try { - stream.respond(responseHeaders); - stream.end(result.body); + releaseAdmissionSlot = await admissionGate.acquire(admissionAbortController.signal); } catch { - // The peer reset the stream (RST_STREAM) between dispatch and response. - // One stream's write fault stays local to that stream. + // The stream closed while it waited in the queue: there is nothing + // left to answer. + stream.removeListener("close", abortAdmissionWaitOnClose); + return; + } + stream.removeListener("close", abortAdmissionWaitOnClose); + + try { + let body: Buffer; + try { + body = await bodyCapture.awaitBody(requestBodyTimeoutMs, requestBodyLifetimeCeilingMs); + } catch (error) { + respondJson(stream, 413, { error: error instanceof Error ? error.message : String(error) }); + await waitForStreamDrain(stream, responseDrainTimeoutMs); + return; + } + + let result: Http2BridgeForwardResult; + try { + result = await options.forwardRequest({ + method, + pathname: parsedPath.value.pathname, + query: parsedPath.value.query, + headers: sanitizedHeaders, + body, + }); + } catch (error) { + respondJson(stream, 502, { error: error instanceof Error ? error.message : String(error) }); + await waitForStreamDrain(stream, responseDrainTimeoutMs); + return; + } + + if (stream.destroyed || stream.closed) return; + const responseHeaders: http2.OutgoingHttpHeaders = { ":status": result.status }; + for (const [key, value] of Object.entries(result.headers ?? {})) { + if (key.toLowerCase() === "content-length") continue; + responseHeaders[key] = value; + } + try { + stream.respond(responseHeaders); + stream.end(result.body); + // `stream.end()` returns before the peer has read the response: wait + // for the stream to actually close before the outer `finally` frees + // the slot, so the response bytes stay accounted for in host memory + // until they truly leave the host. + await waitForStreamDrain(stream, responseDrainTimeoutMs); + } catch { + // The peer reset the stream (RST_STREAM) between dispatch and response. + // One stream's write fault stays local to that stream. + } + } finally { + releaseAdmissionSlot(); } }