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

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -1622,6 +1622,12 @@
         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 @@
         {
             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/js/web/fetch/fetch.stream.test.ts

@@ -1428,4 +1428,49 @@ describe.concurrent("fetch() with streaming", () => {
     expect(new TextDecoder().decode(result.value!)).toBe("hello\n");
     server.kill("SIGTERM");
   });
+
+  it("reader.cancel() aborts the request and triggers the server's stream cancel (#33227)", 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]);
+  });
 });