fetch: fence HTTP-thread FetchTasklet callbacks against worker VM dealloc

src/jsc/VirtualMachine.rs

@@ -69,6 +69,81 @@ pub type ExceptionList = Vec<crate::schema_api::JsException>;
 // VirtualMachine struct (file-level @This())
 // ──────────────────────────────────────────────────────────────────────────
 
+/// Arc'd per-VM shutdown flag + reader fence for threads that may outlive the
+/// VM. A worker's [`WebWorker::shutdown`] raw-`dealloc`s the `VirtualMachine`
+/// while the shared HTTP client thread can still be delivering `FetchTasklet`
+/// callbacks that dereference it; those callbacks hold a clone of this Arc
+/// and bracket every VM access with [`Self::try_begin_vm_read`] /
+/// [`Self::end_vm_read`] so shutdown can spin the readers out first.
+///
+/// [`WebWorker::shutdown`]: crate::web_worker::WebWorker::shutdown
+pub struct CrossThreadShutdownSignal {
+    shutting_down: core::sync::atomic::AtomicBool,
+    readers: core::sync::atomic::AtomicUsize,
+    is_main_thread: bool,
+}
+
+impl CrossThreadShutdownSignal {
+    fn new(is_main_thread: bool) -> std::sync::Arc<Self> {
+        std::sync::Arc::new(Self {
+            shutting_down: core::sync::atomic::AtomicBool::new(false),
+            readers: core::sync::atomic::AtomicUsize::new(0),
+            is_main_thread,
+        })
+    }
+
+    #[inline]
+    pub fn is_shutting_down(&self) -> bool {
+        self.shutting_down
+            .load(core::sync::atomic::Ordering::Acquire)
+    }
+
+    #[inline]
+    pub fn is_main_thread(&self) -> bool {
+        self.is_main_thread
+    }
+
+    pub fn mark_shutting_down(&self) {
+        self.shutting_down
+            .store(true, core::sync::atomic::Ordering::SeqCst);
+    }
+
+    pub fn wait_for_readers(&self) {
+        debug_assert!(self.is_shutting_down());
+        while self.readers.load(core::sync::atomic::Ordering::SeqCst) > 0 {
+            std::hint::spin_loop();
+        }
+    }
+
+    /// On `true` the caller may dereference the owning VM until the paired
+    /// [`Self::end_vm_read`]; on `false` the VM is (or is about to be) freed
+    /// and no `end_vm_read` is owed. SeqCst on all four ops (this increment +
+    /// flag load, and the writer's flag store + reader-count load) is the
+    /// Dekker-style fence that keeps "reader saw `false`" ordered before
+    /// "writer saw `readers == 0`".
+    #[inline]
+    #[must_use]
+    pub fn try_begin_vm_read(&self) -> bool {
+        self.readers
+            .fetch_add(1, core::sync::atomic::Ordering::SeqCst);
+        if self
+            .shutting_down
+            .load(core::sync::atomic::Ordering::SeqCst)
+        {
+            self.readers
+                .fetch_sub(1, core::sync::atomic::Ordering::SeqCst);
+            return false;
+        }
+        true
+    }
+
+    #[inline]
+    pub fn end_vm_read(&self) {
+        self.readers
+            .fetch_sub(1, core::sync::atomic::Ordering::SeqCst);
+    }
+}
+
 #[derive(Default)]
 pub struct EntryPointResult {
     pub value: crate::strong::Optional, // jsc.Strong.Optional
@@ -207,6 +282,9 @@ pub struct VirtualMachine {
     pub(crate) hide_bun_stackframes: bool,
 
     pub is_shutting_down: bool,
+    /// See [`CrossThreadShutdownSignal`]. `Option` only so `destroy()` can
+    /// release the strong ref (the box is raw-`dealloc`'d, no field `Drop`s).
+    pub cross_thread_shutdown: Option<std::sync::Arc<CrossThreadShutdownSignal>>,
     /// Set once `on_exit()` has finished draining `RareData::cleanup_hooks`.
     /// After this point the cleanup-hook list is never iterated again, so
     /// pushing to it (e.g. from a deferred N-API finalizer scheduled during
@@ -979,6 +1057,13 @@ impl VirtualMachine {
         self.is_shutting_down
     }
 
+    #[inline]
+    pub fn cross_thread_shutdown(&self) -> &std::sync::Arc<CrossThreadShutdownSignal> {
+        self.cross_thread_shutdown
+            .as_ref()
+            .expect("cross_thread_shutdown is Some from init() to destroy()")
+    }
+
     pub fn has_run_cleanup_hooks(&self) -> bool {
         self.has_run_cleanup_hooks
     }
@@ -1488,6 +1573,7 @@ impl VirtualMachine {
         }
 
         self.is_shutting_down = true;
+        self.cross_thread_shutdown().mark_shutting_down();
 
         // Make sure we run new cleanup hooks introduced by running cleanup
         // hooks.
@@ -2110,6 +2196,8 @@ impl VirtualMachine {
             addr_of_mut!((*vm).resolved_path_dups).write(Vec::new());
             addr_of_mut!((*vm).macros).write(Default::default());
             addr_of_mut!((*vm).macro_entry_points).write(Default::default());
+            addr_of_mut!((*vm).cross_thread_shutdown)
+                .write(Some(CrossThreadShutdownSignal::new(opts.is_main_thread)));
             addr_of_mut!((*vm).auto_killer).write(Default::default());
             addr_of_mut!((*vm).commonjs_custom_extensions).write(Default::default());
             addr_of_mut!((*vm).entry_point).write(Default::default());
@@ -4463,6 +4551,8 @@ impl VirtualMachine {
         // proxy strings; `ProxyEnvStorage: Default` so take()+drop suffices.
         drop(core::mem::take(&mut self.proxy_env_storage));
 
+        drop(self.cross_thread_shutdown.take());
+
         // The VM box is `dealloc`'d raw by the worker (see `web_worker.rs`
         // section 5) so field `Drop`s never run; reclaim the boxed
         // `ModuleLoader` payloads explicitly. `eval_source.contents` may be

src/jsc/web_worker.rs

@@ -1248,6 +1248,7 @@ impl WebWorker {
             // re-sets it for the JSC VM teardown.
             vm.jsc_vm().clear_has_termination_request();
             vm.is_shutting_down = true;
+            vm.cross_thread_shutdown().mark_shutting_down();
             vm.on_exit();
             if let Some(hooks) = runtime_hooks() {
                 (hooks.cron_clear_all_teardown)(vm);
@@ -1302,6 +1303,10 @@ impl WebWorker {
             // or observes m_isShuttingDown under m_lock and drops. Idempotent;
             // teardownJSCVM sets it again.
             Bun__JSCTaskScheduler__markShuttingDown(vm.global());
+            // Per-worker fence for HTTP-thread FetchTasklet callbacks (the
+            // main-thread equivalent is `bun_http::shutdown_for_exit()`, a
+            // process-global one-shot). See `CrossThreadShutdownSignal`.
+            vm.cross_thread_shutdown().wait_for_readers();
             // Reclaim queued CppTasks (the per-worker stdio/messaging
             // MessagePort drain tasks that can be in self.tasks mid-tick when
             // terminate() lands, and any Worker dispatchExit close task from a

src/runtime/webcore/fetch/FetchTasklet.rs

@@ -17,7 +17,7 @@ use bun_http::{
 };
 use bun_io::KeepAlive;
 use bun_jsc::debugger::AsyncTaskTracker;
-use bun_jsc::virtual_machine::VirtualMachine;
+use bun_jsc::virtual_machine::{CrossThreadShutdownSignal, VirtualMachine};
 use bun_jsc::{
     self as jsc, GlobalRef, JSGlobalObject, JSValue, JsResult, StringJsc, StrongOptional,
 };
@@ -68,6 +68,9 @@ pub struct FetchTasklet {
     pub(crate) result: HTTPClientResult<'static>,
     pub(crate) metadata: Option<HTTPResponseMetadata>,
     pub(crate) javascript_vm: &'static VirtualMachine,
+    /// `javascript_vm` dangles once a worker VM is `dealloc`'d; HTTP-thread
+    /// callbacks fence every VM deref through this (see the struct doc).
+    pub(crate) vm_shutdown_signal: std::sync::Arc<CrossThreadShutdownSignal>,
     pub global_this: GlobalRef,
     pub(crate) request_body: HTTPRequestBody,
     // ThreadSafeStreamBuffer is intrusively refcounted (`ref_count: AtomicU32`,
@@ -397,7 +400,11 @@ impl FetchTasklet {
             return;
         }
         let self_ = Self::from_raw_ref(this);
-        if self_.javascript_vm.is_shutting_down() {
+        // The enqueued `deinit_callback` may free `this` before we return, so
+        // snapshot what's needed past the enqueue into locals now.
+        let signal = std::sync::Arc::clone(&self_.vm_shutdown_signal);
+        let vm = self_.javascript_vm;
+        if !signal.try_begin_vm_read() {
             // SAFETY: last ref; exclusive access. `deinit()` would run
             // `clear_data()` + `Drop` for the JSC `Strong`/`Weak` fields, which
             // reach into the VM's StrongRootBlock list / WeakSet from this
@@ -411,9 +418,10 @@ impl FetchTasklet {
         // `from_callback` heap-allocates a fresh `ConcurrentTaskItem`; the queue
         // takes ownership of it.
         Self::enqueue_concurrent(
-            self_.javascript_vm,
+            vm,
             ConcurrentTask::from_callback(this, FetchTasklet::deinit_callback),
         );
+        signal.end_vm_read();
     }
 
     // ConcurrentTask::from_callback takes `fn(*mut T) -> bun_event_loop::JsResult<()>`
@@ -518,19 +526,26 @@ impl FetchTasklet {
     ///     (`on_response_finalize`) registered against `this`, so freeing the
     ///     box before `destructOnExit` sweeps the Response is a UAF.
     ///
-    /// Park the intact box on the JS thread via
+    /// Main-thread VM: park the intact box on the JS thread via
     /// `bun_http::defer_shutdown_reclaim`; the drain runs from
     /// `global_exit()` after the HTTP thread has parked but before
     /// `destructOnExit`, so `deinit()` there can release every handle on the
     /// right thread and the Weak is cleared before its referent is finalized.
     ///
+    /// Worker VM: no such drain exists and the JSC heap is (about to be)
+    /// freed, so a parked `deinit()` would dereference dead handles. Leak the
+    /// box; the large buffers were already released by the caller.
+    ///
     /// SAFETY: `this` must be the last reference (ref_count == 0) and have
     /// been allocated via heap::alloc.
     unsafe fn dealloc_for_shutdown(this: *mut FetchTasklet) {
         bun_output::scoped_log!(FetchTasklet, "deallocForShutdown");
         // SAFETY: caller contract — `this` is live with ref_count == 0.
         unsafe { (*this).ref_count.assert_no_refs() };
-        http::defer_shutdown_reclaim(this.cast(), FetchTasklet::deinit_erased);
+        // SAFETY: caller contract — `this` is live with ref_count == 0.
+        if unsafe { (*this).vm_shutdown_signal.is_main_thread() } {
+            http::defer_shutdown_reclaim(this.cast(), FetchTasklet::deinit_erased);
+        }
     }
 
     unsafe fn deinit_erased(this: *mut c_void) {
@@ -2027,6 +2042,7 @@ impl FetchTasklet {
             result: HTTPClientResult::default(),
             metadata: None,
             javascript_vm: jsc_vm,
+            vm_shutdown_signal: std::sync::Arc::clone(jsc_vm.cross_thread_shutdown()),
             global_this: GlobalRef::from(global_this),
             request_body: fetch_options.body,
             request_body_streaming_buffer: None,
@@ -2271,7 +2287,7 @@ impl FetchTasklet {
     /// This is ALWAYS called from the http thread and we cannot touch the buffer here because is locked
     fn on_write_request_data_drain(this: *mut FetchTasklet) {
         let this_ref = Self::from_raw_ref(this);
-        if this_ref.javascript_vm.is_shutting_down() {
+        if !this_ref.vm_shutdown_signal.try_begin_vm_read() {
             return;
         }
         // ref until the main thread callback is called
@@ -2282,6 +2298,7 @@ impl FetchTasklet {
             this_ref.javascript_vm,
             ConcurrentTask::from_callback(this, FetchTasklet::resume_request_data_stream),
         );
+        this_ref.vm_shutdown_signal.end_vm_read();
     }
 
     /// This is ALWAYS called from the main thread
@@ -2649,7 +2666,7 @@ impl FetchTasklet {
             }
         }
         // will deinit when done with the http client (when is_done = true)
-        if task_ref.javascript_vm.is_shutting_down() {
+        if !task_ref.vm_shutdown_signal.try_begin_vm_read() {
             // VM teardown: the JS-thread side will never drain this buffer (its
             // on_progress_update bails the same way), so free the body bytes now.
             task_ref.scheduled_response_buffer = MutableString::default();
@@ -2689,6 +2706,7 @@ impl FetchTasklet {
         // `ct` is the inline `concurrent_task` field of the heap tasklet; the
         // queue takes ownership of its `next` link.
         Self::enqueue_concurrent(task_ref.javascript_vm, ct);
+        task_ref.vm_shutdown_signal.end_vm_read();
 
         task_ref.mutex.unlock();
         // we are done with the http client so we can deref our side

test/js/web/workers/worker-terminate-lifetime.test.ts

@@ -455,3 +455,103 @@ test.skipIf(!isDebug)(
   },
   120_000,
 );
+
+// Regression: FetchTasklet holds a lifetime-erased &'static VirtualMachine and
+// the shared HTTP client thread read it (is_shutting_down /
+// enqueue_task_concurrent) after WebWorker::shutdown had dealloc'd the worker's
+// VM storage, taking the whole process down (SIGSEGV on release, ASAN
+// heap-use-after-free on debug). All four shutdown doors funnel through the
+// same WebWorker::shutdown, so the fence is door-agnostic; the test matrix
+// proves it. ASAN-gated: the read is one byte from freed memory, which
+// release builds can survive.
+describe.skipIf(!isASAN)(
+  "worker shutdown with fetch() in flight does not read the freed worker VM from the HTTP thread",
+  () => {
+    // workerExit is inlined into the worker body; parentAction replaces
+    // terminate() when the worker ends itself.
+    const doors: { door: string; workerExit: string; parentAction: string }[] = [
+      { door: "terminate()", workerExit: "", parentAction: "await w.terminate();" },
+      { door: "process.exit()", workerExit: "setTimeout(() => process.exit(0), d.T);", parentAction: "" },
+      { door: "uncaught throw", workerExit: "setTimeout(() => { throw new Error('boom'); }, d.T);", parentAction: "" },
+      {
+        door: "unhandled rejection",
+        workerExit: "setTimeout(() => Promise.reject(new Error('boom')), d.T);",
+        parentAction: "",
+      },
+    ];
+    for (const { door, workerExit, parentAction } of doors) {
+      test.concurrent(
+        door,
+        async () => {
+          await using proc = Bun.spawn({
+            cmd: [
+              bunExe(),
+              "-e",
+              `
+              const { Worker } = require("node:worker_threads");
+              const server = Bun.serve({
+                hostname: "127.0.0.1",
+                port: 0,
+                fetch(req) {
+                  if (new URL(req.url).pathname === "/health") return new Response("ok");
+                  // Long trickle so HTTP-thread callbacks for this request keep
+                  // arriving past the worker's VM dealloc.
+                  const enc = new TextEncoder();
+                  return new Response(new ReadableStream({ async start(c) {
+                    for (let i = 0; i < 200; i++) { c.enqueue(enc.encode("chunk" + i + "\\n")); await Bun.sleep(2); }
+                    c.close();
+                  } }));
+                },
+              });
+              const base = "http://127.0.0.1:" + server.port;
+              // 10 lanes of back-to-back fetches, mixed body consumption: half
+              // buffer the whole body, half read one chunk and release the
+              // reader so the stream is still draining when the worker exits.
+              const src =
+                'const { parentPort, workerData: d } = require("node:worker_threads");' +
+                'async function lane(l) { for (let i = 0; ; i++) { try {' +
+                '  const r = await fetch(d.base + "/slow?l=" + l + "&i=" + i);' +
+                '  if (i & 1) { const rd = r.body.getReader(); await rd.read(); rd.releaseLock(); }' +
+                '  else await r.arrayBuffer(); } catch {} } }' +
+                'for (let l = 0; l < 10; l++) lane(l);' +
+                'parentPort.postMessage("up");' +
+                ${JSON.stringify(workerExit)};
+              function ready(w) {
+                return new Promise((res, rej) => {
+                  w.once("message", res);
+                  w.once("error", rej);
+                  w.once("exit", c => rej(new Error("worker exited " + c + " before ready")));
+                });
+              }
+              for (let r = 0; r < ${rounds * 2}; r++) {
+                const T = 60 + ((r * 37) % 200);
+                const w = new Worker(src, { eval: true, workerData: { base, T } });
+                await ready(w);
+                w.on("error", () => {});
+                const exited = new Promise(res => w.once("exit", res));
+                ${parentAction ? `await Bun.sleep(T); ${parentAction}` : ""}
+                await exited;
+                // Keep-alive pool must stay healthy across the shutdown.
+                const t = await fetch(base + "/health").then(x => x.text());
+                if (t !== "ok") throw new Error("pool unhealthy after round " + r);
+              }
+              server.stop(true);
+              console.log("survived");
+            `,
+            ],
+            env: bunEnv,
+            stdout: "pipe",
+            stderr: "pipe",
+          });
+
+          const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+          // Check stderr first: on failure the sanitizer report is the useful part.
+          expect(stderr).toBe("");
+          expect(stdout).toBe("survived\n");
+          expect(exitCode).toBe(0);
+        },
+        timeout,
+      );
+    }
+  },
+);