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

src/runtime/server/RequestContext.rs

@@ -650,15 +650,21 @@ 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() {
+            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>) {
@@ -2630,6 +2636,20 @@ where
         }
     }
 
+    /// Drops the handler's result as for any aborted request; `to_async` delivers the missed close.
+    #[cold]
+    fn on_connection_closed_during_dispatch(&self, this: &ThisServer, response_value: JSValue) {
+        ctx_log!("connection closed during dispatch");
+        if let Some(promise) = response_value.as_any_promise() {
+            // Subscribing to it would have made a later rejection handled.
+            promise.set_handled(this.global_this().vm());
+        }
+        match (self.req.get(), self.request_mut()) {
+            (Some(req), Some(request)) => self.to_async(req, request),
+            _ => 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 +2678,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.

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,136 @@
     socket.destroy();
   }
 });
+
+// A request subscribes to its connection's close only once its dispatch is
+// over (to_async). A synchronous wait on a promise runs the event loop from
+// inside the dispatch, and that nested run can dispatch the close itself, with
+// nobody subscribed. Before the fix the close was lost: request.signal never
+// fired, the context stayed pending, and a Promise<Response> settling later
+// rendered into the freed socket. bun:test's .resolves is such a wait, which
+// is how a test that does `await expect(fetch(...)).rejects.toThrow()` from
+// inside a handler's microtask checkpoint hit this.
+function waitSync(promise: Promise<void>) {
+  expect(promise).resolves.toBeUndefined();
+}
+
+// Ends the client side of an open request and returns only once the server
+// has closed its side: the client's 'close' needs the server's FIN, so when
+// the wait returns, the server socket's close has been dispatched.
+function endClientAndWaitForServerClose(client: import("node:net").Socket, clientClosed: Promise<void>) {
+  client.end();
+  waitSync(clientClosed);
+}
+
+const closeWindows: Array<[string, (run: () => void) => void]> = [
+  ["inside the handler", run => run()],
+  ["in the handler's microtask checkpoint", run => queueMicrotask(run)],
+];
+const handlerResults: Array<[string, () => Response | Promise<Response>]> = [
+  ["a pending Promise<Response>", () => new Promise<Response>(() => {})],
+  ["a Response", () => new Response("nobody is listening")],
+];
+const closeDuringDispatchCases = closeWindows.flatMap(([where, closeWhen]) =>
+  handlerResults.map(([what, result]) => [where, what, closeWhen, result] as const),
+);
+
+test.each(closeDuringDispatchCases)(
+  "connection closed %s, handler returns %s: the request is aborted",
+  async (_where, _what, closeWhen, result) => {
+    const aborted: Array<{ url: string; host: string | null }> = [];
+    const { promise: closedDuringDispatch, resolve: signalClosed } = Promise.withResolvers<void>();
+    let client: import("node:net").Socket;
+    let clientClosed: Promise<void>;
+
+    using server = Bun.serve({
+      port: 0,
+      idleTimeout: 0,
+      fetch(req) {
+        // Read lazily from inside the listener: an abort delivered this way
+        // must still see the request's url and headers, like any other abort.
+        req.signal.addEventListener("abort", () => aborted.push({ url: req.url, host: req.headers.get("host") }), {
+          once: true,
+        });
+        closeWhen(() => {
+          endClientAndWaitForServerClose(client, clientClosed);
+          signalClosed();
+        });
+        return result();
+      },
+    });
+
+    client = connect(Number(server.port), "127.0.0.1", () => {
+      client.write("GET /closed-during-dispatch HTTP/1.1\r\nHost: example.com\r\n\r\n");
+    });
+    client.on("error", () => {});
+    clientClosed = new Promise<void>(resolve => client.once("close", () => resolve()));
+
+    await closedDuringDispatch;
+    // That settled from inside the dispatch. The abort is delivered when the
+    // dispatch finishes, so look only once an immediate queued here has run.
+    await new Promise(resolve => setImmediate(resolve));
+    expect(aborted).toEqual([{ url: "http://example.com/closed-during-dispatch", host: "example.com" }]);
+    await stopAndAssertDrained(server);
+  },
+);
+
+// Same window while the request body is still arriving and the handler has a
+// read parked on it: the abort has to reject that read as well (the context
+// is not dead at abort time, so this takes on_abort's other branch).
+test.each(closeWindows)(
+  "connection closed %s while the body is in flight: the request is aborted and the pending read rejects",
+  async (_where, closeWhen) => {
+    const events: string[] = [];
+    const { promise: closedDuringDispatch, resolve: signalClosed } = Promise.withResolvers<void>();
+    const { promise: bodyRead, resolve: signalBodyRead } = Promise.withResolvers<string>();
+    let client: import("node:net").Socket;
+    let clientClosed: Promise<void>;
+
+    using server = Bun.serve({
+      port: 0,
+      idleTimeout: 0,
+      fetch(req) {
+        req.signal.addEventListener("abort", () => events.push("abort"), { once: true });
+        req.text().then(
+          () => signalBodyRead("resolved"),
+          e => signalBodyRead(`rejected: ${(e as Error).name}`),
+        );
+        closeWhen(() => {
+          endClientAndWaitForServerClose(client, clientClosed);
+          signalClosed();
+        });
+        return new Promise<Response>(() => {});
+      },
+    });
+
+    client = connect(Number(server.port), "127.0.0.1", () => {
+      // Declares 1000 bytes and sends 10, so the body stays in flight.
+      client.write("POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 1000\r\n\r\n0123456789");
+    });
+    client.on("error", () => {});
+    clientClosed = new Promise<void>(resolve => client.once("close", () => resolve()));
+
+    await closedDuringDispatch;
+    await new Promise(resolve => setImmediate(resolve));
+    expect(events).toEqual(["abort"]);
+    expect(await bodyRead).toBe("rejected: AbortError");
+    await stopAndAssertDrained(server);
+  },
+);
+
+test("a Promise<Response> that settles after its connection closed during dispatch is a no-op", async () => {
+  // In a subprocess: on an unfixed build the late resolve renders into the
+  // socket uSockets freed at the end of the tick (heap-use-after-free under
+  // ASAN), and the abort never fires.
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), join(import.meta.dir, "serve-close-during-dispatch-late-resolve-fixture.ts")],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  expect(stderr).toBe("");
+  expect(JSON.parse(stdout.trim())).toEqual({ abortCount: 1, pendingAfterResolve: 0 });
+  expect(exitCode).toBe(0);
+}, 30_000);