Bun.serve: end the request when an HTTP/2 or HTTP/3 stream closes after a streamed response

packages/bun-uws/src/Http3Context.h

@@ -58,8 +58,9 @@ struct Http3Context {
             Http3ResponseData *rd = res->getHttpResponseData();
             /* Fire onAborted for both real aborts and post-completion stream
              * teardown. The handler distinguishes via hasResponded(); for the
-             * completed case it just drops its pointer so it doesn't outlive
-             * this destructor. */
+             * completed case it ends the request here, because this is its
+             * last notification and its pointer must not outlive this
+             * destructor. */
             if (rd->onAborted) {
                 rd->onAborted(res, rd->userData);
             }

src/runtime/server/RequestContext.rs

@@ -1283,27 +1283,19 @@ where
         }
     }
 
-    /// HTTP/1 only: `end_stream()` for a response the JS sink already fully
-    /// ended (`HTTPServerWritable::ended_response`). HTTP/1's uWS `markDone()`
-    /// drops its `onAborted` on end, so nothing nulls `self.resp` if the peer
-    /// closes afterwards: by the time the parked stream-resolution microtask
-    /// runs, uSockets may already have freed the socket
-    /// (`us_internal_free_closed_sockets`) or recycled it onto the next
-    /// keep-alive request. Release the handle without dereferencing it. The
-    /// `clear_on_data()`/`clear_aborted()`/`clear_timeout()` calls
-    /// `detach_response()` would make are already covered by `markDone()`,
-    /// which is also why a later `server.stop(true)` cannot reach `on_abort`:
-    /// callers come here even once the server is terminated.
+    /// `end_stream()` for a response the sink already ended
+    /// (`HTTPServerWritable::ended_response`), where `resp` must no longer be
+    /// dereferenced. `markDone()` already did what `detach_response()` would.
     ///
-    /// HTTP/2 and HTTP/3 must never reach this. `Http{2,3}Response::markDone()`
-    /// deliberately leave `onAborted` armed so the stream teardown can notify
-    /// the holder, which also proves `resp` is still alive here (`on_abort`
-    /// nulls it first). They therefore need `end_stream()`'s
-    /// `detach_response()` to disarm that callback before the context is
-    /// released, or the later stream teardown invokes it on a freed pool slot.
+    /// HTTP/1 arrives from the stream's resolve/reject reaction, by then
+    /// uSockets may have freed or recycled the socket. HTTP/2 and HTTP/3
+    /// arrive from `on_abort`, which is the last call on a freed stream.
+    ///
+    /// An H2/H3 reaction must use `end_stream()` instead: its `resp` is still
+    /// alive, and `detach_response()` has to disarm the still-armed
+    /// `onAborted` before the context goes, or it fires on a freed pool slot.
     pub(crate) fn end_already_responded_stream(&self) {
         ctx_log!("endAlreadyRespondedStream");
-        debug_assert!(!MUX);
         // `resp` may be freed (see above); the sink resumed it at `ended_response = true`.
         self.flags.set_request_body_paused(false);
         if self.resp.take().is_some() {
@@ -1461,20 +1453,29 @@ where
         let pinned = RequestContextRef::pin(this);
         let this = pinned.ctx();
         debug_assert!(this.resp.get().is_some());
-        // An HTTP/2 or HTTP/3 stream is destroyed once both sides finish,
-        // so this also fires after a successful end(). HTTP/1 sockets persist
-        // for keep-alive, so the equivalent never happens there. Drop the
-        // pointer; everything else cleans up via the resolve/reject path.
+        debug_assert!(this.server.get().is_some());
+        let server = this.server();
+        let vm = server.vm();
+        // Both arms below can reject a parked request-body read and drain.
+        // The held count keeps that drain from checkpointing mid-frame.
+        let _entered = vm.enter_event_loop_scope_without_checkpoint();
+        // An H2/H3 stream is destroyed once both sides finish, so this also
+        // fires after a successful end(); HTTP/1 keep-alive sockets never do.
+        // Only the sink ends a response without `detach_response()`, and its
+        // reaction does nothing once `resp` is gone, so end the request here.
         if MUX {
             // SAFETY: FFI handle
             if resp.has_responded() {
-                this.resp.set(None);
-                this.flags.set_has_abort_handler(false);
+                // The sink can outlive this call; `resp` dies with the stream.
+                if let Some(wrapper) = this.sink_mut() {
+                    wrapper.sink.res = None;
+                }
+                // Releases the base ref itself, so no `_ref` adoption here.
+                this.end_already_responded_stream();
                 return;
             }
         }
         debug_assert!(!this.flags.aborted());
-        debug_assert!(this.server.get().is_some());
         // mark request as aborted
         this.flags.set_aborted(true);
         let abort = this.additional_on_abort.replace(None);
@@ -1485,12 +1486,7 @@ where
 
         this.detach_response();
         let any_js_calls = core::cell::Cell::new(false);
-        let server = this.server();
-        let vm = server.vm();
         let global_this = server.global_this();
-        // Entered for the abort listeners below, and (dropped last) for the
-        // drains below and in the release of `_ref`.
-        let _entered = vm.enter_event_loop_scope_without_checkpoint();
         let _ref = RequestContextRef::adopt(this.as_ctx_ptr());
         // This is a task in the event loop.
         // If we called into JavaScript, we must drain the microtask queue.

src/runtime/webcore/streams.rs

@@ -1170,15 +1170,13 @@ pub struct HTTPServerWritable<const SSL: bool> {
     pub(crate) source_pending_pull: bool,
     pub(crate) end_len: usize,
     /// This sink fully ended the uWS response (`res.end()` / a completed
-    /// `res.try_end()`). On HTTP/1 uWS `markDone()` drops `onAborted` at that
-    /// point, so the owning `RequestContext` is never told if the peer closes
-    /// afterwards and its `resp` must not be dereferenced again: by the time
-    /// the parked stream-resolution microtask runs, uSockets may already have
-    /// freed the socket (`us_internal_free_closed_sockets`) or recycled it
-    /// onto the next keep-alive request. `handle_resolve_stream` /
-    /// `handle_reject_stream` consult this instead of reading the response's
-    /// state. HTTP/1 only; see `end_already_responded_stream` for why
-    /// `Http3Response::markDone()` makes the H3 `resp` still safe to use.
+    /// `res.try_end()`). On HTTP/1 that drops `onAborted`, so the owning
+    /// `RequestContext` is never told the peer went away and must not
+    /// dereference `resp` again: uSockets may have freed or recycled the
+    /// socket by the time the stream reaction runs. `handle_resolve_stream` /
+    /// `handle_reject_stream` read this instead of the response state, and
+    /// act on it only for HTTP/1. An H2/H3 `resp` is still alive there, and
+    /// `on_abort` ends those requests when the transport frees the stream.
     pub(crate) ended_response: bool,
 
     pub(crate) on_first_write: Option<fn(Option<*mut c_void>)>,

test/js/bun/http/serve-http2.test.ts

@@ -855,4 +855,120 @@ describe("Bun.serve http2 in-process", () => {
     expect(aborted).toBe(1);
     await new Promise<void>(r => session.close(() => r()));
   });
+
+  // A 4-byte stream window stalls the body after its first bytes. One
+  // WINDOW_UPDATE then lets the writable callback send the rest and end the
+  // response, and the same socket event frees the finished stream, before the
+  // promise reaction of the body stream can run: the free has to end the request.
+  test("a streamed response body that ends from the writable callback ends its request", async () => {
+    await using server = Bun.serve({
+      port: 0,
+      http2: true,
+      fetch: () =>
+        new Response(
+          new ReadableStream({
+            async pull(controller) {
+              controller.enqueue(new TextEncoder().encode("streamed-past-the-window"));
+              controller.close();
+            },
+          }),
+        ),
+    });
+
+    const SETTINGS_INITIAL_WINDOW_SIZE = 0x4;
+    const settings = Buffer.alloc(6);
+    settings.writeUInt16BE(SETTINGS_INITIAL_WINDOW_SIZE, 0);
+    settings.writeUInt32BE(4, 2);
+    const raw = await RawH2.connect(server.port, false, { settings });
+    raw.headers(1, baseHeaders("/"));
+    const first = await raw.waitFor(f => f.type === T.DATA && f.streamId === 1);
+    const increment = Buffer.alloc(4);
+    increment.writeUInt32BE(1024);
+    raw.write(frame(T.WINDOW_UPDATE, 0, 1, increment));
+    const body = await raw.body(1);
+
+    expect({ first: first.payload.toString(), body: body.toString(), pending: server.pendingRequests }).toEqual({
+      first: "stre",
+      body: "streamed-past-the-window",
+      pending: 0,
+    });
+    raw.close();
+    // A request that never ends keeps a graceful stop pending.
+    await server.stop();
+  });
+
+  // The stream close is the only notification left when the pump promise never
+  // settles, so this case cannot be handled in its resolve reaction. The end
+  // has to come from a later microtask: an end inside the first pull() leaves
+  // the response already finished when the stream is attached, which takes a
+  // different path. HTTP/1 has the same shape and still leaks there, because
+  // its socket is not freed per request.
+  test("a direct stream whose pull() never settles after it ended the response", async () => {
+    await using server = Bun.serve({
+      port: 0,
+      http2: true,
+      fetch: () =>
+        new Response(
+          new ReadableStream({
+            type: "direct",
+            async pull(controller) {
+              controller.write("streamed");
+              await Promise.resolve();
+              controller.end();
+              await new Promise<never>(() => {});
+            },
+          }),
+        ),
+    });
+
+    const session = await connectH2(server.port, false);
+    const bodies: string[] = [];
+    for (let i = 0; i < 3; i++) bodies.push((await request(session, { ":path": "/" })).body.toString());
+
+    expect({ bodies, pendingRequests: server.pendingRequests }).toEqual({
+      bodies: ["streamed", "streamed", "streamed"],
+      pendingRequests: 0,
+    });
+    await new Promise<void>(r => session.close(() => r()));
+    await server.stop();
+  });
+
+  // stop(true) inside pull() closes the socket at once, so the stream teardown
+  // reports while the response is still being attached. The request then ends
+  // from inside that attach, and the sink it leaves behind is cleaned up by the
+  // caller. A late write on the controller must still fail cleanly.
+  test("stop(true) inside pull() after the response ended", async () => {
+    let lateWriteFailed = false;
+    await using server = Bun.serve({
+      port: 0,
+      http2: true,
+      fetch: () =>
+        new Response(
+          new ReadableStream({
+            type: "direct",
+            async pull(controller) {
+              controller.write("streamed");
+              controller.end();
+              server.stop(true);
+              try {
+                controller.write("late");
+                await controller.flush();
+              } catch {
+                lateWriteFailed = true;
+              }
+            },
+          }),
+        ),
+    });
+
+    const session = await connectH2(server.port, false);
+    // The abrupt stop races the response, so the client may or may not see it.
+    await request(session, { ":path": "/" }).catch(() => {});
+    session.destroy();
+
+    expect({ lateWriteFailed, pendingRequests: server.pendingRequests }).toEqual({
+      lateWriteFailed: true,
+      pendingRequests: 0,
+    });
+  });
 });

test/js/bun/http/serve-http3.test.ts

@@ -1490,6 +1490,88 @@ describe("Bun.serve HTTP/3 lifecycle", () => {
     expect(exitCode).toBe(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
+  // callback. The same tick closes the finished stream, before the promise
+  // reaction of the body stream can run: the close has to end the request.
+  // Each case has its own server, so its request is the first on a connection.
+  describe("a streamed response body ends its request when the stream closes", () => {
+    const bodies: Record<string, () => Bun.BodyInit> = {
+      "ReadableStream": () =>
+        new ReadableStream({
+          async pull(controller) {
+            controller.enqueue(new TextEncoder().encode("streamed"));
+            controller.close();
+          },
+        }),
+      "direct ReadableStream": () =>
+        new ReadableStream({
+          type: "direct",
+          async pull(controller) {
+            controller.write("streamed");
+            await controller.end();
+          },
+        }),
+      "async generator": () =>
+        (async function* () {
+          yield "streamed";
+        })(),
+    };
+
+    const requests: Record<string, RequestInit> = {
+      "GET": {},
+      "POST with a request body the handler never reads": { method: "POST", body: "request-content" },
+    };
+
+    describe.each(Object.keys(bodies))("%s", kind => {
+      test.each(Object.keys(requests))("%s", async request => {
+        await using server = Bun.serve({
+          port: 0,
+          tls,
+          http3: true,
+          http1: false,
+          fetch: () => new Response(bodies[kind]()),
+        });
+
+        const text = await fetchH3(server.port, "/", requests[request]).then(res => res.text());
+        expect({ text, pendingRequests: server.pendingRequests }).toEqual({ text: "streamed", pendingRequests: 0 });
+        // A request that never ends keeps a graceful stop pending.
+        await server.stop();
+      });
+    });
+
+    // The stream close is the only notification left when the pump promise
+    // never settles, so this case cannot be handled in its resolve reaction.
+    // The end comes from a later microtask: an end inside the first pull()
+    // leaves the response already finished when the stream is attached, which
+    // takes a different path.
+    test("a direct stream whose pull() never settles after it ended the response", async () => {
+      await using server = Bun.serve({
+        port: 0,
+        tls,
+        http3: true,
+        http1: false,
+        fetch: () =>
+          new Response(
+            new ReadableStream({
+              type: "direct",
+              async pull(controller) {
+                controller.write("streamed");
+                await Promise.resolve();
+                controller.end();
+                await new Promise<never>(() => {});
+              },
+            }),
+          ),
+      });
+
+      const text = await fetchH3(server.port, "/").then(res => res.text());
+      expect({ text, pendingRequests: server.pendingRequests }).toEqual({ text: "streamed", pendingRequests: 0 });
+      await server.stop();
+    });
+  });
+
   // C: req.signal fires when the client resets the H3 stream mid-request.
   test("req.signal aborts on client RST", async () => {
     const script = `