diff --git a/src/event_loop/AnyEventLoop.rs b/src/event_loop/AnyEventLoop.rs index 026d61e84f40..6b3eafbda1cc 100644 --- a/src/event_loop/AnyEventLoop.rs +++ b/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(), diff --git a/src/event_loop/lib.rs b/src/event_loop/lib.rs index cd48a014ae4e..1ebce6e2941a 100644 --- a/src/event_loop/lib.rs +++ b/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); + 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; diff --git a/src/jsc/ConcurrentPromiseTask.rs b/src/jsc/ConcurrentPromiseTask.rs index 051d0517f22a..ff387a050986 100644 --- a/src/jsc/ConcurrentPromiseTask.rs +++ b/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. diff --git a/src/jsc/CppTask.rs b/src/jsc/CppTask.rs index 7d3f74544b10..2e504361a575 100644 --- a/src/jsc/CppTask.rs +++ b/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, diff --git a/src/jsc/RuntimeTranspilerStore.rs b/src/jsc/RuntimeTranspilerStore.rs index f64c9320c3e8..0982aaa37299 100644 --- a/src/jsc/RuntimeTranspilerStore.rs +++ b/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); } diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index 9012d07194bb..21d2debce24f 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -68,6 +68,15 @@ pub type ExceptionList = Vec; // 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, /// 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` 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 diff --git a/src/jsc/WorkTask.rs b/src/jsc/WorkTask.rs index e9617030938d..c0243ef7508a 100644 --- a/src/jsc/WorkTask.rs +++ b/src/jsc/WorkTask.rs @@ -120,6 +120,8 @@ impl WorkTask { 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 WorkTask { // `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(); } } diff --git a/src/jsc/any_task_job.rs b/src/jsc/any_task_job.rs index 3dddfb2f96ba..31616cc941ef 100644 --- a/src/jsc/any_task_job.rs +++ b/src/jsc/any_task_job.rs @@ -114,6 +114,11 @@ impl AnyTaskJob { // 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 AnyTaskJob { // `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<()> { diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index d12b7963a69f..fbf23ab00116 100644 --- a/src/jsc/event_loop.rs +++ b/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 struct EventLoop { 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 @@ 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, @@ -213,8 +226,9 @@ unsafe extern "Rust" { /// 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,7 +760,12 @@ impl EventLoop { /// 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 = Vec::new(); while let Some(task) = self.tasks.read_item() { @@ -754,7 +773,7 @@ impl EventLoop { // 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 @@ impl EventLoop { 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 @@ 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() => diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index 7f30e0f913e2..ef61609523e9 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -73,6 +73,12 @@ use crate::{self as jsc, JSGlobalObject, JSValue, JsError, LogJsc}; bun_core::define_scoped_log!(log, Worker, hidden); +/// How long `shutdown()` waits for `EventLoop::outstanding_offthread` to +/// reach zero before leaking the VM instead. The cancel fan-out bounds the +/// common cases; the deadline only fires for pathological jobs (a blocked +/// filesystem op, an addon `execute` that never returns). +const OFFTHREAD_JOB_WAIT_MS: u64 = 10_000; + // ---- Immutable after `create()` (safe from any thread) ---------------------- pub struct WebWorker { @@ -1226,6 +1232,10 @@ impl WebWorker { let mut arena = self.arena.replace(None); let env_loader = self.worker_env_loader.replace(core::ptr::null_mut()); + // `false` when the off-thread fence timed out: a job can still reach + // VM-owned memory, so every free past step 3 is skipped. + let mut drained = true; + // ---- 1. Unpublish vm ------------------------------------------------ self.vm_lock.lock(); // vm_lock held; this is the unpublish point. @@ -1303,6 +1313,26 @@ impl WebWorker { // or observes m_isShuttingDown under m_lock and drops. Idempotent; // teardownJSCVM sets it again. Bun__JSCTaskScheduler__markShuttingDown(vm.global()); + // Off-thread job fence: jobs this VM handed to other threads + // (WorkPool bodies, HTTP-thread fetch/S3 callbacks, the bundler + // thread) hold raw pointers into the VM box / EventLoop / JSC + // heap, and the gates above only cover the C++ posters. Cancel + // what can be cancelled, then wait for `outstanding_offthread` to + // hit zero before anything below frees that memory. Completions + // posted during the wait are reclaimed unrun by the drain below, + // with JSC still alive. On timeout steps 3/5 leak the VM instead. + vm.event_loop_shared().request_offthread_cancel(); + vm.run_terminate_cancel_hooks(); + drained = vm + .event_loop_shared() + .wait_for_offthread_jobs(OFFTHREAD_JOB_WAIT_MS); + if !drained { + log!( + "[{}] off-thread jobs outstanding after {}ms; leaking the worker VM", + self.execution_context_id, + OFFTHREAD_JOB_WAIT_MS + ); + } // 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 @@ -1310,7 +1340,8 @@ impl WebWorker { // ~JSEventListener Weak<> handles, and after teardownJSCVM the // worker VM is dealloc'd-without-Drop so anything still in // self.tasks leaks. Mirrors the global_exit() ordering. - vm.event_loop_mut().release_queued_tasks_for_shutdown(); + vm.event_loop_mut() + .release_queued_tasks_for_shutdown(drained); if let Some(rare) = vm.rare_data.as_deref_mut() { rare.release_js_handles(); } @@ -1319,17 +1350,21 @@ impl WebWorker { } // ---- 3. JSC VM teardown -------------------------------------------- + // Skipped on fence timeout: a straggler may still read JSC-heap + // memory, so the heap is leaked along with the VM box in step 5. if let Some(global) = global_object { - // `JSGlobalObject` is an opaque ZST handle; `opaque_ref` is the - // centralised non-null deref proof (JSC VM still alive here). - WebWorker__teardownJSCVM(JSGlobalObject::opaque_ref(global)); + if drained { + // `JSGlobalObject` is an opaque ZST handle; `opaque_ref` is the + // centralised non-null deref proof (JSC VM still alive here). + WebWorker__teardownJSCVM(JSGlobalObject::opaque_ref(global)); + } } // The finalizers JSC just ran close the sockets that `close_all_socket_groups` leaves // alone (a Listener owns its listen socket and closes it in `finalize`). `us_socket_close` // only queues onto `loop->data.closed_head`; step 5's `on_thread_exit()` frees the loop // out from under whatever is still queued, so drain it now, while the loop is alive. - if !vm_ptr.is_null() { + if !vm_ptr.is_null() && drained { // SAFETY: `vm_ptr` was unpublished under `vm_lock`; sole owner, `destroy()` is below. unsafe { (*vm_ptr).uws_loop_mut().drain_closed_sockets() }; } @@ -1356,12 +1391,17 @@ impl WebWorker { unsafe { (*loop_).internal_loop_data.jsc_vm = core::ptr::null_mut() }; } #[cfg(windows)] - { + if drained { // Per-thread libuv loop teardown; closes any handles still open on // this worker's loop and drops the thread-local pointer. bun_sys::windows::libuv::Loop::shutdown(); } - if !vm_ptr.is_null() { + if !vm_ptr.is_null() && !drained { + // Fence timeout: leak the VM box, loops, JSC heap, and cloned env; + // only clear the thread-local binding (the thread is exiting). + virtual_machine::VMHolder::set_vm(None); + } + if !vm_ptr.is_null() && drained { // SAFETY: vm_ptr valid; sole owner. unsafe { (*vm_ptr).destroy() }; // Reclaim the boxes allocated on the global @@ -1388,7 +1428,7 @@ impl WebWorker { } } // Reclaim the cloned env (`heap::alloc`'d in `start_vm()`; see field doc). - if !env_loader.is_null() { + if !env_loader.is_null() && drained { // SAFETY: `heap::alloc`'d in `start_vm`; sole owner; the VM is // gone so its raw `transpiler.env` borrow is dead. drop(unsafe { bun_core::heap::take(env_loader) }); @@ -1404,8 +1444,17 @@ impl WebWorker { // skipped on glibc; under BUN_DESTRUCT_VM_ON_EXIT it would also gate // on `!bun_is_exiting()`. Everything that registers polls on the loop // (gc_controller, sockets, timers) has been deinit'd above. - bun_uws::on_thread_exit(); - drop(arena.take()); + // Leak-path exception: a straggler's completion post still calls + // `wakeup()` through the leaked VM's loop pointer, and the arena backs + // VM-reachable allocations, so both must stay allocated. + if drained { + bun_uws::on_thread_exit(); + drop(arena.take()); + } else { + // Deliberate leak; `ManuallyDrop` rather than `mem::forget` for + // clippy::mem_forget. + let _leaked = core::mem::ManuallyDrop::new(arena.take()); + } // We MUST NOT call `pthread_exit` here — // glibc's `pthread_exit` throws a `__forced_unwind` diff --git a/src/runtime/api/Archive.rs b/src/runtime/api/Archive.rs index bf84e474a71f..a0969bf402dd 100644 --- a/src/runtime/api/Archive.rs +++ b/src/runtime/api/Archive.rs @@ -729,6 +729,9 @@ impl AsyncTask { } fn schedule(this: *mut Self) { + // Paired with the `offthread_job_end` in `run_callback`. + // SAFETY: `this` is alive; `vm` is the live owning VM (set in `create`). + unsafe { (*(*this).vm).event_loop_shared().offthread_job_begin() }; // SAFETY: `this` is alive (owned by the task system) until run_from_js drops it; // task field is intrusive and stable since `this` is heap-allocated. WorkPool::schedule(unsafe { &raw mut (*this).task }); @@ -759,10 +762,14 @@ impl AsyncTask { unsafe { (*this).ctx.run() }; // SAFETY: vm points to the live owning VM; concurrent_task is intrusive on the same allocation. unsafe { + // Snapshot: the JS thread may free `*this` once the enqueue lands. + let vm = (*this).vm; let ct = core::ptr::NonNull::from( (*this).concurrent_task.from(this, AutoDeinit::ManualDeinit), ); - (*(*this).vm).enqueue_task_concurrent(ct); + (*vm).enqueue_task_concurrent(ct); + // Last VM access; releases the fence taken in `schedule`. + (*vm).event_loop_shared().offthread_job_end(); } } diff --git a/src/runtime/api/js_bundle_completion_task.rs b/src/runtime/api/js_bundle_completion_task.rs index dfb79bb5c37e..e53a8c7a1126 100644 --- a/src/runtime/api/js_bundle_completion_task.rs +++ b/src/runtime/api/js_bundle_completion_task.rs @@ -145,6 +145,11 @@ pub(crate) fn create_and_schedule_completion_task( // conditions from creating two let _ = WorkPool::get(); + // Paired with the `offthread_job_end` in `complete_on_bundle_thread`: the + // bundle thread reads VM-owned state (`env`, transpiler, plugins) until then. + // SAFETY: `event_loop` is the live JS-thread loop (checked non-null above). + unsafe { (*event_loop).offthread_job_begin() }; + bun_bundler::bundle_v2::singleton::enqueue::(completion); // SAFETY: `completion` is live (refcount==1); `vm` outlives this call. @@ -995,9 +1000,11 @@ impl CompletionStruct for JSBundleCompletionTask { // `jsc_event_loop` is a `BackRef` — safe Deref. // `ConcurrentTask::create` heap-allocates a fresh task; the // queue takes ownership of it. + let jsc_event_loop = self.jsc_event_loop; let this = std::ptr::from_mut::(self); - self.jsc_event_loop - .enqueue_task_concurrent(jsc::ConcurrentTask::create(jsc::Task::init(this))); + jsc_event_loop.enqueue_task_concurrent(jsc::ConcurrentTask::create(jsc::Task::init(this))); + // Last VM access, via the local copy (the JS thread may free `*this` now). + jsc_event_loop.offthread_job_end(); } fn set_result(&mut self, result: BundleV2Result) { self.result = result; diff --git a/src/runtime/crypto/PasswordObject.rs b/src/runtime/crypto/PasswordObject.rs index 0ef0218193f9..2a522d0f3370 100644 --- a/src/runtime/crypto/PasswordObject.rs +++ b/src/runtime/crypto/PasswordObject.rs @@ -573,7 +573,16 @@ impl PasswordJob { // is a false positive on this macro contract. #[allow(clippy::boxed_local)] fn run_owned(mut self: Box) { - let value = self.op.compute(&self.password); + let event_loop = self.event_loop; + // Teardown in progress: skip the deliberately-slow KDF; the shutdown + // drain reclaims the queued result unread via `release_unrun`. + // SAFETY: the fence taken in `JSPasswordObject::run` keeps the loop + // alive for this read. + let value = if unsafe { (*event_loop).offthread_cancel_requested() } { + Err(HashError::JSTerminated) + } else { + self.op.compute(&self.password) + }; let result = Box::new(PasswordResult:: { value, promise: core::mem::take(&mut self.promise), @@ -583,10 +592,12 @@ impl PasswordJob { // SAFETY: `event_loop` was stored from the JS-thread VM and outlives the // job; ownership of `result` transfers to the event loop here. unsafe { - (*self.event_loop).enqueue_task_concurrent(ConcurrentTask::create( + (*event_loop).enqueue_task_concurrent(ConcurrentTask::create( bun_event_loop::Task::from_boxed(result), )); } + // SAFETY: `event_loop` outlives the job (fence still held). Last VM access. + unsafe { (*event_loop).offthread_job_end() }; // `self: Box` drops here; Drop runs secure_zero on password (+op). } } @@ -603,6 +614,18 @@ impl bun_event_loop::Taskable for PasswordResult { } impl PasswordResult { + /// Shutdown-drain counterpart of [`Self::run_from_js`]: unref the loop + /// keep-alive and free the box without settling the promise. No JS runs. + /// + /// # Safety + /// `this` must be the queued box from `PasswordJob::run_owned`; sole owner. + pub(crate) unsafe fn release_unrun(this: *mut Self) { + // SAFETY: caller contract. + let mut this = unsafe { bun_core::heap::take(this) }; + this.r#ref.unref(bun_io::js_vm_ctx()); + // promise Strong + value drop with the box. + } + pub(crate) fn run_from_js(this: *mut Self) -> Result<(), jsc::JsTerminated> { // SAFETY: `this` was produced by heap::into_raw in `run_owned` and the // event loop hands sole ownership to this callback. Reclaim the Box once @@ -668,6 +691,11 @@ impl JSPasswordObject { task: WorkPoolTask::default(), }); job.r#ref.ref_(bun_io::js_vm_ctx()); + // Paired with the `offthread_job_end` in `run_owned`. + global_object + .bun_vm() + .event_loop_shared() + .offthread_job_begin(); WorkPool::schedule_owned(job); Ok(promise_value) diff --git a/src/runtime/dispatch.rs b/src/runtime/dispatch.rs index 7a9f2bb1618d..323aa42a0d48 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -1265,8 +1265,13 @@ unsafe fn __bun_tick_queue_with_count( /// `destructOnExit`. Releases the boxes and JSC handles the dispatch path /// would have dropped. Tags not yet listed leak their box at exit; add them /// as LSan surfaces them. +/// +/// `offthread_drained` is `false` only on the worker fence-timeout leak path, +/// where an off-thread job may still be mid-post; arms whose box the posting +/// thread can touch again (the multi-post S3 streaming task) must requeue in +/// that case. #[unsafe(no_mangle)] -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 { use bun_event_loop::task_tag; match task.tag { // `callback` (HTTP thread) won the `has_schedule_callback` CAS and @@ -1302,13 +1307,12 @@ fn __bun_release_task_at_shutdown(task: bun_event_loop::Task) -> bool { } // `AsyncFSTask`s are `Box::leak`'d in `create()` and freed by // `destroy()` (called from `run_from_js_thread`'s scopeguard). - // `destroy()` resets `JSPromiseStrong` (touches the StrongRootBlock list) - // and unrefs the loop `KeepAlive`, both of which are still valid - // here — we're before `destructOnExit`. Before - // `release_queued_tasks_for_shutdown` existed these boxes stayed - // reachable via `concurrent_tasks` (rooted by the static `VMHolder`), - // so LSan didn't flag them; the drain unhooks that root and surfaces - // the real leak. + // `release_at_shutdown` first frees result heap that `fs_to_js` + // would have transferred to JS (a terminated worker's in-flight + // readFile otherwise strands its whole result buffer), then + // `destroy()` resets `JSPromiseStrong` (touches the StrongRootBlock + // list) and unrefs the loop `KeepAlive`, both of which are still + // valid here — we're before `destructOnExit`. for_each_fs_async_op!(__fs_pat) => { macro_rules! __fs_destroy { ($($tag:ident $ty:ident;)*) => { match task.tag { @@ -1317,7 +1321,7 @@ fn __bun_release_task_at_shutdown(task: bun_event_loop::Task) -> bool { // `AsyncFSTask::create`. The work-pool callback ran // (it posted this entry) so the threadpool no longer // holds the embedded `task` field. - unsafe { fs_async::$ty::destroy(task.ptr.cast::()) }; + unsafe { fs_async::$ty::release_at_shutdown(task.ptr.cast::()) }; })* // SAFETY: outer arm guard proves one of the table tags matched. _ => unsafe { core::hint::unreachable_unchecked() }, @@ -1341,6 +1345,157 @@ fn __bun_release_task_at_shutdown(task: bun_event_loop::Task) -> bool { unsafe { Bun__deleteDeferredWorkTask(task.ptr.cast::()) }; true } + task_tag::NapiAsyncWork => { + // SAFETY: tag identifies pointee; the pool-thread callback already + // posted this entry (`outstanding_offthread` barrier). + unsafe { napi_async_work::release_for_shutdown(task.ptr.cast::()) }; + true + } + // Async `node:zlib` completion: release `write()`'s acquisitions + // (Strong handle, pinned buffers, poll_ref, +1 ref) without calling + // the JS write/error callbacks; JSC is still live here. + task_tag::NativeZlib | task_tag::NativeBrotli | task_tag::NativeZstd => { + macro_rules! release_compression { + ($T:ty) => { + // SAFETY: tag identifies pointee; live m_ctx payload kept + // alive by `write()`'s `ref_()`. + unsafe { + node_zlib_binding::CompressionStream::<$T>::release_unrun( + task.ptr.cast::<$T>(), + ) + } + }; + } + match task.tag { + task_tag::NativeZlib => release_compression!(NativeZlib), + task_tag::NativeBrotli => release_compression!(NativeBrotli), + task_tag::NativeZstd => release_compression!(NativeZstd), + // SAFETY: outer arm guard proves one of the three tags matched. + _ => unsafe { core::hint::unreachable_unchecked() }, + } + true + } + // `run_from_js` early-outs on `is_shutting_down` (already set on both + // drain paths) and frees the box, so the erased dispatch here is a + // pure release: poll unref + `Drop for C`, no user code. + task_tag::AnyTaskJob => { + // SAFETY: §Dispatch — `task.ptr` is a live heap `AnyTaskJob` + // enqueued by `AnyTaskJob::run_task`; the erased entry frees it. + let _ = unsafe { bun_jsc::any_task_job::dispatch_erased(task.ptr) }; + true + } + task_tag::PasswordHashResult => { + // SAFETY: tag identifies pointee; boxed by `PasswordJob::run_owned`. + unsafe { + crate::crypto::password_object::PasswordResult::< + crate::crypto::password_object::HashOp, + >::release_unrun(task.ptr.cast()) + }; + true + } + task_tag::PasswordVerifyResult => { + // SAFETY: tag identifies pointee; boxed by `PasswordJob::run_owned`. + unsafe { + crate::crypto::password_object::PasswordResult::< + crate::crypto::password_object::VerifyOp, + >::release_unrun(task.ptr.cast()) + }; + true + } + // `ConcurrentPromiseTask` completions: unref + `destroy` (drops the + // ctx box and the promise `Strong`; JSC still live, no JS runs). + task_tag::AsyncGlobWalkTask + | task_tag::AsyncImageTask + | task_tag::AsyncTransformTask + | task_tag::CopyFilePromiseTask => { + macro_rules! release_promise_task { + ($ty:ty) => {{ + let t = task.ptr.cast::<$ty>(); + // SAFETY: tag identifies pointee; the pool callback posted + // this entry, so the pool no longer touches it. + unsafe { + (*t).ref_.unref(bun_io::js_vm_ctx()); + bun_jsc::concurrent_promise_task::ConcurrentPromiseTask::destroy(t); + } + }}; + } + match task.tag { + task_tag::AsyncGlobWalkTask => release_promise_task!(AsyncGlobWalkTask<'_>), + task_tag::AsyncImageTask => release_promise_task!(AsyncImageTask<'_>), + task_tag::AsyncTransformTask => release_promise_task!(AsyncTransformTask<'_>), + task_tag::CopyFilePromiseTask => release_promise_task!(CopyFilePromiseTask<'_>), + // SAFETY: outer arm guard proves one of the four tags matched. + _ => unsafe { core::hint::unreachable_unchecked() }, + } + true + } + // The HTTP thread's final callback posted these and made its last + // access before the fence released; we own the box. `Drop` is + // JS-free (poll unref + `clear_data`). + task_tag::S3HttpSimpleTask => { + // SAFETY: tag identifies pointee; sole owner (see above). + drop(unsafe { bun_core::heap::take(task.ptr.cast::()) }); + true + } + task_tag::S3HttpDownloadStreamingTask => { + // Unlike the single-post tags above, the streaming task posts per + // chunk and the HTTP thread keeps touching the box until its + // final callback's `offthread_job_end`. Only a drained fence + // proves that callback finished; otherwise requeue so the box + // stays allocated (it is then leaked with the VM). + if offthread_drained { + // SAFETY: tag identifies pointee; the fence drained, so the + // HTTP thread made its last access; sole owner (see above). + drop(unsafe { + bun_core::heap::take(task.ptr.cast::()) + }); + true + } else { + false + } + } + // `cancelled` short-circuits `on_complete` right after the poll + // unref, so no JS runs; adopting the enqueue's +1 frees the box. + task_tag::JSBundleCompletionTask => { + let c = task + .ptr + .cast::(); + // SAFETY: tag identifies pointee; the bundle thread's last access + // ended when it posted this entry. + unsafe { + (*c).cancelled = true; + let _ = + crate::api::js_bundle_completion_task::JSBundleCompletionTask::on_complete_anytask(c); + } + true + } + // `run_from_js`'s `is_shutting_down()` early-out is a pure release: + // `heap::take` + keep-alive unref + `Drop` for the ctx (which can + // hold the whole archive output) and the promise `Strong`; no JS. + task_tag::ArchiveExtractTask + | task_tag::ArchiveBlobTask + | task_tag::ArchiveWriteTask + | task_tag::ArchiveFilesTask => { + // Tag identifies each pointee; the pool callback posted this + // entry, so the pool no longer touches it. + let _ = match task.tag { + task_tag::ArchiveExtractTask => { + ArchiveAsyncTask::run_from_js(task.ptr.cast::()) + } + task_tag::ArchiveBlobTask => { + ArchiveAsyncTask::run_from_js(task.ptr.cast::()) + } + task_tag::ArchiveWriteTask => { + ArchiveAsyncTask::run_from_js(task.ptr.cast::()) + } + task_tag::ArchiveFilesTask => { + ArchiveAsyncTask::run_from_js(task.ptr.cast::()) + } + // SAFETY: outer arm guard proves one of the four tags matched. + _ => unsafe { core::hint::unreachable_unchecked() }, + }; + true + } // Same reclaim `drop_concurrent_cpp_tasks` performs, but for tasks // that were already batch-moved into `self.tasks`. Must run before // JSC teardown: a Worker `dispatchExit` lambda's `~Ref` walks diff --git a/src/runtime/napi/napi_body.rs b/src/runtime/napi/napi_body.rs index 12cd0b609cce..bb9acaea0ab2 100644 --- a/src/runtime/napi/napi_body.rs +++ b/src/runtime/napi/napi_body.rs @@ -1758,7 +1758,8 @@ impl napi_async_work { env: unsafe { NapiEnvRef::clone_from_raw(env.as_mut_ptr()) }, execute, // SAFETY: `event_loop()` is the live JS-thread loop (non-null, - // stable address) and outlives every napi_async_work. + // stable address); `schedule()`'s `offthread_job_begin()` keeps it + // alive across a worker `terminate()` until `run()` ends the job. event_loop: unsafe { bun_ptr::BackRef::from_raw(global.bun_vm().event_loop()) }, complete, data, @@ -1777,12 +1778,28 @@ impl napi_async_work { drop(unsafe { bun_core::heap::take(this) }); } + /// Shutdown-drain release: unref the loop `KeepAlive` only. The box is + /// addon-owned (`napi_create_async_work` hands the handle out and only + /// `napi_delete_async_work` may free it, possibly from a cleanup hook or + /// finalizer), so it is left to leak with the VM, like the addon's `data`. + /// `complete` is not called (it would run after `NapiEnv::cleanup()`). + /// + /// # Safety + /// `this` must be the heap work popped from the shutdown drain; the pool + /// thread no longer holds it. + pub(crate) unsafe fn release_for_shutdown(this: *mut napi_async_work) { + // SAFETY: see fn contract. + unsafe { core::mem::take(&mut (*this).poll_ref) }.unref(bun_io::js_vm_ctx()); + } + pub(crate) fn schedule(&mut self) { if self.scheduled { return; } self.scheduled = true; self.poll_ref.ref_(bun_io::js_vm_ctx()); + // Matched by `offthread_job_end()` at the end of `run()`. + self.event_loop.offthread_job_begin(); WorkPool::schedule(&raw mut self.task); } @@ -1794,6 +1811,10 @@ impl napi_async_work { fn run(&mut self) { let self_ptr: *mut Self = self; + // The JS thread may free this work right after the enqueue below; + // copy the handle out so the trailing `offthread_job_end()` does not + // touch `self`. + let event_loop = self.event_loop; if let Err(state) = self.status.compare_exchange( AsyncWorkStatus::Pending as u32, AsyncWorkStatus::Started as u32, @@ -1803,11 +1824,11 @@ impl napi_async_work { if state == AsyncWorkStatus::Cancelled as u32 { // `concurrent_task` is the live inline field of this heap work; // the queue takes ownership of its `next` link. - self.event_loop - .enqueue_task_concurrent(core::ptr::NonNull::from( - self.concurrent_task - .from(self_ptr, AutoDeinit::ManualDeinit), - )); + event_loop.enqueue_task_concurrent(core::ptr::NonNull::from( + self.concurrent_task + .from(self_ptr, AutoDeinit::ManualDeinit), + )); + event_loop.offthread_job_end(); return; } } @@ -1817,11 +1838,11 @@ impl napi_async_work { // `concurrent_task` is the live inline field of this heap work; the // queue takes ownership of its `next` link. - self.event_loop - .enqueue_task_concurrent(core::ptr::NonNull::from( - self.concurrent_task - .from(self_ptr, AutoDeinit::ManualDeinit), - )); + event_loop.enqueue_task_concurrent(core::ptr::NonNull::from( + self.concurrent_task + .from(self_ptr, AutoDeinit::ManualDeinit), + )); + event_loop.offthread_job_end(); } pub(crate) fn cancel(&mut self) -> bool { diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index da513be2cf3f..990dc6c8eeaf 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -1015,6 +1015,20 @@ mod _async_tasks { // `bun_sys::Error` frees its path on Drop. task.r#ref.unref(bun_io::js_vm_ctx()); } + + /// Shutdown-drain release for a completion that never reaches + /// `run_from_js_thread`: free result heap that `fs_to_js` would have + /// transferred to JS, then [`Self::destroy`]. + /// + /// SAFETY: same contract as [`Self::destroy`]. + pub(crate) unsafe fn release_at_shutdown(this: *mut Self) { + // SAFETY: caller contract; the drain owns the queued task. + if let Ok(r) = unsafe { &mut (*this).result } { + r.release_unrun(); + } + // SAFETY: caller contract. + unsafe { Self::destroy(this) }; + } } // ────────────────────────────────────────────────────────────────────────── @@ -1155,6 +1169,11 @@ mod _async_tasks { /// Each `ret::*` type implements this by forwarding to its inherent method. pub trait FsReturn { fn fs_to_js(&mut self, global: &JSGlobalObject) -> JsResult; + + /// Free heap that `fs_to_js` would have transferred to JS. Called only + /// by the shutdown drain for completions released unrun (the default + /// is right for types whose `Drop` already frees everything). + fn release_unrun(&mut self) {} } impl FsReturn for JSValue { #[inline] @@ -1197,6 +1216,16 @@ mod _async_tasks { fn fs_to_js(&mut self, global: &JSGlobalObject) -> JsResult { self.to_js(global) } + + fn release_unrun(&mut self) { + // The `Buffer` variant's `Drop` is deliberately a no-op (ownership + // normally transfers to a JS ArrayBuffer in `to_js`); released + // unrun, the bytes must be freed here or a terminated worker + // leaks every in-flight read's result. + if let StringOrBuffer::Buffer(buffer) = self { + buffer.destroy(); + } + } } impl FsReturn for StringOrUndefined { #[inline] @@ -1230,6 +1259,29 @@ mod _async_tasks { let owned = core::mem::replace(self, ret::Readdir::Files(Box::default())); owned.to_js(global) } + + fn release_unrun(&mut self) { + // Entries own refcounts / byte buffers that `to_js` would have + // transferred to JS; mirror `ResultListEntryValue::deinit`. + match self { + ret::Readdir::WithFileTypes(items) => { + for item in items.iter() { + item.deref(); + } + } + ret::Readdir::Buffers(items) => { + for item in items.iter_mut() { + item.destroy(); + } + } + ret::Readdir::Files(items) => { + for item in items.iter() { + item.deref(); + } + } + } + *self = ret::Readdir::Files(Box::default()); + } } impl FsReturn for StatOrNotFound { #[inline] @@ -1309,7 +1361,8 @@ mod _async_tasks { // KeepAlive::ref_ now takes the type-erased aio EventLoopCtx; the JS // event loop is the only one that owns AsyncFSTask/UVFSRequest. task.r#ref.ref_(bun_io::js_vm_ctx()); - let _ = vm; + // Paired with the `offthread_job_end` in `work_pool_callback`. + vm.event_loop_shared().offthread_job_begin(); task.tracker.did_schedule(global_object); let promise = task.promise.value(); WorkPool::schedule(&raw mut bun_core::heap::release(task).task); @@ -1320,28 +1373,38 @@ mod _async_tasks { // SAFETY: `task` points to `Self.task` (container-of). let this = unsafe { Self::from_task_ptr(task) }; - let mut node_fs = NodeFS::default(); - // SAFETY: `this` is the live Box-leaked task; the work-pool thread owns - // it exclusively until the enqueue below hands it to the JS thread. - // `args` and `result` are disjoint fields. - unsafe { - (*this).result = - NodeFS::dispatch::(&mut node_fs, &(*this).args, Flavor::Async); - } - // `sys::Error::path` is `Box<[u8]>` boxed at the - // `errno_sys_p` construction site, so no clone is needed — `node_fs` may drop. - // `bun_vm_concurrently()` skips the JS-thread debug assert and is the // documented accessor for off-thread (work-pool) callers; the // event-loop's concurrent queue is MPSC-safe. - // SAFETY: `this` is still exclusively owned here (see above). + // SAFETY: `this` is the live Box-leaked task; the work-pool thread owns + // it exclusively until the enqueue below hands it to the JS thread. let vm = unsafe { (*this).global_object().bun_vm_concurrently() }; + + // Teardown in progress: skip the op; the shutdown drain reclaims + // the task unrun with `result` still the sentinel. + // SAFETY: the job's own `outstanding_offthread` count keeps the VM + // alive for this read. + if !unsafe { (*vm).event_loop_shared() }.offthread_cancel_requested() { + let mut node_fs = NodeFS::default(); + // SAFETY: `this` is exclusively owned here (see above); `args` + // and `result` are disjoint fields. + unsafe { + (*this).result = + NodeFS::dispatch::(&mut node_fs, &(*this).args, Flavor::Async); + } + // `sys::Error::path` is `Box<[u8]>` boxed at the + // `errno_sys_p` construction site, so no clone is needed — `node_fs` may drop. + } + // SAFETY: VirtualMachine and its event loop are process-static // (LIFETIMES.tsv); the concurrent queue is MPSC-safe. Ownership of // `this` transfers to the JS thread here — no use after this call. unsafe { (*(*vm).event_loop()).enqueue_task_concurrent(ConcurrentTask::create_from(this)); } + // SAFETY: `vm` stays alive until this end call releases the fence + // taken in `create` (last VM access; `this` is not touched). + unsafe { (*vm).event_loop_shared() }.offthread_job_end(); } pub(crate) fn run_from_js_thread(&mut self) -> Result<(), bun_jsc::JsTerminated> { @@ -1399,6 +1462,20 @@ mod _async_tasks { // `bun_sys::Error` frees its path on Drop. task.r#ref.unref(bun_io::js_vm_ctx()); } + + /// Shutdown-drain release for a completion that never reaches + /// `run_from_js_thread`: free result heap that `fs_to_js` would have + /// transferred to JS, then [`Self::destroy`]. + /// + /// SAFETY: same contract as [`Self::destroy`]. + pub(crate) unsafe fn release_at_shutdown(this: *mut Self) { + // SAFETY: caller contract; the drain owns the queued task. + if let Ok(r) = unsafe { &mut (*this).result } { + r.release_unrun(); + } + // SAFETY: caller contract. + unsafe { Self::destroy(this) }; + } } // ────────────────────────────────────────────────────────────────────────── @@ -1620,6 +1697,9 @@ mod _async_tasks { if !IS_SHELL { task.r#ref.ref_(event_loop_handle_to_ctx(task.evtloop)); } + // Paired with the `offthread_job_end` in `on_subtask_done` (the + // copy tree finishes first by the `subtask_count` contract). + task.evtloop.offthread_job_begin(); task.tracker.did_schedule(global_object); let raw = bun_core::heap::release(task); @@ -1655,6 +1735,8 @@ mod _async_tasks { if !IS_SHELL { task.r#ref.ref_(event_loop_handle_to_ctx(task.evtloop)); } + // See `create_with_shell_task`; no-op for the mini loop. + task.evtloop.offthread_job_begin(); let raw = bun_core::heap::release(task); WorkPool::schedule(&raw mut raw.task); @@ -1720,8 +1802,12 @@ mod _async_tasks { // Count reached zero ⇒ exclusive access. `this` carries mutable // provenance from `Box::leak`, so the enqueued callback may safely // form `&mut *this` on the JS thread. - if matches!(this_ref.evtloop, EventLoopHandle::Js { .. }) { - this_ref.evtloop.enqueue_task_concurrent(EventLoopTaskPtr { + // Copy the handle out first: the JS thread may free `*this` as + // soon as the enqueue lands, and the fence release below must not + // touch it. + let evtloop = this_ref.evtloop; + if matches!(evtloop, EventLoopHandle::Js { .. }) { + evtloop.enqueue_task_concurrent(EventLoopTaskPtr { js: ConcurrentTask::from_callback(this, |p| { // SAFETY: `p` is the `Box::leak`'d task; subtask count hit zero so this // JS-thread callback holds the only live reference (exclusive `&mut`). @@ -1730,7 +1816,7 @@ mod _async_tasks { .as_ptr(), }); } else { - this_ref.evtloop.enqueue_task_concurrent(EventLoopTaskPtr { + evtloop.enqueue_task_concurrent(EventLoopTaskPtr { mini: AnyTaskWithExtraContext::from_callback_auto_deinit( this, |p: *mut Self, ctx| { @@ -1740,6 +1826,8 @@ mod _async_tasks { ), }); } + // Last loop access, via the local copy; no-op for the mini loop. + evtloop.offthread_job_end(); } pub(crate) fn run_from_js_thread_mini(&mut self, _: *mut c_void) { @@ -2394,6 +2482,9 @@ mod _async_tasks { pending_err_mutex: bun_threading::Mutex::default(), }); task.r#ref.ref_(bun_io::js_vm_ctx()); + // Paired with the `offthread_job_end` in `finish_concurrently` (the + // scan tree finishes first by the `subtask_count` contract). + vm.event_loop_shared().offthread_job_begin(); task.tracker.did_schedule(global_object); let promise = task.promise.value(); WorkPool::schedule(&raw mut bun_core::heap::release(task).task); @@ -2579,6 +2670,9 @@ mod _async_tasks { std::ptr::from_mut::(self), ))); } + // SAFETY: `vm` stays alive until this end call (last VM access; + // `self` is not touched). + unsafe { (*vm).event_loop_shared() }.offthread_job_end(); } fn clear_result_list(&mut self) { @@ -2679,6 +2773,15 @@ mod _async_tasks { task.free_root_path(); task.clear_result_list(); } + + /// Shutdown-drain alias of [`Self::destroy`], which already releases + /// the result list (`clear_result_list`). + /// + /// SAFETY: same contract as [`Self::destroy`]. + pub(crate) unsafe fn release_at_shutdown(this: *mut Self) { + // SAFETY: caller contract. + unsafe { Self::destroy(this) }; + } } /// Maps a readdir element type to its `ResultListEntryValue` variant. diff --git a/src/runtime/node/node_fs_stat_watcher.rs b/src/runtime/node/node_fs_stat_watcher.rs index 8898164f74bd..b1eacc8fc9f5 100644 --- a/src/runtime/node/node_fs_stat_watcher.rs +++ b/src/runtime/node/node_fs_stat_watcher.rs @@ -1162,6 +1162,10 @@ impl InitialStatTask { // the task lifetime (balanced by `deref()` in run_owned's closed path or // by the main-thread `initial_stat_*_on_main_thread` callbacks). StatWatcher::ref_(watcher); + // SAFETY: per fn contract. Paired with the `offthread_job_end` on + // every exit of `run_owned`; the watcher's `ctx` BackRef does not keep + // the VM alive, this count does. + unsafe { (*watcher).ctx.event_loop_shared().offthread_job_begin() }; WorkPool::schedule_new(InitialStatTask { watcher, task: WorkPoolTask::default(), @@ -1184,11 +1188,17 @@ impl InitialStatTask { // both also deref as shared (R-2), so aliased `&` is sound. // `ParentRef` Deref gives that shared `&`. let this_ref = ParentRef::from(NonNull::new(this).expect("run_owned: watcher")); + // Copied out so `offthread_job_end` below is the job's only VM access + // after the enqueue (or the deref) may hand the watcher away. + let event_loop = this_ref.ctx.event_loop_shared(); if this_ref.closed.load(Ordering::Relaxed) { // Balance the ref() from createAndSchedule(). // SAFETY: `this` is live (ref'd in `create_and_schedule`); we own that ref. StatWatcher::deref(this); + // Releases the count taken in `create_and_schedule` (last VM + // access, via the local). + event_loop.offthread_job_end(); return; } @@ -1216,6 +1226,9 @@ impl InitialStatTask { // ref ownership transferred to main-thread callback // (`initial_stat_*_on_main_thread` calls deref()). Nothing to forget — // `watcher` is a raw pointer. + // Releases the count taken in `create_and_schedule` (last VM access, + // via the local). + event_loop.offthread_job_end(); } } diff --git a/src/runtime/node/node_zlib_binding.rs b/src/runtime/node/node_zlib_binding.rs index e397105e942d..d716fe21156f 100644 --- a/src/runtime/node/node_zlib_binding.rs +++ b/src/runtime/node/node_zlib_binding.rs @@ -464,6 +464,10 @@ impl CompressionStream { callback: Self::async_job_run_task, }); this.poll_ref().with_mut(|p| p.ref_(vm)); + // Hold the worker-shutdown barrier open until the pool thread has + // finished `do_work()` and posted the completion; matched by + // `offthread_job_end()` at the end of `async_job_run`. + vm.event_loop_shared().offthread_job_begin(); WorkPool::schedule(this.task().as_ptr()); Ok(JSValue::UNDEFINED) @@ -496,16 +500,28 @@ impl CompressionStream { NonNull::new(global_this.bun_vm_concurrently()).expect("bun_vm_concurrently"), ); - this_ref.stream().with_mut(|s| s.do_work()); + // Skip the possibly multi-second compression block on teardown; the + // completion enqueued below is released by the shutdown drain. + if !vm.event_loop_shared().offthread_cancel_requested() { + this_ref.stream().with_mut(|s| s.do_work()); + } // SAFETY: `event_loop()` is a self-pointer into a live VM; the // `enqueue_task_concurrent` body only touches the lock-free // `concurrent_tasks` queue (thread-safe). `this` is the heap-allocated // `m_ctx` payload — the matching `ref()` in `write()` keeps it alive - // until `run_from_js_thread` runs and calls `deref()`. + // until `run_from_js_thread` runs and calls `deref()`. Liveness of the + // VM across a worker terminate is guaranteed by the + // `offthread_job_begin()` taken in `write()`: `WebWorker::shutdown` + // blocks on that count reaching zero before freeing the JSC heap or + // the VM box, so `global_this`, `vm` and `event_loop` are all still + // valid here. unsafe { (*vm.event_loop()).enqueue_task_concurrent(ConcurrentTask::create(Task::init(this))); } + // Last VM access; pairs with `offthread_job_begin()` in `write()` and + // releases `WebWorker::shutdown`'s barrier. + vm.event_loop_shared().offthread_job_end(); } /// Dispatched from `dispatch.rs` when the worker-thread `do_work()` posts @@ -587,6 +603,44 @@ impl CompressionStream { unsafe { T::deref(this_ptr) }; } + /// Shutdown-drain counterpart of [`Self::run_from_js_thread`]: releases the + /// resources `write()` acquired (Strong `this_value`, pinned input/output + /// buffers, `poll_ref`, the `ref_()` +1) without invoking JS callbacks. + /// Called from `__bun_release_task_at_shutdown` for a completion that + /// reached the queue after the worker thread stopped ticking. Runs on the + /// worker's JS thread with the JSC heap and VM still live (before + /// `teardownJSCVM`). + /// + /// SAFETY: same contract as [`Self::run_from_js_thread`]. + pub(crate) unsafe fn release_unrun(this_ptr: *mut T) { + let this = ParentRef::from(NonNull::new(this_ptr).expect("release_unrun: this")); + let global: &JSGlobalObject = this.global_this(); + let vm = global.bun_vm(); + + this.write_in_progress().set(false); + + if let Some(this_value) = this.this_value().with_mut(|v| v.try_swap()) { + for pinned in [ + T::pending_input_get_cached(this_value), + T::pending_output_get_cached(this_value), + ] + .into_iter() + .flatten() + { + if pinned.is_cell() { + if let Some(buf) = pinned.as_array_buffer(global) { + buf.unpin(); + } + } + } + } + + this.poll_ref().with_mut(|p| p.unref(vm)); + // SAFETY: matching `ref_()` in `write()`; `this_ptr` is the heap payload + // and is not accessed after this call. + unsafe { T::deref(this_ptr) }; + } + pub(crate) fn write_sync( this: &T, global_this: &JSGlobalObject, diff --git a/src/runtime/shell/builtin/cp.rs b/src/runtime/shell/builtin/cp.rs index 20c43807cd72..86f9b66685e8 100644 --- a/src/runtime/shell/builtin/cp.rs +++ b/src/runtime/shell/builtin/cp.rs @@ -512,6 +512,13 @@ impl ShellCpTask { let st = &raw mut (*this).task; (*st).task.callback = Self::work_pool_callback; (*st).keep_alive.ref_((*st).event_loop.as_event_loop_ctx()); + // Released at the end of `work_pool_callback` on the pool thread + // (the success continuation is covered by `AsyncCpTask`'s own + // fence, and its completion only runs once the JS thread ticks — + // which the shutdown wait never does). This custom scheduler + // bypasses `ShellTask::schedule`, which would otherwise pair with + // `on_finish`'s release. + (*st).event_loop.offthread_job_begin(); WorkPool::schedule(&raw mut (*st).task); } } @@ -529,15 +536,25 @@ impl ShellCpTask { task, ::TASK_OFFSET, ); + // Copy the handle out first: on success `AsyncCpTask` may complete + // and free `*this` (via `cp_on_finish`) before the end call below. + let event_loop = (*this).task.event_loop; if let Some(e) = (*this).run_from_thread_pool_impl() { (*this).err = Some(e); Self::enqueue_to_event_loop(this); } + // Releases the fence taken in `schedule`. On success the + // `AsyncCpTask` created by `run_from_thread_pool_impl` already + // took its own count, so the chain stays covered. + event_loop.offthread_job_end(); } } /// Post this task to the main-thread /// concurrent queue; routed by `dispatch.rs` → [`run_from_main_thread`]. + /// Fence-neutral: the task's count is released at the end of + /// `work_pool_callback`, and the success path calls this from the JS + /// thread (`cp_on_finish`), where no count is held. /// /// # Safety /// `this` is the live `heap::alloc`'d task; not touched again on this @@ -545,7 +562,7 @@ impl ShellCpTask { unsafe fn enqueue_to_event_loop(this: *mut ShellCpTask) { // Reuse the generic `ShellTask` post-back. // SAFETY: caller contract. - unsafe { ShellTask::on_finish::(this) }; + let _ = unsafe { ShellTask::on_finish_no_fence::(this) }; } fn has_trailing_sep(path: &[u8]) -> bool { diff --git a/src/runtime/shell/builtin/rm.rs b/src/runtime/shell/builtin/rm.rs index 4952edb51796..cf15e95d393b 100644 --- a/src/runtime/shell/builtin/rm.rs +++ b/src/runtime/shell/builtin/rm.rs @@ -711,6 +711,10 @@ impl ShellRmTask { let st = &raw mut (*this).task; (*st).task.callback = Self::work_pool_callback; (*st).keep_alive.ref_((*st).event_loop.as_event_loop_ctx()); + // Paired with the `offthread_job_end` in `ShellTask::on_finish` + // (reached via `finish_concurrently`); this custom scheduler + // bypasses `ShellTask::schedule`, which would otherwise take it. + (*st).event_loop.offthread_job_begin(); WorkPool::schedule(&raw mut (*st).task); } } @@ -1389,6 +1393,11 @@ impl DirTask { let will_queue_verbose = tm.opts.verbose && !me.deleted_entries.is_empty(); if will_queue_verbose { tm.pending_main_callbacks.fetch_add(1, Ordering::SeqCst); + // The verbose post's enqueue is a VM access that can run + // AFTER the root completion releases the fence taken in + // `schedule` (both arms below queue after cascading), so + // it carries its own count, released in `queue_for_write`. + tm.event_loop.offthread_job_begin(); } // If we have a parent and we are the last child, now we can delete the parent. @@ -1483,10 +1492,14 @@ impl DirTask { // dropping the ShellRmTask drops the root DirTask, so for the // root `me` may dangle immediately after. let (tm, has_parent) = (me.task_manager, !me.parent_task.is_null()); + let event_loop = (*tm).event_loop; if has_parent { Self::deinit(this); } ShellRmTask::decr_pending_and_maybe_deinit(tm); + // Releases the count taken with the pending bump in + // `post_run` (last loop access, via the local). + event_loop.offthread_job_end(); return; } let event_loop = (*me.task_manager).event_loop; @@ -1504,6 +1517,9 @@ impl DirTask { }, }; event_loop.enqueue_task_concurrent(task_ptr); + // Releases the count taken with the pending bump in `post_run` (last + // loop access, via the local — the main thread may free the task now). + event_loop.offthread_job_end(); } /// Flush verbose output. diff --git a/src/runtime/shell/interpreter.rs b/src/runtime/shell/interpreter.rs index 4e925f76cfb2..650c86fca329 100644 --- a/src/runtime/shell/interpreter.rs +++ b/src/runtime/shell/interpreter.rs @@ -2628,6 +2628,10 @@ impl ShellTask { // `&mut ShellTask` across that call. unsafe { let this = ctx.byte_add(C::TASK_OFFSET).cast::(); + // Paired with the `offthread_job_end` in `on_finish`. Recursive + // (pool-thread) schedules happen under the parent task's count, so + // the count never dips to zero mid-chain. No-op for a mini loop. + (*this).event_loop.offthread_job_begin(); (*this).task.callback = shell_task_trampoline::; WorkPool::schedule(&raw mut (*this).task); } @@ -2644,6 +2648,20 @@ impl ShellTask { /// [`schedule`](Self::schedule); not touched again on the worker thread /// after this returns. pub(crate) unsafe fn on_finish(ctx: *mut C) { + // SAFETY: caller contract. + let event_loop = unsafe { Self::on_finish_no_fence::(ctx) }; + // Last loop access, via the local copy (the main thread may free the task now). + event_loop.offthread_job_end(); + } + + /// [`Self::on_finish`] without the trailing `offthread_job_end`, for + /// callers whose fence was already released on the off thread (the cp + /// builtin, whose success completion is posted from the JS thread). + /// Returns the loop handle copied out before the enqueue. + /// + /// # Safety + /// Same contract as [`Self::on_finish`]. + pub(crate) unsafe fn on_finish_no_fence(ctx: *mut C) -> EventLoopHandle { use bun_event_loop::{ConcurrentTask::AutoDeinit, EventLoopTask, EventLoopTaskPtr}; log!("ShellTask onFinish"); // SAFETY: caller contract — `ctx` embeds `ShellTask` at `TASK_OFFSET`. @@ -2672,6 +2690,7 @@ impl ShellTask { (event_loop, task_ptr) }; event_loop.enqueue_task_concurrent(task_ptr); + event_loop } /// Unrefs the diff --git a/src/runtime/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index f477af732475..31149150ef2d 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -502,6 +502,17 @@ impl FetchTasklet { // SAFETY: caller contract — `this` is live with ref_count == 0. unsafe { (*this).ref_count.assert_no_refs() }; + // Runs on the JS thread (the only caller thread); drop the + // terminate-cancel registration taken in `queue()`. No-op when the + // shutdown fan-out already consumed the list. + // SAFETY: caller contract — `this` is live; `javascript_vm` outlives it. + unsafe { + (*this) + .javascript_vm + .as_mut() + .unregister_terminate_cancel_hook(this.cast()); + } + // SAFETY: this was allocated via heap::alloc in `get()`; ref_count == 0 so exclusive let mut boxed = unsafe { bun_core::heap::take(this) }; boxed.clear_data(); @@ -518,19 +529,29 @@ 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: that drain runs on the main thread at process exit, long + /// after the worker's JSC heap is gone, so a parked `deinit()` would walk + /// freed Strong/Weak storage. Leak the box instead (the caller already + /// released the large buffers). + /// /// 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; the VM is readable here because the + // tasklet's `outstanding_offthread` count is released only after the + // HTTP-thread callback that dropped this last ref returns. + if unsafe { (*this).javascript_vm.is_main_thread } { + http::defer_shutdown_reclaim(this.cast(), FetchTasklet::deinit_erased); + } } unsafe fn deinit_erased(this: *mut c_void) { @@ -561,6 +582,12 @@ impl FetchTasklet { /// `callback` and the JS-thread `on_progress_update`; the JS thread is /// parked in `wait_timeout_while` here, so the load is race-free. /// + /// The `offthread_job_begin` taken in `queue()` is deliberately not + /// released here: this path runs only at process exit, and the fence is + /// awaited only by `WebWorker::shutdown` — whose own fan-out already + /// produced the final `callback` (or timed out and leaked the VM) before + /// the HTTP daemon parks. + /// /// SAFETY: `this` is the live `*mut FetchTasklet` registered as /// `result_callback.ctx` in `get()`; HTTP-thread-only at this point. unsafe fn release_at_shutdown(this: *mut ()) { @@ -2394,11 +2421,27 @@ impl FetchTasklet { // increment ref so we can keep it alive until the http client is done node_ref.ref_(); + // Worker-shutdown fence: held until the HTTP thread's final callback + // (`is_done`) has dropped its ref — see the `offthread_job_end` calls + // in `callback`. Registering the abort lets `WebWorker::shutdown` + // bound that wait by a socket shutdown instead of the transfer. + let vm = node_ref.javascript_vm; + vm.event_loop_shared().offthread_job_begin(); + vm.as_mut() + .register_terminate_cancel_hook(node.cast(), 0, Self::terminate_cancel_hook); http::HTTPThread::schedule(batch); Ok(node) } + /// `TerminateCancelHook::run` for an in-flight fetch: identical to an + /// AbortSignal firing during shutdown. Idempotent (`abort_task` swaps the + /// `aborted` atomic). Runs on the worker's JS thread with the tasklet + /// still registered, hence alive. + fn terminate_cancel_hook(ptr: *mut (), _data: u64) { + Self::from_raw_mut(ptr.cast::()).abort_task(); + } + /// Called from HTTP thread. Handles HTTP events received from socket. /// /// # Safety @@ -2416,6 +2459,11 @@ impl FetchTasklet { // at this point only this thread is accessing result to is no race condition let is_done = !result.has_more; let task_ref = Self::from_raw_mut(task); + // Snapshot for the `offthread_job_end` calls below: on every `is_done` + // exit the trailing `deref_from_thread` may free `*task`, and the end + // call releases `WebWorker::shutdown`'s fence, after which the VM + // itself may be freed — so it must go through this local, last. + let vm = task_ref.javascript_vm; task_ref.mutex.lock(); // we need to unlock before task.deref(); @@ -2489,6 +2537,8 @@ impl FetchTasklet { if is_done { // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held. FetchTasklet::deref_from_thread(task); + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); } return; } @@ -2541,6 +2591,8 @@ impl FetchTasklet { if is_done { // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held. FetchTasklet::deref_from_thread(task); + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); } return; } @@ -2575,6 +2627,8 @@ impl FetchTasklet { FetchTasklet::deref_from_thread(task); // SAFETY: second ref still held until this 1→0 transition. FetchTasklet::deref_from_thread(task); + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); } return; } @@ -2593,6 +2647,8 @@ impl FetchTasklet { if is_done { // SAFETY: `task` is the live heap tasklet; HTTP-thread ref held. FetchTasklet::deref_from_thread(task); + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); } } } diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs index 74d062c5d95a..366716a058aa 100644 --- a/src/runtime/webcore/s3/client.rs +++ b/src/runtime/webcore/s3/client.rs @@ -307,12 +307,18 @@ pub(crate) fn list_objects( concurrent_task: Default::default(), proxy_url: Box::default(), body: Box::default(), + signal_store: Default::default(), + signals: Default::default(), poll_ref: bun_io::KeepAlive::init(), })); // SAFETY: just allocated, non-null let task = unsafe { &mut *task_ptr }; task.poll_ref.ref_(bun_io::js_vm_ctx()); + // Wiring `aborted` gives the request a real `async_http_id` and an + // abort-tracker entry; the store lives in the heap task, so the derived + // pointers stay valid for the request's lifetime. + task.signals = task.signal_store.to(); let proxy = proxy_url.unwrap_or(b""); task.proxy_url = if !proxy.is_empty() { @@ -357,6 +363,7 @@ pub(crate) fn list_objects( bun_http::async_http::Options { http_proxy, verbose: Some(vm.get_verbose_fetch()), + signals: Some(task.signals), reject_unauthorized: Some(vm.get_tls_reject_unauthorized()), ..Default::default() }, @@ -366,7 +373,18 @@ pub(crate) fn list_objects( bun_http::http_thread::init(&Default::default()); let mut batch = bun_threading::thread_pool::Batch::default(); // SAFETY: `http` was initialised by `task.http.write(...)` immediately above. + let async_http_id = unsafe { task.http.assume_init_ref() }.async_http_id; + // SAFETY: as above. unsafe { task.http.assume_init_mut() }.schedule(&mut batch); + // Worker-shutdown fence, mirroring `execute_simple_s3_request`: held until + // the final `http_callback`'s `offthread_job_end`. The hook carries the id + // because the HTTP thread overwrites the on-task `http` storage. + vm.event_loop_shared().offthread_job_begin(); + vm.as_mut().register_terminate_cancel_hook( + task_ptr.cast(), + u64::from(async_http_id), + crate::webcore::s3::simple_request::terminate_cancel_hook, + ); bun_http::HTTPThread::schedule(batch); Ok(()) } @@ -1304,6 +1322,17 @@ fn download_stream( bun_http::http_thread::init(&Default::default()); let mut batch = bun_threading::thread_pool::Batch::default(); http.schedule(&mut batch); + // Worker-shutdown fence: held until the HTTP thread's final + // (`has_more == false`) callback — see `offthread_job_end` in + // `S3HttpDownloadStreamingTask::http_callback`. The cancel hook lets + // `WebWorker::shutdown` bound that wait by a socket shutdown; it carries + // the id (not the task) because `http` is overwritten on the HTTP thread. + vm.event_loop_shared().offthread_job_begin(); + vm.as_mut().register_terminate_cancel_hook( + task_ptr.cast(), + u64::from(task.async_http_id), + crate::webcore::s3::download_stream::terminate_cancel_hook, + ); bun_http::HTTPThread::schedule(batch); task_ptr } diff --git a/src/runtime/webcore/s3/download_stream.rs b/src/runtime/webcore/s3/download_stream.rs index 4fa8977496b7..16910908f88f 100644 --- a/src/runtime/webcore/s3/download_stream.rs +++ b/src/runtime/webcore/s3/download_stream.rs @@ -324,6 +324,12 @@ impl S3HttpDownloadStreamingTask { async_http: *mut AsyncHTTP<'static>, result: HTTPClientResult, ) { + let is_done = !result.has_more; + // Snapshot before the enqueue: once the final completion lands the JS + // thread may consume and free `*this`, and the `offthread_job_end` + // below must go through a local. + // SAFETY: `this` is live for the duration of this callback. + let vm = unsafe { (*this).vm.expect("vm set at task creation") }; // SAFETY: `this` is live for the duration of the HTTP request; HTTPThread holds the only // concurrent reference and `mutex` serializes against `on_response`. `async_http` is the // live HTTP-thread copy, non-null for the callback's duration. Borrows scoped to the call. @@ -331,21 +337,48 @@ impl S3HttpDownloadStreamingTask { // we are always unlocked here and its safe to enqueue // SAFETY: same exclusivity as above; `task` is the inline `concurrent_task` field of // this heap request and the queue takes ownership of its `next` link. - let (vm, task) = unsafe { - let task = core::ptr::NonNull::from( + let task = unsafe { + core::ptr::NonNull::from( (*this).concurrent_task.from(this, AutoDeinit::ManualDeinit), - ); - ((*this).vm.expect("vm set at task creation"), task) + ) }; // `vm` is the live per-thread VM BackRef captured at task creation; event_loop // is initialized for the request's lifetime and enqueue is thread-safe (`&self`). vm.event_loop_shared().enqueue_task_concurrent(task); } + if is_done { + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); + } } } +/// `TerminateCancelHook::run` for an in-flight streaming S3 request: set the +/// task's abort signal (fails the request fast if it has not started yet), +/// then ask the HTTP thread to shut the socket down by id; either path +/// produces the final (`has_more == false`) callback that releases the fence. +/// Only the signal store and the id captured at schedule time are touched; +/// the `http` storage is HTTP-thread-owned here. +pub(crate) fn terminate_cancel_hook(ptr: *mut (), async_http_id: u64) { + // SAFETY: `ptr` is the live heap task registered at schedule time; hooks + // and the task's free both run on the owning JS thread, so no race. + unsafe { &(*ptr.cast::()).signal_store } + .aborted + .store(true, core::sync::atomic::Ordering::Release); + #[allow(clippy::cast_possible_truncation)] + bun_http::http_thread().schedule_shutdown_by_id(async_http_id as u32); +} + impl Drop for S3HttpDownloadStreamingTask { fn drop(&mut self) { + // Runs on the JS thread (the task is freed by `on_response` / the + // stream-cancel path); drop the terminate-cancel registration taken at + // schedule time. No-op when the shutdown fan-out consumed the list or + // the inert `Default` placeholder is being dropped (`vm: None`). + if let Some(vm) = self.vm { + vm.as_mut() + .unregister_terminate_cancel_hook(core::ptr::from_mut(self).cast()); + } // KeepAlive::unref now takes an aio EventLoopCtx; the JS-loop ctx is fetched // via the global hook (registered by crate::init) — same pattern as // `S3HttpSimpleTask::drop` in simple_request.rs. diff --git a/src/runtime/webcore/s3/simple_request.rs b/src/runtime/webcore/s3/simple_request.rs index 9f01d31a3df3..68e48e62e6a9 100644 --- a/src/runtime/webcore/s3/simple_request.rs +++ b/src/runtime/webcore/s3/simple_request.rs @@ -7,7 +7,7 @@ use bun_event_loop::{TaskTag, Taskable, task_tag}; use bun_http::async_http::Options as HttpOptions; use bun_http::{ AsyncHTTP, FetchRedirect, HTTPClientResult, HTTPClientResultCallback, Headers, HeadersExt, - Method, + Method, Signals, }; use bun_io::KeepAlive; use bun_jsc::virtual_machine::VirtualMachine; @@ -133,6 +133,11 @@ pub struct S3HttpSimpleTask { /// concurrently for the lifetime of the request, so the task owns its own /// copy instead of borrowing caller memory. pub(crate) body: Box<[u8]>, + /// Backing store for `signals`. Wiring `aborted` gives the request a real + /// `async_http_id` and an abort-tracker entry, which is what lets + /// [`terminate_cancel_hook`] actually cancel it. + pub(crate) signal_store: bun_http::signals::Store, + pub(crate) signals: Signals, pub poll_ref: KeepAlive, } @@ -160,6 +165,8 @@ impl Default for S3HttpSimpleTask { concurrent_task: ConcurrentTask::default(), proxy_url: Box::default(), body: Box::default(), + signal_store: bun_http::signals::Store::default(), + signals: Signals::default(), poll_ref: KeepAlive::default(), } } @@ -471,12 +478,22 @@ impl S3HttpSimpleTask { ((*this).vm.expect("vm set at task creation"), queued) }; vm.event_loop_shared().enqueue_task_concurrent(queued); + // Final callback: last VM access, via the local. + vm.event_loop_shared().offthread_job_end(); } } } impl Drop for S3HttpSimpleTask { fn drop(&mut self) { + // Runs on the JS thread (the task is consumed by `on_response`); drop + // the terminate-cancel registration taken in + // `execute_simple_s3_request`. No-op when the shutdown fan-out + // consumed the list. + if let Some(vm) = self.vm { + vm.as_mut() + .unregister_terminate_cancel_hook(core::ptr::from_mut(self).cast()); + } // Side effects beyond freeing owned fields (which Rust drops automatically): // - poll_ref.unref(vm) // - http.clearData() @@ -501,6 +518,22 @@ impl Drop for S3HttpSimpleTask { } } +/// `TerminateCancelHook::run` for an in-flight simple S3 request: set the +/// task's abort signal (fails the request fast if it has not started yet), +/// then ask the HTTP thread to shut the socket down by id; either path +/// produces the final callback that releases the fence. Only the signal store +/// and the id captured at schedule time are touched; the `http` storage is +/// HTTP-thread-owned here. +pub(crate) fn terminate_cancel_hook(ptr: *mut (), async_http_id: u64) { + // SAFETY: `ptr` is the live heap task registered at schedule time; hooks + // and the task's free both run on the owning JS thread, so no race. + unsafe { &(*ptr.cast::()).signal_store } + .aborted + .store(true, core::sync::atomic::Ordering::Release); + #[allow(clippy::cast_possible_truncation)] + bun_http::http_thread().schedule_shutdown_by_id(async_http_id as u32); +} + // callers in `client.rs` / `multipart.rs` were translated with three different // names for the request-options struct (`Options`, `S3RequestOptions`, `S3SimpleRequestOptions`) // and two for the callback enum. Alias them here so the call sites compile without churn. @@ -624,8 +657,14 @@ pub(crate) fn execute_simple_s3_request( Box::default() }, body: Box::<[u8]>::from(options.body), + signal_store: bun_http::signals::Store::default(), + signals: Signals::default(), poll_ref, }); + // SAFETY: `task_ptr` is freshly heap-allocated and not yet shared; scoped + // exclusive write. The store lives in the heap task, so the derived + // pointers stay valid for the request's lifetime. + unsafe { (*task_ptr).signals = (*task_ptr).signal_store.to() }; // SAFETY: `task_ptr` is a freshly heap-allocated pointer; shared reads only until // the scoped exclusive `http` writes below. let task = unsafe { &*task_ptr }; @@ -670,6 +709,7 @@ pub(crate) fn execute_simple_s3_request( HttpOptions { http_proxy, verbose: Some(verbose), + signals: Some(task.signals), reject_unauthorized: Some(reject_unauthorized), ..Default::default() }, @@ -681,7 +721,19 @@ pub(crate) fn execute_simple_s3_request( bun_http::http_thread::init(&Default::default()); let mut batch = thread_pool::Batch::default(); // SAFETY: `http` was initialised immediately above; scoped exclusive access. + let async_http_id = unsafe { (*task_ptr).http.assume_init_mut() }.async_http_id; + // SAFETY: as above. unsafe { (*task_ptr).http.assume_init_mut() }.schedule(&mut batch); + // Worker-shutdown fence: held until the HTTP thread's final + // (`has_more == false`) callback — see `offthread_job_end` in + // `http_callback`. The cancel hook carries the id (not the task) because + // the HTTP thread overwrites the on-task `http` storage concurrently. + vm.event_loop_shared().offthread_job_begin(); + vm.as_mut().register_terminate_cancel_hook( + task_ptr.cast(), + u64::from(async_http_id), + terminate_cancel_hook, + ); bun_http::HTTPThread::schedule(batch); Ok(()) } diff --git a/test/js/node/zlib/zlib-worker-terminate.test.ts b/test/js/node/zlib/zlib-worker-terminate.test.ts new file mode 100644 index 000000000000..aa94efaba38c --- /dev/null +++ b/test/js/node/zlib/zlib-worker-terminate.test.ts @@ -0,0 +1,90 @@ +import { expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, isDebug } from "harness"; + +// A terminated worker strands some allocations by design (requeued task +// boxes), so the subprocess runs with leak detection off; the oracle here is +// the UAF abort, which is unaffected. +const env = { + ...bunEnv, + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=0"].filter(Boolean).join(":"), +}; + +// Same tolerated pre-existing terminate() abort as +// worker-terminate-offthread.test.ts: strips exactly the known +// assertNoException signature and returns whatever else stderr contained. +function onlyKnownTerminateAssert(stderr: string): string[] { + const lines = stderr.split("\n").filter(line => line !== ""); + if (!stderr.includes("!exception()")) return lines; + return lines.filter( + line => + line !== "ASSERTION FAILED: (null)" && + line !== "!exception()" && + !(line.includes("ExceptionScope.h") && line.includes("assertNoException")) && + !line.includes("no stacktrace available"), + ); +} + +// worker.terminate() while async node:zlib compression is in flight on the +// thread pool must not dereference the worker's freed VM/EventLoop from the +// pool-thread completion. One lane per Native* tag (zlib/brotli/zstd) keeps +// do_work() busy so terminate reliably lands mid-compression. +test("worker.terminate() during in-flight node:zlib async compression does not UAF", async () => { + const ROUNDS = isASAN || isDebug ? 4 : 10; + + const script = /* js */ ` + const { Worker } = require("node:worker_threads"); + const src = \` + const { parentPort } = require("node:worker_threads"); + const zlib = require("node:zlib"); + const { promisify } = require("node:util"); + const gz = promisify(zlib.gzip); + const br = promisify(zlib.brotliCompress); + const df = promisify(zlib.deflate); + const zs = promisify(zlib.zstdCompress); + const big = Buffer.alloc(16 << 20, 0x61); + const lanes = (n, f) => { + for (let i = 0; i < n; i++) + (async () => { for (;;) { try { await f(); } catch {} } })(); + }; + lanes(2, () => gz(big)); + lanes(1, () => br(big.subarray(0, 4 << 20))); + lanes(2, () => df(big.subarray(0, 10 << 20))); + lanes(1, () => zs(big.subarray(0, 4 << 20))); + parentPort.postMessage("up"); + \`; + (async () => { + for (let r = 0; r < ${ROUNDS}; r++) { + const w = new Worker(src, { eval: true }); + await new Promise((resolve, reject) => { + w.once("message", resolve); + w.once("error", reject); + w.once("exit", code => reject(new Error("worker exited " + code + " before ready"))); + }); + await w.terminate(); + } + console.log("ok"); + })().catch(e => { + console.error(e); + process.exit(1); + }); + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).not.toContain("AddressSanitizer"); + if (stderr.includes("assertNoException")) { + expect(onlyKnownTerminateAssert(stderr)).toEqual([]); + } else { + expect(stderr).toBe(""); + expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "ok", exitCode: 0 }); + } + // Worker startup under debug+ASAN is ~1.8s on its own; 4 rounds cannot fit + // the 5s default. Shrinking the buffers to fit loses the race window (0/3 + // repro on the unfixed build at 4 MiB), so the workload stays as-is. +}, 30_000); diff --git a/test/js/web/workers/worker-terminate-offthread.test.ts b/test/js/web/workers/worker-terminate-offthread.test.ts new file mode 100644 index 000000000000..2df7786803fd --- /dev/null +++ b/test/js/web/workers/worker-terminate-offthread.test.ts @@ -0,0 +1,365 @@ +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, tempDir } from "harness"; + +// Worker spawn loops under a debug+ASAN build are slow (worker startup alone +// is ~2s); every test here is a rare, deliberate outlier. +const TIMEOUT = 90_000; + +// This file's oracle is use-after-free, not leak-freedom: a terminated +// worker still strands some allocations by design (requeued task boxes, the +// deliberate fetch-tasklet box leak) and by known pre-existing bugs, so the +// subprocesses run with leak detection off. UAF reports are unaffected. +// Proxy vars cleared because the S3 client does not honor NO_PROXY; an +// inherited proxy would hijack the S3 lane's requests to its local stub. +const env = { + ...bunEnv, + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=0"].filter(Boolean).join(":"), + HTTP_PROXY: undefined, + HTTPS_PROXY: undefined, + http_proxy: undefined, + https_proxy: undefined, +}; + +// Lines of the one tolerated crash signature (see the assertNoException +// comments below); returns whatever else stderr contained. The filter only +// engages when the exact known condition is present, so a different +// assertion (even in the same file) fails the test. +function onlyKnownTerminateAssert(stderr: string): string[] { + const lines = stderr.split("\n").filter(line => line !== ""); + if (!stderr.includes("!exception()")) return lines; + return lines.filter( + line => + line !== "ASSERTION FAILED: (null)" && + line !== "!exception()" && + !(line.includes("ExceptionScope.h") && line.includes("assertNoException")) && + !line.includes("no stacktrace available"), + ); +} + +// Worker teardown must wait for every job the worker's VM handed to another +// thread (WorkPool bodies, webcrypto's phony work queue, the bundler thread) +// before freeing the VM box, its EventLoop, and the JSC heap. Each family +// below keeps lanes of one off-thread job type in flight while the worker is +// torn down at a jittered offset; on an unfixed build the pool thread's +// completion post (or its read of a JSC-heap-backed buffer) is a +// use-after-free that aborts the subprocess under ASAN. +// +// ASAN-gated: the dangling reads are small and release builds can survive +// them silently, so only the sanitizer build proves anything. +describe.skipIf(!isASAN)("worker teardown with off-thread jobs in flight does not UAF", () => { + const ROUNDS = 3; + + // Each family body runs inside the worker. `lanes(n, f)` keeps n lanes of + // the job type permanently in flight; `buf()` makes a JSC-heap-backed + // buffer big enough that the off-thread body is still touching it when + // teardown lands. + const families: Record = { + "crypto.pbkdf2 (AnyTaskJob)": ` + const { pbkdf2 } = require("node:crypto"); + lanes(3, () => new Promise((res, rej) => pbkdf2("pw", "salt", 150000, 64, "sha512", e => e ? rej(e) : res())));`, + "crypto.scrypt (AnyTaskJob)": ` + const { scrypt } = require("node:crypto"); + lanes(2, () => new Promise((res, rej) => scrypt("pw", "salt", 64, { N: 16384, r: 8, p: 1 }, e => e ? rej(e) : res())));`, + "Bun.password.hash (PasswordJob)": ` + lanes(2, () => Bun.password.hash("hunter2", { algorithm: "argon2id", timeCost: 3, memoryCost: 8192 }));`, + "Bun.zstdCompress (AnyTaskJob)": ` + const big = buf(); + lanes(2, () => Bun.zstdCompress(big));`, + // High-entropy input is the slow shape at brotli's default q11: one + // metablock compresses inside a single multi-second native call, so + // teardown lands mid-call instead of between cheap blocks. + "zlib.brotliCompress q11 high-entropy (NativeBrotli)": ` + const zlib = require("node:zlib"); + const rnd = Buffer.alloc(128 << 10); + for (let i = 0; i < rnd.length; i += 4) rnd.writeUInt32LE(Math.imul(i | 1, 2654435761) >>> 0, i); + lanes(2, () => new Promise((res, rej) => zlib.brotliCompress(rnd, e => e ? rej(e) : res())));`, + "Bun.Glob.scan (ConcurrentPromiseTask)": ` + lanes(2, async () => { + const g = new Bun.Glob("**/*"); + for await (const _ of g.scan({ cwd: d.dir })) {} + });`, + "Bun.Transpiler.transform (ConcurrentPromiseTask)": ` + const t = new Bun.Transpiler({ loader: "tsx" }); + const src = 'export function C(){ return
{"x".repeat(3)}
}\\n'.repeat(1500); + lanes(2, () => t.transform(src));`, + "fs.promises read/write (AsyncFSTask)": ` + const fs = require("node:fs/promises"); + const big = Buffer.alloc(12 << 20, 0x61); + lanes(2, async () => { + await fs.writeFile(d.dir + "/blob.bin", big); + await fs.readFile(d.dir + "/blob.bin"); + });`, + "fs.promises.readdir recursive (AsyncReaddirRecursiveTask)": ` + const fs = require("node:fs/promises"); + lanes(2, () => fs.readdir(d.dir, { recursive: true }));`, + "fs.promises.readdir result modes (AsyncFSTask Readdir)": ` + const fs = require("node:fs/promises"); + lanes(1, () => fs.readdir(d.dir + "/tree/a")); + lanes(1, () => fs.readdir(d.dir + "/tree/a", { withFileTypes: true })); + lanes(1, () => fs.readdir(d.dir + "/tree/a", { encoding: "buffer" }));`, + "fs.promises.cp recursive (AsyncCpTask)": ` + const fs = require("node:fs/promises"); + let i = 0; + lanes(2, () => fs.cp(d.dir + "/tree", d.dir + "/copy" + (i++ % 4), { recursive: true, force: true }));`, + "Bun.$ shell builtins (ShellTask + custom rm/cp schedulers)": ` + let i = 0; + lanes(2, async () => { + const n = "sh" + (i++ % 4); + // -v exercises the verbose DirTask posts, which carry their own + // fence counts. + await Bun.$\`mkdir -p \${n}/a/b && cp -R \${n} \${n}c && rm -rfv \${n} \${n}c\`.cwd(d.dir).quiet(); + });`, + "crypto.subtle.digest (ConcurrentCppTask)": ` + const data = buf(); + lanes(3, () => crypto.subtle.digest("SHA-512", data));`, + "Bun.build (JSBundleCompletionTask)": ` + lanes(1, () => Bun.build({ entrypoints: [d.dir + "/entry.ts"], target: "bun", write: false, logLevel: "silent" }).catch(() => {}));`, + "Bun.Archive.blob (Archive AsyncTask)": ` + const archive = new Bun.Archive({ "a.bin": buf(), "b/b.txt": "hello" }); + lanes(2, () => archive.blob());`, + "Bun.write file-to-file (CopyFilePromiseTask)": ` + const fs = require("node:fs/promises"); + let ci = 0; + const setup = fs.writeFile(d.dir + "/copysrc.bin", Buffer.alloc(8 << 20, 0x62)); + lanes(2, async () => { + await setup; + ci++; + await Bun.write(Bun.file(d.dir + "/copydst" + (ci % 4) + ".bin"), Bun.file(d.dir + "/copysrc.bin")); + });`, + "Bun.Image pipeline (AsyncImageTask)": ` + const png = Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==", "base64"); + lanes(2, () => new Bun.Image(png).resize(256, 256).png().bytes());`, + // list() goes through list_objects' own S3HttpSimpleTask construction + // site, not execute_simple_s3_request; the lane pins that site's fence + // (an unbalanced counter here stalls every teardown into the 10s + // deadline, which the dt guard below catches deterministically). + "Bun.S3Client.list (S3HttpSimpleTask via list_objects)": ` + // development: false so the dev-mode "failed" print for exchanges the + // teardown abort kills mid-delivery stays out of stderr. + const srv = Bun.serve({ port: 0, development: false, fetch: () => new Response('false', { headers: { "Content-Type": "application/xml" } }) }); + const s3 = new Bun.S3Client({ accessKeyId: "k", secretAccessKey: "s", region: "us-east-1", bucket: "b", endpoint: srv.url.href }); + // One list must complete before the door fires so its counter release + // has landed; an unbalanced fence then stalls teardown deterministically. + await s3.list().catch(() => {}); + lanes(2, () => s3.list());`, + "dynamic import (RuntimeTranspilerStore)": ` + const fs = require("node:fs/promises"); + let di = 0; + lanes(2, async () => { + di++; + const p = d.dir + "/dyn" + (di % 8) + ".ts"; + await fs.writeFile(p, "export const x" + di + ": number = " + di + ";"); + await import(p + "?v=" + di); + });`, + }; + + // Three teardown doors, all funneling into the same WebWorker::shutdown: + // parent-side terminate(), in-worker process.exit(), in-worker uncaught + // throw. The full door matrix runs for two representative families; the + // rest use terminate(), the door that lands at the most hostile time. + const doors: Record = { + "terminate()": { workerExit: "", parentAction: "await Bun.sleep(T); await w.terminate();" }, + "process.exit()": { workerExit: "setTimeout(() => process.exit(0), d.T);", parentAction: "" }, + "uncaught throw": { workerExit: "setTimeout(() => { throw new Error('boom'); }, d.T);", parentAction: "" }, + }; + const allDoorFamilies = new Set(["crypto.pbkdf2 (AnyTaskJob)", "fs.promises read/write (AsyncFSTask)"]); + + for (const [family, body] of Object.entries(families)) { + for (const [door, { workerExit, parentAction }] of Object.entries(doors)) { + if (door !== "terminate()" && !allDoorFamilies.has(family)) continue; + test.concurrent( + `${family}, ${door}`, + async () => { + using dir = tempDir("worker-offthread", { + "entry.ts": `import { x } from "./dep.ts"; console.log(x);`, + "dep.ts": `export const x: number = ${"1 + ".repeat(500)}1;`, + "tree/a/one.txt": "one", + "tree/a/b/two.txt": "two", + "tree/a/b/c/three.txt": "three", + "tree/four.txt": "four", + }); + + const script = /* js */ ` + const { Worker } = require("node:worker_threads"); + const src = + 'const { parentPort, workerData: d } = require("node:worker_threads");' + + 'const lanes = (n, f) => { for (let i = 0; i < n; i++) (async () => { for (;;) { try { await f(); } catch {} } })(); };' + + 'const buf = () => { const b = new Uint8Array(12 << 20); for (let i = 0; i < b.length; i += 4096) b[i] = i & 0xff; return b; };' + + // Async wrapper so a body can await setup (e.g. one completed + // request) before signaling readiness. + '(async () => {' + ${JSON.stringify(body.trim())} + ';' + + '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"))); + }); + } + (async () => { + for (let r = 0; r < ${ROUNDS}; r++) { + // Jittered teardown offset so some rounds land mid-compute and + // some land with completions already queued. + const T = 10 + ((r * 73) % 180); + const w = new Worker(src, { eval: true, workerData: { dir: ${JSON.stringify(String(dir))}, T } }); + await ready(w); + w.on("error", () => {}); + const exited = new Promise(res => w.once("exit", res)); + // t0 before the door fires: terminate() only resolves once + // the worker exited, so starting the clock after it would + // measure nothing. + const t0 = Date.now(); + ${parentAction} + await exited; + // Teardown waits for in-flight jobs, which are a few seconds + // at worst here; hitting the fence's 10s deadline means an + // unbalanced outstanding_offthread count for this family. T + // is the intentional pre-door delay on every door. + const dt = Date.now() - t0 - T; + if (dt > 9000) throw new Error("teardown stalled " + dt + "ms (unbalanced off-thread fence?)"); + } + console.log("OK"); + })().catch(e => { + console.error(String(e && (e.stack || e))); + process.exit(1); + }); + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // Any sanitizer report is the bug this file exists to catch. + expect(stderr).not.toContain("AddressSanitizer"); + if (stderr.includes("assertNoException")) { + // Debug builds have a separate, pre-existing terminate() bug: the + // TerminationException can materialize mid-dispatch and trip JSC's + // ExceptionScope::assertNoException abort before teardown even + // starts (reproduces on an unfixed-teardown build too, right + // after notifyNeedTermination; tracked separately). Tolerate + // exactly that abort: stripping its lines must leave stderr + // empty, and nothing else is asserted because the abort kills + // the subprocess mid-matrix. + expect(onlyKnownTerminateAssert(stderr)).toEqual([]); + } else { + expect(stderr).toBe(""); + expect(stdout).toBe("OK\n"); + expect(exitCode).toBe(0); + } + }, + TIMEOUT, + ); + } + } +}); + +// FetchTasklet holds a lifetime-erased &'static VirtualMachine and the shared +// HTTP client thread reads it (is_shutting_down / enqueue_task_concurrent) +// after WebWorker::shutdown 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", + () => { + const ROUNDS = 8; + // 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}; 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, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // Any sanitizer report is the bug this block exists to catch. + expect(stderr).not.toContain("AddressSanitizer"); + if (stderr.includes("assertNoException")) { + // Same pre-existing mid-dispatch assertNoException tolerance as + // the family matrix above: only that exact abort may appear. + expect(onlyKnownTerminateAssert(stderr)).toEqual([]); + } else { + expect(stderr).toBe(""); + expect(stdout).toBe("survived\n"); + expect(exitCode).toBe(0); + } + }, + TIMEOUT, + ); + } + }, +);