diff --git a/src/bundler/bundle_v2.rs b/src/bundler/bundle_v2.rs index e1a2f1fda10b..6a955519d44d 100644 --- a/src/bundler/bundle_v2.rs +++ b/src/bundler/bundle_v2.rs @@ -1120,6 +1120,7 @@ pub mod bv2_impl { std::ptr::from_mut::(self).cast::(), ), callback: Self::run_on_js_thread_wrap, + dispose: None, }; let task = bun_event_loop::ConcurrentTask::ConcurrentTask::create(self.js_task.task()); @@ -1250,6 +1251,7 @@ pub mod bv2_impl { std::ptr::from_mut::(self).cast::(), ), callback: Self::run_on_js_thread_wrap, + dispose: None, }; let concurrent_task = bun_event_loop::ConcurrentTask::ConcurrentTask::create(self.js_task.task()); diff --git a/src/event_loop/AnyTask.rs b/src/event_loop/AnyTask.rs index 8663f8f7bf1d..a2d59fba3a42 100644 --- a/src/event_loop/AnyTask.rs +++ b/src/event_loop/AnyTask.rs @@ -18,6 +18,10 @@ pub type JsResult = core::result::Result; pub struct AnyTask { pub ctx: Option>, pub callback: fn(*mut c_void) -> JsResult<()>, + /// Releases a queue-owned `ctx` when the task is reclaimed unrun by the + /// worker-terminate drain (JS thread, VM alive, so plain drop suffices). + /// `None` ⇒ the ctx is owned elsewhere and must not be freed here. + pub dispose: Option, } impl Default for AnyTask { @@ -27,6 +31,7 @@ impl Default for AnyTask { Self { ctx: None, callback: |_| unreachable!("AnyTask.callback was undefined"), + dispose: None, } } } @@ -64,6 +69,22 @@ impl AnyTask { callback, ) }, + dispose: None, } } + + /// [`Self::from_typed`] plus a release for `ctx` if the task is reclaimed + /// unrun by the worker-terminate drain (see the [`Self::dispose`] field). + #[inline] + pub fn from_typed_with_dispose( + ctx: *mut T, + callback: fn(*mut T) -> JsResult<()>, + dispose: fn(*mut T), + ) -> Self { + let mut task = Self::from_typed(ctx, callback); + // SAFETY: same ABI argument as `from_typed`'s callback cast. + task.dispose = + Some(unsafe { core::mem::transmute::(dispose) }); + task + } } diff --git a/src/jsc/AsyncModule.rs b/src/jsc/AsyncModule.rs index 18d2be1e5dba..d4cf76fbcf77 100644 --- a/src/jsc/AsyncModule.rs +++ b/src/jsc/AsyncModule.rs @@ -686,6 +686,7 @@ impl AsyncModule { Self::on_done(p.cast()); Ok(()) }, + dispose: None, }; jsc_vm.enqueue_task(Task::init(&raw mut (*clone).any_task)); } diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index 6cf416165611..7e5569e5ea57 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -274,6 +274,14 @@ pub struct VirtualMachine { pub regular_event_loop: EventLoop, pub event_loop: *mut EventLoop, // BORROW_FIELD — points at sibling regular_event_loop/macro_event_loop + /// Cross-thread producers that enqueue completions back onto this VM's + /// event loop hold this gate open across the enqueue; worker terminate + /// closes it (blocking until every guest leaves) before freeing the VM + /// box. Lives in an `Arc` so guests may outlive the box; `Option` so + /// [`Self::destroy`] can drop the VM's ref explicitly (worker boxes are + /// freed without `Drop`). + pub shutdown_gate: Option>, + pub ref_strings: crate::ref_string::Map, pub ref_strings_mutex: bun_threading::Mutex, @@ -971,6 +979,15 @@ impl VirtualMachine { self.is_shutting_down } + /// The per-VM [`bun_threading::ShutdownGate`]; see the field doc. Clone + /// on the JS thread when scheduling async work that will enqueue back + /// from another thread, and bracket the enqueue with `enter()`/`leave()`. + pub fn shutdown_gate(&self) -> &std::sync::Arc { + self.shutdown_gate + .as_ref() + .expect("shutdown_gate: accessed after destroy()") + } + pub fn has_run_cleanup_hooks(&self) -> bool { self.has_run_cleanup_hooks } @@ -2115,6 +2132,10 @@ impl VirtualMachine { let _ = (*regular).tasks.ensure_unused_capacity(64); addr_of_mut!((*vm).event_loop).write(regular); + addr_of_mut!((*vm).shutdown_gate).write(Some(std::sync::Arc::new( + bun_threading::ShutdownGate::new(), + ))); + // `source_mappings.map` is a sibling-field backref onto // `saved_source_map_table`. addr_of_mut!((*vm).saved_source_map_table) @@ -4372,6 +4393,20 @@ impl VirtualMachine { } /// Worker-thread teardown. pub fn destroy(&mut self) { + // Backstop close of the shutdown gate: refuse new producer guests. + // Worker shutdown already did the full close-and-wait before its + // drain; main-VM exit must not wait (its box is never freed and a + // straggling guest on a parked thread must not hang exit). + if let Some(gate) = &self.shutdown_gate { + if self.is_main_thread { + gate.close_without_waiting(); + } else { + gate.close_and_wait(); + } + } + // Drop the VM's ref explicitly: worker boxes are freed without + // running `Drop`, which would leak the `Arc` allocation. + drop(self.shutdown_gate.take()); self.regular_event_loop.deinit(); self.macro_event_loop.deinit(); diff --git a/src/jsc/any_task_job.rs b/src/jsc/any_task_job.rs index 870232d0fae8..2b5294adb621 100644 --- a/src/jsc/any_task_job.rs +++ b/src/jsc/any_task_job.rs @@ -91,6 +91,7 @@ impl AnyTaskJob { (*job).any_task = AnyTask { ctx: NonNull::new(job.cast::()), callback: |p: *mut c_void| Self::run_from_js(p.cast::()).map_err(Into::into), + dispose: None, }; } // `ctx.init` may throw (e.g. CryptoJob); on error, reclaim the diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index 5c562f1cdd07..6f2c8d050aff 100644 --- a/src/jsc/event_loop.rs +++ b/src/jsc/event_loop.rs @@ -749,6 +749,23 @@ impl EventLoop { self.drop_concurrent_cpp_tasks(); let mut requeue: Vec = Vec::new(); while let Some(task) = self.tasks.read_item() { + // A queue-owned `AnyTask` completion is released via the fn its + // producer registered (plain drop; runs here with the VM alive). + if task.tag == bun_event_loop::task_tag::AnyTask { + // SAFETY: `ptr` is the `*mut AnyTask` registered by + // `AnyTask::task()`. Copy the fields out before calling + // `dispose`: it may free the allocation embedding the AnyTask. + let (dispose, ctx) = unsafe { + let any = &*task.ptr.cast::(); + (any.dispose, any.ctx) + }; + if let (Some(dispose), Some(ctx)) = (dispose, ctx) { + dispose(ctx.as_ptr()); + continue; + } + requeue.push(task); + continue; + } // SAFETY: tag-specific release (drops JSC handles while the VM is // still live); definer in `bun_runtime::dispatch` matches the same // tag set `tick_queue_with_count` does. `false` ⇒ not handled. diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index 718c8f01ff96..1955a56420f7 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -1312,6 +1312,10 @@ impl WebWorker { // or observes m_isShuttingDown under m_lock and drops. Idempotent; // teardownJSCVM sets it again. Bun__JSCTaskScheduler__markShuttingDown(vm.global()); + // Close the VM's shutdown gate and wait for every cross-thread + // producer that entered it: after this no pool thread can + // dereference the (about-to-be-freed) VM box to enqueue. + vm.shutdown_gate().close_and_wait(); // 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 diff --git a/src/runtime/crypto/PasswordObject.rs b/src/runtime/crypto/PasswordObject.rs index a40a2294107d..350cdb410363 100644 --- a/src/runtime/crypto/PasswordObject.rs +++ b/src/runtime/crypto/PasswordObject.rs @@ -550,6 +550,10 @@ struct PasswordJob { password: Box<[u8]>, promise: JSPromiseStrong, event_loop: *mut EventLoop, + /// Cloned from the VM on the JS thread; bracket the `event_loop` + /// dereference in `run_owned` with `enter()`/`leave()` so worker + /// terminate can fence the pool thread out before freeing the VM box. + gate: std::sync::Arc, global: *const JSGlobalObject, r#ref: KeepAlive, task: WorkPoolTask, @@ -573,6 +577,20 @@ impl PasswordJob { #[allow(clippy::boxed_local)] fn run_owned(mut self: Box) { let value = self.op.compute(&self.password); + // argon2/bcrypt take hundreds of ms: by the time the compute finishes + // the worker that scheduled this job may have been terminated and its + // VM box freed. Enter the VM's shutdown gate (cloned into `self.gate`, + // so the gate itself outlives the VM box) and only dereference + // `event_loop` while inside; worker terminate closes-and-waits on the + // gate before the dealloc. + if !self.gate.enter() { + // Worker gone. The promise handle points into the dead JSC VM's + // HandleSet; leak the slot rather than let `Drop` deref it. + let _ = core::mem::ManuallyDrop::new(core::mem::take(&mut self.promise)); + return; + // `self: Box` drops here; Drop runs secure_zero on password + // (+op); `KeepAlive` has no `Drop`; `gate` Arc is released. + } let result = bun_core::heap::into_raw(Box::new(PasswordResult:: { value, task: AnyTask::default(), // overwritten below @@ -583,16 +601,22 @@ impl PasswordJob { // SAFETY: `result` was just heap-allocated and is not yet shared // (enqueue happens after this write). unsafe { - (*result).task = AnyTask::from_typed(result, PasswordResult::::run_from_js_erased); + (*result).task = AnyTask::from_typed_with_dispose( + result, + PasswordResult::::run_from_js_erased, + PasswordResult::::release_unrun, + ); } - // SAFETY: `event_loop` was stored from the JS-thread VM and outlives the - // job; ownership of `result` transfers to the event loop here. `task` is - // an intrusive field at a stable address. + // SAFETY: `event_loop` was stored from the JS-thread VM and is kept + // alive by the gate guest held above; ownership of `result` transfers + // to the event loop here. `task` is an intrusive field at a stable + // address. unsafe { (*self.event_loop).enqueue_task_concurrent(ConcurrentTask::create_from( core::ptr::addr_of_mut!((*result).task), )); } + self.gate.leave(); // `self: Box` drops here; Drop runs secure_zero on password (+op). } } @@ -606,6 +630,13 @@ struct PasswordResult { } impl PasswordResult { + /// Release a queued-but-unrun completion during the worker-terminate + /// drain (JS thread, VM alive): plain drop releases everything. + fn release_unrun(p: *mut Self) { + // SAFETY: queue-owned heap result popped by the drain; sole owner. + drop(unsafe { bun_core::heap::take(p) }); + } + fn run_from_js_erased(p: *mut Self) -> AnyTaskJsResult<()> { Self::run_from_js(p) .map_err(|_: jsc::JsTerminated| bun_event_loop::ErasedJsError::Terminated) @@ -666,12 +697,13 @@ impl JSPasswordObject { let promise = JSPromiseStrong::init(global_object); let promise_value = promise.value(); + let vm = global_object.bun_vm(); let mut job = Box::new(PasswordJob:: { op, password, promise, - // SAFETY: bun_vm() is non-null for a Bun-owned global; VM outlives the job. - event_loop: global_object.bun_vm().event_loop(), + event_loop: vm.event_loop(), + gate: std::sync::Arc::clone(vm.shutdown_gate()), global: std::ptr::from_ref(global_object), r#ref: KeepAlive::default(), task: WorkPoolTask::default(), diff --git a/src/runtime/dns_jsc/dns.rs b/src/runtime/dns_jsc/dns.rs index 88533da4b7ed..c66a7319f576 100644 --- a/src/runtime/dns_jsc/dns.rs +++ b/src/runtime/dns_jsc/dns.rs @@ -403,6 +403,7 @@ pub(super) mod lib_uv_backend { (*holder).task = jsc::AnyTask::AnyTask { ctx: NonNull::new(holder.cast()), callback: Holder::run, + dispose: None, }; (*this) .head diff --git a/src/runtime/node/node_fs_stat_watcher.rs b/src/runtime/node/node_fs_stat_watcher.rs index ef7e7092700c..818010ddbfa7 100644 --- a/src/runtime/node/node_fs_stat_watcher.rs +++ b/src/runtime/node/node_fs_stat_watcher.rs @@ -330,6 +330,7 @@ impl StatWatcherScheduler { (*holder_ptr).task = AnyTask { ctx: core::ptr::NonNull::new(holder_ptr.cast()), callback: update_timer, + dispose: None, }; (*this) .vm diff --git a/src/runtime/socket/WindowsNamedPipeContext.rs b/src/runtime/socket/WindowsNamedPipeContext.rs index e2c53939c4c0..c7d868860fde 100644 --- a/src/runtime/socket/WindowsNamedPipeContext.rs +++ b/src/runtime/socket/WindowsNamedPipeContext.rs @@ -369,6 +369,7 @@ impl WindowsNamedPipeContext { Self::run_event(ctx.cast::()); Ok(()) }, + dispose: None, }; // SAFETY: `this` is freshly allocated uninit storage exclusively owned here; we write diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 4b3e7361e4ae..9f83d64bb1b0 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -4575,6 +4575,7 @@ pub fn js_upgrade_duplex_to_tls( DuplexUpgradeContext::run_event(p.cast::()); Ok(()) }, + dispose: None, }); ptr::addr_of_mut!((*duplex_context).task_event).write(EventState::StartTLS); // When `owned_ctx` is set, `runEvent` builds from it and ignores diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index d71ee5a77bc9..428a6287ec00 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1469,6 +1469,7 @@ impl JSValkeyClient { Holder::run(p.cast::()); Ok(()) }, + dispose: None, }; } diff --git a/src/threading/ShutdownGate.rs b/src/threading/ShutdownGate.rs new file mode 100644 index 000000000000..57e3ed86af3e --- /dev/null +++ b/src/threading/ShutdownGate.rs @@ -0,0 +1,139 @@ +//! A counted guest gate protecting an allocation shared with other threads. +//! +//! Guests bracket every access to the protected memory with `enter()`/ +//! `leave()`. The owner calls `close_and_wait()` before freeing the memory: +//! it refuses new guests, then blocks until every guest inside has left. +//! The gate itself must live OUTSIDE the protected allocation (e.g. in an +//! `Arc` cloned by each guest) so that a late `enter()` after the close is a +//! safe "gate closed" answer instead of a use-after-free. + +use core::sync::atomic::{AtomicU32, Ordering}; + +use crate::futex; + +/// Bit 0: closed. Remaining bits: number of guests currently inside ×2. +const CLOSED: u32 = 1; +const GUEST: u32 = 2; + +#[derive(Default)] +pub struct ShutdownGate { + state: AtomicU32, +} + +impl ShutdownGate { + pub const fn new() -> Self { + Self { + state: AtomicU32::new(0), + } + } + + /// Try to enter the gate. Returns `false` if the gate is closed — the + /// protected memory may already be freed and must not be touched. + #[must_use] + pub fn enter(&self) -> bool { + // Optimistic add, undo on closed: `close_and_wait` tolerates the + // transient count because it re-reads state until it settles. + let prev = self.state.fetch_add(GUEST, Ordering::Acquire); + if prev & CLOSED == 0 { + return true; + } + self.leave(); + false + } + + /// Leave the gate. Must pair with an `enter()` that returned `true` (or + /// the internal undo above). After this call the guest must not touch the + /// protected memory again. + pub fn leave(&self) { + let prev = self.state.fetch_sub(GUEST, Ordering::Release); + if prev == CLOSED | GUEST { + // Last guest out of a closed gate: wake `close_and_wait` (all + // waiters — `close_and_wait` is callable from several owners). + futex::wake(&self.state, u32::MAX); + } + } + + /// Set the CLOSED bit without waiting for guests. Only sound when the + /// protected memory is never actually freed (the main VM's box lives for + /// the process) — new guests are refused, in-flight ones finish on their + /// own time. + pub fn close_without_waiting(&self) { + self.state.fetch_or(CLOSED, Ordering::AcqRel); + } + + /// Close the gate and block until every guest has left. After this + /// returns, no guest is inside and none can enter; the protected memory + /// may be freed. Idempotent; must never be called from a guest section. + pub fn close_and_wait(&self) { + let mut state = self.state.fetch_or(CLOSED, Ordering::AcqRel) | CLOSED; + while state != CLOSED { + let _ = futex::wait(&self.state, state, None); + state = self.state.load(Ordering::Acquire); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::AtomicUsize; + + #[test] + fn close_and_wait_drains_racing_guests() { + for _ in 0..64 { + let gate = Arc::new(ShutdownGate::new()); + let inside = Arc::new(AtomicUsize::new(0)); + let guests: Vec<_> = (0..8) + .map(|_| { + let (gate, inside) = (Arc::clone(&gate), Arc::clone(&inside)); + std::thread::spawn(move || { + for _ in 0..500 { + if !gate.enter() { + return; + } + inside.fetch_add(1, Ordering::SeqCst); + std::hint::spin_loop(); + inside.fetch_sub(1, Ordering::SeqCst); + gate.leave(); + } + }) + }) + .collect(); + // A second concurrent closer: close_and_wait must be callable + // from several owners and both must drain. + let second_closer = { + let gate = Arc::clone(&gate); + std::thread::spawn(move || gate.close_and_wait()) + }; + gate.close_and_wait(); + // After close_and_wait returns, no guest is inside and none can enter. + assert_eq!(inside.load(Ordering::SeqCst), 0); + assert!(!gate.enter()); + gate.close_and_wait(); // idempotent + second_closer.join().unwrap(); + for g in guests { + g.join().unwrap(); + } + } + } + + #[test] + fn enter_is_rejected_while_a_guest_holds_the_gate_closed() { + let gate = Arc::new(ShutdownGate::new()); + assert!(gate.enter()); + let closer = { + let gate = Arc::clone(&gate); + std::thread::spawn(move || gate.close_and_wait()) + }; + // The closer sets CLOSED immediately; wait until new entries bounce. + while gate.enter() { + gate.leave(); + std::hint::spin_loop(); + } + // Release the in-flight guest so the blocked closer drains. + gate.leave(); + closer.join().unwrap(); + assert!(!gate.enter()); + } +} diff --git a/src/threading/lib.rs b/src/threading/lib.rs index ee6283c2ed4e..6a24a457f491 100644 --- a/src/threading/lib.rs +++ b/src/threading/lib.rs @@ -13,6 +13,8 @@ pub mod reset_event; pub mod rwlock; #[path = "Semaphore.rs"] pub mod semaphore; +#[path = "ShutdownGate.rs"] +pub mod shutdown_gate; #[path = "ThreadPool.rs"] pub mod thread_pool; pub mod work_pool; @@ -35,6 +37,7 @@ pub use mutex::{Mutex, MutexGuard}; pub use reset_event::ResetEvent; pub use rwlock::{RwLock, RwLockReadGuard, RwLockWriteGuard}; pub use semaphore::Semaphore; +pub use shutdown_gate::ShutdownGate; pub use thread_pool::ThreadPool; pub use unbounded_queue::{Link, Linked, UnboundedQueue}; pub use wait_group::WaitGroup; diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 9d476d1f3d53..07eac75116de 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -121,6 +121,57 @@ test( timeout, ); +// Regression: Bun.password.hash() runs argon2/bcrypt on the shared work pool +// and enqueues its completion back through a raw *mut EventLoop captured at +// schedule time. worker.terminate() freed the VM box mid-hash, then the pool +// thread dereferenced the freed event loop (heap-use-after-free in +// EventLoop::enqueue_task_concurrent on a Bun Pool thread). +test( + "terminate() while Bun.password.hash() is in flight does not UAF the worker VM", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + // Keep several argon2/bcrypt jobs in flight continuously so at least + // one is mid-compute on the pool when terminate() frees the worker VM. + const workerSrc = + "const { parentPort } = require('node:worker_threads');" + + "const lane = f => (async () => { for (;;) { try { await f(); } catch {} } })();" + + "for (let i = 0; i < 3; i++) lane(() => Bun.password.hash('hunter2', { algorithm: 'argon2id', memoryCost: 1 << 12, timeCost: 2 }));" + + "for (let i = 0; i < 3; i++) lane(() => Bun.password.hash('pw', { algorithm: 'bcrypt', cost: 6 }));" + + "parentPort.postMessage('up');"; + for (let r = 0; r < ${rounds}; r++) { + const w = new Worker(workerSrc, { eval: true }); + await new Promise((res, rej) => { + w.once("message", res); + w.once("error", rej); + w.once("exit", c => rej(new Error("worker exited (" + c + ") before 'up'"))); + }); + // Vary the delay so terminate scans across the hash-compute window. + await Bun.sleep(20 + (r * 37) % 120); + await w.terminate(); + } + console.log("done"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // stderr before stdout/exitCode: on failure the sanitizer/crash report is + // the useful output. + expect(stderr).toBe(""); + expect(stdout).toBe("done\n"); + expect(exitCode).toBe(0); + }, + timeout, +); + // Regression: the per-VM c-ares channel was destroyed in deinit_runtime_state // (RuntimeState drop) AFTER JSC teardown and RareData.file_polls drop. // ares_destroy() synchronously fires EDESTRUCTION query callbacks and socket-