fetch: hold FetchTasklet mutex through deref_from_thread so HTTP thread is never the final deref

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -391,7 +391,15 @@ impl FetchTasklet {
         if !unsafe { bun_ptr::ThreadSafeRefCount::<Self>::release(this) } {
             return;
         }
+        // The 1→0 transition should only be reachable at shutdown. The
+        // `callback` paths that call this hold the tasklet mutex, which
+        // blocks on_progress_update from releasing the JS-side initial ref,
+        // so at least one other ref is always live there.
         let self_ = Self::from_raw_ref(this);
+        debug_assert!(
+            self_.javascript_vm.is_shutting_down(),
+            "FetchTasklet::deref_from_thread reached 1->0 outside shutdown",
+        );
         if self_.javascript_vm.is_shutting_down() {
             // SAFETY: last ref; exclusive access. `deinit()` would run
             // `clear_data()` + `Drop` for the JSC `Strong`/`Weak` fields, which
@@ -401,10 +409,9 @@ impl FetchTasklet {
             unsafe { FetchTasklet::dealloc_for_shutdown(this) };
             return;
         }
-        // this is really unlikely to happen, but can happen
-        // lets make sure that we always call deinit from main thread
-        // `from_callback` heap-allocates a fresh `ConcurrentTaskItem`; the queue
-        // takes ownership of it.
+        // Defensive fallback for release builds; unreachable per the
+        // is_shutting_down() assert above. Bounce deinit to the JS thread
+        // via a fresh heap `ConcurrentTaskItem` that the queue owns.
         Self::enqueue_concurrent(
             self_.javascript_vm,
             ConcurrentTask::from_callback(this, FetchTasklet::deinit_callback),
@@ -2182,8 +2189,10 @@ impl FetchTasklet {
         let task_ref = Self::from_raw_mut(task);
 
         task_ref.mutex.lock();
-        // we need to unlock before task.deref();
-        // explicit unlock + deref at end instead of nested defers.
+        // The mutex stays held through deref_from_thread at every exit so the
+        // HTTP-side deref is never the 1→0 transition (on_progress_update
+        // needs this mutex to release the JS-side initial ref).
+        //
         // Sync HTTP-thread state back into the JS-side instance via an
         // explicit field-subset copy (`AsyncHTTP` is not `Copy`:
         // `HTTPClient: Drop`, owned Vecs); see `AsyncHTTP::sync_progress_from`
@@ -2261,11 +2270,8 @@ impl FetchTasklet {
             }
             if success && task_ref.result.has_more {
                 // we are ignoring the body so we should not receive more data, so will only signal when result.has_more = true
+                // `has_more` is true here so `is_done` is always false; unlock only.
                 task_ref.mutex.unlock();
-                if is_done {
-                    // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held.
-                    FetchTasklet::deref_from_thread(task);
-                }
                 return;
             }
         } else {
@@ -2287,11 +2293,16 @@ impl FetchTasklet {
             Ordering::Relaxed,
         ) {
             if has_schedule_callback {
-                task_ref.mutex.unlock();
+                // Deref while still holding the mutex. on_progress_update
+                // (the only releaser of the JS-side initial ref) needs this
+                // mutex, so the initial ref is still held here and this
+                // deref is never the 1→0 transition.
                 if is_done {
                     // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held.
                     FetchTasklet::deref_from_thread(task);
                 }
+                // SAFETY: `task` is still live (initial ref still held).
+                Self::from_raw_ref(task).mutex.unlock();
                 return;
             }
         }
@@ -2337,13 +2348,17 @@ impl FetchTasklet {
         // queue takes ownership of its `next` link.
         Self::enqueue_concurrent(task_ref.javascript_vm, ct);
 
-        task_ref.mutex.unlock();
-        // we are done with the http client so we can deref our side
-        // this is a atomic operation and will enqueue a task to deinit on the main thread
+        // Deref while still holding the mutex. on_progress_update (the only
+        // releaser of the JS-side initial ref) needs this mutex, so the
+        // initial ref is still held here and this deref is never the 1→0
+        // transition — deref_from_thread therefore never schedules
+        // deinit_callback from this path.
         if is_done {
             // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held.
             FetchTasklet::deref_from_thread(task);
         }
+        // SAFETY: `task` is still live (initial ref still held).
+        Self::from_raw_ref(task).mutex.unlock();
     }
 }
 

test/js/web/fetch/fetch-tasklet-deref-race.test.ts

@@ -0,0 +1,33 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+import path from "node:path";
+
+// The HTTP thread's result callback must hold the tasklet mutex through its
+// deref_from_thread call so it is never the 1->0 transition. When it was
+// not, the deref could schedule a deinit_callback task that later observed
+// a nonzero refcount and panicked with
+// "assertion failed: self.raw_count.load(Ordering::SeqCst) == 0".
+//
+// The race window is a handful of instructions between mutex.unlock() and
+// deref_from_thread() on the HTTP thread, so this test is best-effort: it
+// exercises many concurrent fetch + abort cycles under load and asserts
+// the process completes. It does not deterministically reproduce the crash
+// on an unfixed build; a debug_assert in deref_from_thread documents the
+// invariant the mutex ordering enforces.
+test("FetchTasklet HTTP-thread deref is never the final ref", async () => {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), path.join(import.meta.dir, "fetch-tasklet-deref-race-fixture.ts")],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  // Include stderr in the failure message for diagnostics without asserting
+  // it is exactly empty (debug/ASAN builds may emit benign warnings).
+  expect({ stdout: stdout.trim(), exitCode, stderr }).toMatchObject({
+    stdout: "ok",
+    exitCode: 0,
+  });
+});