Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions src/jsc/VirtualMachine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::sync::Arc<bun_threading::ShutdownGate>>,

pub ref_strings: crate::ref_string::Map,
pub ref_strings_mutex: bun_threading::Mutex,

Expand Down Expand Up @@ -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<bun_threading::ShutdownGate> {
self.shutdown_gate
.as_ref()
.expect("shutdown_gate: accessed after destroy()")
}

pub fn has_run_cleanup_hooks(&self) -> bool {
self.has_run_cleanup_hooks
}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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();

Expand Down
4 changes: 4 additions & 0 deletions src/jsc/web_worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
31 changes: 26 additions & 5 deletions src/runtime/crypto/PasswordObject.rs
Original file line number Diff line number Diff line change
Expand Up @@ -550,6 +550,10 @@ struct PasswordJob<Op: PasswordOp> {
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<bun_threading::ShutdownGate>,
global: *const JSGlobalObject,
r#ref: KeepAlive,
task: WorkPoolTask,
Expand All @@ -573,6 +577,20 @@ impl<Op: PasswordOp> PasswordJob<Op> {
#[allow(clippy::boxed_local)]
fn run_owned(mut self: Box<Self>) {
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<Self>` 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::<Op> {
value,
task: AnyTask::default(), // overwritten below
Expand All @@ -585,14 +603,16 @@ impl<Op: PasswordOp> PasswordJob<Op> {
unsafe {
(*result).task = AnyTask::from_typed(result, PasswordResult::<Op>::run_from_js_erased);
}
// 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<Self>` drops here; Drop runs secure_zero on password (+op).
}
}
Expand Down Expand Up @@ -666,12 +686,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> {
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(),
Expand Down
139 changes: 139 additions & 0 deletions src/threading/ShutdownGate.rs
Original file line number Diff line number Diff line change
@@ -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());
}
}
3 changes: 3 additions & 0 deletions src/threading/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
51 changes: 51 additions & 0 deletions test/js/web/workers/worker-terminate-lifetime.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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-
Expand Down
Loading