From 3d906a788d745095f7f8ab376362bf44da562bde Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:17:08 +0000 Subject: [PATCH] Bun.serve http2: close a connection gracefully with a FIN, not with close(2) A graceful stop() on Bun.serve({ http2: true }) cut a response that was in flight when the client sent a frame (WINDOW_UPDATE, PING) while it read the response. Http2Context called close(2) as soon as its own buffer was empty. The kernel still held the end of the response and the GOAWAY. A frame from the client, unread at that moment or sent later, then made the kernel reset the connection and drop them. Http2Connection now has one graceful close, finish(), and one forceful close, abort(). Every close site in Http2Context.h goes through one of them. finish() sends only the FIN, behind everything queued, and keeps the fd. Input is dropped from then on, and the loop closes the socket on the peer's FIN. The wait is bounded by the socket timeout: the server's idleTimeout, and 8 s when that is shorter or off. More input than the connection window allowed (16 MiB, and 1 MiB more) also closes the connection. finish() runs for a drained going-away connection (stop() with a stream in flight, a client GOAWAY), for an idle connection in stop() and closeIdleConnections(), for an idle timeout with nothing in flight, and for a connection error whose GOAWAY the kernel took. A client GOAWAY is now answered with GOAWAY(NO_ERROR) before the FIN. closeIdle() no longer closes a connection whose last bytes the kernel did not take yet. abort() closes the fd before it returns, also over TLS, where a graceful close alone waits for the peer's close_notify. stop(true), an idle timeout with streams open and an expired wait use it. It calls us_socket_close_now, whose two usockets hunks are taken unchanged from #44618. The http2 test fixture no longer imports "harness" and reads stdin through Bun.stdin. Both took seconds in a debug build, in every fixture a test starts. Co-authored-by: Jarred Sumner --- docs/runtime/http/server.mdx | 4 +- packages/bun-types/serve.d.ts | 7 +- packages/bun-usockets/src/libusockets.h | 2 + packages/bun-usockets/src/socket.c | 6 + packages/bun-uws/src/App.h | 2 +- packages/bun-uws/src/Http2Context.h | 139 ++++++-- test/js/bun/http/serve-http2-fixture.ts | 95 +++++- test/js/bun/http/serve-http2-helpers.ts | 74 +++- .../js/bun/http/serve-http2-lifecycle.test.ts | 317 +++++++++++++++++- 9 files changed, 599 insertions(+), 47 deletions(-) diff --git a/docs/runtime/http/server.mdx b/docs/runtime/http/server.mdx index 2c375f63faeb..5addc4569f35 100644 --- a/docs/runtime/http/server.mdx +++ b/docs/runtime/http/server.mdx @@ -330,6 +330,8 @@ await server.stop(true); By default, `stop()` allows in-flight requests and WebSocket connections to complete. The server closes idle keep-alive connections immediately. Connections with a request in flight close once the server has sent their response. Pass `true` to immediately terminate all connections instead. The returned promise resolves once every connection has closed. +An HTTP/2 connection gets a `GOAWAY` frame and the server ends its side of the connection. The connection closes when the client closes its own side, or after `idleTimeout` (after 8 seconds when that is shorter or disabled). + ### `server.closeIdleConnections()` To close keep-alive connections that are not currently serving a request, without stopping the server: @@ -339,7 +341,7 @@ const closed = server.closeIdleConnections(); console.log(`closed ${closed} idle connections`); ``` -It returns the number of connections it closed. Connections with a request in flight and open WebSockets are untouched, and the server keeps accepting new connections. This method mirrors `node:http`'s `server.closeIdleConnections()`, which returns nothing. +It returns the number of connections it closed. An HTTP/2 connection counts when the server ends its side of it. Connections with a request in flight and open WebSockets are untouched, and the server keeps accepting new connections. This method mirrors `node:http`'s `server.closeIdleConnections()`, which returns nothing. ### `server.ref()` and `server.unref()` diff --git a/packages/bun-types/serve.d.ts b/packages/bun-types/serve.d.ts index 356c15491724..9e0534efa7b6 100644 --- a/packages/bun-types/serve.d.ts +++ b/packages/bun-types/serve.d.ts @@ -937,6 +937,10 @@ declare module "bun" { * request in flight close as soon as their response completes. That means * it may take some time before all network activity stops. * + * An HTTP/2 connection gets a GOAWAY frame and the server ends its side of + * it. The connection is closed when the client closes its own side, or + * after `idleTimeout` (after 8 seconds when that is shorter or disabled). + * * The returned promise resolves once every connection is closed. * * @param closeActiveConnections Immediately terminate in-flight requests, websockets, and stop accepting new connections. @@ -951,7 +955,8 @@ declare module "bun" { * In-flight requests and open WebSockets are untouched, and the server * keeps accepting new connections. * - * @returns The number of connections that were closed. + * @returns The number of connections that were closed. An HTTP/2 + * connection counts when the server ends its side of it. */ closeIdleConnections(): number; diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index a11ffb59fa34..64723dabbb4f 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -705,6 +705,8 @@ int us_socket_ssl_handshake_callback_has_fired(us_socket_r s) nonnull_fn_decl; unsigned int us_socket_ssl_spill_pending(us_socket_r s) nonnull_fn_decl; struct us_socket_t *us_socket_close(us_socket_r s, int code, void *reason) __attribute__((nonnull(1))); +/* Closes before it returns. A TLS socket first sends what the kernel takes now: its unsent ciphertext, then close_notify. */ +struct us_socket_t *us_socket_close_now(us_socket_r s) nonnull_fn_decl; int us_socket_local_port(us_socket_r s) nonnull_fn_decl; int us_socket_remote_port(us_socket_r s) nonnull_fn_decl; diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 284dab2bf9c3..d89112f2e37d 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -364,6 +364,12 @@ __attribute__((always_inline)) struct us_socket_t *us_socket_close(struct us_soc return us_internal_socket_close_raw(s, code, reason); } +struct us_socket_t *us_socket_close_now(struct us_socket_t *s) { + /* The graceful close sends what it can and may then wait, for the spill or for the peer. A close defers at most once. */ + us_socket_close(s, LIBUS_SOCKET_CLOSE_CODE_CLEAN_SHUTDOWN, NULL); + return us_socket_close(s, LIBUS_SOCKET_CLOSE_CODE_FAST_SHUTDOWN, NULL); +} + struct us_socket_t *us_socket_pair(struct us_socket_group_t *group, unsigned char kind, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR *fds) { #if defined(LIBUS_USE_LIBUV) || defined(WIN32) return 0; diff --git a/packages/bun-uws/src/App.h b/packages/bun-uws/src/App.h index 6396a840d0d3..025837dff877 100644 --- a/packages/bun-uws/src/App.h +++ b/packages/bun-uws/src/App.h @@ -417,7 +417,7 @@ struct TemplatedApp { /** Closes all connections connected to this server which are not sending a request or waiting for a response. Does not close the listen socket. * With closeWhenIdle set, connections that are busy right now are marked to close as soon as their in-flight work completes (graceful shutdown); * upgraded WebSockets and CONNECT/Upgrade tunnels never become idle, so they are left alone either way. - * Returns the number of connections closed. */ + * Returns the number of connections closed. An HTTP/2 connection counts when its graceful close starts (Http2Connection::finish). */ size_t closeIdle(bool closeWhenIdle = false) { auto *group = httpContext->getSocketGroup(); struct us_socket_t *s = group->head_sockets; diff --git a/packages/bun-uws/src/Http2Context.h b/packages/bun-uws/src/Http2Context.h index 31c094d215a8..7d8762fb1725 100644 --- a/packages/bun-uws/src/Http2Context.h +++ b/packages/bun-uws/src/Http2Context.h @@ -112,6 +112,13 @@ static constexpr size_t MAX_QUEUED_CONTROL = 1u << 20; /* RST_STREAM token bucket (CVE-2023-44487 class): burst, refill per second. */ static constexpr double RESET_BURST = 1000; static constexpr double RESET_REFILL_PER_SEC = 33; +/* After Http2Connection::finish() the peer has the server's idleTimeout to + * answer our FIN, and this long when that is shorter or off (usockets ticks + * every 4 s, so 4 to 8 s). */ +static constexpr unsigned LINGER_TIMEOUT_S = 8; +/* Input dropped after finish() before the peer is taken to ignore our FIN: + * the DATA our connection window still allows it, and room for its frames. */ +static constexpr uint32_t LINGER_MAX_DISCARD = LOCAL_CONNECTION_WINDOW_SIZE + (1u << 20); static constexpr char CLIENT_PREFACE[] = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; static constexpr uint32_t CLIENT_PREFACE_LEN = 24; @@ -386,6 +393,11 @@ struct Http2Connection { uint32_t lastProcessedStreamId = 0; bool goawaySent = false; bool goawayReceived = false; + /* finish() sent our FIN behind everything queued. No frame is read or + * written after it; the fd closes on the peer's FIN or from onTimeout. */ + bool finSent = false; + /* Input dropped since finish(); see LINGER_MAX_DISCARD. */ + uint32_t lingerDiscarded = 0; double resetTokens = http2::RESET_BURST; std::chrono::steady_clock::time_point resetRefilledAt = std::chrono::steady_clock::now(); @@ -582,9 +594,28 @@ struct Http2Connection { * stream's onWritable. */ inline void scheduleFlush(); + /* A going-away connection has nothing left in user space: time for + * finish(). The kernel may still hold all of it. */ bool drainedAfterGoaway() { - return !closed && streams.empty() && (goawaySent || goawayReceived) && out->length() == 0; - } + return !closed && streams.empty() && (goawaySent || goawayReceived) && !finSent && + out->length() == 0 && us_socket_ssl_spill_pending(s) == 0; + } + + /* The graceful close, for a connection where drainedAfterGoaway() holds. + * close(2) on a socket the peer still sends to makes the kernel reset + * the connection and drop what it has not delivered yet: the end of a + * response, the GOAWAY. So only our FIN goes out, behind everything + * queued, and the fd stays open. Http2Context::onData drops what the + * peer sends from here on, the loop closes a shut-down socket on the + * peer's FIN, and Http2Context::onTimeout closes it when the peer does + * not answer in time. */ + inline void finish(); + /* The forceful close. The fd is closed on return, also over TLS, where a + * graceful close alone waits for the peer's close_notify. Bytes already + * in the kernel still go out. `this` may be deleted on return. */ + us_socket_t *abort() { return us_socket_close_now(s); } + /* Retire every open stream and tell its holder (a connection error). */ + inline void abortStreams(); bool takeResetToken() { auto now = std::chrono::steady_clock::now(); @@ -785,26 +816,28 @@ struct Http2Context { forEachConnection([](Http2Connection *conn) { if (!conn->goawaySent) conn->writeGoaway(http2::ERR_NO_ERROR); conn->flush(); - us_socket_close(conn->s, 0, nullptr); + conn->abort(); }); } - /* Close connections with nothing in flight (after a GOAWAY). With + /* GOAWAY the connections with nothing in flight and finish() them. With * `closeWhenIdle`, busy connections also get GOAWAY so no new streams - * start and they close once their last stream retires (graceful stop); - * without it they are left alone. */ + * start and they finish once their last stream retires (graceful stop); + * without it they are left alone. Returns how many it finished: the + * socket of each is released when its peer answers the FIN. A GOAWAY + * the kernel did not take yet leaves the connection to the epilogue. */ size_t closeIdle(bool closeWhenIdle) { - size_t closedNow = 0; + size_t finished = 0; forEachConnection([&](Http2Connection *conn) { if (!conn->streams.empty() && !closeWhenIdle) return; if (!conn->goawaySent) conn->writeGoaway(http2::ERR_NO_ERROR); conn->flush(); - if (conn->streams.empty()) { - us_socket_close(conn->s, 0, nullptr); - closedNow++; + if (conn->drainedAfterGoaway()) { + conn->finish(); + finished++; } }); - return closedNow; + return finished; } /* Installed by the embedder: arrange for sweep() to run soon, outside the @@ -862,7 +895,7 @@ struct Http2Context { conn->returnSharedOut(); conn->busy--; if (!sweepConnection(conn)) continue; - if (conn->drainedAfterGoaway()) us_socket_close(conn->s, 0, nullptr); + if (conn->drainedAfterGoaway()) conn->finish(); } } @@ -899,8 +932,13 @@ struct Http2Context { /* ── socket vtable ────────────────────────────────────────────────── */ static us_socket_t *onData(us_socket_t *s, char *data, int length) { - if (us_socket_is_shut_down(s)) return s; Http2Connection *conn = connection(s); + if (conn->finSent) { + /* Read and dropped, so that these bytes never meet a close(2). A + * peer that sends this much after our FIN is not going to stop. */ + conn->lingerDiscarded += (uint32_t) length; + return conn->lingerDiscarded > http2::LINGER_MAX_DISCARD ? conn->abort() : s; + } conn->borrowSharedOut(); conn->onData(data, (size_t) length); return s; @@ -947,10 +985,12 @@ struct Http2Context { return s; } - /* No traffic either way for idleTimeoutS (or a drained going-away - * connection): streams in flight are aborted through onClose. */ + /* No traffic either way for idleTimeoutS: streams in flight are aborted + * through onClose. */ static us_socket_t *onTimeout(us_socket_t *s) { Http2Connection *conn = connection(s); + /* The peer did not answer our FIN in time. */ + if (conn->finSent) return conn->abort(); conn->busy++; std::vector open = conn->streams; for (Http2Response *stream : open) { @@ -963,17 +1003,23 @@ struct Http2Context { } if (!conn->goawaySent) conn->writeGoaway(http2::ERR_NO_ERROR); conn->flush(); - return us_socket_close(s, 0, nullptr); + /* An idle connection whose GOAWAY the kernel took: a request that + * crosses it on the wire must not reset it away. */ + if (conn->drainedAfterGoaway()) { + conn->finish(); + return s; + } + return conn->abort(); } static us_socket_t *onEnd(us_socket_t *s) { Http2Connection *conn = connection(s); conn->flush(); - return us_socket_close(s, 0, nullptr); + return conn->abort(); } /* After user code ran from a socket event: push bytes out, free what - * retired, close a drained going-away connection. */ + * retired, finish a drained going-away connection. */ static us_socket_t *epilogue(Http2Connection *conn, us_socket_t *s) { if (conn->closed) { conn->returnSharedOut(); @@ -985,7 +1031,7 @@ struct Http2Context { conn->busy--; conn->returnSharedOut(); if (!sweepConnection(conn)) return s; - if (conn->drainedAfterGoaway()) return us_socket_close(s, 0, nullptr); + if (conn->drainedAfterGoaway()) conn->finish(); return s; } @@ -1025,9 +1071,51 @@ inline void Http2Connection::writeSettings() { } inline void Http2Connection::touch() { + /* After finish() the socket timeout is the linger's. */ + if (finSent) return; us_socket_timeout(s, idleTimeoutS); } +inline void Http2Connection::finish() { + if (!goawaySent) { + /* The peer's GOAWAY brought us here. Ours goes out before the FIN. */ + writeGoaway(http2::ERR_NO_ERROR); + flush(); + if (!drainedAfterGoaway()) return; + } + us_socket_shutdown(s); + finSent = true; + /* No frame is parsed again. A frame handler that got us here from a + * callback does not read its payload after it. */ + in.clear(); + headerBlock.clear(); + us_socket_timeout(s, std::max(ctx->idleTimeoutS, http2::LINGER_TIMEOUT_S)); +} + +inline void Http2Connection::abortStreams() { + /* Every stream is dead before the first holder hears of it, as in + * onClose: an abort listener finds no sibling left to write to. */ + std::vector open; + open.swap(streams); + writable.clear(); + for (Http2Response *stream : open) { + stream->dead = true; + stream->wantsWrite = false; + stream->data.onWritable = nullptr; + stream->data.inStream = nullptr; + pendingFree.push_back(stream); + } + for (Http2Response *stream : open) { + /* onClose told the rest. */ + if (closed) return; + if (stream->data.onAborted) { + auto cb = stream->data.onAborted; + stream->data.onAborted = nullptr; + cb(stream, stream->data.userData); + } + } +} + inline void Http2Connection::recomputeIdleTimeout() { /* Most permissive among open streams (0 = never); a stream that never * asked counts as the context default, as does a connection with none. */ @@ -1237,8 +1325,14 @@ inline void Http2Connection::streamMaybeClosed(Http2Response *stream) { inline bool Http2Connection::connectionError(http2::ErrorCode code) { if (!goawaySent) writeGoaway(code); flush(); - us_socket_shutdown(s); - us_socket_close(s, 0, nullptr); + /* The peer is sending, by definition: a close(2) now can reset the + * GOAWAY away before it reads it. When the kernel took the GOAWAY, end + * every stream and finish() instead. */ + if (!closed && out->length() == 0 && us_socket_ssl_spill_pending(s) == 0) { + abortStreams(); + if (drainedAfterGoaway()) finish(); + } + if (!closed && !finSent) abort(); return false; } @@ -1294,7 +1388,8 @@ inline void Http2Connection::onData(const char *data, size_t length) { return; } if (avail - consumed < http2::FRAME_HEADER_SIZE + flen) break; - if (!handleFrame(type, flags, streamId, h + http2::FRAME_HEADER_SIZE, flen) || closed) { + /* finSent: a handler stopped the server, or the frame was a connection error. */ + if (!handleFrame(type, flags, streamId, h + http2::FRAME_HEADER_SIZE, flen) || closed || finSent) { busy--; Http2Context::epilogue(this, s); return; diff --git a/test/js/bun/http/serve-http2-fixture.ts b/test/js/bun/http/serve-http2-fixture.ts index fa546836c2c9..d3ecb70af4ad 100644 --- a/test/js/bun/http/serve-http2-fixture.ts +++ b/test/js/bun/http/serve-http2-fixture.ts @@ -2,14 +2,17 @@ // per describe block, driven over real sockets by the test. Prints its port as // JSON on stdout once listening and stops on stdin EOF. // -// bun serve-http2-fixture.ts --big-file [--tls] [--no-http1] [--http3] [--idle-timeout ] +// bun serve-http2-fixture.ts --big-file [--tls ] [--no-http1] [--http3] [--idle-timeout ] +// +// The certificate comes in a file and not from "harness": that import alone +// takes seconds in a debug build, in every fixture a test starts. import { serve } from "bun"; -import { tls as tlsCert } from "harness"; +import { existsSync, readFileSync, writeSync } from "node:fs"; import { parseArgs } from "node:util"; const { values: args } = parseArgs({ options: { - tls: { type: "boolean", default: false }, + tls: { type: "string" }, http1: { type: "boolean", default: true }, http3: { type: "boolean", default: false }, "idle-timeout": { type: "string", default: "30" }, @@ -21,6 +24,9 @@ const bigFile = args["big-file"]!; const big = Buffer.alloc(5 * 1024 * 1024, "abcdefghijklmnop"); let lateRead: PromiseWithResolvers | undefined; +let gate: PromiseWithResolvers | undefined; +// A marker the test waits for. Synchronous, so that it is out before the handler goes on. +const mark = (marker: string) => writeSync(2, marker + "\n"); const makeRoutes = () => ({ "/api/:id": (req: Bun.BunRequest<"/api/:id">) => @@ -40,7 +46,7 @@ const makeRoutes = () => ({ const server = serve({ port: 0, - tls: args.tls ? tlsCert : undefined, + tls: args.tls ? JSON.parse(readFileSync(args.tls, "utf8")) : undefined, http2: true, http1: args.http1, http3: args.http3, @@ -274,8 +280,66 @@ async function handler(req: Request, server: Bun.Server): Promise server.stop(), 0); + setTimeout(() => server.stop().then(() => mark("STOPPED")), 0); return new Response("stopping"); + // ?n= bytes, once /hold lets it answer. + case "/gate": { + gate ??= Promise.withResolvers(); + mark("GATED"); + await gate.promise; + return new Response(Buffer.alloc(Number(url.searchParams.get("n")), "g")); + } + // Keeps the event loop stopped until the test creates ?file= (and for ?ms= at least), so + // that what the clients write meanwhile is still unread in the kernel when ?do= runs. + // Then /gate answers. + case "/hold": { + const file = url.searchParams.get("file")!; + const until = performance.now() + Number(url.searchParams.get("ms") ?? "0"); + mark("HOLDING"); + while (!existsSync(file) || performance.now() < until) Bun.sleepSync(1); + let result = "held"; + switch (url.searchParams.get("do")) { + case "stop": + server.stop().then(() => mark("STOPPED")); + break; + case "close-idle": + result += server.closeIdleConnections() + "," + server.closeIdleConnections(); + break; + } + gate?.resolve(); + gate = undefined; + return new Response(result); + } + // Never answers. When the client resets the stream, the abort listener stops the server: + // the connection ends from inside that callback. + case "/abort-stop": { + req.signal.addEventListener("abort", () => { + server.stop().then(() => mark("STOPPED")); + server.closeIdleConnections(); + }); + return new Promise(() => {}); + } + // The next three arm a usockets fault in this process (ASAN builds). The module is not + // imported at the top: every other test would pay for loading it. + // Arms ?rule=, then calls closeIdleConnections() twice. + case "/fault-close-idle": { + require("bun:internal-for-testing").socketFaultInjection.set(JSON.parse(url.searchParams.get("rule")!)); + return new Response("counts " + server.closeIdleConnections() + "," + server.closeIdleConnections()); + } + // Answers with ?n= bytes that send() does not take. In the next macrotask, when the response + // is with the connection and its stream has retired, stops the server and lets send() work again. + case "/respond-blocked": { + const fault = require("bun:internal-for-testing").socketFaultInjection; + fault.set({ syscall: "send", action: "zero", repeat: -1 }); + setImmediate(() => { + server.stop().then(() => mark("STOPPED")); + fault.clear(); + }); + return new Response(Buffer.alloc(Number(url.searchParams.get("n")), "g")); + } + case "/arm": + require("bun:internal-for-testing").socketFaultInjection.set(JSON.parse(url.searchParams.get("rule")!)); + return new Response("armed"); case "/reload": server.reload({ routes: { ...makeRoutes(), "/reloaded-route": new Response("after-reload") }, @@ -306,8 +370,19 @@ async function handler(req: Request, server: Bun.Server): Promise { - server.stop(true); - process.stderr.write("", () => process.exit(0)); -}); -process.stdin.resume(); +// Bun.stdin and not process.stdin: the node stream takes more than a second to +// load in a debug build, and every test waits for this process to exit. +(async () => { + for await (const _ of Bun.stdin.stream()) { + } + // stop(true) closes every connection before it returns, so its promise + // settles within the next turns of the event loop, whatever the clients do. + let settled = false; + server.stop(true).then(() => (settled = true)); + setImmediate(() => + setImmediate(() => { + mark(settled ? "STOPPED-AT-ONCE" : "STOP-PENDING"); + process.exit(0); + }), + ); +})(); diff --git a/test/js/bun/http/serve-http2-helpers.ts b/test/js/bun/http/serve-http2-helpers.ts index f9b9e869db2f..a90e703dcc03 100644 --- a/test/js/bun/http/serve-http2-helpers.ts +++ b/test/js/bun/http/serve-http2-helpers.ts @@ -1,5 +1,5 @@ import { createHash } from "crypto"; -import { bunEnv, bunExe, tempDir } from "harness"; +import { bunEnv, bunExe, tempDir, tls as tlsCert } from "harness"; import http2 from "node:http2"; import net from "node:net"; import { join } from "node:path"; @@ -14,6 +14,8 @@ export type Fixture = { port: number; proc: Bun.Subprocess<"pipe", "pipe", "pipe">; stderr: () => string; + /** Resolves once the fixture has written `marker` to stderr `count` times. */ + stderrHas(marker: string, count?: number): Promise; [Symbol.asyncDispose](): Promise; }; @@ -30,7 +32,10 @@ export type FixtureOptions = { export async function startFixture(opts: FixtureOptions): Promise { // The file behind the Bun.file routes. Per fixture, so that no module-level // hook has to outlive the test file that first imported this module. - const dir = tempDir("serve-http2", { "big.bin": Buffer.alloc(3 * 1024 * 1024 + 17, "0123456789") }); + const dir = tempDir("serve-http2", { + "big.bin": Buffer.alloc(3 * 1024 * 1024 + 17, "0123456789"), + "tls.json": JSON.stringify(tlsCert), + }); try { return await spawnFixture(opts, dir); } catch (err) { @@ -49,7 +54,7 @@ async function spawnFixture(opts: FixtureOptions, dir: ReturnType void)[] = []; + const wakeStderrWaiters = () => { + const waiters = stderrWaiters; + stderrWaiters = []; + for (const wake of waiters) wake(); + }; (async () => { - for await (const chunk of proc.stderr) stderr += new TextDecoder().decode(chunk); + for await (const chunk of proc.stderr) { + stderr += new TextDecoder().decode(chunk); + wakeStderrWaiters(); + } + stderrDone = true; + wakeStderrWaiters(); })(); const reader = proc.stdout.getReader(); let line = ""; @@ -75,6 +92,12 @@ async function spawnFixture(opts: FixtureOptions, dir: ReturnType stderr, + async stderrHas(marker, count = 1) { + while (stderr.split(marker).length - 1 < count) { + if (stderrDone) throw new Error(`fixture exited before writing ${count} x ${marker}: ${stderr}`); + await new Promise(r => stderrWaiters.push(r)); + } + }, async [Symbol.asyncDispose]() { proc.stdin.end(); await proc.exited; @@ -226,26 +249,42 @@ export class RawH2 { private buf = Buffer.alloc(0); private waiters: (() => void)[] = []; closed = false; - - static async connect(port: number, secure: boolean, opts: { sendPreface?: boolean; settings?: Buffer } = {}) { + /** The server's FIN arrived: the stream ended in order, with nothing lost behind it. */ + ended = false; + /** Code of the socket error, such as ECONNRESET when the server's kernel reset the connection. */ + error: string | undefined; + + /** `allowHalfOpen` keeps this side open after the server's FIN, like a client that never closes. */ + static async connect( + port: number, + secure: boolean, + opts: { sendPreface?: boolean; settings?: Buffer; allowHalfOpen?: boolean } = {}, + ) { const c = new RawH2(); + const allowHalfOpen = opts.allowHalfOpen ?? false; await new Promise((resolve, reject) => { const onErr = (e: Error) => reject(e); if (secure) { - c.socket = tls.connect({ port, host: "127.0.0.1", ALPNProtocols: ["h2"], rejectUnauthorized: false }, () => - resolve(), - ); + // The option reaches the net.Socket under the TLS socket. The types of tls.connect() do not list it. + const options = { port, host: "127.0.0.1", ALPNProtocols: ["h2"], rejectUnauthorized: false, allowHalfOpen }; + c.socket = tls.connect(options as tls.ConnectionOptions, () => resolve()); } else { - c.socket = net.connect({ port, host: "127.0.0.1" }, () => resolve()); + c.socket = net.connect({ port, host: "127.0.0.1", allowHalfOpen }, () => resolve()); } c.socket.once("error", onErr); }); c.socket.on("data", (d: Buffer) => c.onData(d)); + c.socket.on("end", () => { + c.ended = true; + c.wake(); + }); c.socket.on("close", () => { c.closed = true; c.wake(); }); - c.socket.on("error", () => {}); + c.socket.on("error", (e: NodeJS.ErrnoException) => { + c.error = e.code ?? e.message; + }); if (opts.sendPreface ?? true) { c.socket.write(PREFACE); c.socket.write(frame(T.SETTINGS, 0, 0, opts.settings ?? Buffer.alloc(0))); @@ -279,6 +318,19 @@ export class RawH2 { this.socket.write(b); } + /** Resolves once the kernel has taken `b`. */ + writeFlushed(b: Buffer) { + return new Promise((resolve, reject) => this.socket.write(b, err => (err ? reject(err) : resolve()))); + } + + /** Resolves at the server's FIN; throws if the connection closed without one. */ + async waitForEnd() { + while (!this.ended) { + if (this.closed) throw new Error(`connection closed without a FIN (${this.error}); got ${this.describe()}`); + await new Promise(r => this.waiters.push(r)); + } + } + /** Resolve with the first frame matching `pred`, waiting for more data as needed. */ async waitFor(pred: (f: RawFrame) => boolean): Promise { for (;;) { diff --git a/test/js/bun/http/serve-http2-lifecycle.test.ts b/test/js/bun/http/serve-http2-lifecycle.test.ts index 311403949613..316d6722dd41 100644 --- a/test/js/bun/http/serve-http2-lifecycle.test.ts +++ b/test/js/bun/http/serve-http2-lifecycle.test.ts @@ -1,6 +1,9 @@ +import { socketFaultInjection as fault } from "bun:internal-for-testing"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, tempDir } from "harness"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { writeFileSync } from "node:fs"; import http2 from "node:http2"; +import { join } from "node:path"; import tls from "node:tls"; import { F, @@ -411,3 +414,315 @@ describe("Bun.serve http2 lifecycle", () => { await proc.exited; }); }); + +// A server that answers a connection it is done with by close(2) resets it when the client still +// sends, and the reset drops what the kernel has not delivered yet: the end of a response, the +// GOAWAY. These tests put the server in exactly that state without a timing assumption. Either a +// frame is unread in the server's kernel when the server acts (the /hold handler keeps the fixture's +// event loop stopped until the test creates a file), or the client does not read and sends its +// frame after the server's FIN. Each client must then see every byte, the GOAWAY, and a FIN. +const holdPath = (dir: string, action: string, ms = 0) => + `/hold?do=${action}&ms=${ms}&file=${encodeURIComponent(join(dir, "release"))}`; +const release = (dir: string) => writeFileSync(join(dir, "release"), ""); +const ping = () => frame(T.PING, 0, 0, Buffer.from("8 bytes!")); + +/** A raw client whose windows take a whole response, so flow control never holds one back. */ +async function connectWide(port: number, secure: boolean, opts: { allowHalfOpen?: boolean } = {}) { + const raw = await RawH2.connect(port, secure, { settings: Buffer.from([0, 4, 0x40, 0, 0, 0]), ...opts }); + raw.write(frame(T.WINDOW_UPDATE, 0, 0, Buffer.from([0x40, 0, 0, 0]))); + await raw.waitFor(f => f.type === T.SETTINGS && (f.flags & F.ACK) !== 0); + return raw; +} + +/** The connection ended in order: GOAWAY(`code`) for `lastStreamId`, then the server's FIN. */ +async function expectGoawayThenFin(raw: RawH2, lastStreamId: number, code = 0) { + expect(await raw.goaway()).toEqual({ lastStreamId, code }); + await raw.waitForEnd(); + expect(raw.error).toBeUndefined(); +} + +describe.each([ + ["h2c", false], + ["TLS", true], +] as const)("Bun.serve http2 graceful close over %s", (_, secure) => { + const bodySize = 1024 * 1024; + const connectRaw = (port: number, opts: { allowHalfOpen?: boolean } = {}) => connectWide(port, secure, opts); + + test("stop() with a response in flight: a frame the client sent meanwhile does not cut it", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const raw = await connectRaw(fx.port); + raw.headers(1, baseHeaders("/gate?n=" + bodySize)); + await fx.stderrHas("GATED"); + // Same connection: the response completes inside this connection's own socket event. + raw.headers(3, baseHeaders(holdPath(String(dir), "stop"))); + await fx.stderrHas("HOLDING"); + await raw.writeFlushed(ping()); + release(String(dir)); + expect((await raw.body(1)).length).toBe(bodySize); + expect((await raw.body(3)).toString()).toBe("held"); + await expectGoawayThenFin(raw, 3); + await fx.stderrHas("STOPPED"); + }); + + test("stop() from another connection: the response ends outside a socket event and is not cut", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const raw = await connectRaw(fx.port); + raw.headers(1, baseHeaders("/gate?n=" + bodySize)); + await fx.stderrHas("GATED"); + const control = await connectRaw(fx.port); + control.headers(1, baseHeaders(holdPath(String(dir), "stop"))); + await fx.stderrHas("HOLDING"); + await raw.writeFlushed(ping()); + release(String(dir)); + expect((await raw.body(1)).length).toBe(bodySize); + await expectGoawayThenFin(raw, 1); + expect((await control.body(1)).toString()).toBe("held"); + await expectGoawayThenFin(control, 1); + await fx.stderrHas("STOPPED"); + }); + + test("stop() with a response in flight: frames the client sends after the server's FIN do not cut it", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const raw = await connectRaw(fx.port); + raw.headers(1, baseHeaders("/gate?n=" + bodySize)); + await fx.stderrHas("GATED"); + const control = await connectRaw(fx.port); + control.headers(1, baseHeaders(holdPath(String(dir), "stop"))); + await fx.stderrHas("HOLDING"); + // This side reads nothing from here on: its kernel takes one receive window of the response + // and the server's kernel keeps the rest, also after the server is done with the connection. + raw.socket.pause(); + release(String(dir)); + // The server ends `raw` first and `control` after it: control's FIN says `raw` has its FIN. + expect((await control.body(1)).toString()).toBe("held"); + await expectGoawayThenFin(control, 1); + // Nothing was unread when the server ended `raw`. These frames arrive afterwards. A reset + // that answers the first one fails the second write. + await raw.writeFlushed(ping()); + await raw.writeFlushed(ping()); + raw.socket.resume(); + expect((await raw.body(1)).length).toBe(bodySize); + await expectGoawayThenFin(raw, 1); + await fx.stderrHas("STOPPED"); + }); + + test("stop() with a response larger than the kernel buffers: a frame for every chunk read does not cut it", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const size = 16 * 1024 * 1024; + const raw = await connectRaw(fx.port); + raw.headers(1, baseHeaders("/gate?n=" + size)); + await fx.stderrHas("GATED"); + raw.headers(3, baseHeaders(holdPath(String(dir), "stop"))); + await fx.stderrHas("HOLDING"); + // Like a client that sends WINDOW_UPDATE or PING while it downloads. + raw.socket.on("data", () => { + if (!raw.ended && !raw.closed) raw.write(ping()); + }); + release(String(dir)); + expect((await raw.body(1)).length).toBe(size); + await expectGoawayThenFin(raw, 3); + // 16 MiB is what a TLS connection needs to show the cut, and seconds in a debug build. + }, 20000); + + test("stop() ends an idle connection with GOAWAY and a FIN when a frame crosses it", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const idle = await connectRaw(fx.port); + const control = await connectRaw(fx.port); + control.headers(1, baseHeaders(holdPath(String(dir), "stop"))); + await fx.stderrHas("HOLDING"); + await idle.writeFlushed(ping()); + release(String(dir)); + await expectGoawayThenFin(idle, 0); + expect((await control.body(1)).toString()).toBe("held"); + await expectGoawayThenFin(control, 1); + await fx.stderrHas("STOPPED"); + }); + + test("closeIdleConnections() ends an idle connection with GOAWAY and a FIN, and counts it once", async () => { + await using fx = await startFixture({ tls: secure }); + using dir = tempDir("serve-http2-hold", {}); + const idle = await connectRaw(fx.port); + const control = await connectRaw(fx.port); + control.headers(1, baseHeaders(holdPath(String(dir), "close-idle"))); + await fx.stderrHas("HOLDING"); + await idle.writeFlushed(ping()); + release(String(dir)); + await expectGoawayThenFin(idle, 0); + // Two calls in a row: the first one finishes the idle connection, the second one leaves it alone. + expect((await control.body(1)).toString()).toBe("held1,0"); + // The busy connection was spared and the server still serves it. + control.headers(3, baseHeaders("/hello")); + expect((await control.body(3)).toString()).toBe("hello"); + control.close(); + }); + + test("a client GOAWAY is answered with GOAWAY and a FIN once its streams are done", async () => { + await using fx = await startFixture({ tls: secure }); + const raw = await connectRaw(fx.port); + raw.headers(1, baseHeaders("/slow?ms=50")); + raw.write(frame(T.GOAWAY, 0, 0, Buffer.alloc(8))); + expect((await raw.body(1)).toString()).toBe("slow"); + await expectGoawayThenFin(raw, 1); + }); + + test("a connection error ends the streams, reports its code and ends with a FIN; later frames are not reset", async () => { + await using fx = await startFixture({ tls: secure }); + const raw = await connectRaw(fx.port, { allowHalfOpen: true }); + raw.headers(1, baseHeaders("/abort")); + // The connection window already holds 2^30: one more of these overflows it. The frame comes + // in two reads, so the server finds the error in a frame it had to put together. The answer + // to the PING says that the server has read up to the first part. + const overflow = frame(T.WINDOW_UPDATE, 0, 0, Buffer.from([0x7f, 0xff, 0xff, 0xff])); + raw.write(Buffer.concat([frame(T.GOAWAY, 0, 0, Buffer.alloc(8)), ping(), overflow.subarray(0, 5)])); + await raw.waitFor(f => f.type === T.PING && (f.flags & F.ACK) !== 0); + raw.write(overflow.subarray(5)); + await expectGoawayThenFin(raw, 1, 3); + await fx.stderrHas("ABORTED"); + // This side is still open and keeps sending. A reset that answers the first frame fails the second write. + await raw.writeFlushed(ping()); + await raw.writeFlushed(ping()); + raw.socket.end(); + await raw.waitForClose(); + expect(raw.error).toBeUndefined(); + }); + + test("stop(true) closes a lingering connection before it returns", async () => { + await using fx = await startFixture({ tls: secure }); + const raw = await connectRaw(fx.port, { allowHalfOpen: true }); + raw.headers(1, baseHeaders("/stop")); + expect((await raw.body(1)).toString()).toBe("stopping"); + await expectGoawayThenFin(raw, 1); + // This side never answers the FIN. The fixture reports whether stop(true) waited for it. + fx.proc.stdin.end(); + await fx.stderrHas("STOPPED-AT-ONCE"); + raw.close(); + }); + + test("stop(true) closes a connection with a stream open before it returns, whatever the client does", async () => { + await using fx = await startFixture({ tls: secure }); + const raw = await connectRaw(fx.port, { allowHalfOpen: true }); + raw.headers(1, baseHeaders("/abort")); + await raw.writeFlushed(ping()); + await raw.waitFor(f => f.type === T.PING && (f.flags & F.ACK) !== 0); + fx.proc.stdin.end(); + await fx.stderrHas("STOPPED-AT-ONCE"); + expect(fx.stderr()).toContain("ABORTED"); + expect((await raw.goaway()).code).toBe(0); + raw.close(); + }); +}); + +describe("Bun.serve http2 graceful close, the wait for the client's FIN", () => { + // usockets ticks timeouts in 4 s steps, so these tests need real seconds. The client never closes + // its side, so only the server's own bound ends the connection. stop() is called from the abort + // listener of a stream the client resets: the connection then ends inside that callback, and the + // stream's own timeout must not replace the bound afterwards. + for (const [name, secure, idleTimeout, atLeastMs] of [ + // Idle timeouts are off: the bound is 8 s, which is 4 to 8 s on the 4 s tick. + ["h2c, idleTimeout 0", false, 0, 3000], + // The idle timeout is longer than 8 s, so it is the bound: 12 to 16 s. + ["TLS, idleTimeout 16", true, 16, 9000], + ] as const) { + test(`a client that never closes holds stop() until the bound and no longer (${name})`, async () => { + await using fx = await startFixture({ tls: secure, idleTimeout }); + const raw = await connectWide(fx.port, secure, { allowHalfOpen: true }); + raw.headers(1, baseHeaders("/abort-stop")); + // The reset comes in two reads with a frame behind it: the server ends the connection from + // inside a frame that it had to put together, and must not go on to the next one. The + // answer to the PING says that the server has read up to the first part. + const reset = frame(T.RST_STREAM, 0, 1, Buffer.from([0, 0, 0, 8])); + raw.write(Buffer.concat([ping(), reset.subarray(0, 5)])); + await raw.waitFor(f => f.type === T.PING && (f.flags & F.ACK) !== 0); + raw.write(Buffer.concat([reset.subarray(5), ping()])); + await expectGoawayThenFin(raw, 1); + const finAt = performance.now(); + // The promise of stop() stays pending while the connection lingers, so that a process + // that exits after `await server.stop()` cannot cut what the client still reads. + await fx.stderrHas("STOPPED"); + expect(performance.now() - finAt).toBeGreaterThan(atLeastMs); + raw.close(); + }, 40000); + } + + test("a client that keeps sending after the server's FIN is closed when it has sent more than a window", async () => { + await using fx = await startFixture({ tls: false }); + const raw = await connectWide(fx.port, false, { allowHalfOpen: true }); + raw.headers(1, baseHeaders("/stop")); + expect((await raw.body(1)).toString()).toBe("stopping"); + await expectGoawayThenFin(raw, 1); + // The server reads and drops what still comes, up to the 16 MiB its connection window allowed + // and 1 MiB more. Past that it closes, long before the wait for this side's FIN ends. + raw.write(Buffer.alloc(18 * 1024 * 1024)); + await raw.waitForClose(); + await fx.stderrHas("STOPPED"); + }); + + test("an idle timeout ends the connection with GOAWAY and a FIN when a frame crosses it", async () => { + // idleTimeout 2 is one tick. /hold keeps the event loop stopped for more than a tick, so the + // timeout of the idle connection fires right after it, with this client's frame still unread. + await using fx = await startFixture({ tls: false, idleTimeout: 2 }); + using dir = tempDir("serve-http2-hold", {}); + const idle = await connectWide(fx.port, false); + const control = await connectWide(fx.port, false); + control.headers(1, baseHeaders(holdPath(String(dir), "none", 4200))); + await fx.stderrHas("HOLDING"); + await idle.writeFlushed(ping()); + release(String(dir)); + await expectGoawayThenFin(idle, 0); + expect((await control.body(1)).toString()).toBe("held"); + control.close(); + }, 20000); +}); + +// These need a build with usockets fault injection (ASAN builds): a send() that takes nothing, or +// a few bytes only, is the kernel of a peer that does not read. +describe.skipIf(!fault.available() || isWindows)( + "Bun.serve http2 graceful close when the kernel does not take the bytes", + () => { + const rule = (r: object) => encodeURIComponent(JSON.stringify(r)); + + test.each([ + // send() takes nothing, twice: the GOAWAY stays in the connection's buffer. + ["h2c", false, { syscall: "send", action: "zero", repeat: 2 }], + // send() takes 5 bytes once: the rest of the sealed GOAWAY record stays in the TLS spill. + ["TLS", true, { syscall: "send", action: "short", bytes: 5, repeat: 1 }], + ] as const)( + "closeIdleConnections() over %s neither counts nor ends a connection whose GOAWAY is not out; both follow", + async (_, secure, blocked) => { + await using fx = await startFixture({ tls: secure }); + const idle = await connectWide(fx.port, secure); + const control = await connectWide(fx.port, secure); + control.headers(1, baseHeaders("/fault-close-idle?rule=" + rule(blocked))); + await expectGoawayThenFin(idle, 0); + expect((await control.body(1)).toString()).toBe("counts 0,0"); + control.close(); + }, + ); + + test("stop() does not cut a response that is still in the connection's buffer", async () => { + await using fx = await startFixture({ tls: false }); + const raw = await connectWide(fx.port, false); + raw.headers(1, baseHeaders("/respond-blocked?n=100000")); + expect((await raw.body(1)).length).toBe(100000); + await expectGoawayThenFin(raw, 1); + await fx.stderrHas("STOPPED"); + }); + + test("a connection error whose GOAWAY the kernel does not take closes the connection and ends its streams", async () => { + await using fx = await startFixture({ tls: false }); + const raw = await connectWide(fx.port, false); + raw.headers(1, baseHeaders("/abort")); + // after: 1 lets the answer to this request out. The next send() is the GOAWAY of the error. + raw.headers(3, baseHeaders("/arm?rule=" + rule({ syscall: "send", action: "zero", after: 1, repeat: 1 }))); + expect((await raw.body(3)).toString()).toBe("armed"); + raw.write(frame(T.WINDOW_UPDATE, 0, 0, Buffer.from([0x7f, 0xff, 0xff, 0xff]))); + await raw.waitForClose(); + await fx.stderrHas("ABORTED"); + }); + }, +);