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

src/event_loop/AnyEventLoop.rs

@@ -450,6 +450,21 @@ impl EventLoopHandle {
         }
     }
 
+    /// `jsc::EventLoop::offthread_job_begin` for the `Js` arm; no-op for
+    /// `Mini` (a mini loop is never torn down by `worker.terminate()`).
+    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() };
+        // Worker teardown in progress: skip the compute (the promise will
+        // never settle; 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,9 @@ impl<'a, Context: ConcurrentPromiseTaskContext> ConcurrentPromiseTask<'a, Contex
     }
 
     pub fn schedule(&mut self) {
+        // Holds the worker-shutdown fence open until `on_finish` has posted
+        // the completion (see `EventLoop::outstanding_offthread`).
+        self.event_loop.offthread_job_begin();
         WorkPool::schedule(&raw mut self.task);
     }
 
@@ -115,6 +124,10 @@ 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`
+        // as soon as the enqueue lands); releases the worker-shutdown fence
+        // taken in `schedule`.
+        event_loop.offthread_job_end();
     }
 
     /// Frees the heap allocation backing this task.

src/jsc/CppTask.rs

@@ -80,6 +80,9 @@ impl ConcurrentCppTask {
         unsafe { EventLoopTaskNoContext::run(cpp_task) };
         if let Some(vm) = maybe_vm {
             vm.event_loop_shared().unref_concurrently();
+            // Last VM access; releases the worker-shutdown fence taken in
+            // `ConcurrentCppTask__createAndRun`.
+            vm.event_loop_shared().offthread_job_end();
         }
     }
 }
@@ -91,6 +94,9 @@ 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();
+        // Holds the worker-shutdown fence open while the pool runs the C++
+        // task body against this VM (see `EventLoop::outstanding_offthread`).
+        vm.event_loop_shared().offthread_job_begin();
     }
     WorkPool::schedule_new(ConcurrentCppTask {
         cpp_task,

src/jsc/VirtualMachine.rs

@@ -68,6 +68,16 @@ pub type ExceptionList = Vec<crate::schema_api::JsException>;
 // VirtualMachine struct (file-level @This())
 // ──────────────────────────────────────────────────────────────────────────
 
+/// One entry in [`VirtualMachine::terminate_cancel_hooks`]: an erased pointer
+/// to the in-flight operation plus a cancel fn. `data` carries per-entry
+/// payload the 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 +227,13 @@ pub struct VirtualMachine {
     pub(crate) hide_bun_stackframes: bool,
 
     pub is_shutting_down: bool,
+    /// JS-thread-only registry of in-flight off-thread operations that
+    /// `WebWorker::shutdown` should cancel before waiting on
+    /// [`EventLoop::outstanding_offthread`] (in-flight `fetch`/S3 HTTP work
+    /// registers an abort here so the wait is bounded by socket shutdown, not
+    /// by the transfer). Entries are keyed by `ptr`; owners unregister when
+    /// the operation's JS-side handle is released.
+    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 +1006,43 @@ impl VirtualMachine {
         self.is_shutting_down
     }
 
+    /// Register an off-thread operation for the terminate-time cancel fan-out
+    /// (see the [`Self::terminate_cancel_hooks`] field doc). JS thread only.
+    pub fn register_terminate_cancel_hook(
+        &mut self,
+        ptr: *mut (),
+        data: u64,
+        run: fn(*mut (), u64),
+    ) {
+        self.terminate_cancel_hooks
+            .push(TerminateCancelHook { ptr, data, run });
+    }
+
+    /// Remove the hook registered with `ptr`. No-op when absent (the fan-out
+    /// leaves entries in place; owners released afterwards must still call
+    /// this). JS thread only.
+    pub fn unregister_terminate_cancel_hook(&mut self, ptr: *mut ()) {
+        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. Hooks stay
+    /// registered (each `run` must be idempotent); the list is never walked
+    /// again — the VM is torn down right after.
+    pub fn run_terminate_cancel_hooks(&mut self) {
+        // Hooks may re-enter `self` (an abort can schedule follow-up work),
+        // so iterate a moved-out list rather than borrowing the field.
+        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
     }
@@ -2112,6 +2166,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 +4418,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,9 @@ impl<Context: WorkTaskContext> WorkTask<Context> {
 
     pub fn schedule(this: &mut Self) {
         this.ref_.ref_(Async::js_vm_ctx());
+        // Holds the worker-shutdown fence open until `on_finish` has posted
+        // the completion (see `EventLoop::outstanding_offthread`).
+        this.event_loop.offthread_job_begin();
         this.async_task_tracker.did_schedule(this.global_this.get());
         WorkPool::schedule(&raw mut this.task);
     }
@@ -138,5 +141,9 @@ 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`
+        // as soon as the enqueue lands); releases the worker-shutdown fence
+        // taken in `schedule`.
+        event_loop.offthread_job_end();
     }
 }

src/jsc/any_task_job.rs

@@ -114,6 +114,12 @@ 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()) };
+        // Holds the worker-shutdown fence open until `run_task` has posted the
+        // completion (see `EventLoop::outstanding_offthread`).
+        // 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 +147,21 @@ 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);
+        // Worker teardown in progress: skip the compute entirely (these jobs
+        // include deliberately slow KDFs — pbkdf2/scrypt — that would
+        // otherwise stall `terminate()`). `run_from_js` early-outs on
+        // `is_shutting_down`, so `then` never observes the missing result.
+        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
+        // as soon as the enqueue lands); releases the worker-shutdown fence
+        // taken in `schedule`.
+        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,22 @@ pub struct EventLoop {
 
     pub entered_event_loop_count: isize,
     pub concurrent_ref: AtomicI32,
+    /// Count of off-thread jobs (WorkPool, HTTP thread, bundler thread) whose
+    /// worker-side body can still dereference this `EventLoop`, the owning
+    /// `VirtualMachine`, or JSC-heap memory (completion posts via
+    /// `enqueue_task_concurrent`, buffers pinned by the scheduling call).
+    /// `WebWorker::shutdown` waits for this to reach zero before
+    /// `WebWorker__teardownJSCVM` frees the JSC heap and the raw VM dealloc
+    /// frees this struct; 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
+    /// [`Self::outstanding_offthread`]. Off-thread job bodies that can skip
+    /// their (expensive) work load this first and complete immediately; the
+    /// completion is reclaimed unrun by the shutdown drain. Advisory: a body
+    /// that never checks it is still memory-safe, just slower to drain.
+    pub offthread_cancel: AtomicBool,
     /// Atomic nullable pointer to the next-due `WTFTimer`.
     ///
     /// Note (§Dispatch): payload is `*mut ()` — the real
@@ -128,6 +144,8 @@ impl Default for EventLoop {
             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,
@@ -1005,6 +1023,65 @@ impl EventLoop {
         self.wakeup();
     }
 
+    /// JS-thread: call when handing a job to another thread (`WorkPool`,
+    /// `HTTPThread::schedule`, the bundler thread) whose off-thread body will
+    /// dereference this `EventLoop` / the owning `VirtualMachine` / the JSC
+    /// heap. Rule of thumb: every `KeepAlive::ref_` paired with an off-thread
+    /// schedule gets a begin at the same site. Paired with exactly one
+    /// [`Self::offthread_job_end`] on the off thread; see
+    /// [`Self::outstanding_offthread`].
+    #[inline]
+    pub fn offthread_job_begin(&self) {
+        self.outstanding_offthread.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Off-thread: call after the job's last access to this `EventLoop` / VM /
+    /// JSC heap (typically right after the completion
+    /// `enqueue_task_concurrent`). Must be invoked through a pointer copied to
+    /// a local before that last access: once the count can reach zero the
+    /// JS-thread completion may free the job, and once the waiter observes
+    /// zero it frees the VM. The `Release` store pairs with
+    /// [`Self::wait_for_offthread_jobs`]'s `Acquire` load so the waiter cannot
+    /// observe zero until every prior access is visible.
+    #[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 every
+    /// outstanding [`Self::offthread_job_begin`] to be matched by
+    /// [`Self::offthread_job_end`]. Returns `true` when the count reached
+    /// zero — the VM may be freed — and `false` on timeout, in which case the
+    /// caller must leak the VM (and everything an off-thread job can still
+    /// reach) instead of freeing it.
+    ///
+    /// The wait polls with a short `Futex` timeout rather than being woken by
+    /// `offthread_job_end`: a wake there would touch `self` after the
+    /// `fetch_sub` that is contractually the job's last access.
+    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 +1422,8 @@ bun_event_loop::link_impl_JsEventLoop! {
         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() =>