From 729a51600551973120d115d816569275cb5756c4 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:40:44 +0000 Subject: [PATCH] test(fetch-http2-client): fetch in-process instead of spawning a child per test Most tests in this file only exercise the h2 client after ALPN has picked h2, which protocol: "http2" reaches through the same ClientSession code as the env flag, so they now fetch in-process against their own server instead of spawning a debug child each. The tests that are about a startup knob (env flag, --experimental-http2-fetch, protocol: "http1.1", the h1 fallback, BUN_CONFIG_HTTP_IDLE_TIMEOUT) and the use-after-free regression keep their subprocess. 60 spawns per run become 13, which also makes the ASAN spawn semaphore unnecessary. The idle-timeout test, which has to hold requests for 10s, is declared first so its hold overlaps the rest of the file's concurrent group instead of running after it. Timer-based waits are replaced with the event they were waiting for: a barrier request on the same connection for the RST_STREAM and stale-frame assertions, a PING round trip for the Expect: 100-continue proof, the server session's close for the GOAWAY reconnect, and the accept events for the coalescing tests. Assertions that only checked "rejected" or an exit code now check the exact error code, body, headers, session counts or the frames the client put on the wire. --- test/js/web/fetch/fetch-http2-client.test.ts | 1935 +++++++----------- 1 file changed, 755 insertions(+), 1180 deletions(-) diff --git a/test/js/web/fetch/fetch-http2-client.test.ts b/test/js/web/fetch/fetch-http2-client.test.ts index 96a5f9de2641..cbec8d71cefe 100644 --- a/test/js/web/fetch/fetch-http2-client.test.ts +++ b/test/js/web/fetch/fetch-http2-client.test.ts @@ -1,11 +1,31 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, isASAN, tls } from "harness"; +import { bunEnv, bunExe, tls } from "harness"; import { once } from "node:events"; import http2 from "node:http2"; import https from "node:https"; +import net from "node:net"; import nodetls from "node:tls"; import zlib from "node:zlib"; +// Most tests here fetch in-process with `protocol: "http2"`. That pins the +// ALPN offer to h2 only; once the handshake has picked h2 it runs the exact +// same ClientSession code as the BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT / +// --experimental-http2-fetch path (the flags only add http/1.1 to the offer +// and an h1 fallback). Every server below listens on its own ephemeral port, +// so each test gets a fresh session/pool entry even though they share one +// process. The tests that are about the negotiation itself (the env flag, the +// CLI flag, protocol: "http1.1", the h1 fallback, BUN_CONFIG_HTTP_IDLE_TIMEOUT) +// and the one whose failure mode is a crash spawn a subprocess instead, since +// those knobs are read at process startup. +const h2 = { protocol: "http2", tls: { rejectUnauthorized: false } } as const; +// h2 failures carry their name in `code`; aborts are DOMExceptions whose +// `code` is the legacy numeric ABORT_ERR, so fall back to `name` for those. +const errcode = (e: any) => (typeof e?.code === "string" ? e.code : e?.name); +const statusOrCode = (p: Promise) => p.then(r => r.status, errcode); +// A response whose failure may surface either from fetch() itself or from the +// body read, depending on whether the server's frames landed in one TLS read. +const bodyOrCode = (p: Promise) => p.then(r => r.text(), errcode).catch(errcode); + // allowHTTP1: false forces the server to reject anything that didn't // negotiate "h2" via ALPN, so these tests only pass when fetch actually // speaks HTTP/2 on the wire. @@ -25,17 +45,45 @@ function makeH2Server( return server; } +// Listens on an ephemeral port, runs `fn`, then destroys every session the +// server accepted: an in-process fetch() keeps its h2 session pooled after the +// response, and server.close() alone would leave those connections open. +async function listenH2(server: http2.Http2SecureServer, fn: (url: string) => Promise) { + const sessions = new Set(); + server.on("session", s => sessions.add(s)); + server.listen(0); + await once(server, "listening"); + const { port } = server.address() as import("node:net").AddressInfo; + try { + await fn(`https://localhost:${port}`); + } finally { + for (const s of sessions) s.destroy(); + server.close(); + } +} + async function withH2Server( handler: (req: http2.Http2ServerRequest, res: http2.Http2ServerResponse) => void, fn: (url: string, server: http2.Http2SecureServer) => Promise, ) { const server = makeH2Server({}, handler); - server.listen(0); + await listenH2(server, url => fn(url, server)); +} + +// Same for servers that are not http2 servers (https / raw tls / plain tcp): +// sockets still open when `fn` returns are destroyed. The url is built on +// 127.0.0.1 so connection counts are exact: "localhost" resolves to both +// loopback addresses and fetch races a TCP connect to each of them. +async function listenTcp(server: net.Server, scheme: "http" | "https", fn: (url: string) => Promise) { + const sockets = new Set(); + server.on("connection", s => sockets.add(s)); + server.listen(0, "127.0.0.1"); await once(server, "listening"); const { port } = server.address() as import("node:net").AddressInfo; try { - await fn(`https://localhost:${port}`, server); + await fn(`${scheme}://127.0.0.1:${port}`); } finally { + for (const s of sockets) s.destroy(); server.close(); } } @@ -68,6 +116,12 @@ const hpackStatus = (code: 100 | 200 | 204 | 404) => const hpackLit = (name: string, value: string) => Buffer.concat([Buffer.from([0x10, name.length]), Buffer.from(name), Buffer.from([value.length]), Buffer.from(value)]); +const FRAME_DATA = 0; +const FRAME_HEADERS = 1; +const FLAG_END_STREAM = 1; + +type RawFrame = { id: number; type: number; flags: number; len: number }; + type RawConn = { socket: nodetls.TLSSocket; settings(): void; @@ -75,35 +129,29 @@ type RawConn = { data(streamId: number, chunk: string | Buffer, endStream?: boolean): void; rst(streamId: number, code: number): void; goaway(lastId: number, code: number): void; + /** Sends a PING and resolves once the client ACKs it. The client answers + * frames in arrival order, so the ACK proves every frame it wrote before + * seeing the PING has already reached the `onFrame` handler. */ + ping(): Promise; + /** Called for every frame the client sends on a non-zero stream id. */ + onFrame?: (f: RawFrame) => void; }; type RawState = { connections: number; rst: Array<{ id: number; code: number }>; - /** Resolves once every accepted socket has emitted `close` — i.e. all - * in-flight client frames have been delivered to the `data` handler. - * Tests asserting on server-side capture (`state.rst`) should await this - * instead of racing the subprocess exit. */ - allClosed: () => Promise; + /** Every stream-addressed frame received from the client, in wire order. */ + frames: RawFrame[]; }; async function withRawH2Server( onStream: (conn: RawConn, streamId: number, connIndex: number) => void, fn: (url: string, state: RawState) => Promise, ) { - const closed: Promise[] = []; - const state: RawState = { - connections: 0, - rst: [], - allClosed: () => Promise.all(closed).then(() => {}), - }; + const state: RawState = { connections: 0, rst: [], frames: [] }; const server = nodetls.createServer({ ...tls, ALPNProtocols: ["h2"] }, socket => { const connIndex = state.connections++; - // Track teardown with a promise that resolves on "close" even when the - // client tears the connection down abortively (RST → ECONNRESET emits an - // "error" first). `once(socket, "close")` would reject in that case, - // making allClosed() throw and leaving unhandled rejections behind. - closed.push(new Promise(resolve => socket.once("close", resolve))); + const pingWaiters: Array<() => void> = []; const conn: RawConn = { socket, settings: () => socket.write(frame(4, 0, 0)), @@ -113,6 +161,11 @@ async function withRawH2Server( socket.write(frame(0, end ? 1 : 0, id, typeof chunk === "string" ? Buffer.from(chunk) : chunk)), rst: (id, code) => socket.write(frame(3, 0, id, u32be(code))), goaway: (lastId, code) => socket.write(frame(7, 0, 0, Buffer.concat([u32be(lastId), u32be(code)]))), + ping: () => + new Promise(resolve => { + pingWaiters.push(resolve); + socket.write(frame(6, 0, 0, Buffer.alloc(8))); + }), }; let buf = Buffer.alloc(0); let prefaceSeen = false; @@ -133,62 +186,144 @@ async function withRawH2Server( const payload = buf.subarray(9, 9 + len); buf = buf.subarray(9 + len); if (type === 4 && !(flags & 1)) socket.write(frame(4, 1, 0)); // ack their SETTINGS + if (type === 6 && flags & 1) pingWaiters.shift()?.(); + if (id !== 0) { + const f = { id, type, flags, len }; + state.frames.push(f); + conn.onFrame?.(f); + } if (type === 1) onStream(conn, id, connIndex); // HEADERS opens a stream if (type === 3) state.rst.push({ id, code: payload.readUInt32BE(0) }); } }); socket.on("error", () => {}); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await fn(`https://localhost:${port}`, state); - } finally { - server.close(); - } -} - -// Each test spawns a fresh subprocess so the BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT -// env var is read at startup. With describe.concurrent + max_concurrency=20 that -// peaks at ~8GB of debug subprocesses, which under ASAN (~2-3x) OOM-kills the -// 16GB/0-swap runner; fully serialising under ASAN instead pushes the 50+ spawns -// past the 3-minute file timeout. Cap live subprocesses to a few under ASAN so -// describe.concurrent can still overlap them without exhausting memory. Non-ASAN -// keeps the cap at max_concurrency so the semaphore is a no-op. -const slotLimit = isASAN ? 4 : 20; -let live = 0; -const waiters: Array<() => void> = []; -async function spawnCapped(options: Parameters[0]) { - if (live >= slotLimit) { - await new Promise(r => waiters.push(r)); // slot handed off to us, `live` already counts it - } else { - live++; - } - const proc = Bun.spawn(options); - proc.exited.finally(() => { - const next = waiters.shift(); - if (next) - next(); // hand the slot off directly; `live` stays as-is - else live--; - }); - return proc; + await listenTcp(server, "https", url => fn(url, state)); } -function spawnFetch(script: string) { - return spawnCapped({ - cmd: [bunExe(), "--no-warnings", "-e", script], - env: { - ...bunEnv, - BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT: "1", - NODE_TLS_REJECT_UNAUTHORIZED: "0", - }, +// The handful of tests that need a knob read at process startup run fetch in +// a child. `env` holds exactly those knobs; by default that is the h2 env flag. +const h2EnvFlag = { BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT: "1" }; +function spawnFetch(script: string, env: Record = h2EnvFlag, ...flags: string[]) { + return Bun.spawn({ + cmd: [bunExe(), "--no-warnings", ...flags, "-e", script], + env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0", ...env }, stdout: "pipe", stderr: "pipe", }); } +async function collect(proc: ReturnType) { + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout: stdout.trim(), stderr, exitCode }; +} + +// https://github.com/oven-sh/bun/issues/16682 (h2 aggregate path): the +// session's shared socket timer is the max over every attached client's +// effective idle deadline. +// +// Declared before everything else on purpose: it has to hold its requests +// open for 10s of wall time, and as the first member of the file's concurrent +// group that hold overlaps the rest of the file instead of being added to it. +test.concurrent( + "h2: per-request `timeout` extends the session idle deadline, and {timeout:false} is not killed by a sibling's shorter explicit timeout", + async () => { + const HOLD_MS = 10_000; + const holdTimers = new Set>(); + const server = makeH2Server({}, (_req, res) => { + // Hold every request idle past uSockets' worst-case firing window for a + // 1s short-tick timer (~5s), then respond. + const timer = setTimeout(() => { + holdTimers.delete(timer); + try { + res.end("hello"); + } catch {} + }, HOLD_MS); + holdTimers.add(timer); + }); + try { + await listenH2(server, async url => { + const run = (idleDefault: string, body: string) => + collect( + spawnFetch( + /* js */ ` + const url = ${JSON.stringify(url)}; + const get = init => fetch(url, { tls: { rejectUnauthorized: false }, ...init }) + .then(r => r.text(), e => "ERR:" + (e?.code ?? e?.name ?? e)); + ${body} + `, + { ...h2EnvFlag, BUN_CONFIG_HTTP_IDLE_TIMEOUT: idleDefault }, + ), + ); + const [extendsDefault, floorsSibling, disarmsOnGlobalZero] = await Promise.all([ + // Global idle default = 1s. `{timeout:60000}` must extend the shared + // socket's deadline past the 10s hold; the `{timeout:false}` sibling + // coalesces onto the same session and rides along. + run( + "1", + /* js */ ` + const [longTimeout, noTimeout] = await Promise.all([ + get({ timeout: 60_000 }), + get({ timeout: false }), + ]); + console.log(JSON.stringify({ longTimeout, noTimeout })); + `, + ), + // Global idle default = 20s. `{timeout:false}` contributes 0 to the + // session max and the `{timeout:1000}` sibling contributes 1s; the + // session must floor at the 20s global default so the no-timeout + // stream is not killed by the sibling's short explicit deadline. + run( + "20", + /* js */ ` + const [noTimeout, shortTimeout] = await Promise.all([ + get({ timeout: false }), + get({ timeout: 1000 }), + ]); + console.log(JSON.stringify({ noTimeout, shortTimeout })); + `, + ), + // Global idle default = 0 (disabled). A plain fetch with no `timeout` + // option inherits effective deadline 0 without setting the + // `disable_timeout` flag; the session must still disarm rather than + // letting the `{timeout:1000}` sibling arm the shared socket. + run( + "0", + /* js */ ` + const [plain, shortTimeout] = await Promise.all([ + get(undefined), + get({ timeout: 1000 }), + ]); + console.log(JSON.stringify({ plain, shortTimeout })); + `, + ), + ]); + expect(extendsDefault).toEqual({ + stdout: JSON.stringify({ longTimeout: "hello", noTimeout: "hello" }), + stderr: "", + exitCode: 0, + }); + expect(floorsSibling).toEqual({ + stdout: JSON.stringify({ noTimeout: "hello", shortTimeout: "hello" }), + stderr: "", + exitCode: 0, + }); + expect(disarmsOnGlobalZero).toEqual({ + stdout: JSON.stringify({ plain: "hello", shortTimeout: "hello" }), + stderr: "", + exitCode: 0, + }); + }); + } finally { + for (const timer of holdTimers) clearTimeout(timer); + } + }, + 60_000, +); + describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT)", () => { + // The two round-trips below keep the env-flag (ALPN "h2, http/1.1") path + // covered end to end; the rest of the file fetches in-process. test("GET: status, headers and body round-trip", async () => { await withH2Server( (req, res) => { @@ -200,7 +335,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL res.end("hello over h2"); }, async url => { - await using proc = await spawnFetch(` + await using proc = spawnFetch(` const res = await fetch(${JSON.stringify(url)} + "/hello?x=1", { headers: { "X-Foo": "bar" }, tls: { rejectUnauthorized: false }, @@ -216,10 +351,9 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL body, })); `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - const out = JSON.parse(stdout); - expect(out).toEqual({ + expect(JSON.parse(stdout)).toEqual({ status: 201, ct: "text/plain", seenPath: "/hello?x=1", @@ -241,21 +375,24 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL req.on("data", c => (body += c)); req.on("end", () => { res.writeHead(200, { "content-type": "application/json" }); - res.end(JSON.stringify({ got: body, method: req.method })); + res.end(JSON.stringify({ got: body, method: req.method, httpVersion: req.httpVersion })); }); }, async url => { - await using proc = await spawnFetch(` + await using proc = spawnFetch(` const res = await fetch(${JSON.stringify(url)} + "/echo", { method: "POST", body: "the payload", tls: { rejectUnauthorized: false }, }); - process.stdout.write(await res.text()); + console.log(JSON.stringify({ status: res.status, echoed: await res.json() })); `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - expect(JSON.parse(stdout)).toEqual({ got: "the payload", method: "POST" }); + expect(JSON.parse(stdout)).toEqual({ + status: 200, + echoed: { got: "the payload", method: "POST", httpVersion: "2.0" }, + }); expect(exitCode).toBe(0); }, ); @@ -269,15 +406,9 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL res.end(big); }, async url => { - await using proc = await spawnFetch(` - const res = await fetch(${JSON.stringify(url)}, { tls: { rejectUnauthorized: false } }); - const buf = await res.arrayBuffer(); - console.log(buf.byteLength); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe(String(big.length)); - expect(exitCode).toBe(0); + const res = await fetch(url, h2); + expect(res.status).toBe(200); + expect(await res.text()).toBe(big); }, ); }); @@ -291,14 +422,12 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL res.end(gz); }, async url => { - await using proc = await spawnFetch(` - const res = await fetch(${JSON.stringify(url)}, { tls: { rejectUnauthorized: false } }); - process.stdout.write(await res.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe(payload); - expect(exitCode).toBe(0); + const res = await fetch(url, h2); + expect({ status: res.status, ct: res.headers.get("content-type"), body: await res.text() }).toEqual({ + status: 200, + ct: "text/plain", + body: payload, + }); }, ); }); @@ -307,41 +436,33 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL let sessions = 0; let maxOpen = 0; let open = 0; + const held: Array<() => void> = []; const server = makeH2Server(); server.on("session", () => sessions++); server.on("stream", (stream, headers) => { + const path = String(headers[":path"]); + const answer = () => { + stream.respond({ ":status": 200 }); + stream.end(path); + }; + if (path === "/warmup") return answer(); open++; maxOpen = Math.max(maxOpen, open); stream.on("close", () => open--); - // Hold each stream briefly so all 8 are open at once. - setTimeout(() => { - stream.respond({ ":status": 200 }); - stream.end(String(headers[":path"])); - }, 100); + // Hold every stream of the burst until all 8 are open at once, then + // answer them: maxOpen === 8 below is only reachable by multiplexing. + held.push(answer); + if (held.length === 8) for (const a of held.splice(0)) a(); + }); + await listenH2(server, async url => { + // Warmup so the session exists before the concurrent burst. + expect(await fetch(`${url}/warmup`, h2).then(r => r.text())).toBe("/warmup"); + const results = await Promise.all( + Array.from({ length: 8 }, (_, i) => fetch(`${url}/${i}`, h2).then(r => r.text())), + ); + expect(results).toEqual(["/0", "/1", "/2", "/3", "/4", "/5", "/6", "/7"]); + expect({ sessions, maxOpen }).toEqual({ sessions: 1, maxOpen: 8 }); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - // Warmup so the session exists before the concurrent burst. - await fetch(url + "/warmup", opts).then(r => r.text()); - const results = await Promise.all( - Array.from({ length: 8 }, (_, i) => fetch(url + "/" + i, opts).then(r => r.text())) - ); - console.log(results.join(",")); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("/0,/1,/2,/3,/4,/5,/6,/7"); - expect(exitCode).toBe(0); - expect(sessions).toBe(1); - expect(maxOpen).toBe(8); - } finally { - server.close(); - } }); test("POST with ReadableStream body streams as raw DATA frames", async () => { @@ -356,29 +477,22 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL }); }, async url => { - await using proc = await spawnFetch(` - const chunks = ["alpha-", "bravo-", "charlie-", "delta-", "echo"]; - const body = new ReadableStream({ - async pull(ctrl) { - for (const c of chunks) { - ctrl.enqueue(new TextEncoder().encode(c)); - await new Promise(r => setTimeout(r, 5)); - } - ctrl.close(); - }, - }); - const res = await fetch("${url}/stream", { - method: "POST", - body, - duplex: "half", - tls: { rejectUnauthorized: false }, - }); - console.log(res.status, res.headers.get("x-len"), await res.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 30 alpha-bravo-charlie-delta-echo"); - expect(exitCode).toBe(0); + const chunks = ["alpha-", "bravo-", "charlie-", "delta-", "echo"]; + // One chunk per pull(): each is handed to the client on its own turn, + // so the upload goes out as a sequence of DATA frames. + const body = new ReadableStream({ + pull(ctrl) { + const c = chunks.shift(); + if (c === undefined) ctrl.close(); + else ctrl.enqueue(new TextEncoder().encode(c)); + }, + }); + const res = await fetch(`${url}/stream`, { ...h2, method: "POST", body, duplex: "half" }); + expect({ status: res.status, len: res.headers.get("x-len"), body: await res.text() }).toEqual({ + status: 200, + len: "30", + body: "alpha-bravo-charlie-delta-echo", + }); // No chunked-encoding artifacts leaked into the framed body. expect(received).toBe("alpha-bravo-charlie-delta-echo"); }, @@ -400,49 +514,30 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.end(Buffer.concat(chunks)); }); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}/"; - const opts = { tls: { rejectUnauthorized: false } }; - const N = 24, M = 24; - // Warmup so the h2 session exists and SETTINGS have been exchanged - // before the concurrent burst; otherwise requests can fan out to - // additional connections while the first is still handshaking. - await fetch(url, { ...opts, method: "POST", body: "warmup" }).then(r => r.text()); - const results = await Promise.all( - Array.from({ length: N }, (_, i) => { - let k = 0; - const body = new ReadableStream({ - pull(ctrl) { - if (k < M) { - ctrl.enqueue(new TextEncoder().encode(i + ":" + k + ",")); - k++; - } else ctrl.close(); - }, - }); - return fetch(url, { ...opts, method: "POST", body, duplex: "half" }).then(r => r.text()); - }), - ); - const bad = results - .map((got, i) => { - const want = Array.from({ length: M }, (_, k) => i + ":" + k + ",").join(""); - return got === want ? null : i + " got=" + JSON.stringify(got); - }) - .filter(Boolean); - if (bad.length) throw new Error("mismatch: " + bad.join(" | ")); - console.log("ok", results.length); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("ok 24"); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + const N = 24, + M = 24; + // Warmup so the h2 session exists and SETTINGS have been exchanged + // before the concurrent burst; otherwise requests can fan out to + // additional connections while the first is still handshaking. + expect(await fetch(url, { ...h2, method: "POST", body: "warmup" }).then(r => r.text())).toBe("warmup"); + const results = await Promise.all( + Array.from({ length: N }, (_, i) => { + let k = 0; + const body = new ReadableStream({ + pull(ctrl) { + if (k < M) ctrl.enqueue(new TextEncoder().encode(`${i}:${k++},`)); + else ctrl.close(); + }, + }); + return fetch(url, { ...h2, method: "POST", body, duplex: "half" }).then(r => r.text()); + }), + ); + expect(results).toEqual( + Array.from({ length: N }, (_, i) => Array.from({ length: M }, (_, k) => `${i}:${k},`).join("")), + ); expect(sessions).toBe(1); - } finally { - server.close(); - } + }); }); test("POST with ReadableStream body larger than initial send window", async () => { @@ -456,28 +551,17 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL }); }, async url => { - await using proc = await spawnFetch(` - // 256 KiB > 64 KiB default INITIAL_WINDOW_SIZE: requires the - // client to honour the server's WINDOW_UPDATE before continuing. - const buf = new Uint8Array(256 * 1024).fill(0x61); - const body = new ReadableStream({ - start(ctrl) { - for (let i = 0; i < 4; i++) ctrl.enqueue(buf.subarray(i * 65536, (i + 1) * 65536)); - ctrl.close(); - }, - }); - const res = await fetch("${url}/big", { - method: "POST", - body, - duplex: "half", - tls: { rejectUnauthorized: false }, - }); - console.log(res.status, await res.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 262144"); - expect(exitCode).toBe(0); + // 256 KiB > 64 KiB default INITIAL_WINDOW_SIZE: requires the + // client to honour the server's WINDOW_UPDATE before continuing. + const buf = new Uint8Array(256 * 1024).fill(0x61); + const body = new ReadableStream({ + start(ctrl) { + for (let i = 0; i < 4; i++) ctrl.enqueue(buf.subarray(i * 65536, (i + 1) * 65536)); + ctrl.close(); + }, + }); + const res = await fetch(`${url}/big`, { ...h2, method: "POST", body, duplex: "half" }); + expect({ status: res.status, body: await res.text() }).toEqual({ status: 200, body: "262144" }); }, ); }); @@ -490,27 +574,14 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.respond({ ":status": 200 }); stream.end(String(headers[":path"])); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - // No warmup: all 12 race the same fresh handshake. - const results = await Promise.all( - Array.from({ length: 12 }, (_, i) => fetch(url + "/" + i, opts).then(r => r.text())) - ); - console.log(results.sort().join(",")); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("/0,/1,/10,/11,/2,/3,/4,/5,/6,/7,/8,/9"); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + // No warmup: all 12 race the same fresh handshake. + const results = await Promise.all( + Array.from({ length: 12 }, (_, i) => fetch(`${url}/${i}`, h2).then(r => r.text())), + ); + expect(results).toEqual(Array.from({ length: 12 }, (_, i) => `/${i}`)); expect(sessions).toBe(1); - } finally { - server.close(); - } + }); }); test("abort sends RST_STREAM; siblings on the session survive", async () => { @@ -527,36 +598,22 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.end("survivor"); } }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - // Warmup so /slow, /fast, /after share one session. - await fetch(url + "/warmup", opts).then(r => r.text()); - const ac = new AbortController(); - const slow = fetch(url + "/slow", { ...opts, signal: ac.signal }).catch(e => "aborted:" + e.name); - const fast = fetch(url + "/fast", opts).then(r => r.text()); - await fast; - ac.abort(); - await slow; - const after = await fetch(url + "/after", opts).then(r => r.text()); - console.log([await slow, await fast, after].join(",")); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("aborted:AbortError,survivor,survivor"); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + // Warmup so /slow, /fast, /after share one session. + expect(await fetch(`${url}/warmup`, h2).then(r => r.text())).toBe("survivor"); + const ac = new AbortController(); + const slow = fetch(`${url}/slow`, { ...h2, signal: ac.signal }).then(() => "resolved", errcode); + const fast = await fetch(`${url}/fast`, h2).then(r => r.text()); + ac.abort(); + expect(await slow).toBe("AbortError"); + const after = await fetch(`${url}/after`, h2).then(r => r.text()); + expect({ fast, after }).toEqual({ fast: "survivor", after: "survivor" }); // Aborting one stream must not tear down the connection: all four - // requests rode one session, and /slow's stream closed (RST_STREAM) - // while /fast and /after on the same session completed. + // requests rode one session, and /slow's stream was closed with + // RST_STREAM(CANCEL) while /fast and /after on the same session completed. expect(sessions).toBe(1); - await slowClosed; - } finally { - server.close(); - } + expect(await slowClosed).toBe(http2.constants.NGHTTP2_CANCEL); + }); }); test("server SETTINGS_MAX_CONCURRENT_STREAMS=1 is honoured per session", async () => { @@ -569,36 +626,25 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL open++; perSessionMax[idx] = Math.max(perSessionMax[idx], open); stream.on("close", () => open--); + // Keep the stream open for a moment so a client that ignored the cap + // would be caught with a second stream open on this session. setTimeout(() => { stream.respond({ ":status": 200 }); stream.end("x"); }, 30); }); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - // First request alone so the server's SETTINGS arrives before the - // burst, then fire 4 concurrently against the cap. - await fetch(url, opts).then(r => r.text()); - await Promise.all(Array.from({ length: 4 }, () => fetch(url, opts).then(r => r.text()))); - console.log("ok"); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("ok"); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + // First request alone so the server's SETTINGS arrives before the + // burst, then fire 4 concurrently against the cap. + expect(await fetch(url, h2).then(r => r.text())).toBe("x"); + const burst = await Promise.all(Array.from({ length: 4 }, () => fetch(url, h2).then(r => r.text()))); + expect(burst).toEqual(["x", "x", "x", "x"]); // The cap is per-connection: no session may ever see >1 open stream. // Excess concurrent requests fan out to additional connections. - for (const max of perSessionMax) expect(max).toBe(1); expect(perSessionMax.length).toBeGreaterThanOrEqual(1); - } finally { - server.close(); - } + expect(perSessionMax).toEqual(perSessionMax.map(() => 1)); + }); }); test("keep-alive: sequential requests reuse one h2 session", async () => { @@ -611,34 +657,22 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.respond({ ":status": 200, "content-type": "text/plain" }); stream.end(`req=${headers[":path"]}`); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - for (let i = 0; i < 4; i++) { - const r = await fetch(url + "/" + i, opts); - console.log(await r.text()); - } - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim().split("\n")).toEqual(["req=/0", "req=/1", "req=/2", "req=/3"]); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + const bodies: string[] = []; + for (let i = 0; i < 4; i++) bodies.push(await fetch(`${url}/${i}`, h2).then(r => r.text())); + expect(bodies).toEqual(["req=/0", "req=/1", "req=/2", "req=/3"]); expect(sessions).toBe(1); // stream ids must be fresh odd numbers on the reused session expect(seen).toEqual([1, 3, 5, 7]); - } finally { - server.close(); - } + }); }); test("GOAWAY after a request: next request reconnects", async () => { - let sessions = 0; + // One "closed by the peer" promise per accepted session, armed as soon as + // the session exists so a close that lands before we look is not missed. + const sessionClosed: Promise[] = []; const server = makeH2Server(); - server.on("session", () => sessions++); + server.on("session", s => sessionClosed.push(new Promise(resolve => s.once("close", resolve)))); server.on("stream", (stream, headers) => { const session = stream.session!; stream.respond({ ":status": 200 }); @@ -647,26 +681,16 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL session.goaway(http2.constants.NGHTTP2_NO_ERROR, stream.id); } }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - const a = await (await fetch(url + "/first", opts)).text(); - await Bun.sleep(50); - const b = await (await fetch(url + "/second", opts)).text(); - console.log(a + "," + b); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("ok,ok"); - expect(exitCode).toBe(0); - expect(sessions).toBe(2); - } finally { - server.close(); - } + await listenH2(server, async url => { + const a = await fetch(`${url}/first`, h2).then(r => r.text()); + // A GOAWAY'd session can't be pooled, so the client closes it as soon as + // it has processed the frame; once that close has arrived the next + // request cannot land on the old session, so it has to open a new one. + await sessionClosed[0]; + const b = await fetch(`${url}/second`, h2).then(r => r.text()); + expect([a, b]).toEqual(["ok", "ok"]); + expect(sessionClosed).toHaveLength(2); + }); }); test("response body larger than initial window triggers WINDOW_UPDATE", async () => { @@ -677,17 +701,11 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL res.end(big); }, async url => { - await using proc = await spawnFetch(` - const res = await fetch(${JSON.stringify(url)}, { tls: { rejectUnauthorized: false } }); - const buf = new Uint8Array(await res.arrayBuffer()); - let ok = buf.length === ${big.length}; - for (let i = 0; ok && i < buf.length; i += 4096) ok = buf[i] === 0x61; - console.log(ok ? "ok" : "bad:" + buf.length); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("ok"); - expect(exitCode).toBe(0); + const res = await fetch(url, h2); + const body = await res.bytes(); + expect(res.status).toBe(200); + expect(body.byteLength).toBe(big.byteLength); + expect(big.equals(body)).toBe(true); }, ); }); @@ -699,62 +717,55 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.on("wantTrailers", () => stream.sendTrailers({ "x-trailer": "hello" })); stream.end("body-text"); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const r = await fetch("https://localhost:${port}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 body-text"); - expect(exitCode).toBe(0); - } finally { - server.close(); - } + await listenH2(server, async url => { + const r = await fetch(url, h2); + expect({ status: r.status, ct: r.headers.get("content-type"), body: await r.text() }).toEqual({ + status: 200, + ct: "text/plain", + body: "body-text", + }); + // Trailer fields must not be merged into the response headers. + expect(r.headers.get("x-trailer")).toBeNull(); + }); }); // Bun's node:http2 server currently emits an empty DATA+END_STREAM for // stream.close(code) rather than RST_STREAM, so this also covers the - // RFC 9113 §8.1 "DATA before HEADERS" stream-error case. + // RFC 9113 §8.1 "DATA before HEADERS" stream-error case (HTTP2ProtocolError); + // once it sends a real RST_STREAM the request fails with HTTP2StreamReset. test("server-reset stream fails that request; sibling on the session survives", async () => { let sessions = 0; + let good: http2.ServerHttp2Stream | undefined; + let badReset = false; + const answerGood = () => { + if (!good || !badReset) return; + good.respond({ ":status": 200 }); + good.end("ok"); + }; const server = makeH2Server(); server.on("session", () => sessions++); server.on("stream", (stream, headers) => { stream.on("error", () => {}); if (headers[":path"] === "/bad") { stream.close(http2.constants.NGHTTP2_PROTOCOL_ERROR); - return; + badReset = true; + } else { + good = stream; } - setTimeout(() => { - stream.respond({ ":status": 200 }); - stream.end("ok"); - }, 50); + // /good is only answered once /bad has been reset, so its response is + // queued behind the reset on the one connection: receiving it proves + // the client processed the reset without dropping the session. + answerGood(); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const opts = { tls: { rejectUnauthorized: false } }; - const [good, bad] = await Promise.allSettled([ - fetch(url + "/good", opts).then(r => r.text()), - fetch(url + "/bad", opts).then(r => r.text()), - ]); - console.log(good.status, good.value, bad.status); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("fulfilled ok rejected"); - expect(exitCode).toBe(0); + await listenH2(server, async url => { + const [goodResult, badResult] = await Promise.all([ + fetch(`${url}/good`, h2).then(r => r.text()), + bodyOrCode(fetch(`${url}/bad`, h2)), + ]); + expect(goodResult).toBe("ok"); + expect(badResult).toMatch(/^HTTP2(ProtocolError|StreamReset)$/); expect(sessions).toBe(1); - } finally { - server.close(); - } + }); }); test("connection-specific request headers are stripped before HPACK", async () => { @@ -765,34 +776,22 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.respond({ ":status": 200 }); stream.end(); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const r = await fetch("https://localhost:${port}", { - headers: { - "x-keep": "me", - "Connection": "keep-alive", - "Keep-Alive": "timeout=5", - "Proxy-Connection": "x", - "Transfer-Encoding": "chunked", - "Upgrade": "ws", - }, - tls: { rejectUnauthorized: false }, - }); - console.log(r.status); - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("200"); - expect(exitCode).toBe(0); - expect(seen).toContain("x-keep"); - for (const bad of ["connection", "keep-alive", "proxy-connection", "transfer-encoding", "upgrade"]) { - expect(seen).not.toContain(bad); - } - } finally { - server.close(); - } + await listenH2(server, async url => { + const sent = { + "x-keep": "me", + "Connection": "keep-alive", + "Keep-Alive": "timeout=5", + "Proxy-Connection": "x", + "Transfer-Encoding": "chunked", + "Upgrade": "ws", + }; + const r = await fetch(url, { ...h2, headers: sent }); + expect(r.status).toBe(200); + // Of the headers we sent, only the non-connection-specific one reached + // the server (fetch adds its own accept/user-agent/etc. on top). + const ours = Object.keys(sent).map(k => k.toLowerCase()); + expect(seen.filter(k => ours.includes(k))).toEqual(["x-keep"]); + }); }); test("multiple Set-Cookie response headers survive HPACK decode", async () => { @@ -801,21 +800,11 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL stream.respond({ ":status": 200, "set-cookie": ["a=b", "c=d", "e=f"] }); stream.end(); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const r = await fetch("https://localhost:${port}", { tls: { rejectUnauthorized: false } }); - console.log(JSON.stringify(r.headers.getSetCookie())); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe(`["a=b","c=d","e=f"]`); - expect(exitCode).toBe(0); - } finally { - server.close(); - } + await listenH2(server, async url => { + const r = await fetch(url, h2); + expect(r.status).toBe(200); + expect(r.headers.getSetCookie()).toEqual(["a=b", "c=d", "e=f"]); + }); }); describe("raw frame server", () => { @@ -828,16 +817,15 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(204), { endStream: true }); }, async (url, state) => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("204"); - expect(exitCode).toBe(0); - expect(attempts).toBe(2); - expect(state.connections).toBe(1); + const r = await fetch(url, h2); + expect({ status: r.status, body: await r.text(), attempts, connections: state.connections }).toEqual({ + status: 204, + body: "", + attempts: 2, + connections: 1, + }); + // The retry must use a fresh stream id on the same connection. + expect(state.frames.filter(f => f.type === FRAME_HEADERS).map(f => f.id)).toEqual([1, 3]); }, ); }); @@ -850,13 +838,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.rst(id, http2.constants.NGHTTP2_REFUSED_STREAM); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", String(e).includes("Refused")); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected true"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2RefusedStream"); // initial + 5 retries expect(attempts).toBe(6); }, @@ -871,13 +853,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.rst(id, http2.constants.NGHTTP2_PROTOCOL_ERROR); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected"); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2StreamReset"); expect(attempts).toBe(1); }, ); @@ -896,15 +872,12 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.data(id, "second-conn", true); }, async (url, state) => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 second-conn"); - expect(exitCode).toBe(0); - expect(state.connections).toBe(2); + const r = await fetch(url, h2); + expect({ status: r.status, body: await r.text(), connections: state.connections }).toEqual({ + status: 200, + body: "second-conn", + connections: 2, + }); }, ); }); @@ -917,16 +890,15 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.rst(id, http2.constants.NGHTTP2_REFUSED_STREAM); }, async url => { - await using proc = await spawnFetch(` - const body = new ReadableStream({ start(c) { c.enqueue(new Uint8Array([1,2,3])); c.close(); } }); - try { - await fetch("${url}", { method: "POST", body, duplex: "half", tls: { rejectUnauthorized: false } }); - console.log("ok"); - } catch (e) { console.log("rejected"); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected"); - expect(exitCode).toBe(0); + const body = new ReadableStream({ + start(c) { + c.enqueue(new Uint8Array([1, 2, 3])); + c.close(); + }, + }); + expect(await statusOrCode(fetch(url, { ...h2, method: "POST", body, duplex: "half" }))).toBe( + "HTTP2RefusedStream", + ); expect(attempts).toBe(1); }, ); @@ -943,14 +915,8 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.socket.write(frame(0, 0x8 | 0x1, id, payload)); }, async url => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 padded-body"); - expect(exitCode).toBe(0); + const r = await fetch(url, h2); + expect({ status: r.status, body: await r.text() }).toEqual({ status: 200, body: "padded-body" }); }, ); }); @@ -969,14 +935,12 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL ); }, async url => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, r.headers.get("x-after"), await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 100 final"); - expect(exitCode).toBe(0); + const r = await fetch(url, h2); + expect({ status: r.status, after: r.headers.get("x-after"), body: await r.text() }).toEqual({ + status: 200, + after: "100", + body: "final", + }); }, ); }); @@ -985,16 +949,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL await withRawH2Server( (conn, id) => conn.headers(id, hpackStatus(100), { endStream: true }), async url => { - await using proc = await spawnFetch(` - try { - await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log("resolved"); - } catch (e) { console.log("rejected", e?.code ?? e?.name); } - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1005,16 +960,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.socket.write(Buffer.concat([frame(1, 4, id, hpackStatus(100)), frame(0, 1, id, Buffer.from("body"))])); }, async url => { - await using proc = await spawnFetch(` - try { - await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log("resolved"); - } catch (e) { console.log("rejected", e?.code ?? e?.name); } - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1031,113 +977,76 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL ); }, async url => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, r.headers.get("x-real"), await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 yes body"); - expect(exitCode).toBe(0); + const r = await fetch(url, h2); + expect({ + status: r.status, + real: r.headers.get("x-real"), + trailer: r.headers.get("x-trailer"), + body: await r.text(), + }).toEqual({ status: 200, real: "yes", trailer: null, body: "body" }); }, ); }); test("Expect: 100-continue withholds the body until 100 arrives", async () => { - const seen: { id: number; type: number; len: number }[] = []; - const server = nodetls.createServer({ ...tls, ALPNProtocols: ["h2"] }, socket => { - let buf = Buffer.alloc(0); - let prefaceSeen = false; - let sent100 = false; - socket.on("data", chunk => { - buf = Buffer.concat([buf, chunk]); - if (!prefaceSeen) { - if (buf.length < 24) return; - buf = buf.subarray(24); - prefaceSeen = true; - socket.write(frame(4, 0, 0)); - } - while (buf.length >= 9) { - const len = buf.readUIntBE(0, 3); - if (buf.length < 9 + len) return; - const type = buf[3], - flags = buf[4], - id = buf.readUInt32BE(5) & 0x7fffffff; - buf = buf.subarray(9 + len); - if (id !== 0) seen.push({ id, type, len }); - if (type === 4 && !(flags & 1)) socket.write(frame(4, 1, 0)); - if (type === 1 && !sent100) { - sent100 = true; - // Prove no DATA preceded the 100 by responding only after a tick. - setTimeout(() => socket.write(frame(1, 4, id, hpackStatus(100))), 20); - } - if (type === 0 && flags & 1) { - socket.write(frame(1, 4, id, hpackStatus(200))); - socket.write(frame(0, 1, id, Buffer.from("got-body"))); + let dataFramesBefore100 = -1; + await withRawH2Server( + async (conn, id) => { + let dataFrames = 0; + conn.onFrame = f => { + if (f.id !== id || f.type !== FRAME_DATA) return; + dataFrames++; + if (f.flags & FLAG_END_STREAM) { + conn.headers(id, hpackStatus(200)); + conn.data(id, "got-body", true); } - } - }); - socket.on("error", () => {}); - }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const r = await fetch("https://localhost:${port}", { + }; + // A client that ignored Expect would have written the DATA frames right + // behind HEADERS, i.e. ahead of its answer to this PING; so once the + // ACK is back, any DATA that was going to arrive before the 100 has. + await conn.ping(); + dataFramesBefore100 = dataFrames; + conn.headers(id, hpackStatus(100)); + }, + async (url, state) => { + const r = await fetch(url, { + ...h2, method: "POST", headers: { Expect: "100-continue" }, body: "twenty-chars-body!!!", - tls: { rejectUnauthorized: false }, }); - console.log(r.status, await r.text()); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 got-body"); - expect(exitCode).toBe(0); - // First per-stream frame must be HEADERS; no DATA until after the 100. - expect(seen[0].type).toBe(1); - const firstData = seen.findIndex(f => f.type === 0); - expect(firstData).toBeGreaterThan(0); - expect(seen[firstData].len).toBe(20); - } finally { - server.close(); - } + expect({ status: r.status, body: await r.text() }).toEqual({ status: 200, body: "got-body" }); + expect(dataFramesBefore100).toBe(0); + // HEADERS (stream left open for the body), then, only after the 100, + // the whole body in one DATA frame carrying END_STREAM. + expect(state.frames.map(f => ({ type: f.type, len: f.len, endStream: f.flags & FLAG_END_STREAM }))).toEqual([ + { type: FRAME_HEADERS, len: expect.any(Number), endStream: 0 }, + { type: FRAME_DATA, len: 20, endStream: FLAG_END_STREAM }, + ]); + }, + ); }); test("Expect: 100-continue with final status before 100 skips body upload", async () => { - let dataBytes = 0; await withRawH2Server( (conn, id) => { - // Reject immediately without 100; client should half-close with - // an empty DATA+END_STREAM rather than uploading the body. - conn.headers(id, hpackStatus(404), { endStream: true }); - conn.socket.on("data", chunk => { - // crude: count any DATA frame payloads on this socket after reject - let b = chunk; - while (b.length >= 9) { - const len = b.readUIntBE(0, 3); - if (b[3] === 0 && (b.readUInt32BE(5) & 0x7fffffff) === id) dataBytes += len; - b = b.subarray(9 + len); - } - }); + // Reject stream 1 immediately without a 100; the barrier request on + // stream 3 is answered normally. + conn.headers(id, id === 1 ? hpackStatus(404) : hpackStatus(204), { endStream: true }); }, - async url => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { - method: "POST", - headers: { Expect: "100-continue" }, - body: Buffer.alloc(50000, "x").toString(), - tls: { rejectUnauthorized: false }, - }); - console.log(r.status); - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("404"); - expect(exitCode).toBe(0); - // Body was withheld; only the empty END_STREAM DATA frame allowed. - expect(dataBytes).toBe(0); + async (url, state) => { + const body = Buffer.alloc(50000, "x").toString(); + const r = await fetch(url, { ...h2, method: "POST", headers: { Expect: "100-continue" }, body }); + expect(r.status).toBe(404); + // Anything the client wrote for stream 1 is queued ahead of the + // barrier's HEADERS(3) on the one connection, so the 204 coming back + // means the stream-1 frame log below is complete. + expect((await fetch(url, h2)).status).toBe(204); + expect(state.connections).toBe(1); + // Body was withheld: no DATA payload bytes were ever sent for stream 1. + const stream1 = state.frames.filter(f => f.id === 1); + expect(stream1[0].type).toBe(FRAME_HEADERS); + expect(stream1.filter(f => f.type === FRAME_DATA).reduce((n, f) => n + f.len, 0)).toBe(0); }, ); }); @@ -1149,16 +1058,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.data(id, "short", true); }, async url => { - await using proc = await spawnFetch(` - try { - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - await r.text(); - console.log("ok"); - } catch (e) { console.log("rejected", String(e).includes("ContentLength")); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected true"); - expect(exitCode).toBe(0); + expect(await bodyOrCode(fetch(url, h2))).toBe("HTTP2ContentLengthMismatch"); }, ); }); @@ -1172,15 +1072,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, Buffer.concat([hpackStatus(200), hpackLit("content-length", "42")]), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log("ok", r.status, (await r.text()).length); - } catch (e) { console.log("rejected", String(e).includes("ContentLength")); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected true"); - expect(exitCode).toBe(0); + expect(await bodyOrCode(fetch(url, h2))).toBe("HTTP2ContentLengthMismatch"); }, ); }); @@ -1191,39 +1083,41 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackLit("content-type", "text/plain"), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected"); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); test("Content-Length satisfied before END_STREAM doesn't dereference a freed client", async () => { - // Server sends body in one DATA frame without END_STREAM, then a - // separate empty DATA(END_STREAM). The first frame fully satisfies - // Content-Length, so progressUpdate fires and the JS callback frees - // the AsyncHTTP; the second frame must not touch a stale client ptr. + // Server sends the body in a DATA frame without END_STREAM; the frame + // fully satisfies Content-Length, so the response completes and the JS + // callback frees the AsyncHTTP. The stray empty DATA(END_STREAM) for that + // stream is only sent once the client opens stream 3, i.e. after the + // first response has been consumed in JS, and it is queued ahead of the + // stream-3 response, so the second fetch resolving proves the stale + // stream-1 frame was processed. Runs in a subprocess because the + // failure mode is a crash. await withRawH2Server( (conn, id) => { - conn.headers(id, Buffer.concat([hpackStatus(200), hpackLit("content-length", "5")])); - conn.data(id, "hello", false); - // brief gap so the two frames hit separate onData calls - setTimeout(() => conn.data(id, "", true), 30); + if (id === 1) { + conn.headers(id, Buffer.concat([hpackStatus(200), hpackLit("content-length", "5")])); + conn.data(id, "hello", false); + } else { + conn.data(1, "", true); + conn.headers(id, hpackStatus(204), { endStream: true }); + } }, - async url => { - await using proc = await spawnFetch(` - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); + async (url, state) => { + await using proc = spawnFetch(` + const r = await fetch(${JSON.stringify(url)}, { protocol: "http2", tls: { rejectUnauthorized: false } }); console.log(r.status, await r.text()); - await Bun.sleep(80); - console.log("survived"); + console.log((await fetch(${JSON.stringify(url)}, { protocol: "http2", tls: { rejectUnauthorized: false } })).status); `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("200 hello\nsurvived"); + const { stdout, stderr, exitCode } = await collect(proc); + expect(stderr).toBe(""); + expect(stdout).toBe("200 hello\n204"); expect(exitCode).toBe(0); + expect(state.connections).toBe(1); }, ); }); @@ -1239,15 +1133,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { - const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log("status", r.status); - } catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1261,13 +1147,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2FlowControlError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2FlowControlError"); }, ); }); @@ -1280,13 +1160,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1301,13 +1175,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1318,18 +1186,11 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL // buffered (the unbounded path would let a peer balloon read_buffer to // ~16 MiB). await withRawH2Server( - (conn, id) => { - void id; + conn => { conn.socket.write(frame(0, 0, 1, Buffer.alloc(16385))); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2FrameSizeError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2FrameSizeError"); }, ); }); @@ -1342,13 +1203,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1361,13 +1216,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1380,13 +1229,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2FrameSizeError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2FrameSizeError"); }, ); }); @@ -1399,13 +1242,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, hpackStatus(200), { endStream: true }); }, async url => { - await using proc = await spawnFetch(` - try { await fetch("${url}", { tls: { rejectUnauthorized: false } }); console.log("ok"); } - catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2ProtocolError"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2ProtocolError"); }, ); }); @@ -1425,20 +1262,27 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL } }, async (url, state) => { - await using proc = await spawnFetch(` - const body = new ReadableStream({ - async start(ctrl) { ctrl.enqueue(new Uint8Array([1,2,3])); await new Promise(r => setTimeout(r, 60_000)); }, - }); - const r = await fetch("${url}/upload", { method: "POST", body, duplex: "half", tls: { rejectUnauthorized: false } }); - console.log(r.status, r.url.endsWith("/target")); - process.exit(0); - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 true"); - await state.allClosed(); - expect(state.rst).toEqual([{ id: 1, code: 8 }]); - expect(exitCode).toBe(0); + // Never closes: the 303 has to cancel the upload. + const body = new ReadableStream({ + start(ctrl) { + ctrl.enqueue(new Uint8Array([1, 2, 3])); + }, + pull() { + return new Promise(() => {}); + }, + }); + const r = await fetch(`${url}/upload`, { ...h2, method: "POST", body, duplex: "half" }); + expect({ status: r.status, redirected: r.redirected, url: r.url }).toEqual({ + status: 200, + redirected: true, + url: `${url}/target`, + }); + // The follow-up GET rode the same connection, and its HEADERS(3) was + // queued behind RST_STREAM(1), so the 200 arriving means the RST did. + expect({ rst: state.rst, connections: state.connections }).toEqual({ + rst: [{ id: 1, code: http2.constants.NGHTTP2_CANCEL }], + connections: 1, + }); }, ); }); @@ -1453,33 +1297,20 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.headers(id, Buffer.concat([hpackStatus(200), hpackLit("content-encoding", "gzip")])); conn.data(id, Buffer.from("not gzip")); } else { - // Barrier request — see subprocess comment below. + // Barrier request, see below. conn.headers(id, hpackStatus(204), { endStream: true }); } }, async (url, state) => { - await using proc = await spawnFetch(` - const tls = { rejectUnauthorized: false }; - try { await (await fetch("${url}", { tls })).arrayBuffer(); console.log("ok"); } - catch (e) { console.log("rejected", e.code ?? e.message); } - // Second request on the same pooled session acts as a delivery - // barrier: RST_STREAM(1) is queued ahead of HEADERS(3) on the one - // socket, so the 204 arriving back proves the RST reached the - // server. Without this the subprocess can exit while the RST is - // still in the TLS/TCP send buffer, which Windows drops on - // process death (abortive close) — leaving state.rst empty. - console.log((await fetch("${url}", { tls })).status); - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - const out = stdout.trim().split("\n"); - expect(out[0]).toContain("rejected"); - expect(out[1]).toBe("204"); - expect(exitCode).toBe(0); - await state.allClosed(); + expect(await bodyOrCode(fetch(url, h2))).toBe("ZlibError"); + // Second request on the same pooled session acts as a delivery + // barrier: RST_STREAM(1) is queued ahead of HEADERS(3) on the one + // socket, so the 204 arriving back proves the RST reached the server. + expect((await fetch(url, h2)).status).toBe(204); // 0x8 = CANCEL. connections=1 proves the barrier rode the same // socket, so ordering actually applies. expect({ rst: state.rst, connections: state.connections }).toEqual({ - rst: [{ id: 1, code: 8 }], + rst: [{ id: 1, code: http2.constants.NGHTTP2_CANCEL }], connections: 1, }); }, @@ -1492,15 +1323,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL conn.rst(id, 0); }, async url => { - await using proc = await spawnFetch(` - try { - await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log("ok"); - } catch (e) { console.log("rejected", e.code); } - `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("rejected HTTP2StreamReset"); - expect(exitCode).toBe(0); + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2StreamReset"); }, ); }); @@ -1513,31 +1336,19 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL sock.end("HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 2\r\n\r\nok"); }); server.on("tlsClientError", () => {}); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `console.log(await fetch("https://localhost:${port}", { tls: { rejectUnauthorized: false } }).then(r => r.text()));`, - ], - env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + await listenTcp(server, "https", async url => { + await using proc = spawnFetch( + `console.log(await fetch(${JSON.stringify(url)}, { tls: { rejectUnauthorized: false } }).then(r => r.text()));`, + {}, + ); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - expect(stdout.trim()).toBe("ok"); + expect(stdout).toBe("ok"); // The server prefers h2; if the client had offered it, ALPN would have // selected it and the HTTP/1.1 response above would have failed parse. expect(alpn).toBe("http/1.1"); expect(exitCode).toBe(0); - } finally { - server.close(); - } + }); }); test("--experimental-http2-fetch enables h2 without the env flag", async () => { @@ -1549,22 +1360,15 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL async url => { // No BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT in env; the CLI flag // alone should make ALPN offer h2. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "--experimental-http2-fetch", - "-e", - `const r = await fetch("${url}", { tls: { rejectUnauthorized: false } }); - console.log(r.status, await r.text());`, - ], - env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + await using proc = spawnFetch( + `const r = await fetch(${JSON.stringify(url)}, { tls: { rejectUnauthorized: false } }); + console.log(r.status, await r.text());`, + {}, + "--experimental-http2-fetch", + ); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 2.0"); + expect(stdout).toBe("200 2.0"); expect(exitCode).toBe(0); }, ); @@ -1578,21 +1382,14 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL }, async url => { // No BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT in env. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `const r = await fetch("${url}", { protocol: "http2", tls: { rejectUnauthorized: false } }); - console.log(r.status, await r.text());`, - ], - env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + await using proc = spawnFetch( + `const r = await fetch(${JSON.stringify(url)}, { protocol: "http2", tls: { rejectUnauthorized: false } }); + console.log(r.status, await r.text());`, + {}, + ); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - expect(stdout.trim()).toBe("200 2.0"); + expect(stdout).toBe("200 2.0"); expect(exitCode).toBe(0); }, ); @@ -1617,69 +1414,33 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL }); }, async url => { - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `const payload = Buffer.alloc(${size}, "abcdefghij"); - const r = await fetch("${url}", { - method: "POST", - body: payload, - compress: "gzip", - protocol: "http2", - tls: { rejectUnauthorized: false }, - }); - const decoded = Buffer.from(await r.arrayBuffer()); - console.log(JSON.stringify({ - recvLen: Number(r.headers.get("x-recv-len")), - encoding: r.headers.get("x-recv-encoding"), - contentLength: r.headers.get("x-recv-content-length"), - match: decoded.equals(payload), - }));`, - ], - env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - const out = JSON.parse(stdout.trim()); - expect(out.encoding).toBe("gzip"); - expect(out.recvLen).toBeLessThan(size); - expect(out.contentLength).toBe(String(out.recvLen)); - expect(out.match).toBe(true); - expect(exitCode).toBe(0); + const payload = Buffer.alloc(size, "abcdefghij"); + const r = await fetch(url, { ...h2, method: "POST", body: payload, compress: "gzip" }); + const decoded = Buffer.from(await r.arrayBuffer()); + const recvLen = Number(r.headers.get("x-recv-len")); + expect(recvLen).toBeGreaterThan(0); + expect(recvLen).toBeLessThan(size); + expect({ + status: r.status, + encoding: r.headers.get("x-recv-encoding"), + contentLength: r.headers.get("x-recv-content-length"), + match: decoded.equals(payload), + }).toEqual({ status: 200, encoding: "gzip", contentLength: String(recvLen), match: true }); }, ); }); test("protocol:'http2' against an h1-only server fails with HTTP2Unsupported", async () => { - const server = https.createServer({ ...tls }, (_req, res) => res.end("h1")); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `try { - await fetch("https://localhost:${port}", { protocol: "http2", tls: { rejectUnauthorized: false } }); - console.log("unexpected-ok"); - } catch (e) { console.log(e.code || String(e)); }`, - ], - env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toContain("HTTP2Unsupported"); - expect(exitCode).toBe(0); - } finally { - server.close(); - } + let requests = 0; + const server = https.createServer({ ...tls }, (_req, res) => { + requests++; + res.end("h1"); + }); + await listenTcp(server, "https", async url => { + expect(await statusOrCode(fetch(url, h2))).toBe("HTTP2Unsupported"); + // Pinned to h2, the request must not be downgraded and sent over h1. + expect(requests).toBe(0); + }); }); test("ALPN h1 result re-dispatches coalesced waiters in parallel, not serial", async () => { @@ -1689,7 +1450,7 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL let active = 0; let peak = 0; const { promise, resolve } = Promise.withResolvers(); - const server = https.createServer({ ...tls }, (req, res) => { + const server = https.createServer({ ...tls }, (_req, res) => { active++; peak = Math.max(peak, active); if (active === 5) resolve(); @@ -1698,44 +1459,47 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL active--; }); }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}/"; + await listenTcp(server, "https", async url => { + await using proc = spawnFetch(` + const url = ${JSON.stringify(url)}; const tls = { rejectUnauthorized: false }; const rs = await Promise.all(Array.from({ length: 5 }, () => fetch(url, { tls }).then(r => r.text()))); console.log(rs.join(",")); `); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toBe("ok,ok,ok,ok,ok"); + const { stdout, stderr, exitCode } = await collect(proc); + expect(stderr).toBe(""); + expect(stdout).toBe("ok,ok,ok,ok,ok"); expect(exitCode).toBe(0); // If waiters re-coalesced, peak would be 1 (sequential); 5 means all // five connections were open before any response was written. expect(peak).toBe(5); - } finally { - server.close(); - } + }); }); test('protocol: "http1.1" overrides the env flag and pins ALPN to http/1.1', async () => { // Server is h2-only: the unpinned fetch (env flag on) negotiates h2, while // the pinned fetch advertises only http/1.1 and is rejected at ALPN — // proving the pin actually reached the ClientHello. + let requests = 0; await withH2Server( - (req, res) => res.end(req.httpVersion), + (req, res) => { + requests++; + res.end(req.httpVersion); + }, async url => { - await using proc = await spawnFetch(` + await using proc = spawnFetch(` + const url = ${JSON.stringify(url)}; const tls = { rejectUnauthorized: false }; - const a = await fetch("${url}", { tls }).then(r => r.text()); - const b = await fetch("${url}", { protocol: "http1.1", tls }).then(r => r.text(), e => "rejected"); + const a = await fetch(url, { tls }).then(r => r.text()); + const b = await fetch(url, { protocol: "http1.1", tls }).then(r => "resolved:" + r.status, () => "rejected"); console.log(a, b); `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const { stdout, stderr, exitCode } = await collect(proc); expect(stderr).toBe(""); - expect(stdout.trim()).toBe("2.0 rejected"); + expect(stdout).toBe("2.0 rejected"); expect(exitCode).toBe(0); + // The pinned request never got past the handshake. + expect(requests).toBe(1); }, ); }); @@ -1743,67 +1507,49 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL test('protocol: "http2" on a plain http:// URL fails with HTTP2Unsupported', async () => { // h2c is out of scope; without an explicit check the request would // silently complete over HTTP/1.1. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `try { - await fetch("http://127.0.0.1:1/", { protocol: "http2" }); - console.log("unexpected-ok"); - } catch (e) { console.log(e.code || String(e)); }`, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stdout.trim()).toContain("HTTP2Unsupported"); - expect(exitCode).toBe(0); + let connections = 0; + const server = net.createServer(sock => { + connections++; + sock.end("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"); + }); + await listenTcp(server, "http", async url => { + expect(await statusOrCode(fetch(url, { protocol: "http2" }))).toBe("HTTP2Unsupported"); + // The request is refused before anything is sent: a fresh connection to + // the same server over h1 works and is the first one it sees. + expect(await fetch(url).then(r => r.text())).toBe("ok"); + expect(connections).toBe(1); + }); }); test("abort while coalesced onto an in-flight TLS connect resolves promptly", async () => { // Leader's TLS handshake never completes (server is plain TCP), so its // PendingConnect stays open. The waiter has no abort-tracker entry and // would otherwise wait for the leader before observing the abort. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `import net from "node:net"; - let conns = 0; - const { promise: accepted, resolve } = Promise.withResolvers(); - const server = net.createServer(sock => { conns++; sock.on("error", () => {}); resolve(); }); - server.listen(0, "127.0.0.1"); - await new Promise(r => server.on("listening", r)); - const url = "https://127.0.0.1:" + server.address().port + "/"; - const opts = { protocol: "http2", tls: { rejectUnauthorized: false } }; - const leader = fetch(url, opts).catch(() => {}); - const ac = new AbortController(); - const waiter = fetch(url, { ...opts, signal: ac.signal }).then( - () => "unexpected-ok", - e => e?.name || String(e), - ); - // Once the server has accepted the leader's TCP connection both - // fetches have been processed on the http thread (PendingConnect - // creation is synchronous in connect()). The settle window lets a - // non-coalesced waiter's connect land so conns reflects it. - await accepted; - await Bun.sleep(100); - ac.abort(); - console.log(await waiter, "conns=" + conns); - void leader; - process.exit(0);`, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("AbortError conns=1"); - expect(exitCode).toBe(0); + let conns = 0; + const { promise: accepted, resolve: onAccept } = Promise.withResolvers(); + const server = net.createServer(sock => { + conns++; + sock.on("error", () => {}); + onAccept(sock); + }); + await listenTcp(server, "https", async url => { + const leader = fetch(url, h2).then(() => "resolved", errcode); + const ac = new AbortController(); + const waiter = fetch(url, { ...h2, signal: ac.signal }).then(() => "resolved", errcode); + // Once the server has accepted the leader's TCP connection both fetches + // have been processed on the http thread (PendingConnect creation is + // synchronous in connect()). + const leaderSocket = await accepted; + ac.abort(); + // The leader is still stuck in its handshake at this point, so this can + // only settle if the abort was observed while coalesced. + expect(await waiter).toBe("AbortError"); + // Killing the leader's socket fails the leader; the aborted waiter must + // not be re-dispatched as a new connect in the process. + leaderSocket.destroy(); + expect(await leader).not.toBe("resolved"); + expect(conns).toBe(1); + }); }); test("SETTINGS_HEADER_TABLE_SIZE=0: encoder emits a Dynamic Table Size Update so request 2+ decodes", async () => { @@ -1812,30 +1558,21 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL // size-update opcode. nghttp2 (which backs node:http2) enforces this and // closes the connection with COMPRESSION_ERROR if the opcode is missing, // so requests after the first hang/fail without the fix. + let sessions = 0; const server = makeH2Server({ settings: { headerTableSize: 0 } }, (_req, res) => { res.writeHead(200); res.end("ok"); }); + server.on("session", () => sessions++); server.on("sessionError", () => {}); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const url = "https://localhost:${port}"; - const tls = { rejectUnauthorized: false }; - for (let i = 0; i < 3; i++) { - const res = await fetch(url, { tls }); - console.log(i, res.status, await res.text()); - } - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim()).toBe("0 200 ok\n1 200 ok\n2 200 ok"); - expect(exitCode).toBe(0); - } finally { - server.close(); - } + await listenH2(server, async url => { + const results: string[] = []; + for (let i = 0; i < 3; i++) results.push(await bodyOrCode(fetch(url, h2))); + expect(results).toEqual(["ok", "ok", "ok"]); + // All three header blocks were decoded by the same (shrunk) decoder; a + // reconnect per request would have passed the body check vacuously. + expect(sessions).toBe(1); + }); }); test("303 to a streaming POST over HTTP/1.1 closes the socket instead of pooling it mid-chunked-body", async () => { @@ -1844,52 +1581,40 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL // body fall through to the keep-alive pool even though the chunked // upload's terminating 0\r\n\r\n was never written. The follow-up GET // must open a fresh connection. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `import net from "node:net"; - let conns = 0; - const server = net.createServer(sock => { - const idx = conns++; - let buf = ""; - sock.on("data", c => { - buf += c; - if (idx === 0 && buf.includes("\\r\\n\\r\\n") && !sock.replied) { - sock.replied = true; - sock.write("HTTP/1.1 303 See Other\\r\\nLocation: /target\\r\\nConnection: keep-alive\\r\\nContent-Length: 0\\r\\n\\r\\n"); - } - if (buf.includes("GET /target")) { - sock.end("HTTP/1.1 200 OK\\r\\nConnection: close\\r\\nContent-Length: 6\\r\\n\\r\\nconn=" + idx); - } - }); - sock.on("error", () => {}); - }); - server.listen(0, "127.0.0.1"); - await new Promise(r => server.on("listening", r)); - const url = "http://127.0.0.1:" + server.address().port; - const body = new ReadableStream({ - async start(ctrl) { - ctrl.enqueue(new Uint8Array([1, 2, 3, 4])); - // Never close — the 303 cancels the upload. - await new Promise(r => setTimeout(r, 60_000)); - }, - }); - const res = await fetch(url + "/upload", { method: "POST", body, duplex: "half" }); - console.log(res.status, await res.text(), "conns=" + conns); - process.exit(0);`, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - // conn=1 (zero-indexed) and conns=2 prove the follow-up GET opened a - // fresh socket; the bug would show conn=0 or hang. - expect(stdout.trim()).toBe("200 conn=1 conns=2"); - expect(exitCode).toBe(0); + let conns = 0; + const server = net.createServer(sock => { + const idx = conns++; + let buf = ""; + let replied = false; + sock.on("data", c => { + buf += c; + if (idx === 0 && buf.includes("\r\n\r\n") && !replied) { + replied = true; + sock.write( + "HTTP/1.1 303 See Other\r\nLocation: /target\r\nConnection: keep-alive\r\nContent-Length: 0\r\n\r\n", + ); + } + if (buf.includes("GET /target")) { + sock.end(`HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 6\r\n\r\nconn=${idx}`); + } + }); + sock.on("error", () => {}); + }); + await listenTcp(server, "http", async url => { + // Never closes; the 303 cancels the upload. + const body = new ReadableStream({ + start(ctrl) { + ctrl.enqueue(new Uint8Array([1, 2, 3, 4])); + }, + pull() { + return new Promise(() => {}); + }, + }); + const res = await fetch(`${url}/upload`, { method: "POST", body, duplex: "half" }); + // conn=1 (zero-indexed) and conns=2 prove the follow-up GET opened a + // fresh socket; the bug would show conn=0 or hang. + expect({ status: res.status, body: await res.text(), conns }).toEqual({ status: 200, body: "conn=1", conns: 2 }); + }); }); test("leader abort does not fail a coalesced force_http2 waiter with HTTP2Unsupported", async () => { @@ -1898,47 +1623,29 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL // sits in handshake; the waiter coalesces onto its PendingConnect. When // the leader is aborted, the waiter must retry as the new leader (a // second TCP connect) rather than be told the server lacks h2. - await using proc = await spawnCapped({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - `import net from "node:net"; - let conns = 0; - const { promise: accepted, resolve } = Promise.withResolvers(); - const server = net.createServer(sock => { conns++; sock.on("error", () => {}); resolve(); }); - server.listen(0, "127.0.0.1"); - await new Promise(r => server.on("listening", r)); - const url = "https://127.0.0.1:" + server.address().port + "/"; - const opts = { protocol: "http2", tls: { rejectUnauthorized: false } }; - const leaderAc = new AbortController(); - const leader = fetch(url, { ...opts, signal: leaderAc.signal }).catch(e => e?.name || String(e)); - const waiterAc = new AbortController(); - const waiter = fetch(url, { ...opts, signal: waiterAc.signal }).then( - () => "unexpected-ok", - e => (typeof e?.code === "string" ? e.code : e?.name) || String(e), - ); - await accepted; - await Bun.sleep(100); - const before = conns; - leaderAc.abort(); - await leader; - await Bun.sleep(100); - const after = conns; - waiterAc.abort(); - console.log(await waiter, "before=" + before, "after=" + after); - process.exit(0);`, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - // before=1 proves the waiter coalesced; after=2 proves it retried as - // the new leader instead of being failed with HTTP2Unsupported. - expect(stdout.trim()).toBe("AbortError before=1 after=2"); - expect(exitCode).toBe(0); + let conns = 0; + const accepts = [Promise.withResolvers(), Promise.withResolvers()]; + const server = net.createServer(sock => { + sock.on("error", () => {}); + accepts[conns++]?.resolve(); + }); + await listenTcp(server, "https", async url => { + const leaderAc = new AbortController(); + const leader = fetch(url, { ...h2, signal: leaderAc.signal }).then(() => "resolved", errcode); + const waiterAc = new AbortController(); + const waiter = fetch(url, { ...h2, signal: waiterAc.signal }).then(() => "resolved", errcode); + await accepts[0].promise; + // Only the leader has connected: the waiter coalesced onto it. + expect(conns).toBe(1); + leaderAc.abort(); + expect(await leader).toBe("AbortError"); + // The waiter must now show up as a second TCP connect while still + // pending; a regressed client settles it (HTTP2Unsupported) instead. + expect(await Promise.race([accepts[1].promise.then(() => "second connect"), waiter])).toBe("second connect"); + expect(conns).toBe(2); + waiterAc.abort(); + expect(await waiter).toBe("AbortError"); + }); }); // Cloudflare sends its SETTINGS frame as TLS 1.3 0.5-RTT data, so the @@ -1949,182 +1656,50 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL // the 0.5-RTT write window, so this hits a real Cloudflare-fronted origin — // tolerate network blips by only failing on the specific regression code. test("GET https://registry.npmjs.org over protocol: http2", async () => { - await using proc = await spawnFetch(` - try { - const r = await fetch("https://registry.npmjs.org", { protocol: "http2" }); - console.log("status", r.status); - } catch (e) { console.log("error", e?.code ?? e?.name ?? String(e)); } - `); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(exitCode).toBe(0); - const out = stdout.trim(); + const out = await statusOrCode(fetch("https://registry.npmjs.org", { protocol: "http2" })); // The bug under test surfaces as Malformed_HTTP_Response — DNS/connect // failures or 5xx are environmental, not regressions. - expect(out).not.toContain("Malformed_HTTP_Response"); - if (!out.startsWith("status")) { + expect(out).not.toBe("Malformed_HTTP_Response"); + if (typeof out !== "number") { console.warn(`skipping live h2 assertion: ${out}`); return; } - expect(out).toBe("status 200"); + expect(out).toBe(200); }); }); -// Serial: this test reads the subprocess's stderr stream mid-flight to gate -// the server-side body write, which interferes with sibling tests' TLS -// handshakes under describe.concurrent on aarch64/musl. -test("await fetch() over HTTP/2 resolves on headers, before a content-length body is fully received", async () => { - let heldStream: http2.ServerHttp2Stream | undefined; - const server = makeH2Server(); - server.on("stream", stream => { - stream.on("error", () => {}); - stream.respond({ ":status": 200, "content-length": "262144" }); - stream.write(Buffer.alloc(64 * 1024)); - // The remaining 192 KiB is written from the test body once the - // subprocess proves it already has the Response and a first read. - heldStream = stream; - }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - await using proc = await spawnFetch(` - const r = await fetch("https://localhost:${port}", { tls: { rejectUnauthorized: false } }); - const reader = r.body.getReader(); - let n = 0; - const { value } = await reader.read(); - n += value.byteLength; - // Signal the server (via stderr so the test can unblock the rest). - process.stderr.write("first-chunk\\n"); +test.concurrent( + "await fetch() over HTTP/2 resolves on headers, before a content-length body is fully received", + async () => { + const { promise: held, resolve: hold } = Promise.withResolvers(); + const server = makeH2Server(); + server.on("stream", stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-length": "262144" }); + stream.write(Buffer.alloc(64 * 1024)); + // The remaining 192 KiB is written from the test body once the client + // has the Response and a first chunk in hand. + hold(stream); + }); + await listenH2(server, async url => { + const r = await fetch(url, h2); + expect({ status: r.status, contentLength: r.headers.get("content-length") }).toEqual({ + status: 200, + contentLength: "262144", + }); + const reader = r.body!.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + let n = first.value!.byteLength; + // Both the Response and a body chunk were delivered while the server was + // still holding back most of the body; only now does the rest go out. + (await held).end(Buffer.alloc(192 * 1024)); while (true) { const { value, done } = await reader.read(); - if (value) n += value.byteLength; if (done) break; + n += value.byteLength; } - console.log(r.status, r.headers.get("content-length"), n); - `); - // Wait for the subprocess to prove it received the Response + first chunk - // BEFORE the server has written the full body. - let stderr = ""; - for await (const chunk of proc.stderr) { - stderr += Buffer.from(chunk).toString(); - if (stderr.includes("first-chunk")) break; - } - heldStream!.end(Buffer.alloc(192 * 1024)); - const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); - expect(stdout.trim()).toBe("200 262144 262144"); - expect(exitCode).toBe(0); - } finally { - heldStream?.destroy(); - server.close(); - } -}); - -// https://github.com/oven-sh/bun/issues/16682 (h2 aggregate path): the -// session's shared socket timer is the max over every attached client's -// effective idle deadline. -test("h2: per-request `timeout` extends the session idle deadline, and {timeout:false} is not killed by a sibling's shorter explicit timeout", async () => { - const HOLD_MS = 10_000; - const holdTimers = new Set>(); - const server = makeH2Server({}, (_req, res) => { - // Hold every request idle past uSockets' worst-case firing window for a - // 1s short-tick timer (~5s), then respond. - const timer = setTimeout(() => { - holdTimers.delete(timer); - try { - res.end("hello"); - } catch {} - }, HOLD_MS); - holdTimers.add(timer); - }); - server.listen(0); - await once(server, "listening"); - const { port } = server.address() as import("node:net").AddressInfo; - try { - const run = async (idleDefault: string, body: string) => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "--no-warnings", - "-e", - /* js */ ` - const url = "https://localhost:${port}"; - const get = init => fetch(url, { tls: { rejectUnauthorized: false }, ...init }) - .then(r => r.text(), e => "ERR:" + (e?.code ?? e?.name ?? e)); - ${body} - `, - ], - env: { - ...bunEnv, - BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CLIENT: "1", - BUN_CONFIG_HTTP_IDLE_TIMEOUT: idleDefault, - }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - return { stdout: stdout.trim(), stderr, exitCode }; - }; - const [extendsDefault, floorsSibling, disarmsOnGlobalZero] = await Promise.all([ - // Global idle default = 1s. `{timeout:60000}` must extend the shared - // socket's deadline past the 10s hold; the `{timeout:false}` sibling - // coalesces onto the same session and rides along. - run( - "1", - /* js */ ` - const [longTimeout, noTimeout] = await Promise.all([ - get({ timeout: 60_000 }), - get({ timeout: false }), - ]); - console.log(JSON.stringify({ longTimeout, noTimeout })); - `, - ), - // Global idle default = 20s. `{timeout:false}` contributes 0 to the - // session max and the `{timeout:1000}` sibling contributes 1s; the - // session must floor at the 20s global default so the no-timeout - // stream is not killed by the sibling's short explicit deadline. - run( - "20", - /* js */ ` - const [noTimeout, shortTimeout] = await Promise.all([ - get({ timeout: false }), - get({ timeout: 1000 }), - ]); - console.log(JSON.stringify({ noTimeout, shortTimeout })); - `, - ), - // Global idle default = 0 (disabled). A plain fetch with no `timeout` - // option inherits effective deadline 0 without setting the - // `disable_timeout` flag; the session must still disarm rather than - // letting the `{timeout:1000}` sibling arm the shared socket. - run( - "0", - /* js */ ` - const [plain, shortTimeout] = await Promise.all([ - get(undefined), - get({ timeout: 1000 }), - ]); - console.log(JSON.stringify({ plain, shortTimeout })); - `, - ), - ]); - expect(extendsDefault).toEqual({ - stdout: JSON.stringify({ longTimeout: "hello", noTimeout: "hello" }), - stderr: "", - exitCode: 0, - }); - expect(floorsSibling).toEqual({ - stdout: JSON.stringify({ noTimeout: "hello", shortTimeout: "hello" }), - stderr: "", - exitCode: 0, - }); - expect(disarmsOnGlobalZero).toEqual({ - stdout: JSON.stringify({ plain: "hello", shortTimeout: "hello" }), - stderr: "", - exitCode: 0, + expect(n).toBe(262144); }); - } finally { - for (const timer of holdTimers) clearTimeout(timer); - server.close(); - } -}, 60_000); + }, +);