Wait for off-thread jobs before freeing a terminated worker's VM

src/event_loop/AnyEventLoop.rs

@@ -450,6 +450,20 @@ impl EventLoopHandle {
         }
     }
 
+    /// No-op for `Mini`: `worker.terminate()` never tears down a mini loop.
+    pub fn offthread_job_begin(self) {
+        if let EventLoopHandle::Js { owner } = self {
+            owner.offthread_job_begin();
+        }
+    }
+
+    /// Pair of [`Self::offthread_job_begin`]; see that method.
+    pub fn offthread_job_end(self) {
+        if let EventLoopHandle::Js { owner } = self {
+            owner.offthread_job_end();
+        }
+    }
+
     pub fn r#loop(self) -> *mut UwsLoop {
         match self {
             EventLoopHandle::Js { owner } => owner.uws_loop(),

src/event_loop/lib.rs

@@ -60,6 +60,8 @@ bun_dispatch::link_interface! {
         fn exit();
         fn enqueue_task(task: Task);
         fn enqueue_task_concurrent(task: core::ptr::NonNull<ConcurrentTask::ConcurrentTask>);
+        fn offthread_job_begin();
+        fn offthread_job_end();
         fn env() -> *mut bun_dotenv::Loader;
         fn top_level_dir() -> *const [u8];
         fn create_null_delimited_env_map() -> Result<bun_dotenv::NullDelimitedEnvMap, bun_core::AllocError>;

src/jsc/ConcurrentPromiseTask.rs

@@ -83,9 +83,15 @@ impl<'a, Context: ConcurrentPromiseTaskContext> ConcurrentPromiseTask<'a, Contex
         // field, so `from_task_ptr` recovers the live heap `Self` parent,
         // exclusively owned by the work pool for this callback's duration.
         let this = unsafe { Self::from_task_ptr(task) };
-        // SAFETY: `this` is alive for the duration of the thread-pool callback;
-        // exclusively owned by the work pool at this point.
-        unsafe { (*this).ctx.run() };
+        // Teardown in progress: skip the compute; the shutdown drain reclaims
+        // the task unrun.
+        // SAFETY: `this` is alive (see above); the job's own
+        // `outstanding_offthread` count keeps the loop alive for this read.
+        if !unsafe { (*this).event_loop }.offthread_cancel_requested() {
+            // SAFETY: `this` is alive for the duration of the thread-pool
+            // callback; exclusively owned by the work pool at this point.
+            unsafe { (*this).ctx.run() };
+        }
         Self::on_finish(this);
     }
 
@@ -97,6 +103,8 @@ impl<'a, Context: ConcurrentPromiseTaskContext> ConcurrentPromiseTask<'a, Contex
     }
 
     pub fn schedule(&mut self) {
+        // Paired with the `offthread_job_end` in `on_finish`.
+        self.event_loop.offthread_job_begin();
         WorkPool::schedule(&raw mut self.task);
     }
 
@@ -115,6 +123,8 @@ impl<'a, Context: ConcurrentPromiseTaskContext> ConcurrentPromiseTask<'a, Contex
         // `task` is the live `concurrent_task` field of the heap-allocated
         // job; the queue takes ownership of its intrusive `next` link.
         event_loop.enqueue_task_concurrent(task);
+        // Last VM access, via the local copy (the JS thread may free `*this` now).
+        event_loop.offthread_job_end();
     }
 
     /// Frees the heap allocation backing this task.

src/jsc/CppTask.rs

@@ -80,6 +80,8 @@ impl ConcurrentCppTask {
         unsafe { EventLoopTaskNoContext::run(cpp_task) };
         if let Some(vm) = maybe_vm {
             vm.event_loop_shared().unref_concurrently();
+            // Last VM access.
+            vm.event_loop_shared().offthread_job_end();
         }
     }
 }
@@ -91,6 +93,8 @@ extern "C" fn ConcurrentCppTask__createAndRun(cpp_task: *mut EventLoopTaskNoCont
     // the centralised non-null deref proof. C++ just handed it over.
     if let Some(vm) = EventLoopTaskNoContext::opaque_ref(cpp_task).get_vm() {
         vm.event_loop_shared().ref_concurrently();
+        // Paired with the `offthread_job_end` in `run_owned`.
+        vm.event_loop_shared().offthread_job_begin();
     }
     WorkPool::schedule_new(ConcurrentCppTask {
         cpp_task,

src/jsc/RuntimeTranspilerStore.rs

@@ -496,6 +496,9 @@ impl TranspilerJob {
         // SAFETY: vm outlives the job; event_loop() returns the live self-pointer.
         unsafe { &*(*vm).event_loop() }
             .enqueue_task_concurrent(ConcurrentTask::create_from(transpiler_store));
+        // Last VM access, via the local; releases the fence taken in `schedule`.
+        // SAFETY: the job's own `outstanding_offthread` count kept `vm` alive.
+        unsafe { (*vm).event_loop_shared() }.offthread_job_end();
     }
 
     fn run_from_js_thread(&mut self) -> JsResult<()> {
@@ -559,6 +562,9 @@ impl TranspilerJob {
         // `EventLoopCtx` vtable; resolve it via the `get_vm_ctx` hook (registered by
         // `bun_runtime::init`).
         self.poll_ref.ref_(get_vm_ctx(AllocatorType::Js));
+        // Paired with the `offthread_job_end` in `dispatch_to_main_thread`.
+        // SAFETY: `vm` is the live owning VM (BACKREF — the VM owns the store).
+        unsafe { (*self.vm).event_loop_shared() }.offthread_job_begin();
         WorkPool::schedule(&raw mut self.work_task);
     }
 

src/jsc/VirtualMachine.rs

@@ -68,6 +68,15 @@ pub type ExceptionList = Vec<crate::schema_api::JsException>;
 // VirtualMachine struct (file-level @This())
 // ──────────────────────────────────────────────────────────────────────────
 
+/// One entry in [`VirtualMachine::terminate_cancel_hooks`]. `data` carries
+/// payload the cancel fn must not read from `ptr` (e.g. an `async_http_id`
+/// whose on-task storage the HTTP thread mutates concurrently).
+pub struct TerminateCancelHook {
+    pub ptr: *mut (),
+    pub data: u64,
+    pub run: fn(*mut (), u64),
+}
+
 #[derive(Default)]
 pub struct EntryPointResult {
     pub value: crate::strong::Optional, // jsc.Strong.Optional
@@ -217,6 +226,11 @@ pub struct VirtualMachine {
     pub(crate) hide_bun_stackframes: bool,
 
     pub is_shutting_down: bool,
+    /// JS-thread-only registry of cancels for `WebWorker::shutdown` to run
+    /// before waiting on `EventLoop::outstanding_offthread` (fetch/S3 register
+    /// an abort here so the wait is bounded by socket shutdown, not by the
+    /// transfer). Keyed by `ptr`; owners unregister on JS-side release.
+    pub(crate) terminate_cancel_hooks: Vec<TerminateCancelHook>,
     /// 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
@@ -989,6 +1003,49 @@ impl VirtualMachine {
         self.is_shutting_down
     }
 
+    /// See [`Self::terminate_cancel_hooks`]. JS thread only. No-op on the
+    /// main-thread VM: only `WebWorker::shutdown` runs the fan-out, so the
+    /// registry (and the O(n) unregister scan) would be dead weight on the
+    /// outbound-fetch hot path there.
+    pub fn register_terminate_cancel_hook(
+        &mut self,
+        ptr: *mut (),
+        data: u64,
+        run: fn(*mut (), u64),
+    ) {
+        if self.is_main_thread() {
+            return;
+        }
+        self.terminate_cancel_hooks
+            .push(TerminateCancelHook { ptr, data, run });
+    }
+
+    /// Remove the hook registered with `ptr`; no-op when absent (the fan-out
+    /// empties the list, and the main-thread VM never registers). JS thread
+    /// only.
+    pub fn unregister_terminate_cancel_hook(&mut self, ptr: *mut ()) {
+        if self.is_main_thread() {
+            return;
+        }
+        if let Some(i) = self
+            .terminate_cancel_hooks
+            .iter()
+            .position(|h| h.ptr == ptr)
+        {
+            self.terminate_cancel_hooks.swap_remove(i);
+        }
+    }
+
+    /// Worker-shutdown cancel fan-out: run every registered hook (each `run`
+    /// must be idempotent; the VM is torn down right after).
+    pub fn run_terminate_cancel_hooks(&mut self) {
+        // Moved out because hooks may re-enter `self`.
+        let hooks = core::mem::take(&mut self.terminate_cancel_hooks);
+        for hook in &hooks {
+            (hook.run)(hook.ptr, hook.data);
+        }
+    }
+
     pub fn has_run_cleanup_hooks(&self) -> bool {
         self.has_run_cleanup_hooks
     }
@@ -1610,7 +1667,10 @@ impl VirtualMachine {
             // without it the tasklet ⇄ `Box<AsyncHTTP>` cycle leaks. Must
             // precede `destructOnExit` so `FetchTasklet::deinit` can drop its
             // JSC `Strong`/`Weak` handles against a live heap.
-            self.event_loop_mut().release_queued_tasks_for_shutdown();
+            // `offthread_drained: true` — the HTTP daemon just parked, so
+            // every posting thread has made its last access.
+            self.event_loop_mut()
+                .release_queued_tasks_for_shutdown(true);
 
             if let Some(rare) = self.rare_data.as_deref_mut() {
                 rare.release_js_handles();
@@ -2112,6 +2172,7 @@ impl VirtualMachine {
             // their validity invariants even when len/cap are 0. Write the
             // canonical empty value via `ptr::write` (no Drop of zeroed bytes).
             addr_of_mut!((*vm).preload).write(Vec::new());
+            addr_of_mut!((*vm).terminate_cancel_hooks).write(Vec::new());
             addr_of_mut!((*vm).argv).write(Vec::new());
             addr_of_mut!((*vm).resolved_path_dups).write(Vec::new());
             addr_of_mut!((*vm).macros).write(Default::default());
@@ -4363,6 +4424,8 @@ impl VirtualMachine {
         // proxy strings; `ProxyEnvStorage: Default` so take()+drop suffices.
         drop(core::mem::take(&mut self.proxy_env_storage));
 
+        drop(core::mem::take(&mut self.terminate_cancel_hooks));
+
         // 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/WorkTask.rs

@@ -120,6 +120,8 @@ impl<Context: WorkTaskContext> WorkTask<Context> {
 
     pub fn schedule(this: &mut Self) {
         this.ref_.ref_(Async::js_vm_ctx());
+        // Paired with the `offthread_job_end` in `on_finish`.
+        this.event_loop.offthread_job_begin();
         this.async_task_tracker.did_schedule(this.global_this.get());
         WorkPool::schedule(&raw mut this.task);
     }
@@ -138,5 +140,7 @@ impl<Context: WorkTaskContext> WorkTask<Context> {
         // `task` is the inline `concurrent_task` field of the live
         // heap-allocated `*this`; `event_loop` is the JS-thread loop stored at init.
         event_loop.enqueue_task_concurrent(task);
+        // Last VM access, via the local copy (the JS thread may free `*this` now).
+        event_loop.offthread_job_end();
     }
 }

src/jsc/any_task_job.rs

@@ -114,6 +114,11 @@ impl<C: AnyTaskJobCtx> AnyTaskJob<C> {
         // pointer handed to the pool is derived from the raw `this` and nothing
         // touches the job afterwards.
         unsafe { (*this).poll.ref_(bun_io::js_vm_ctx()) };
+        // Paired with the `offthread_job_end` in `run_task`.
+        // SAFETY: `this` is live (caller contract).
+        unsafe { (*this).vm }
+            .event_loop_shared()
+            .offthread_job_begin();
         // SAFETY: `this` is live; the pointer handed to the pool is derived
         // from the raw `this` and nothing touches the job after the schedule.
         WorkPool::schedule(unsafe { &raw mut (*this).task });
@@ -141,11 +146,17 @@ impl<C: AnyTaskJobCtx> AnyTaskJob<C> {
         // `run_from_js` reclaims it.
         let job = unsafe { &mut *Self::from_task_ptr(task) };
         let vm = job.vm;
-        job.ctx.run(vm.global);
+        // Teardown in progress: skip the compute (pbkdf2/scrypt are
+        // deliberately slow); `run_from_js` early-outs on `is_shutting_down`.
+        if !vm.event_loop_shared().offthread_cancel_requested() {
+            job.ctx.run(vm.global);
+        }
         // `ConcurrentTask::create` heap-allocates a fresh task; the queue takes
         // ownership of it.
         vm.event_loop_shared()
             .enqueue_task_concurrent(ConcurrentTask::create(Task::init(std::ptr::from_mut(job))));
+        // Last VM access, via the `vm` local (the JS thread may free the job now).
+        vm.event_loop_shared().offthread_job_end();
     }
 
     fn run_from_js(this: *mut Self) -> JsResult<()> {

src/jsc/event_loop.rs

@@ -10,7 +10,7 @@
 //! poll deadline). See PORTING.md §Dispatch.
 
 use core::ptr::NonNull;
-use core::sync::atomic::{AtomicI32, AtomicPtr, Ordering};
+use core::sync::atomic::{AtomicBool, AtomicI32, AtomicPtr, AtomicU32, Ordering};
 
 use bun_io::{self as Async, Waker};
 use bun_uws as uws;
@@ -88,6 +88,17 @@
 
     pub entered_event_loop_count: isize,
     pub concurrent_ref: AtomicI32,
+    /// Count of off-thread jobs (WorkPool, HTTP thread, bundler thread) whose
+    /// body can still dereference this `EventLoop`, the owning
+    /// `VirtualMachine`, or JSC-heap memory. `WebWorker::shutdown` waits for
+    /// zero before freeing any of those; without the wait every such job is a
+    /// use-after-free when `worker.terminate()` lands mid-flight. Bracket
+    /// with [`Self::offthread_job_begin`] / [`Self::offthread_job_end`].
+    pub outstanding_offthread: AtomicU32,
+    /// Set by `WebWorker::shutdown` before it waits on the count above. Job
+    /// bodies that can skip their (expensive) work check this first; the
+    /// unrun completion is reclaimed by the shutdown drain. Advisory only.
+    pub offthread_cancel: AtomicBool,
     /// Atomic nullable pointer to the next-due `WTFTimer`.
     ///
     /// Note (§Dispatch): payload is `*mut ()` — the real
@@ -128,6 +139,8 @@
             uws_loop: (),
             entered_event_loop_count: 0,
             concurrent_ref: AtomicI32::new(0),
+            outstanding_offthread: AtomicU32::new(0),
+            offthread_cancel: AtomicBool::new(false),
             imminent_gc_timer: AtomicPtr::new(core::ptr::null_mut()),
             #[cfg(unix)]
             signal_handler: None,
@@ -213,8 +226,9 @@
     /// must be left in the queue (it stays reachable from the static-rooted
     /// VM box, which is the pre-`532a5411961b` behaviour for tags that don't
     /// own JSC handles or whose callback isn't safe to no-op-dispatch).
+    /// `offthread_drained`: see `release_queued_tasks_for_shutdown`.
     /// Defined in `bun_runtime::dispatch`. Link-time resolved.
-    fn __bun_release_task_at_shutdown(task: bun_event_loop::Task) -> bool;
+    fn __bun_release_task_at_shutdown(task: bun_event_loop::Task, offthread_drained: bool) -> bool;
 }
 
 #[inline]
@@ -746,15 +760,20 @@
     /// pre-`532a5411961b` state). Consuming them silently here unhooked that
     /// root and surfaced the boxes as direct leaks (e.g. `AnyTaskJob<_>`); the
     /// definer can't safely dispatch every erased callback at shutdown.
-    pub fn release_queued_tasks_for_shutdown(&mut self) {
+    /// `offthread_drained`: whether every off-thread job has released its
+    /// [`Self::outstanding_offthread`] count (`false` only on the worker
+    /// fence-timeout leak path). Arms whose safety depends on the posting
+    /// thread having finished (the multi-post S3 streaming task) requeue
+    /// instead of freeing when it is `false`.
+    pub fn release_queued_tasks_for_shutdown(&mut self, offthread_drained: bool) {
         self.drop_concurrent_cpp_tasks();
         let mut requeue: Vec<bun_event_loop::Task> = Vec::new();
         while let Some(task) = self.tasks.read_item() {
             // SAFETY: tag-specific release (drops JSC handles while the VM is
             // still live); definer in `bun_runtime::dispatch` matches the same
             // tag set `tick_queue_with_count` does. `false` ⇒ not handled.
             let consumed = task.tag != bun_event_loop::task_tag::ManagedTask
-                && unsafe { __bun_release_task_at_shutdown(task) };
+                && unsafe { __bun_release_task_at_shutdown(task, offthread_drained) };
             if !consumed {
                 requeue.push(task);
             }
@@ -1005,6 +1024,53 @@
         self.wakeup();
     }
 
+    /// JS-thread: call when handing a job to another thread whose body will
+    /// dereference this `EventLoop` / the VM / the JSC heap (rule of thumb:
+    /// every `KeepAlive::ref_` paired with an off-thread schedule). Paired
+    /// with exactly one [`Self::offthread_job_end`] on the off thread.
+    #[inline]
+    pub fn offthread_job_begin(&self) {
+        self.outstanding_offthread.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Off-thread: call after the job's last VM access (typically right after
+    /// the completion enqueue), through a pointer copied to a local first —
+    /// once the waiter observes zero it frees the VM. The `Release` store
+    /// pairs with [`Self::wait_for_offthread_jobs`]'s `Acquire` load.
+    #[inline]
+    pub fn offthread_job_end(&self) {
+        self.outstanding_offthread.fetch_sub(1, Ordering::Release);
+    }
+
+    #[inline]
+    pub fn offthread_cancel_requested(&self) -> bool {
+        self.offthread_cancel.load(Ordering::Acquire)
+    }
+
+    pub fn request_offthread_cancel(&self) {
+        self.offthread_cancel.store(true, Ordering::Release);
+    }
+
+    /// Worker-shutdown barrier: wait (bounded by `timeout_ms`) for the count
+    /// to reach zero. `true` ⇒ the VM may be freed; `false` (timeout) ⇒ the
+    /// caller must leak it. Polls with a short `Futex` timeout instead of a
+    /// wake from `offthread_job_end`, which must not touch `self` after its
+    /// `fetch_sub`.
+    pub fn wait_for_offthread_jobs(&self, timeout_ms: u64) -> bool {
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(timeout_ms);
+        loop {
+            let n = self.outstanding_offthread.load(Ordering::Acquire);
+            if n == 0 {
+                return true;
+            }
+            if std::time::Instant::now() >= deadline {
+                return false;
+            }
+            // 1ms re-check bounds the added terminate() latency.
+            let _ = bun_threading::Futex::wait(&self.outstanding_offthread, n, Some(1_000_000));
+        }
+    }
+
     pub fn ref_concurrently(&self) {
         let _ = self.concurrent_ref.fetch_add(1, Ordering::SeqCst);
         self.wakeup();
@@ -1345,6 +1411,8 @@
         exit() => (*this).exit(),
         enqueue_task(task) => (*this).enqueue_task(task),
         enqueue_task_concurrent(task) => (*this).enqueue_task_concurrent(task),
+        offthread_job_begin() => (*this).offthread_job_begin(),
+        offthread_job_end() => (*this).offthread_job_end(),
         env() => (*this).vm_ref().transpiler.env,
         top_level_dir() => core::ptr::from_ref::<[u8]>((*this).vm_ref().top_level_dir()),
         create_null_delimited_env_map() =>