From 12f97efa9b8016fe84709bf11d28f24608f3a856 Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Wed, 26 Aug 2026 22:05:42 +0000 Subject: [PATCH] feat(adapter-utils): hold one HTTP/2 bridge admission slot for the whole host hold time Wire the admission gate from the prior phase into the bridge server. Every admitted stream now takes one gate slot after the route check, and holds it until the response stream closes or a drain bound frees it, so the gate's process-wide caps carry real weight against host memory instead of sitting unused. The body read now starts capturing data in the same synchronous turn as the stream event, ahead of the admission wait, and arms its idle and lifetime timers only after admission grants a slot. Node does not replay a request body the read was not yet listening for, so a body read that began only after an async admission wait could stall a request that had already fully arrived. Co-authored-by: Paperclip --- .../src/http2-bridge-server.test.ts | 425 ++++++++++++++++++ .../adapter-utils/src/http2-bridge-server.ts | 267 +++++++++-- 2 files changed, 659 insertions(+), 33 deletions(-) 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(); } }