diff --git a/packages/adapter-utils/src/http2-bridge-server.test.ts b/packages/adapter-utils/src/http2-bridge-server.test.ts index 86377297b59..04a34a418eb 100644 --- a/packages/adapter-utils/src/http2-bridge-server.test.ts +++ b/packages/adapter-utils/src/http2-bridge-server.test.ts @@ -1,3 +1,4 @@ +import { EventEmitter } from "node:events"; import { duplexPair } from "node:stream"; import type { Duplex } from "node:stream"; import http2 from "node:http2"; @@ -8,6 +9,7 @@ import { buildHttp2BridgeForwardUrl, classifyStreamAgainstGoaway, createHttp2BridgeServer, + discardHttp2StreamBody, parseCanonicalBridgeRequestPath, wrapDuplexChannelAsNodeDuplex, DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES, @@ -26,6 +28,7 @@ import { HTTP2_BRIDGE_SERVER_OPTIONS, HTTP2_BRIDGE_STREAM_RESET_BURST, HTTP2_BRIDGE_STREAM_RESET_RATE, + type Http2BridgeBodyBounds, type Http2BridgeForwardRequest, type Http2BridgeForwardResult, type Http2BridgeGoawayRecord, @@ -1736,6 +1739,62 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { } }); + it("denied streams never take an admission slot, even many at once across two server instances sharing a one-slot gate", async () => { + const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 1, maxSessions: 4 }); + const forwarderTracker = createForwarderCallTracker(); + const forwardRequest = async (): Promise => { + forwarderTracker.markCalled(); + return { status: 200 }; + }; + const serverA = bindTestServer({ admissionGate, forwardRequest }); + const serverB = bindTestServer({ admissionGate, forwardRequest }); + const rawClientA = connectRawClient(serverA.clientSide); + const rawClientB = connectRawClient(serverB.clientSide); + const denyOneStream = (rawClient: http2.ClientHttp2Session): Promise => + new Promise((resolve, reject) => { + // No `authorization` header: the token check denies every one of + // these streams before the admission gate ever runs. + const req = rawClient.request({ ":method": "POST", ":path": "/api/issues/abc/comments" }); + let status = 0; + req.on("response", (headers) => { + status = Number(headers[":status"]) || 0; + }); + req.on("error", reject); + req.on("close", () => resolve(status)); + req.resume(); + // A body on every one of six concurrent denied streams: if a + // denial ever took the one shared slot, five of the six would + // queue behind it instead of answering at once. + req.write(Buffer.alloc(200_000, "d")); + req.end(); + }); + try { + const startedAtMs = Date.now(); + const statuses = await Promise.all([ + denyOneStream(rawClientA), + denyOneStream(rawClientA), + denyOneStream(rawClientA), + denyOneStream(rawClientB), + denyOneStream(rawClientB), + denyOneStream(rawClientB), + ]); + // Every denied stream answered at once: none of them queued behind + // the one-slot gate. + expect(Date.now() - startedAtMs).toBeLessThan(2_000); + expect(statuses).toEqual([401, 401, 401, 401, 401, 401]); + expect(forwarderTracker.called).toBe(false); + // The gate count returns to its start value: a denial never moved + // it at all. + expect(admissionGate.activeCount).toBe(0); + expect(admissionGate.queuedCount).toBe(0); + } finally { + rawClientA.close(); + rawClientB.close(); + await serverA.handle.close(); + await serverB.handle.close(); + } + }); + it("the default caps keep the named byte total inside the reserved budget", () => { // 64 admitted streams x 868,351 bytes, plus 8 live sessions x // 16,777,216 bytes: 189,792,192 named bytes in total. This stays @@ -1787,3 +1846,53 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => { }); }); }); + +describe("discardHttp2StreamBody", () => { + /** A minimal fake `ServerHttp2Stream`: only the surface + * `discardHttp2StreamBody` touches — `on`/`once` for `data`, `end`, + * `error`, `aborted`, and `close`, plus a `destroy` a test can observe. */ + function fakeDeniedStream(): http2.ServerHttp2Stream & { destroyCallCount: number } { + const emitter = new EventEmitter() as unknown as http2.ServerHttp2Stream & { destroyCallCount: number }; + emitter.destroyCallCount = 0; + (emitter as unknown as { destroy: () => void }).destroy = () => { + emitter.destroyCallCount += 1; + emitter.emit("close"); + }; + return emitter; + } + + it("resolves with no buffer once the body ends, however many bytes the peer delivered", async () => { + const stream = fakeDeniedStream(); + const bounds: Http2BridgeBodyBounds = { maxBodyBytes: 1_000_000, idleTimeoutMs: 5_000, lifetimeCeilingMs: 60_000 }; + const discarded = discardHttp2StreamBody(stream, bounds); + + let deliveredBytes = 0; + const chunk = Buffer.alloc(300_000, "x"); + for (let i = 0; i < 3; i += 1) { + stream.emit("data", chunk); + deliveredBytes += chunk.byteLength; + } + stream.emit("end"); + + const outcome = await discarded; + // The helper hands back nothing: a buffer-returning helper would + // instead resolve here with a 900,000-byte `Buffer`, one copy of every + // chunk this denied stream ever delivered. + expect(outcome).toBeUndefined(); + expect(deliveredBytes).toBe(900_000); + expect(stream.destroyCallCount).toBe(0); + }); + + it("still destroys the stream once the running byte count passes maxBodyBytes", async () => { + const stream = fakeDeniedStream(); + const bounds: Http2BridgeBodyBounds = { maxBodyBytes: 10, idleTimeoutMs: 5_000, lifetimeCeilingMs: 60_000 }; + const rejection = discardHttp2StreamBody(stream, bounds).catch((error: unknown) => error); + + stream.emit("data", Buffer.alloc(20, "y")); + + const error = await rejection; + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toMatch(/exceeded the configured size limit/i); + expect(stream.destroyCallCount).toBe(1); + }); +}); diff --git a/packages/adapter-utils/src/http2-bridge-server.ts b/packages/adapter-utils/src/http2-bridge-server.ts index 89bc3eb15dd..ff4d0387448 100644 --- a/packages/adapter-utils/src/http2-bridge-server.ts +++ b/packages/adapter-utils/src/http2-bridge-server.ts @@ -657,23 +657,27 @@ export interface Http2BridgeBodyBounds { } /** - * Read one request body, bounded on size and on two independent time - * bounds: an idle bound and a total-lifetime ceiling. See + * Discard one denied request's body, bounded on size and on two independent + * time bounds: an idle bound and a total-lifetime ceiling. See * {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS} and * {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS} for why the * server enforces each one. * + * A denial never reads its own body content, so this function keeps no + * chunk: it counts bytes to enforce `bounds.maxBodyBytes`, then drops each + * chunk at once. The host holds one running number per denied stream, not a + * copy of the body. + * * The `close` listener is the settle-of-last-resort: it fires whenever the * 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. */ -function readHttp2StreamBody( +export function discardHttp2StreamBody( stream: http2.ServerHttp2Stream, bounds: Http2BridgeBodyBounds, -): Promise { +): Promise { return new Promise((resolve, reject) => { - const chunks: Buffer[] = []; let totalBytes = 0; let settled = false; let idleTimer: ReturnType; @@ -709,12 +713,13 @@ function readHttp2StreamBody( stream.destroy(); return; } - chunks.push(chunk); // The chunk is real progress, so the peer is not stalled: reset the // idle bound. The lifetime ceiling timer above does not reset here. + // The chunk itself is dropped here: a denial keeps no request body + // byte in host memory. armIdleTimer(); }); - stream.once("end", () => settle(() => resolve(Buffer.concat(chunks)))); + stream.once("end", () => settle(() => resolve())); 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.")))); @@ -889,14 +894,17 @@ function respondJson(stream: http2.ServerHttp2Stream, status: number, body: unkn } /** - * Answer a denied stream, then consume and discard its request body under - * the same bounds ({@link Http2BridgeBodyBounds}) an authenticated request - * gets. A denied request's body content never reaches the forward handler, - * but the inbound half of the stream still needs a bound: without one, a - * 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. + * Answer a denied stream, then discard its request body under the same + * bounds ({@link Http2BridgeBodyBounds}) an authenticated request gets. A + * denied request's body content never reaches the forward handler, but the + * inbound half of the stream still needs a bound: without one, a 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. This one function is every denial branch's only path to a body + * read, so every denial — an invalid token, a malformed path, or a denied + * route — discards through the same non-retaining helper and takes no + * admission-gate slot. Nothing awaits the discard; the caller has already + * answered the request and moves on to the next stream. */ function denyRequest( stream: http2.ServerHttp2Stream, @@ -906,7 +914,7 @@ function denyRequest( ): void { respondJson(stream, status, body); if (stream.destroyed || stream.closed) return; - readHttp2StreamBody(stream, bounds).catch(() => { + discardHttp2StreamBody(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. @@ -933,7 +941,7 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) 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 + // Built one time and passed to every discardHttp2StreamBody() and // denyRequest() call below, so every stream on this server enforces the // same bounds. const bodyBounds: Http2BridgeBodyBounds = { @@ -1104,7 +1112,8 @@ export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions) // timeout has not yet fired) would hold this wait open forever. The // grace timer bounds it: past `closeGraceMs`, the server force-destroys // the session, which ends its streams at once and settles their body - // reads through the `close` backstop in `readHttp2StreamBody`. + // reads through the `close` backstop in `discardHttp2StreamBody` and + // `beginHttp2BridgeStreamBodyCapture`. await Promise.all( [...activeSessions].map( (session) =>