fetch: abort the request when the response body reader is cancelled

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -1622,6 +1622,17 @@ impl FetchTasklet {
         if this.signal_store.body_receive_mode() == BodyReceiveMode::Ignore {
             return;
         }
+        // reader.cancel() on a still-arriving body aborts the fetch so the
+        // connection closes and the server observes the cancellation. A fully
+        // received body is left to drain/cleanup so the socket stays reusable.
+        // `result` is replaced wholesale by the HTTP thread under `mutex`, so
+        // read `has_more` under the lock like the other JS-thread callbacks.
+        this.mutex.lock();
+        let has_more = this.result.has_more;
+        this.mutex.unlock();
+        if has_more {
+            this.abort_task();
+        }
         this.ignore_remaining_response_body(false);
     }
 
@@ -1766,14 +1777,22 @@ impl FetchTasklet {
         }
         // enabling streaming will make the http thread to drain into the main thread (aka stop buffering)
         // without a stream ref, response body or response instance alive it will just ignore the result
+        // An aborted fetch (reader.cancel() on an in-flight body, or an
+        // AbortSignal) is already shutting the connection down; don't re-arm the
+        // receive or resume draining, which would read the rest of an unbounded
+        // body and hold the socket open (drain_events resumes before shutdowns).
+        let aborted = self.signal_store.aborted.load(Ordering::Relaxed);
         if self
             .signal_store
             .set_receive_mode_terminal(BodyReceiveMode::Ignore)
+            && !aborted
         {
             self.schedule_receive_resume();
         }
         if let Some(http_) = self.http.as_mut() {
-            http_.enable_response_body_streaming();
+            if !aborted {
+                http_.enable_response_body_streaming();
+            }
         }
         // we should not keep the process alive if we are ignoring the body
         let _ = self.javascript_vm;
@@ -2254,7 +2273,12 @@ impl FetchTasklet {
     }
 
     pub(crate) fn abort_task(&mut self) {
-        self.signal_store.aborted.store(true, Ordering::Relaxed);
+        // Idempotent: reader.cancel() and an AbortSignal can both reach here for
+        // the same fetch. Only the first abort enqueues a shutdown; a second
+        // would append a redundant ShutdownMessage for an already-closing socket.
+        if self.signal_store.aborted.swap(true, Ordering::Relaxed) {
+            return;
+        }
         self.tracker.did_cancel(&self.global_this);
 
         if let Some(http_) = self.http.as_mut() {

test/js/web/fetch/fetch-backpressure.test.ts

@@ -300,21 +300,21 @@
 });
 
 describe.concurrent("fetch() receive backpressure — streaming consumer shapes", () => {
-  test("reader.cancel() resumes the socket so the abandoned body drains", async () => {
+  test("reader.cancel() aborts the in-flight request instead of draining the abandoned body", async () => {
     await using server = await serve("h1");
     const r1 = await fetch(server.url, { keepalive: true });
     const reader = r1.body!.getReader();
     await reader.read();
     await Bun.sleep(50);
     await reader.cancel();
-    // The cancelled body must drain before the connection is poolable;
-    // poll server.sent() instead of asserting a connection count.
-    while (server.sent() < TOTAL) await Bun.sleep(5);
+    // Cancelling an in-flight body aborts the request (matching Node/Deno and
+    // browsers, see #33227): the connection closes, so the server never sends
+    // the whole body. A subsequent request still succeeds on a fresh connection.
     const buf = await (await fetch(server.url, { keepalive: true })).arrayBuffer();
-    expect({ byteLength: buf.byteLength, sent: server.sent() }).toEqual({
-      byteLength: TOTAL,
-      sent: 2 * TOTAL,
-    });
+    expect(buf.byteLength).toBe(TOTAL);
+    // Drain-for-keepalive would push the first body to completion too, reaching
+    // 2 * TOTAL; the abort leaves the first response only partially sent.
+    expect(server.sent()).toBeLessThan(2 * TOTAL);
   });
 
   test("res.body.tee() both branches drain", async () => {

test/regression/issue/33227.test.ts

@@ -0,0 +1,45 @@
+import { sleep } from "bun";
+import { expect, test } from "bun:test";
+
+// https://github.com/oven-sh/bun/issues/33227
+test("reader.cancel() aborts the fetch and triggers the server's stream cancel", async () => {
+  const { promise: sawAbort, resolve: onAbort } = Promise.withResolvers<void>();
+  const { promise: sawCancel, resolve: onCancel } = Promise.withResolvers<void>();
+  const encoder = new TextEncoder();
+
+  using server = Bun.serve({
+    port: 0,
+    fetch(request) {
+      request.signal.addEventListener("abort", () => onAbort());
+      let count = 0;
+      return new Response(
+        new ReadableStream({
+          // Stream forever (paced) so the body is always in flight: the unfixed
+          // drain path keeps the connection open and actively reading, so the
+          // server never sees a close unless the cancel actually aborts.
+          async pull(controller) {
+            controller.enqueue(encoder.encode(`data: ${count++}\n\n`));
+            await sleep(10);
+          },
+          cancel() {
+            onCancel();
+          },
+        }),
+        { headers: { "Content-Type": "text/event-stream" } },
+      );
+    },
+  });
+
+  const res = await fetch(`http://localhost:${server.port}`, { method: "POST" });
+  const reader = res.body!.getReader();
+
+  const first = await reader.read();
+  expect(first.done).toBe(false);
+  expect(first.value!.byteLength).toBeGreaterThan(0);
+
+  await reader.cancel();
+
+  // Cancelling the reader must abort the fetch and close the connection, so the
+  // server observes request.signal's abort and runs the body stream's cancel().
+  await Promise.all([sawAbort, sawCancel]);
+});