diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index a5454542d838..2f1d40cb060e 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -650,15 +650,25 @@ where server.vm().as_mut().event_loop_mut().drain_microtasks() } + /// Runs `on_abort` itself (may free `self`) if a nested event loop run already closed the socket. pub(crate) fn set_abort_handler(&self) { if self.flags.has_abort_handler() { return; } - if let Some(resp) = self.resp.get() { - self.flags.set_has_abort_handler(true); - // SAFETY: FFI handle valid while resp is Some - resp.on_aborted(|this, resp| Self::on_abort(this, resp), self.as_ctx_ptr()); + let Some(resp) = self.resp.get() else { + return; + }; + self.flags.set_has_abort_handler(true); + if resp.is_closed() { + // `req` is still set only while the dispatch is on the stack: snapshot as `to_async` would have. + if let (Some(req), Some(request)) = (self.req.get(), self.request_mut()) { + self.to_async_without_abort_handler(req, request); + } + Self::on_abort(self.as_ctx_ptr(), resp); + return; } + // SAFETY: FFI handle valid while resp is Some + resp.on_aborted(|this, resp| Self::on_abort(this, resp), self.as_ctx_ptr()); } pub(crate) fn set_cookies(&self, cookie_map: Option<*mut CookieMap>) { @@ -785,7 +795,7 @@ where pub(crate) fn should_render_missing(&self) -> bool { // If we did not respond yet, we should render missing // To allow this all the conditions above should be true: - // 1 - still has a response (not detached) + // 1 - still has a response (not detached, socket still open) // 2 - not aborted // 3 - not marked completed // 4 - not marked pending @@ -831,7 +841,7 @@ where "no sendfile context" }, ); - self.resp.get().is_some() + self.resp.get().is_some_and(|resp| !resp.is_closed()) && !self.flags.aborted() && !self.flags.has_marked_complete() && !self.flags.has_marked_pending() @@ -2028,6 +2038,8 @@ where debug_assert!(this.server.get().is_some()); let global_this = this.server().global_this(); + // Armed here, not in `to_async()`: `stop(true)` inside `pull()` must reach `on_abort`; see `end_already_responded_stream`. + this.set_abort_handler(); if this.is_aborted_or_ended() { crate::dispatch::fold(stream.cancel(global_this)); this.response_body_readable_stream_ref @@ -2060,11 +2072,6 @@ where this.render_metadata(); } - // Before `pull()` runs, not in `to_async()`: a `server.stop(true)` inside - // `pull()` has to reach `on_abort`, and once the stream has completed the - // response, uWS `markDone()` must have dropped these for good (the flag - // makes the later `to_async()` a no-op); see `end_already_responded_stream`. - this.set_abort_handler(); resp.on_writable( |this, off, resp| Self::on_writable_response_stream(this, off, resp), this.as_ctx_ptr(), @@ -2630,6 +2637,17 @@ where } } + /// Drops the callback's result as for any aborted request; `set_abort_handler` delivers the missed close. + #[cold] + fn on_connection_closed_during_dispatch(&self, this: &ThisServer, result: JSValue) { + ctx_log!("connection closed during dispatch"); + if let Some(promise) = result.as_any_promise() { + // Subscribing to it would have made a later rejection handled. + promise.set_handled(this.global_this().vm()); + } + self.set_abort_handler(); + } + // Each HTTP request or TCP socket connection is effectively a "task". // // However, unlike the regular task queue, we don't drain the microtask @@ -2658,6 +2676,10 @@ where if ctx.drain_microtasks().is_err() || ctx.is_aborted_or_ended() { return; } + if ctx.resp.get().is_some_and(|resp| resp.is_closed()) { + ctx.on_connection_closed_during_dispatch(this, response_value); + return; + } // if you return a Response object or a Promise // but you upgraded the connection to a WebSocket // just ignore the Response object. It doesn't do anything. @@ -3605,6 +3627,10 @@ where if self.is_aborted_or_ended() || self.did_upgrade_web_socket() { return; } + if self.resp.get().is_some_and(|resp| resp.is_closed()) { + self.on_connection_closed_during_dispatch(server, result); + return; + } if !result.is_empty_or_undefined_or_null() { if let Some(err) = result.to_error() { self.finish_running_error_handler(err, status); diff --git a/src/uws_sys/Response.rs b/src/uws_sys/Response.rs index e70180e69f5b..260ffafcde06 100644 --- a/src/uws_sys/Response.rs +++ b/src/uws_sys/Response.rs @@ -159,6 +159,12 @@ impl Response { self.state().is_http_connection_close() } + /// Valid after the close callback: uSockets frees a closed socket only when the outermost tick ends. + pub(crate) fn is_closed(&self) -> bool { + // Same view as `downcast_socket`: the response handle is the socket. + us_socket_t::opaque_ref(std::ptr::from_ref::(self).cast::()).is_closed() + } + pub(crate) fn prepare_for_sendfile(&mut self) { c::uws_res_prepare_for_sendfile(Self::ssl_flag(), self.as_raw()) } @@ -835,6 +841,11 @@ impl AnyResponse { any_dispatch!(self, |r| r.should_close_connection()) } + /// See `Response::is_closed`; always `false` for HTTP/3 (see `h3::Response::is_closed`). + pub fn is_closed(self) -> bool { + any_dispatch!(self, |r| r.is_closed()) + } + pub fn try_end(self, data: &[u8], total_size: usize, close_connection: bool) -> bool { any_dispatch!(self, |r| r.try_end(data, total_size, close_connection)) } diff --git a/src/uws_sys/h3.rs b/src/uws_sys/h3.rs index 6d86cfd74bec..7cf9119eb722 100644 --- a/src/uws_sys/h3.rs +++ b/src/uws_sys/h3.rs @@ -178,6 +178,10 @@ impl Response { pub(crate) fn should_close_connection(&mut self) -> bool { self.state().is_http_connection_close() } + /// `us_quic_on_close` frees the stream right after its close callback: a live handle is never closed. + pub(crate) fn is_closed(&self) -> bool { + false + } pub(crate) fn is_corked(&self) -> bool { false } diff --git a/test/js/bun/http/serve-pending-promise-abort-leak.test.ts b/test/js/bun/http/serve-pending-promise-abort-leak.test.ts index adcf4e41e6eb..82fd67d25297 100644 --- a/test/js/bun/http/serve-pending-promise-abort-leak.test.ts +++ b/test/js/bun/http/serve-pending-promise-abort-leak.test.ts @@ -607,3 +607,58 @@ test("413 on a chunked upload frees the context while the handler promise stays socket.destroy(); } }); + +// A request subscribes to its connection's close only once its dispatch is +// over (to_async), so a close that lands before that was lost. server.stop(true) +// inside the handler closes the connection right there. A request with its body +// in flight then went async on the closed socket and was parked forever: no +// abort, a pending body read that never settled, pendingRequests stuck at 1, +// and a stop() promise that never resolved. A request without a body rendered +// a 204 into the closed socket instead of aborting. +const stoppedRequests: Array<[string, string, string[]]> = [ + ["a GET", "GET /stopped HTTP/1.1\r\nHost: example.com\r\n\r\n", ["abort http://example.com/stopped example.com"]], + [ + "a POST with its body in flight", + // Declares 1000 bytes and sends 10. + "POST /stopped HTTP/1.1\r\nHost: example.com\r\nContent-Length: 1000\r\n\r\n0123456789", + ["abort http://example.com/stopped example.com", "text rejected: AbortError"], + ], +]; +test.each(stoppedRequests)("server.stop(true) inside the handler of %s aborts it", async (_what, head, expected) => { + const events: string[] = []; + const { promise: reached, resolve: signalReached, reject: failReached } = Promise.withResolvers(); + let stopped: Promise; + using server = Bun.serve({ + port: 0, + idleTimeout: 0, + fetch(req, srv) { + // url and headers are read lazily from inside the listener: an abort + // delivered this way must still see them, like any other abort. + req.signal.addEventListener("abort", () => events.push(`abort ${req.url} ${req.headers.get("host")}`), { + once: true, + }); + if (req.method === "POST") { + req.text().then( + () => events.push("text resolved"), + e => events.push(`text rejected: ${(e as Error).name}`), + ); + } + stopped = srv.stop(true); + signalReached(); + return new Promise(() => {}); + }, + }); + + const client = connect(Number(server.port), "127.0.0.1", () => client.write(head)); + // A reset after the server closed the connection is expected; a failure + // before the handler ran is not. + client.on("error", failReached); + + await reached; + // The abort is delivered as the dispatch finishes; an immediate queued from + // inside it runs after that. + await new Promise(resolve => setImmediate(resolve)); + expect(events).toEqual(expected); + expect(server.pendingRequests).toBe(0); + await stopped!; +});