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

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -1622,6 +1622,12 @@ 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.
+        if this.result.has_more {
+            this.abort_task();
+        }
         this.ignore_remaining_response_body(false);
     }
 
@@ -1772,8 +1778,14 @@ impl FetchTasklet {
         {
             self.schedule_receive_resume();
         }
+        let aborted = self.signal_store.aborted.load(Ordering::Relaxed);
         if let Some(http_) = self.http.as_mut() {
-            http_.enable_response_body_streaming();
+            // An aborted fetch (reader.cancel() on an in-flight body, or an
+            // AbortSignal) is already shutting the connection down; draining
+            // would read the rest of an unbounded body and hold the socket open.
+            if !aborted {
+                http_.enable_response_body_streaming();
+            }
         }
         // we should not keep the process alive if we are ignoring the body
         let _ = self.javascript_vm;

test/regression/issue/33227.test.ts

@@ -0,0 +1,47 @@
+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 { promise: holdOpen } = 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({
+          async pull(controller) {
+            if (count < 8) {
+              controller.enqueue(encoder.encode(`data: ${count++}\n\n`));
+            } else {
+              // Keep the body in-flight (never closes) so the client is
+              // cancelling a response that is still arriving.
+              await holdOpen;
+            }
+          },
+          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]);
+});