napi: settle or refuse async work queued during VM teardown instead of taking a ticket

src/jsc/VmHandle.rs

@@ -409,6 +409,14 @@ impl VmHandle {
         self.0.state() == State::Open
     }
 
+    /// Is the wait over? Past it nothing started for the VM runs any more:
+    /// [`VirtualMachine::ticket`] panics and a queued task is only released.
+    /// Any thread; meaningful on the JS thread (a final collection's finalizer).
+    #[inline]
+    pub fn closed(&self) -> bool {
+        self.0.state() == State::Closed
+    }
+
     pub(crate) fn tickets_outstanding(&self) -> u32 {
         self.0.tickets.load(Ordering::SeqCst)
     }
@@ -503,19 +511,30 @@ impl VmHandle {
 impl VirtualMachine {
     /// JS thread: a ticket for work about to leave this thread — this VM, and
     /// the loop it is currently ticking. Hold it in the in-flight operation
-    /// and drop it after the completion is posted. Infallible until the wait
-    /// has finished (after which nothing on this thread starts off-thread work).
+    /// and drop it after the completion is posted.
+    ///
+    /// Panics, in release too, once the VM is [`closed`](Self::closed): the
+    /// wait is over, so nothing would wait for this ticket and its completion
+    /// would land on a loop teardown is freeing. Callers that can run that late
+    /// (finalizers) check `closed()` and refuse the work; the panic names one
+    /// that did not.
     #[track_caller]
     #[inline]
     pub fn ticket(&self) -> Ticket {
         let h = self.handle_ref();
         h.assert_js_thread();
-        debug_assert!(
-            h.0.state() != State::Closed,
+        assert!(
+            !h.closed(),
             "off-thread work started after the VM finished draining"
         );
         Ticket::issue(&h.0, self.current_loop_kind())
     }
+
+    /// [`VmHandle::closed`] for this VM.
+    #[inline]
+    pub fn closed(&self) -> bool {
+        self.handle_ref().closed()
+    }
 }
 
 // ── Test suite only: deterministic late completions ───────────────────────

src/jsc/event_loop.rs

@@ -49,6 +49,9 @@ pub struct EventLoop {
     /// Set when teardown releases the queue: from then on `enqueue_task`
     /// releases instead of parking (nothing will tick this loop again).
     closed_for_tasks: bool,
+    /// Set while `release_queued_tasks` is draining: a task enqueued by one
+    /// it releases is parked for that same drain (see `enqueue_task`).
+    releasing_tasks: bool,
 
     /// setImmediate() gets it's own two task queues
     /// When you call `setImmediate` in JS, it queues to the start of the next tick
@@ -119,6 +122,7 @@ impl Default for EventLoop {
         Self {
             tasks: Queue::init(),
             closed_for_tasks: false,
+            releasing_tasks: false,
             immediate_tasks: Vec::new(),
             next_immediate_tasks: Vec::new(),
             yield_tasks: Vec::new(),
@@ -719,10 +723,14 @@ impl EventLoop {
     }
 
     pub fn enqueue_task(&mut self, task: Task) {
-        if self.closed_for_tasks {
+        if self.closed_for_tasks && !self.releasing_tasks {
             // Teardown already released the queue and this loop never ticks
             // again: release the task now, as `release_queued_tasks` would have
             // — the queue owns refusal, like `VmHandle::post` does off-thread.
+            // While that release is still draining, the task is parked for it
+            // instead, so a released task that enqueues another (a napi
+            // `complete` queueing more work) has it released after itself
+            // rather than inside itself.
             // SAFETY: JS thread, JSC heap alive (teardown phase B/C).
             unsafe { __bun_release_task_unrun(task) };
             return;
@@ -758,16 +766,35 @@ impl EventLoop {
     /// JSC heap alive, children joined): called on every turn of the wait, and
     /// once more after `Closed`.
     pub fn release_queued_tasks(&mut self) {
-        self.closed_for_tasks = true;
-        self.take_concurrent_tasks();
-        let _ = self.promote_yield_tasks();
-        while let Some(task) = self.tasks.read_item() {
+        // R-2 noalias mitigation — see `run_callback` above. `&mut self` is
+        // `noalias` and `__bun_release_task_unrun` receives nothing derived
+        // from it, but a release re-enters this loop through
+        // `vm.event_loop_mut()`: `enqueue_task` reads the two flags stored
+        // below and, while we are draining, pushes onto `tasks`, which the next
+        // `read_item` must observe. So everything here goes through a
+        // laundered pointer, re-escaped after every release.
+        let mut this: *mut Self = core::hint::black_box(core::ptr::from_mut(self));
+        // SAFETY: `this` is the unique live `EventLoop`; every access through
+        // it is a short-lived borrow that ends before a release runs.
+        unsafe {
+            (*this).closed_for_tasks = true;
+            (*this).take_concurrent_tasks();
+            let _ = (*this).promote_yield_tasks();
+            (*this).releasing_tasks = true;
+        }
+        // SAFETY: as above.
+        while let Some(task) = unsafe { (*this).tasks.read_item() } {
             // SAFETY: JS thread, heap alive; `task` just left the queue.
             unsafe { __bun_release_task_unrun(task) };
+            this = core::hint::black_box(this);
+        }
+        // SAFETY: as above. Pending immediates likewise: cancelling one drops
+        // its keep-alive on this thread's loop, so it happens now, not after
+        // the loop is gone.
+        unsafe {
+            (*this).releasing_tasks = false;
+            (*this).release_pending_immediates();
         }
-        // Pending immediates likewise: cancelling one drops its keep-alive on
-        // this thread's loop, so it happens now, not after the loop is gone.
-        self.release_pending_immediates();
     }
 
     /// Cancel (never run) every queued ImmediateObject; each cancel drops the

src/runtime/api/JSTranspiler.rs

@@ -651,7 +651,8 @@ impl Config {
 // This is going to be hard to not leak
 /// `transpiler.transform()` off the JS thread. The parse/print state points
 /// into the owning `JSTranspiler`'s config (its `Transpiler` is bit-copied),
-/// which the job's Js side keeps alive and the pool borrow keeps valid.
+/// which the `Strong` on the job's Js side keeps alive until the job is back
+/// on the JS thread; `run` reads it while the job's ticket keeps the VM alive.
 pub(crate) struct TransformTask {
     pub input_code: bun_jsc::ThreadSafe<StringOrBuffer>,
     pub output_code: BunString,
@@ -663,8 +664,8 @@ pub(crate) struct TransformTask {
     pub loader: Loader,
     pub replace_exports: bun_ast::runtime::ReplaceableExportMap,
 }
-// SAFETY: see the type doc — VM-owned config is read only under the pool
-// borrow; everything else is owned.
+// SAFETY: see the type doc — the wrapper's config (behind `transpiler` and
+// `tsconfig`) is read only in `run`, under the job's ticket; the rest is owned.
 unsafe impl Send for TransformTask {}
 
 #[derive(bun_jsc::JsAffine)]

src/runtime/image/Image.rs

@@ -1398,8 +1398,10 @@ pub struct PipelineTask {
     auto_orient: bool,
     result: TaskResult,
 }
-// SAFETY: `input` borrows bytes that are pinned (`Pin`) or owned by the Image
-// the job's Js side keeps alive; read only under the pool borrow. The rest is owned.
+// SAFETY: `input` borrows bytes pinned by `Pin` or, like its path, owned by the
+// Image `PendingTask` holds; both sit on the job's Js side, dropped there only
+// after the pool has posted this task back, and the pool reads them in `run`,
+// under the job's ticket. The rest is owned.
 unsafe impl Send for PipelineTask {}
 
 /// The JS-thread half of a scheduled `PipelineTask`.

src/runtime/napi/napi_body.rs

@@ -43,9 +43,10 @@ impl JSValueNapiExt for JSValue {
 // `Taskable` impls for the napi heap tasks dispatched through the JS event loop.
 impl Taskable for napi_async_work {
     const TAG: TaskTag = task_tag::NapiAsyncWork;
-    /// Work the pool handed back during teardown: its `complete` callback is
-    /// how the addon learns the outcome and frees the work (Node calls it from
-    /// environment cleanup too); script it tries to run is refused at the boundary.
+    /// Work handed back during teardown (by the pool, or by `schedule` itself
+    /// once script was forbidden): its `complete` callback is how the addon
+    /// learns the outcome and frees the work (Node calls it from environment
+    /// cleanup too); script it tries to run is refused at the boundary.
     unsafe fn release_unrun(this: *mut Self) {
         let vm = VirtualMachine::get().as_mut();
         let global = vm.global();
@@ -143,6 +144,12 @@ impl NapiEnv {
         Self::set_last_error(Some(self), NapiStatus::pending_exception)
     }
 
+    /// Node's status for a call the environment can no longer serve because it
+    /// is shutting down.
+    pub(crate) fn cannot_run_js(&self) -> napi_status {
+        Self::set_last_error(Some(self), NapiStatus::cannot_run_js)
+    }
+
     /// Checks both `env->m_pendingException` (set by `napi_throw*`) and the JSC
     /// VM exception slot. This is the gate Node.js's `NAPI_PREAMBLE` enforces.
     pub(crate) fn has_pending_exception(&self) -> bool {
@@ -1851,17 +1858,44 @@ impl napi_async_work {
         drop(unsafe { bun_core::heap::take(this) });
     }
 
-    pub(crate) fn schedule(&mut self) {
-        if self.scheduled {
-            return;
+    /// `false`: refused untouched, the work stays the addon's to delete.
+    ///
+    /// # Safety
+    /// `this` is a live work on its JS thread. It may be gone when this
+    /// returns `true`: its `complete` (which addons delete the work from) can
+    /// run inside the call.
+    pub(crate) unsafe fn schedule(this: *mut Self) -> bool {
+        // SAFETY: fn contract. The work is only ever reached through `this`,
+        // one statement at a time, so no reference to it is live across
+        // `enqueue_task`, which may free it.
+        unsafe {
+            if (*this).scheduled {
+                return true;
+            }
+            let vm = VirtualMachine::get();
+            if vm.closed() {
+                // A finalizer of the collection destroying the heap:
+                // `vm.ticket()` would panic, and the closed queue would run
+                // `complete` right here, mid-collection.
+                return false;
+            }
+            (*this).scheduled = true;
+            if !vm.script_allowed() {
+                // The pool would only hand it back cancelled; skip the pool and
+                // the ticket. Whether the queue runs it or teardown releases
+                // it, `complete` gets `napi_cancelled` as before.
+                let _ = (*this).cancel();
+                vm.event_loop_mut().enqueue_task(Task::init(this));
+                return true;
+            }
+            (*this).poll_ref.ref_(bun_io::js_vm_ctx());
+            // The work object belongs to the addon and `execute` receives this
+            // env, so the VM waits for it (Node likewise settles its threadpool
+            // requests before an environment is freed).
+            (*this).ticket = Some(vm.ticket());
+            WorkPool::schedule(&raw mut (*this).task);
+            true
         }
-        self.scheduled = true;
-        self.poll_ref.ref_(bun_io::js_vm_ctx());
-        // The work object belongs to the addon and `execute` receives this
-        // env, so the VM waits for it (Node likewise settles its threadpool
-        // requests before an environment is freed).
-        self.ticket = Some(self.global.bun_vm().ticket());
-        WorkPool::schedule(&raw mut self.task);
     }
 
     pub(crate) unsafe fn run_from_thread_pool(task: *mut WorkPoolTask) {
@@ -2222,12 +2256,18 @@ extern "C" fn napi_delete_async_work(env_: napi_env, work_: *mut napi_async_work
 extern "C" fn napi_queue_async_work(env_: napi_env, work_: *mut napi_async_work) -> napi_status {
     bun_output::scoped_log!(napi, "napi_queue_async_work");
     let env = get_env!(env_);
-    // SAFETY: `work_` is null or the `napi_async_work` we allocated in `napi_create_async_work`.
-    let Some(work) = (unsafe { work_.as_mut() }) else {
+    if work_.is_null() {
         return env.invalid_arg();
-    };
-    debug_assert!(core::ptr::eq(env.to_js(), work.global.as_ptr()));
-    work.schedule();
+    }
+    // SAFETY: non-null `work_` is the `napi_async_work` we allocated in
+    // `napi_create_async_work`, live until `schedule` (which may free it: its
+    // `complete` can run inside) takes it; no reference to it is formed here.
+    unsafe {
+        debug_assert!(core::ptr::eq(env.to_js(), (*work_).global.as_ptr()));
+        if !napi_async_work::schedule(work_) {
+            return env.cannot_run_js();
+        }
+    }
     env.ok()
 }
 

src/runtime/webcore/CompressionStreamCoder.rs

@@ -579,7 +579,8 @@ pub(crate) enum AsyncInput {
     Owned(Vec<u8>),
 }
 // SAFETY: `Pinned.ptr` is a backing store pinned + protected by the paired
-// `PinnedChunk` for as long as the job lives; read only under the pool borrow.
+// `PinnedChunk` on the job's Js side, dropped there only after the pool has
+// posted the job back; the pool reads it in `run`, under the job's ticket.
 unsafe impl Send for AsyncInput {}
 
 /// The pin + GC protection on a chunk whose bytes went to the pool; released