Bun.serve: abort a request whose connection closed while its handler was being dispatched

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<Response>
         // 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);

src/uws_sys/Response.rs

@@ -159,6 +159,12 @@ impl<const SSL: bool> Response<SSL> {
         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>(self).cast::<us_socket_t>()).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))
     }

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
     }

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<void>();
+  let stopped: Promise<void>;
+  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<Response>(() => {});
+    },
+  });
+
+  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!;
+});