diff --git a/packages/bun-usockets/src/bsd.c b/packages/bun-usockets/src/bsd.c index a1190b3a6515..86410f6f9073 100644 --- a/packages/bun-usockets/src/bsd.c +++ b/packages/bun-usockets/src/bsd.c @@ -153,6 +153,9 @@ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int fl int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags, int max_packets) { if (max_packets > LIBUS_UDP_RECV_COUNT) max_packets = LIBUS_UDP_RECV_COUNT; + ssize_t injected = 0; int unused = 0; + if (US_FAULT_CHECK(US_FAULT_RECVMSG, fd, injected, unused)) return (int) injected; + (void)unused; #if defined(_WIN32) for (int i = 0; i < LIBUS_UDP_RECV_COUNT; i++) { while (1) { diff --git a/packages/bun-usockets/src/quic.c b/packages/bun-usockets/src/quic.c index c6cb52a21617..952dd380eb6e 100644 --- a/packages/bun-usockets/src/quic.c +++ b/packages/bun-usockets/src/quic.c @@ -58,6 +58,10 @@ struct us_quic_socket_context_s { /* us_quic_socket_context_shutdown ran while `processing` was set; * us_quic_engine_leave finishes it. */ int shutdown_pending; + /* us_quic_listen_socket_close ran, or the last UDP fd closed under live + * conns, while `processing` was set; us_quic_engine_leave runs + * us_quic_abort_conns. */ + int abort_pending; int is_client; unsigned int conn_count; unsigned int conn_ext_size; @@ -84,6 +88,7 @@ struct us_quic_socket_context_s { void (*on_hsk_done)(us_quic_socket_t *, int); void (*on_goaway)(us_quic_socket_t *); void (*on_close)(us_quic_socket_t *); + void (*on_released)(us_quic_socket_context_t *); void (*on_stream_open)(us_quic_stream_t *, int); void (*on_stream_headers)(us_quic_stream_t *); void (*on_stream_data)(us_quic_stream_t *, const char *, unsigned int, int); @@ -221,6 +226,19 @@ static void us_quic_send_unsent(us_quic_socket_context_t *ctx) { us_quic_engine_leave(ctx); } +/* Closes every conn at once, with the engine off the stack. Under + * process_conns lsquic_conn_abort only marks a conn, and cooldown skips a conn + * that is already going away: abort again, so the tick below closes them all. */ +static void us_quic_abort_conns(us_quic_socket_context_t *ctx) { + if (!ctx->engine) return; + for (us_quic_socket_t *qs = ctx->conns; qs; qs = qs->next) { + if (qs->conn) lsquic_conn_abort(qs->conn); + } + lsquic_engine_cooldown(ctx->engine); + us_quic_process(ctx); + us_quic_send_unsent(ctx); +} + /* ───── packets out ───── */ static inline socklen_t sa_len(const struct sockaddr *sa) { @@ -310,10 +328,10 @@ static int us_quic_sendmmsg(int fd, struct mmsghdr *mm, unsigned k) { * retires that one datagram through ci_packet_too_large and feeds DPLPMTUD. * * A datagram the kernel refuses for any other reason (EPERM from a firewall - * rule, EACCES, ENETUNREACH, ...) is counted as sent. To lsquic that is a - * packet lost on the wire: the loss timer retransmits it with backoff, and the - * handshake or idle timeout ends a connection that never gets through. Neither - * remaining outcome fits: + * rule, EACCES, ENETUNREACH, ...) is counted as sent, and so is one whose UDP + * socket has closed. To lsquic that is a packet lost on the wire: the loss + * timer retransmits it with backoff, and the handshake or idle timeout ends a + * connection that never gets through. Neither remaining outcome fits: * - With the errno passed through, lsquic closes the connection that owns * specs[sent]. On a shared unconnected UDP socket Linux reports a pending * ICMP error on the next send to *any* destination, so the error isn't @@ -330,7 +348,7 @@ static int us_quic_packets_out(void *out_ctx, const struct lsquic_out_spec *spec struct mmsghdr mm[BATCH]; while (sent < n) { us_quic_listen_socket_t *ls = (us_quic_listen_socket_t *) specs[sent].peer_ctx; - if (!ls->udp) { errno = EBADF; break; } + if (!ls->udp) { sent++; continue; } int fd = us_poll_fd((struct us_poll_t *) ls->udp); unsigned k = 0; while (k < BATCH && sent + k < n && specs[sent + k].peer_ctx == (void *) ls) { @@ -365,7 +383,7 @@ static int us_quic_packets_out(void *out_ctx, const struct lsquic_out_spec *spec #else for (; sent < n; sent++) { us_quic_listen_socket_t *ls = (us_quic_listen_socket_t *) specs[sent].peer_ctx; - if (!ls->udp) { errno = EBADF; break; } + if (!ls->udp) continue; LIBUS_SOCKET_DESCRIPTOR fd = us_poll_fd((struct us_poll_t *) ls->udp); if (us_quic_send_one(fd, &specs[sent]) < 0) { int stop = us_quic_send_would_block(errno) || errno == EMSGSIZE; @@ -378,16 +396,13 @@ static int us_quic_packets_out(void *out_ctx, const struct lsquic_out_spec *spec #endif /* EMSGSIZE goes to lsquic as it is. Everything else that stops the loop is - * backpressure, or a UDP socket that is already closed: report EAGAIN, so - * the engine pauses and resumes from on_drain → send_unsent_packets (or - * from its one-second failsafe when there is no socket left to poll). */ + * backpressure: report EAGAIN, so the engine pauses and resumes from + * on_drain → send_unsent_packets. */ if (sent < n && errno != EMSGSIZE) { errno = EAGAIN; us_quic_listen_socket_t *ls = (us_quic_listen_socket_t *) specs[sent].peer_ctx; - if (ls->udp) { - us_poll_change((struct us_poll_t *) ls->udp, ls->ctx->loop, - LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); - } + us_poll_change((struct us_poll_t *) ls->udp, ls->ctx->loop, + LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE); } return (int) sent; } @@ -415,6 +430,18 @@ static void us_quic_udp_on_drain(struct us_udp_socket_t *u) { us_quic_send_unsent(ls->ctx); } +/* Nothing of this context is on the wire: no UDP socket, no connection. */ +static int us_quic_is_released(us_quic_socket_context_t *ctx) { + return !ctx->listeners && ctx->conn_count == 0; +} + +/* The last UDP fd and the last conn go in either order: a drain closes the fd + * with the last conn, us_quic_release_pending_port and the loop's close on a + * socket error take the fd from under live conns. Both ends call this. */ +static void us_quic_report_if_released(us_quic_socket_context_t *ctx) { + if (ctx->on_released && us_quic_is_released(ctx)) ctx->on_released(ctx); +} + static void us_quic_udp_on_close(struct us_udp_socket_t *u) { us_quic_listen_socket_t *ls = (us_quic_listen_socket_t *) us_udp_socket_user(u); us_quic_socket_context_t *ctx = ls->ctx; @@ -429,6 +456,14 @@ static void us_quic_udp_on_close(struct us_udp_socket_t *u) { } ls->next = ctx->closed_listeners; ctx->closed_listeners = ls; + /* Conns with no UDP fd left have no wire, and the owner waits for the + * release: close them, now or when the engine leaves. */ + if (ctx->on_released && !ctx->listeners && ctx->conn_count) { + if (ctx->processing) ctx->abort_pending = 1; + else us_quic_abort_conns(ctx); + return; + } + us_quic_report_if_released(ctx); } /* ───── SSL ───── */ @@ -635,10 +670,14 @@ static void us_quic_on_conn_closed(lsquic_conn_t *conn) { #endif ctx->conn_count--; us_internal_disable_sweep_timer(ctx->loop); - /* During graceful drain the UDP fd is the only thing left holding the - * loop; release it when the last conn closes so the process can exit. */ - if (ctx->closing && ctx->conn_count == 0) { - while (ctx->listeners) us_udp_socket_close(ctx->listeners->udp); + if (ctx->conn_count == 0) { + if (ctx->closing && ctx->listeners) { + /* The graceful drain ends here: release the UDP fd, the only thing + * left holding the loop. us_quic_udp_on_close reports the release. */ + while (ctx->listeners) us_udp_socket_close(ctx->listeners->udp); + } else { + us_quic_report_if_released(ctx); + } } } @@ -900,6 +939,9 @@ void us_quic_socket_context_shutdown(us_quic_socket_context_t *ctx) { void us_quic_socket_context_free(us_quic_socket_context_t *ctx) { if (!ctx) return; ctx->closing = 1; + /* The owner is going away with the context: the fd and conn teardown below + * must not call into it. */ + ctx->on_released = NULL; struct us_loop_t *loop = ctx->loop; for (us_quic_socket_context_t **pp = &loop->data.quic_head; *pp; pp = &(*pp)->next) { if (*pp == ctx) { *pp = ctx->next; break; } @@ -1001,18 +1043,11 @@ us_quic_listen_socket_t *us_quic_socket_context_listen( return ls; } -static void us_quic_listen_socket_finish_close(us_quic_listen_socket_t *ls) { - if (ls->ctx->engine) { - lsquic_engine_cooldown(ls->ctx->engine); - us_quic_process(ls->ctx); - us_quic_send_unsent(ls->ctx); - } - /* on_conn_closed releases every fd when the last conn of a closing ctx goes. */ - if (ls->udp) us_udp_socket_close(ls->udp); -} - void us_quic_listen_socket_close(us_quic_listen_socket_t *ls) { - if (!ls || !ls->udp) return; + if (!ls) return; + us_quic_socket_context_t *ctx = ls->ctx; + /* The fd can be gone with conns left (us_quic_report_if_released): abort those. */ + if (!ls->udp && !ctx->conns) return; /* Send CONNECTION_CLOSE on every live conn before the fd disappears; * cooldown alone only schedules GOAWAY, which leaves peers waiting on * in-flight streams that this abrupt close will never serve. @@ -1026,26 +1061,26 @@ void us_quic_listen_socket_close(us_quic_listen_socket_t *ls) { * abrupt stop() went silent and pooled clients reused the dead session * until idle-timeout. IFC_ABORTED takes the IFC_IMMEDIATE_CLOSE_FLAGS * path which always packs CONNECTION_CLOSE. */ - if (ls->ctx->engine) { - for (us_quic_socket_t *qs = ls->ctx->conns; qs; qs = qs->next) { - if (qs->conn) lsquic_conn_abort(qs->conn); - } + if (ctx->engine && ctx->processing) { /* stop(true) from a request handler, or from a microtask it resolved. * The aborted conns only pack their CONNECTION_CLOSE once the running * tick resumes, so the fd has to outlive that call. */ - if (ls->ctx->processing) { - ls->close_pending = 1; - return; + for (us_quic_socket_t *qs = ctx->conns; qs; qs = qs->next) { + if (qs->conn) lsquic_conn_abort(qs->conn); } + ctx->abort_pending = 1; + if (ls->udp) ls->close_pending = 1; + return; } - us_quic_listen_socket_finish_close(ls); + us_quic_abort_conns(ctx); + if (ls->udp) us_udp_socket_close(ls->udp); } static void us_quic_finish_pending_closes(us_quic_socket_context_t *ctx) { for (us_quic_listen_socket_t *ls = ctx->listeners; ls; ) { if (!ls->close_pending) { ls = ls->next; continue; } ls->close_pending = 0; - us_quic_listen_socket_finish_close(ls); + us_udp_socket_close(ls->udp); /* udp_on_close unlinked `ls`; restart from the head. */ ls = ctx->listeners; } @@ -1058,7 +1093,16 @@ static void us_quic_after_engine_call(us_quic_socket_context_t *ctx) { ctx->shutdown_pending = 0; us_quic_socket_context_finish_shutdown(ctx); } - us_quic_finish_pending_closes(ctx); + if (ctx->abort_pending) { + /* Also for a listener whose fd us_quic_release_pending_port already took. */ + ctx->abort_pending = 0; + us_quic_abort_conns(ctx); + us_quic_finish_pending_closes(ctx); + } +} + +int us_quic_listen_socket_is_released(us_quic_listen_socket_t *ls) { + return us_quic_is_released(ls->ctx); } int us_quic_listen_socket_port(us_quic_listen_socket_t *ls) { @@ -1086,6 +1130,7 @@ DEF_CB(on_open, void (*cb)(us_quic_socket_t *)) DEF_CB(on_hsk_done, void (*cb)(us_quic_socket_t *, int)) DEF_CB(on_goaway, void (*cb)(us_quic_socket_t *)) DEF_CB(on_close, void (*cb)(us_quic_socket_t *)) +DEF_CB(on_released, void (*cb)(us_quic_socket_context_t *)) DEF_CB(on_stream_open, void (*cb)(us_quic_stream_t *, int)) DEF_CB(on_stream_headers, void (*cb)(us_quic_stream_t *)) DEF_CB(on_stream_data, void (*cb)(us_quic_stream_t *, const char *, unsigned int, int)) diff --git a/packages/bun-usockets/src/quic.h b/packages/bun-usockets/src/quic.h index bed47e6bafc3..c912a06077dc 100644 --- a/packages/bun-usockets/src/quic.h +++ b/packages/bun-usockets/src/quic.h @@ -51,7 +51,8 @@ us_quic_socket_context_t *us_create_quic_socket_context( unsigned int ext_size, unsigned int idle_timeout_s); /* Send GOAWAY on every connection and stop accepting new ones; the engine and - * timer keep running so in-flight streams drain. */ + * timer keep running so in-flight streams drain. The UDP socket closes with + * the last connection, and on_released reports it. */ void us_quic_socket_context_shutdown(us_quic_socket_context_t *ctx); void us_quic_socket_context_free(us_quic_socket_context_t *ctx); @@ -69,6 +70,9 @@ us_quic_listen_socket_t *us_quic_socket_context_listen( void us_quic_listen_socket_close(us_quic_listen_socket_t *ls); int us_quic_listen_socket_port(us_quic_listen_socket_t *ls); int us_quic_listen_socket_local_address(us_quic_listen_socket_t *ls, char *buf, int len); +/* 1 once the listener's context has no open UDP socket and no connection left. + * `ls` stays valid after its fd has closed, until us_quic_socket_context_free. */ +int us_quic_listen_socket_is_released(us_quic_listen_socket_t *ls); /* Client engine: one shared lsquic engine for all outbound connections on * `loop`. Per-connection cert verification is decided at connect time. */ @@ -116,6 +120,13 @@ void us_quic_socket_context_on_goaway(us_quic_socket_context_t *ctx, void (*on_goaway)(us_quic_socket_t *)); void us_quic_socket_context_on_close(us_quic_socket_context_t *ctx, void (*on_close)(us_quic_socket_t *)); +/* The context has closed its last UDP socket and its last connection has gone, + * in either order. It runs after the close(2). It can run inside + * lsquic_engine_process_conns, so it must not enter the engine or free the + * context. us_quic_socket_context_free itself does not report. A context with + * this hook closes its connections when its last UDP socket goes first. */ +void us_quic_socket_context_on_released(us_quic_socket_context_t *ctx, + void (*on_released)(us_quic_socket_context_t *)); /* Stream-level callbacks */ void us_quic_socket_context_on_stream_open(us_quic_socket_context_t *ctx, diff --git a/packages/bun-uws/src/Http3App.h b/packages/bun-uws/src/Http3App.h index bde4a4828e01..ef75b28cf532 100644 --- a/packages/bun-uws/src/Http3App.h +++ b/packages/bun-uws/src/Http3App.h @@ -55,6 +55,11 @@ struct H3App { } /* GOAWAY + drain. The engine itself is torn down in the destructor. */ void close() { http3Context->shutdown(); } + /* See us_quic_socket_context_on_released. */ + void onReleased(void (*cb)(void *), void *user) { + http3Context->getContextData()->onReleased = cb; + http3Context->getContextData()->onReleasedUser = user; + } bool addServerNameWithOptions(const char *hostname, SocketContextOptions options) { us_bun_socket_context_options_t raw; memcpy(&raw, &options, sizeof(raw)); diff --git a/packages/bun-uws/src/Http3Context.h b/packages/bun-uws/src/Http3Context.h index 379fc3223fd8..5d746e1b41fa 100644 --- a/packages/bun-uws/src/Http3Context.h +++ b/packages/bun-uws/src/Http3Context.h @@ -67,6 +67,11 @@ struct Http3Context { rd->~Http3ResponseData(); }); + us_quic_socket_context_on_released(ctx, [](us_quic_socket_context_t *ctx) { + Http3ContextData *cd = (Http3ContextData *) us_quic_socket_context_ext(ctx); + if (cd->onReleased) cd->onReleased(cd->onReleasedUser); + }); + return (Http3Context *) ctx; } diff --git a/packages/bun-uws/src/Http3ContextData.h b/packages/bun-uws/src/Http3ContextData.h index 77a555e09038..207b8f5c454c 100644 --- a/packages/bun-uws/src/Http3ContextData.h +++ b/packages/bun-uws/src/Http3ContextData.h @@ -14,6 +14,9 @@ struct Http3ContextData { Http3Request *httpRequest; }; HttpRouter router; + /* H3App::onReleased */ + void (*onReleased)(void *user) = nullptr; + void *onReleasedUser = nullptr; }; } diff --git a/patches/lsquic/going-away-tickable.patch b/patches/lsquic/going-away-tickable.patch new file mode 100644 index 000000000000..fb2bd663df15 --- /dev/null +++ b/patches/lsquic/going-away-tickable.patch @@ -0,0 +1,14 @@ +--- a/src/liblsquic/lsquic_full_conn_ietf.c ++++ b/src/liblsquic/lsquic_full_conn_ietf.c +@@ -3385,6 +3385,11 @@ + end: + conn->ifc_flags |= IFC_GOING_AWAY; + maybe_close_conn(conn); ++ /* bun: a server connection that has not carried a request skips the ++ * GOAWAY write above, and with it the only call that makes the connection ++ * tickable. maybe_close_conn() has just scheduled its CONNECTION_CLOSE, ++ * and only a tick sends it. */ ++ lsquic_engine_add_conn_to_tickable(conn->ifc_enpub, lconn); + } + + diff --git a/scripts/build/deps/lsquic.ts b/scripts/build/deps/lsquic.ts index b60d95f0828d..f24415697106 100644 --- a/scripts/build/deps/lsquic.ts +++ b/scripts/build/deps/lsquic.ts @@ -134,6 +134,11 @@ export const lsquic: Dependency = { // H3_CLOSED_CRITICAL_STREAM and closes the connection, which kills the // requests the graceful stop was draining. Reject only request streams. "patches/lsquic/goaway-accept-uni-streams.patch", + // ci_going_away() skips the GOAWAY write for a server connection that has + // not carried a request, and with it the call that makes the connection + // tickable. Its CONNECTION_CLOSE then waited for the next alarm or packet, + // and a graceful server.stop() waited with it. + "patches/lsquic/going-away-tickable.patch", ], fetchDeps: ["zlib", "lshpack", "lsqpack", "boringssl"], diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index 5cb83b6cf091..fa91642d0598 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -233,6 +233,8 @@ pub(crate) struct NewServer { /// negotiate "h2" (ALPN) or open with the cleartext preface. pub(crate) h2_app: Option<*mut uws_sys::h2::App>, pub(crate) h3_listener: Option<*mut uws_sys::h3::ListenSocket>, + /// `h3_listener` after a stop took it; stays valid after its fd closes, until `h3::App::destroy` frees it. + pub(crate) h3_stopped_listener: Option>, /// Cached `h3=":"; ma=86400` for Alt-Svc on H1 responses; formatted /// once in onH3Listen so renderMetadata doesn't reformat per-request. pub(crate) h3_alt_svc: Box<[u8]>, @@ -261,7 +263,7 @@ pub(crate) struct NewServer { pub(crate) base_url_string_for_joining: Box<[u8]>, pub(crate) config: ServerConfig, pub(crate) pending_requests: core::cell::Cell, - /// Live HTTP connections (accepted, not yet closed or upgraded), fed by + /// Live TCP connections (accepted, not yet closed or upgraded), fed by /// the uWS filter in [`NewServer::listen`]. Any of them can still dispatch /// a handler, so the wrapper stays `Strong` until this drains /// ([`NewServer::is_drained`]); for Bun.serve it also holds the @@ -275,7 +277,7 @@ pub(crate) struct NewServer { /// `AnyServer` handles on the JS thread. pub(crate) active_websocket_count: core::cell::Cell, /// Set across [`NewServer::deinit_if_we_can`] and the synchronous - /// `app.close()` drain in `stop_listening`; lets a nested call (reached + /// close calls in `stop_listening`; lets a nested call (reached /// via a callback the body fires) early-return instead of re-running the /// downgrade/teardown while the outer frame still holds `&mut self`. deinit_running: core::cell::Cell, @@ -545,6 +547,20 @@ impl NewServer { } } + /// quic.c has closed the HTTP/3 UDP socket and its last QUIC connection is gone ([`Self::has_open_h3_transport`]). + extern "C" fn on_h3_released(user_data: *mut c_void) { + // SAFETY: `user_data` is the `*mut Self` registered in `listen()`; quic.c drops the hook + // before it frees the app's context, so the server is alive. Only `&Self` is formed: the + // H3 close calls of `stop_listening` can reach this under its `&mut self`. + if unsafe { &*user_data.cast::() }.deinit_running.get() { + return; + } + // SAFETY: `stop_listening` holds `deinit_running` across those calls, and a nested stop + // restores it, so no `&mut self` is live. This can run inside lsquic's `process_conns`: + // `deinit_if_we_can` reads cells and queues tasks, it does not enter the engine. + unsafe { &mut *user_data.cast::() }.deinit_if_we_can(); + } + /// Build the server's base URL string (`http(s)://host:port/`, or a /// `unix://`/abstract-socket URL) from the configured listen address. /// Errors only on allocation failure. @@ -607,7 +623,7 @@ impl NewServer { /// Returns the wrapper while it is alive (`Strong` or `Weak`) and its VM may still run script, /// else `None`. It stays `Strong` — the WriteBarrier root of the handler shadows — until the - /// server is drained (no listener, request, HTTP/1 connection or websocket left; connections count + /// server is drained (no listener, request, connection or websocket left; TCP connections count /// from accept), so nothing dispatches through a wrapper the GC may have collected. A VM whose /// script gate has closed (a worker that called `process.exit()` or was asked to terminate, still /// draining the current loop tick before its stop phase closes the listener) must not have a @@ -617,9 +633,8 @@ impl NewServer { if !self.vm().script_allowed() { return None; } - // HTTP/3 connections are not counted yet, so a stream on one can still arrive after the drain. debug_assert!( - self.h3_app.is_some() || self.js_value.is_strong() || self.js_value.try_get().is_none(), + self.js_value.is_strong() || self.js_value.try_get().is_none(), "a socket outlived the server's connection accounting" ); self.js_value.try_get() @@ -1655,6 +1670,16 @@ impl NewServer { && !self.has_listener() && !self.has_active_web_sockets() && !self.has_loop_holding_connections() + && !self.has_open_h3_transport() + } + + /// A stopped HTTP/3 listener still has its UDP socket or a QUIC connection; [`Self::on_h3_released`] runs when both are gone. + fn has_open_h3_transport(&self) -> bool { + Self::HAS_H3 + && self + .h3_stopped_listener + // S008: `h3::ListenSocket` is an `opaque_ffi!` ZST — safe deref. + .is_some_and(|h3l| !bun_opaque::opaque_deref_mut(h3l.as_ptr()).is_released()) } /// Nothing is left that can dispatch a handler: [`Self::is_closed`] and @@ -1724,10 +1749,21 @@ impl NewServer { } if Self::HAS_H3 { - if let Some(h3l) = self.h3_listener.take() { + let listening = self.h3_listener.take().and_then(core::ptr::NonNull::new); + if listening.is_some() { + self.h3_stopped_listener = listening; + } + // An abrupt stop also closes what an earlier graceful stop left to drain. + let to_close = if abrupt { + self.h3_stopped_listener + } else { + listening + }; + if let Some(h3l) = to_close { // Graceful: GOAWAY + drain via the still-open UDP socket; the // engine rejects new conns and the timer keeps in-flight streams - // progressing until deinit. Abrupt: close the fd now. + // progressing until the last connection is gone. Abrupt: close the fd now. + let was_running = self.deinit_running.replace(true); if !abrupt { if let Some(h3a) = self.h3_app { // S008: `h3::App` is an `opaque_ffi!` ZST — safe deref. @@ -1735,15 +1771,19 @@ impl NewServer { } } else { // S008: `h3::ListenSocket` is an `opaque_ffi!` ZST — safe deref. - bun_opaque::opaque_deref_mut(h3l).close(); + bun_opaque::opaque_deref_mut(h3l.as_ptr()).close(); } + self.deinit_running.set(was_running); } } let already_terminated = self.flags.contains(ServerFlags::TERMINATED); let Some(listener) = self.listener.take() else { if Self::HAS_H3 && self.h3_app.is_some() { - self.unref(); + // As in the TCP arm below: a graceful stop keeps the ref while QUIC connections drain. + if abrupt || self.is_closed() { + self.unref(); + } self.notify_inspector_server_stopped(); if abrupt { self.flags.insert(ServerFlags::TERMINATED); @@ -1760,10 +1800,10 @@ impl NewServer { } self.flags.insert(ServerFlags::TERMINATED); if let Some(app) = self.app { - self.deinit_running.set(true); + let was_running = self.deinit_running.replace(true); // S012: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. bun_opaque::opaque_deref_mut(app).close(); - self.deinit_running.set(false); + self.deinit_running.set(was_running); } if let Some(ws) = self.config.websocket.as_mut() { ws.handler.server = None; @@ -1817,10 +1857,10 @@ impl NewServer { // reaps it — verified against Node v26. if !self.config.is_node_http_server { if let Some(app) = self.app { - self.deinit_running.set(true); + let was_running = self.deinit_running.replace(true); // S012: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. let _closed = bun_opaque::opaque_deref_mut(app).close_idle_connections(true); - self.deinit_running.set(false); + self.deinit_running.set(was_running); } } } else if !self.flags.contains(ServerFlags::TERMINATED) { @@ -1834,10 +1874,10 @@ impl NewServer { // while this frame still holds `&mut self`. Hold the re-entrance // guard across the drain so that nested call early-returns — // `stop()` runs `deinit_if_we_can` itself right after this returns. - self.deinit_running.set(true); + let was_running = self.deinit_running.replace(true); // S012: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. bun_opaque::opaque_deref_mut(self.app.unwrap()).close(); - self.deinit_running.set(false); + self.deinit_running.set(was_running); // Only clear after the drain — `on_close` defers reach // `on_websocket_closed` through `handler.server`, so wiping it // earlier would strand the live-socket count and the idle pass @@ -2146,6 +2186,8 @@ impl NewServer { server.abort_handle.leave(); if Self::HAS_H3 { + // The app frees this listen socket, and a close callback of the apps below can run `is_closed()`. + server.h3_stopped_listener = None; if let Some(h3a) = server.h3_app.take() { // SAFETY: live H3::App handle owned by this server. unsafe { uws_sys::h3::App::destroy(h3a) }; @@ -2200,6 +2242,7 @@ impl NewServer { h3_app: None, h2_app: None, h3_listener: None, + h3_stopped_listener: None, h3_alt_svc: Box::<[u8]>::default(), js_value: jsc::JsRef::empty(), pending_requests: core::cell::Cell::new(0), @@ -3087,6 +3130,13 @@ impl NewServer { // S012: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. bun_opaque::opaque_deref_mut(app).filter(Self::on_connection_filter, this.cast::()); + if Self::HAS_H3 { + if let Some(h3_app) = this_ref.h3_app { + // S008: `h3::App` is an `opaque_ffi!` ZST — safe deref. + bun_opaque::opaque_deref_mut(h3_app) + .on_released(Self::on_h3_released, this.cast::()); + } + } if this_ref.config.is_node_http_server { // SAFETY: `this` is the live boxed server from `init()`; no other diff --git a/src/uws_sys/h3.rs b/src/uws_sys/h3.rs index caabd299d636..939b3d47cbb2 100644 --- a/src/uws_sys/h3.rs +++ b/src/uws_sys/h3.rs @@ -22,6 +22,10 @@ impl ListenSocket { pub fn close(&mut self) { c::uws_h3_listen_socket_close(self) } + /// The app has no open UDP socket and no QUIC connection left. Valid on a closed listener until the app is destroyed. + pub fn is_released(&mut self) -> bool { + c::uws_h3_listen_socket_is_released(self) + } pub fn get_local_port(&mut self) -> Option { u16::try_from(c::uws_h3_listen_socket_port(self)).ok() } @@ -418,6 +422,11 @@ impl App { pub fn close(&mut self) { c::uws_h3_app_close(self) } + /// `cb(user)` runs when [`ListenSocket::is_released`] becomes true; it must not enter the QUIC engine or destroy the app. + pub fn on_released(&mut self, cb: unsafe extern "C" fn(user: *mut c_void), user: *mut c_void) { + // SAFETY: live handle; cb/user are stored and invoked on this thread. + unsafe { c::uws_h3_app_on_released(self, cb, user) } + } pub fn clear_routes(&mut self) { c::uws_h3_app_clear_routes(self) } @@ -595,6 +604,11 @@ mod c { ) -> *mut App; pub(super) fn uws_h3_app_destroy(app: *mut App); pub(super) safe fn uws_h3_app_close(app: &mut App); + pub(super) fn uws_h3_app_on_released( + app: *mut App, + cb: unsafe extern "C" fn(user: *mut c_void), + user: *mut c_void, + ); pub(super) safe fn uws_h3_app_clear_routes(app: &mut App); pub(super) fn uws_h3_app_add_server_name( app: *mut App, @@ -687,6 +701,7 @@ mod c { len: c_int, ) -> c_int; pub(super) safe fn uws_h3_listen_socket_close(ls: &mut ListenSocket); + pub(super) safe fn uws_h3_listen_socket_is_released(ls: &mut ListenSocket) -> bool; pub(super) safe fn uws_h3_res_state(res: &mut Response) -> State; pub(super) fn uws_h3_res_end(res: *mut Response, p: *const u8, n: usize, close: bool); diff --git a/src/uws_sys/libuwsockets_h3.cpp b/src/uws_sys/libuwsockets_h3.cpp index fa06529014b0..231ab03a2937 100644 --- a/src/uws_sys/libuwsockets_h3.cpp +++ b/src/uws_sys/libuwsockets_h3.cpp @@ -51,6 +51,7 @@ uws_h3_app_t* uws_h3_create_app(struct us_bun_socket_context_options_t options, void uws_h3_app_destroy(uws_h3_app_t* app) { delete (H3App*)app; } void uws_h3_app_close(uws_h3_app_t* app) { ((H3App*)app)->close(); } +void uws_h3_app_on_released(uws_h3_app_t* app, void (*cb)(void*), void* user) { ((H3App*)app)->onReleased(cb, user); } void uws_h3_app_clear_routes(uws_h3_app_t* app) { ((H3App*)app)->clearRoutes(); } bool uws_h3_app_add_server_name(uws_h3_app_t* app, const char* hostname, @@ -97,6 +98,7 @@ int uws_h3_listen_socket_local_address(us_quic_listen_socket_t* ls, char* buf, i return us_quic_listen_socket_local_address(ls, buf, len); } void uws_h3_listen_socket_close(us_quic_listen_socket_t* ls) { us_quic_listen_socket_close(ls); } +bool uws_h3_listen_socket_is_released(us_quic_listen_socket_t* ls) { return us_quic_listen_socket_is_released(ls) != 0; } /* ───── response ───── */ diff --git a/test/js/bun/http/serve-http3.test.ts b/test/js/bun/http/serve-http3.test.ts index d2df224d4f3b..7b86316917f3 100644 --- a/test/js/bun/http/serve-http3.test.ts +++ b/test/js/bun/http/serve-http3.test.ts @@ -1,4 +1,5 @@ import type { ServerWebSocket } from "bun"; +import { socketFaultInjection } from "bun:internal-for-testing"; import { describe, expect, test } from "bun:test"; import { createHash, createPrivateKey, randomBytes, X509Certificate } from "crypto"; import { readFileSync } from "fs"; @@ -1516,6 +1517,444 @@ describe("Bun.serve HTTP/3 lifecycle", () => { }); }); + // A graceful stop() keeps the UDP socket for the connections that still have + // a response to deliver: the last of them goes once the peer has acknowledged + // it, and takes the socket with it. The promise of stop() settles from that + // release, so the port is free for the statement behind the await. Nothing is + // awaited at the top level: only the server keeps the process alive for that + // statement. stop() in the microtask runs inside lsquic's process_conns, the + // later task does not. + const inFlightStop = (beforeStop: string, http1: boolean, afterStop: string) => ` + const tls = ${JSON.stringify(tls)}; + const inHandler = Promise.withResolvers(); + const proceed = Promise.withResolvers(); + const options = { hostname: "127.0.0.1", tls, http3: true, http1: ${http1} }; + const h3 = { protocol: "http3", tls: { rejectUnauthorized: false } }; + const server = Bun.serve({ + ...options, port: 0, + async fetch() { + inHandler.resolve(); + await proceed.promise; + return new Response("first"); + }, + }); + const port = server.port; + const first = fetch("https://127.0.0.1:" + port + "/", h3).then(r => r.text()); + inHandler.promise.then(async () => { + ${beforeStop} + const stopped = server.stop(); + proceed.resolve(); + await stopped; + ${afterStop} + }); + `; + + test.each([ + ["in a microtask of the handler", false, ""], + ["in a later task", false, "await new Promise(resolve => setImmediate(resolve));"], + ["in a later task", true, "await new Promise(resolve => setImmediate(resolve));"], + ])( + "await server.stop() %s with an H3 request in flight, then Bun.serve on the same port binds (http1: %p)", + async (_where, http1, beforeStop) => { + const script = inFlightStop( + beforeStop, + http1, + ` + const second = Bun.serve({ ...options, port, fetch: () => new Response("second") }); + const answers = [await first, await fetch("https://127.0.0.1:" + port + "/", h3).then(r => r.text())]; + await second.stop(true); + console.log(answers.join(",")); + `, + ); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "first,second\n", stderr: "" }); + expect(exitCode).toBe(0); + }, + ); + + // The port is free for every binder, not only for the next HTTP/3 listener. + test("await server.stop() with an H3 request in flight, then Bun.udpSocket on the same port binds", async () => { + const script = inFlightStop( + "", + false, + ` + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + console.log("bound," + (await first)); + `, + ); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "bound,first\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + + describe("server.stop() with nothing to deliver frees the UDP port", () => { + const options = { hostname: "127.0.0.1", tls, http3: true, http1: false, fetch: () => new Response("ok") }; + + test("before it returns, with no H3 connection", async () => { + const first = Bun.serve({ ...options, port: 0 }); + const port = first.port!; + const stopped = first.stop(); + await using second = Bun.serve({ ...options, port }); + expect(second.port).toBe(port); + await stopped; + }); + + test("behind the await, with an idle H3 connection", async () => { + await using first = Bun.serve({ ...options, port: 0 }); + const port = first.port!; + for (let i = 0; i < 2; i++) expect(await fetchH3(port, "/").then(res => res.text())).toBe("ok"); + expect(first.pendingRequests).toBe(0); + await first.stop(); + await using second = Bun.serve({ ...options, port }); + expect(second.port).toBe(port); + }); + + // Such a connection gets no GOAWAY, and lsquic did not tick it: its + // CONNECTION_CLOSE waited for the next alarm. The server can close it at + // once only when the client has acknowledged the server's SETTINGS, which + // has no signal here, so each attempt gives the acknowledgement more time. + test("before it returns, with an H3 connection that has not carried a request", async () => { + let freed = false; + for (let wait = 1; wait <= 128 && !freed; wait *= 2) { + await using server = Bun.serve({ ...options, port: 0 }); + const port = server.port!; + const session = await connect( + { address: "127.0.0.1", port } as SocketAddress, + { alpn: "h3", servername: "localhost", verifyPeer: "manual", onerror() {} } as SessionOptions, + ); + session.closed.catch(() => {}); + try { + await session.opened; + await Bun.sleep(wait); + const stopped = server.stop(); + try { + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + freed = true; + } catch (e) { + if ((e as NodeJS.ErrnoException).code !== "EADDRINUSE") throw e; + } + server.stop(true); + await stopped; + } finally { + if (!session.destroyed) session.close().catch(() => {}); + } + } + expect(freed).toBe(true); + }); + }); + + // A graceful stop() leaves a connection with a request in flight open. A + // later stop(true), or the disposal of the server, must still reach it: abort + // the connection with CONNECTION_CLOSE and close the UDP socket. From a later + // task the socket is closed when the call returns. Inside the handler lsquic's + // process_conns is on the stack and both stops are deferred until it returns. + test.each([ + ["server.stop(true)", "server.stop(true);"], + ["server[Symbol.dispose]()", "server[Symbol.dispose]();"], + ])("server.stop() then %s with a parked H3 request frees the UDP port", async (_how, force) => { + const script = (handler: string, main: string) => ` + const tls = ${JSON.stringify(tls)}; + const inHandler = Promise.withResolvers(); + const server = Bun.serve({ + port: 0, hostname: "127.0.0.1", tls, http3: true, http1: false, + async fetch() { + ${handler} + await new Promise(() => {}); + }, + }); + const port = server.port; + const outcome = fetch("https://127.0.0.1:" + port + "/", { + protocol: "http3", tls: { rejectUnauthorized: false }, + method: "POST", + body: new ReadableStream({ start(c) { c.enqueue("payload"); c.close(); } }), + }).then(r => r.text(), e => e.code); + ${main} + console.log("bound," + (await outcome)); + `; + const scripts = { + "from a later task": script( + `inHandler.resolve();`, + ` + await inHandler.promise; + await new Promise(resolve => setImmediate(resolve)); + const stopped = server.stop(); + ${force} + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + await stopped; + `, + ), + "inside the handler": script( + `inHandler.resolve(server.stop()); ${force}`, + ` + await inHandler.promise; + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + `, + ), + }; + const results: Record = {}; + for (const [where, source] of Object.entries(scripts)) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", source], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + results[where] = { stdout, stderr, exitCode }; + } + const expected = { stdout: "bound,HTTP3StreamReset\n", stderr: "", exitCode: 0 }; + expect(results).toEqual({ "from a later task": expected, "inside the handler": expected }); + }); + + // The context that started the server stops it abruptly through its abort + // handle, which has to reach the connection that the graceful stop left. + test("server.stop() then dispose() of its Bun.ModuleGraph with a parked H3 request frees the UDP port", async () => { + const script = ` + const tls = ${JSON.stringify(tls)}; + const inHandler = Promise.withResolvers(); + const graph = new Bun.ModuleGraph(); + const server = graph.run(() => + Bun.serve({ + port: 0, hostname: "127.0.0.1", tls, http3: true, http1: false, + async fetch() { + inHandler.resolve(); + await new Promise(() => {}); + }, + }), + ); + const port = server.port; + const outcome = fetch("https://127.0.0.1:" + port + "/", { + protocol: "http3", tls: { rejectUnauthorized: false }, + method: "POST", + body: new ReadableStream({ start(c) { c.enqueue("payload"); c.close(); } }), + }).then(r => r.text(), e => e.code); + await inHandler.promise; + await new Promise(resolve => setImmediate(resolve)); + server.stop(); + graph.dispose(); + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + console.log("bound," + (await outcome)); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "bound,HTTP3StreamReset\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + + // The rebind takes the UDP socket from under the connections of the stopped + // server, one of them idle. They cannot send their CONNECTION_CLOSE any more, + // and the promise of stop(true) still has to settle when the handler returns. + test("server.stop(true) then Bun.serve on the same port inside an H3 handler settles the promise of the stop", async () => { + const script = ` + const tls = ${JSON.stringify(tls)}; + const options = { hostname: "127.0.0.1", tls, http3: true, http1: false }; + const h3 = session => ({ protocol: "http3", tls: { rejectUnauthorized: false }, session }); + const first = Bun.serve({ + ...options, port: 0, + fetch(req) { + if (new URL(req.url).pathname !== "/stop") return new Response("idle"); + const stopped = first.stop(true); + const second = Bun.serve({ ...options, port: first.port, fetch: () => new Response("second") }); + stopped.then(async () => { + await second.stop(true); + console.log("the first server stopped"); + process.exit(0); + }); + return new Response("stopping"); + }, + }); + const origin = "https://127.0.0.1:" + first.port; + await fetch(origin + "/idle", h3(new Bun.FetchSession())).then(r => r.text()); + fetch(origin + "/stop", h3(new Bun.FetchSession())).catch(() => {}); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "the first server stopped\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + + // The client is killed while its request is parked, so nothing acknowledges + // the response: the connection drains until lsquic's idle timer closes it or + // an abrupt stop does. Until then a repeated stop() returns the same promise. + test.each([ + ["server.stop(true)", 30, (server: { stop(abrupt: boolean): unknown }) => void server.stop(true)], + ["the idle timeout", 1, () => {}], + ])( + "a graceful stop() whose H3 peer vanished stays pending until %s ends the drain", + async (_how, idleTimeout, end) => { + const inHandler = Promise.withResolvers(); + const proceed = Promise.withResolvers(); + const server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + tls, + http3: true, + http1: false, + idleTimeout, + async fetch() { + inHandler.resolve(); + await proceed.promise; + return new Response("first"); + }, + }); + try { + const port = server.port!; + await using client = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `await fetch("https://127.0.0.1:${port}/", { protocol: "http3", tls: { rejectUnauthorized: false } }).then(r => r.text());`, + ], + env: bunEnv, + stdout: "ignore", + stderr: "inherit", + }); + client.exited.then(code => + inHandler.reject(new Error(`the client exited before its request arrived: ${code}`)), + ); + await inHandler.promise; + client.kill("SIGKILL"); + await client.exited; + const stopped = server.stop(); + proceed.resolve(); + while (server.pendingRequests !== 0) await new Promise(resolve => setImmediate(resolve)); + expect({ same: server.stop() === stopped, status: Bun.peek.status(stopped) }).toEqual({ + same: true, + status: "pending", + }); + end(server); + await stopped; + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + } finally { + server.stop(true); + } + }, + ); + + // The event loop closes a UDP socket whose receive fails, and nothing tells + // the server. The connections have no socket left, so quic.c closes them + // then: a stop() that waited for them to drain would hold its promise for + // the idle timeout. The fault is armed in the server process only. The next + // datagram from a client fails its receive, so the acknowledgement of the + // last response never arrives. A stop() of a server with nothing left + // returns a settled promise, else the pending promise of the first stop(). + test.skipIf(!socketFaultInjection.available()).each([ + ["32 connections, closed before the stop", 32, "", "server.stop();"], + ["1 connection, closed during the drain", 1, "server.stop();", ""], + ])( + "a graceful stop() does not wait for the H3 connections of a UDP socket that the loop closed on a receive error (%s)", + async (_what, connections, inHandler, afterClose) => { + const script = ` + import { socketFaultInjection as fault } from "bun:internal-for-testing"; + const armed = Promise.withResolvers(); + const server = Bun.serve({ + port: 0, hostname: "127.0.0.1", tls: ${JSON.stringify(tls)}, http3: true, http1: false, + fetch(req) { + if (new URL(req.url).pathname !== "/arm") return new Response("ok"); + fault.set({ syscall: "recvmsg", action: "errno", errno: "ENOBUFS" }); + ${inHandler} + armed.resolve(); + return new Response("armed"); + }, + }); + const port = server.port; + console.error("PORT=" + port); + await armed.promise; + while (true) { + try { + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + break; + } catch (e) { + if (e.code !== "EADDRINUSE") throw e; + } + await new Promise(resolve => setImmediate(resolve)); + } + ${afterClose} + console.error("STOPPED " + Bun.peek.status(server.stop())); + `; + await withCustomServer(script, async (port, _send, waitForStderr) => { + using sessions = new DisposableStack(); + const open = Array.from({ length: connections }, () => sessions.use(new Bun.FetchSession())); + const get = (session: Bun.FetchSession, path: string) => + fetchH3(port, path, { session } as RequestInit).then(res => res.text()); + // One at a time, so lsquic keeps its first send batch size: the close + // then covers more connections than one batch holds. + for (const session of open) expect(await get(session, "/")).toBe("ok"); + expect(await get(open[0], "/arm")).toBe("armed"); + expect((await waitForStderr(/STOPPED (\w+)/))[1]).toBe("fulfilled"); + }); + }, + ); + + // Nothing references the stopped server, and the collector must not finalize + // it while a connection still drains: that frees the QUIC engine. + test("a stopped H3 server that nothing references delivers the response in flight across a GC", async () => { + const script = ` + const tls = ${JSON.stringify(tls)}; + const body = Buffer.alloc(1024 * 1024, "abcdefghijklmnop"); + const inHandler = Promise.withResolvers(); + const proceed = Promise.withResolvers(); + let port, stop; + (function () { + const server = Bun.serve({ + port: 0, hostname: "127.0.0.1", tls, http3: true, http1: false, + async fetch() { + inHandler.resolve(); + await proceed.promise; + return new Response(new ReadableStream({ + type: "direct", + async pull(controller) { controller.write(body); await controller.end(); }, + })); + }, + }); + port = server.port; + stop = () => void server.stop(); + })(); + const received = fetch("https://127.0.0.1:" + port + "/", { protocol: "http3", tls: { rejectUnauthorized: false } }) + .then(r => r.bytes()).then(bytes => bytes.length, e => e.code); + await inHandler.promise; + stop(); + stop = undefined; + proceed.resolve(); + for (let i = 0; i < 8; i++) { + Bun.gc(true); + await new Promise(resolve => setImmediate(resolve)); + } + console.log(await received); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "1048576\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + // Each QUIC connection counts as a virtual poll (loop->num_polls); after // server.stop() drains, the last conn close releases the UDP fd and the // loop has no polls left — the process exits without process.exit(). @@ -1555,6 +1994,64 @@ describe("Bun.serve HTTP/3 lifecycle", () => { expect(exitCode).toBe(0); }); + // The server holds the loop until its last QUIC connection has gone, so a + // program whose last act is `await server.stop()` still delivers the response + // that was in flight, and the statement behind the await still runs. 1 MB + // needs several round trips of acknowledgements. + test.each([false, true])( + "a server that stops with a 1 MB H3 response in flight delivers it before the process ends (http1: %p)", + async http1 => { + const expected = Buffer.alloc(1024 * 1024, "abcdefghijklmnop"); + const script = ` + const body = Buffer.alloc(1024 * 1024, "abcdefghijklmnop"); + const server = Bun.serve({ + port: 0, hostname: "127.0.0.1", tls: ${JSON.stringify(tls)}, http3: true, http1: ${http1}, + fetch() { + setImmediate(async () => { + await server.stop(); + (await Bun.udpSocket({ port, hostname: "127.0.0.1" })).close(); + console.log("stopped, port free"); + }); + return new Response(body); + }, + }); + const port = server.port; + console.log("PORT=" + port); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + let stdout = ""; + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + const readUntil = async (done: () => boolean) => { + while (!done()) { + const chunk = await reader.read(); + if (chunk.done) return; + stdout += decoder.decode(chunk.value, { stream: true }); + } + }; + await readUntil(() => /PORT=\d+\n/.test(stdout)); + const port = Number(stdout.match(/PORT=(\d+)/)?.[1]); + expect(port).toBeGreaterThan(0); + const [body, , stderr, exitCode] = await Promise.all([ + fetchH3(port, "/").then(res => res.bytes()), + readUntil(() => false), + proc.stderr.text(), + proc.exited, + ]); + expect({ whole: Buffer.compare(body, expected) === 0, stdout, stderr, exitCode }).toEqual({ + whole: true, + stdout: `PORT=${port}\nstopped, port free\n`, + stderr: "", + exitCode: 0, + }); + }, + ); + // The HEADERS of the first response on a connection wait for the QPACK encoder // stream, and lsquic takes no DATA until they are out. A small streamed body is // therefore written, and the response ended, from the stream's writable