ByteStream: take the pending buffer action before signal_drained() in on_data

src/runtime/webcore/ByteStream.rs

@@ -305,16 +305,16 @@ impl ByteStream {
         }
 
         if self.buffer_action.get().is_some() {
-            self.signal_drained();
             if let streams::Result::Err(err) = &stream {
                 // Explicit post-reject cleanup; runs after `action.reject`
                 // (`?` would skip it).
                 bun_output::scoped_log!(ByteStream, "ByteStream.onData err  action.reject()");
 
                 let global = self.parent_const().global_this();
-                // R-2: move the action out of the cell *before* calling
-                // `reject` (which resolves a JS promise and may re-enter).
+                // R-2: move the action out of the cell *before* `signal_drained`
+                // and `reject`; both can re-enter and consume the slot.
                 let mut action = self.buffer_action.replace(None).unwrap();
+                self.signal_drained();
                 let res = action.reject(global, err);
 
                 self.buffer.with_mut(|b| {
@@ -330,9 +330,16 @@ impl ByteStream {
                 return res;
             }
 
+            // R-2: the drain signal can re-enter and consume `buffer_action`,
+            // so the paths below re-take it with `let`-`else`.
+            self.signal_drained();
+
             if self.has_received_last_chunk.get() {
                 // `defer { this.buffer_action = null; }` — handled by `replace(None)` below.
-                let mut action = self.buffer_action.replace(None).unwrap();
+                let Some(mut action) = self.buffer_action.replace(None) else {
+                    // Consumed re-entrantly during `signal_drained`.
+                    return Ok(());
+                };
 
                 if self.buffer.get().capacity() == 0 && matches!(stream, streams::Result::Done) {
                     bun_output::scoped_log!(

test/js/web/fetch/fetch.stream.test.ts

@@ -1,7 +1,7 @@
 import { Socket } from "bun";
-import { describe, expect, it } from "bun:test";
+import { describe, expect, it, test } from "bun:test";
 import { createReadStream, readFileSync } from "fs";
-import { gcTick, isWindows, tempDirWithFilesAnon } from "harness";
+import { bunEnv, bunExe, gcTick, isWindows, tempDirWithFilesAnon } from "harness";
 import http from "http";
 import type { AddressInfo } from "net";
 import path, { join } from "path";
@@ -1422,3 +1422,24 @@ describe.concurrent("fetch() with streaming", () => {
     server.kill("SIGTERM");
   });
 });
+
+// ByteStream::on_data used to call signal_drained() before taking the pending
+// buffer action out of its cell; the drain signal can re-enter and consume the
+// action, so the unwrap() that followed panicked and killed the process
+// (seen as a crash when aborting fetches with parked reads on streaming
+// bodies). The race is timing-dependent, so this stress fixture exercises the
+// abort paths and asserts every parked consumer settles with exit code 0.
+test.concurrent("aborting streaming fetches with parked body consumers settles them without crashing", async () => {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), join(import.meta.dir, "fetch-abort-parked-reads-fixture.ts")],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  expect(stderr).toBe("");
+  expect(stdout).toBe("done 12\n");
+  expect(exitCode).toBe(0);
+});