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

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -1152,12 +1152,7 @@ impl FetchTasklet {
                             // mark to wait until deinit
                             self.is_waiting_abort = self.result.has_more;
                             self.abort_reason.set(&global_object, check_result);
-                            self.signal_store.aborted.store(true, Ordering::Relaxed);
-                            self.tracker.did_cancel(&self.global_this);
-                            // we need to abort the request
-                            if let Some(http_) = self.http.as_mut() {
-                                http::http_thread().schedule_shutdown(http_);
-                            }
+                            self.abort_task();
                             self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID"));
                             return false;
                         }
@@ -1177,11 +1172,7 @@ impl FetchTasklet {
                             let hostname_err_result = global_object.try_take_exception().unwrap();
                             self.is_waiting_abort = self.result.has_more;
                             self.abort_reason.set(&global_object, hostname_err_result);
-                            self.signal_store.aborted.store(true, Ordering::Relaxed);
-                            self.tracker.did_cancel(&self.global_this);
-                            if let Some(http_) = self.http.as_mut() {
-                                http::http_thread().schedule_shutdown(http_);
-                            }
+                            self.abort_task();
                             self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID"));
                             return false;
                         }
@@ -1202,13 +1193,7 @@ impl FetchTasklet {
                         // mark to wait until deinit
                         self.is_waiting_abort = self.result.has_more;
                         self.abort_reason.set(&global_object, check_result);
-                        self.signal_store.aborted.store(true, Ordering::Relaxed);
-                        self.tracker.did_cancel(&self.global_this);
-
-                        // we need to abort the request
-                        if let Some(http_) = self.http.as_mut() {
-                            http::http_thread().schedule_shutdown(http_);
-                        }
+                        self.abort_task();
                         self.result.fail = Some(err!("ERR_TLS_CERT_ALTNAME_INVALID"));
                         return false;
                     }
@@ -1622,6 +1607,9 @@ impl FetchTasklet {
         if this.signal_store.body_receive_mode() == BodyReceiveMode::Ignore {
             return;
         }
+        // reader.cancel() / body.cancel() aborts the fetch so the server sees the
+        // close (Node/Deno/browsers abort unconditionally). abort_task() is idempotent.
+        this.abort_task();
         this.ignore_remaining_response_body(false);
     }
 
@@ -1766,14 +1754,21 @@ 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 is already shutting down; don't re-arm receive/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 +2249,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,17 @@ describe.concurrent("fetch() receive backpressure — buffered consumers are not
 });
 
 describe.concurrent("fetch() receive backpressure — streaming consumer shapes", () => {
-  test("reader.cancel() resumes the socket so the abandoned body drains", async () => {
+  test("reader.cancel() mid-stream lets a subsequent request complete", 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);
+    // reader.cancel() aborts the in-flight request (#33227), closing the
+    // connection; the client must recover so a later request still completes.
+    // The abort-vs-drain behavior itself is asserted in regression/issue/33227.
     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);
   });
 
   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]);
+});