diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index 5c562f1cdd07..601c759182c9 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::{AtomicI32, AtomicPtr, AtomicU32, Ordering}; use bun_io::{self as Async, Waker}; use bun_uws as uws; @@ -89,6 +89,15 @@ pub struct EventLoop { pub entered_event_loop_count: isize, pub concurrent_ref: AtomicI32, + /// Count of `WorkPool` jobs scheduled from this VM's JS thread whose + /// pool-thread callback has not yet finished its last access to this + /// `EventLoop` (typically `enqueue_task_concurrent`). `WebWorker::shutdown` + /// spins on this reaching zero before freeing the JSC heap and the + /// `VirtualMachine` box that this `EventLoop` is a field of; without that + /// barrier the pool thread's `do_work()` (JSC-heap input/output buffers) + /// and completion post are both use-after-free. Bracket with + /// [`Self::work_pool_task_ref`] / [`Self::work_pool_task_unref`]. + pub work_pool_pending: AtomicU32, /// Atomic nullable pointer to the next-due `WTFTimer`. /// /// Note (§Dispatch): payload is `*mut ()` — the real @@ -129,6 +138,7 @@ impl Default for EventLoop { uws_loop: (), entered_event_loop_count: 0, concurrent_ref: AtomicI32::new(0), + work_pool_pending: AtomicU32::new(0), imminent_gc_timer: AtomicPtr::new(core::ptr::null_mut()), #[cfg(unix)] signal_handler: None, @@ -1005,6 +1015,38 @@ impl EventLoop { self.wakeup(); } + /// JS-thread: call immediately before `WorkPool::schedule` for a task whose + /// pool-thread callback will dereference this `EventLoop` / the owning + /// `VirtualMachine` / the JSC heap (e.g. to post a completion via + /// [`Self::enqueue_task_concurrent`]). Paired with + /// [`Self::work_pool_task_unref`] on the pool thread; see + /// [`Self::work_pool_pending`]. + #[inline] + pub fn work_pool_task_ref(&self) { + self.work_pool_pending.fetch_add(1, Ordering::Relaxed); + } + + /// Pool-thread: call as the last `EventLoop`/VM access in the `WorkPool` + /// callback (after [`Self::enqueue_task_concurrent`]). The `Release` store + /// pairs with [`Self::wait_for_pending_work_pool_tasks`]'s `Acquire` load + /// so `WebWorker::shutdown` cannot observe zero until every prior access + /// is visible, making the subsequent VM dealloc safe. + #[inline] + pub fn work_pool_task_unref(&self) { + self.work_pool_pending.fetch_sub(1, Ordering::Release); + } + + /// Worker-thread shutdown barrier. Spins (yielding) until every + /// outstanding [`Self::work_pool_task_ref`] has been matched by + /// [`Self::work_pool_task_unref`]. Each pending task is one bounded + /// compression/IO step, so the wait is bounded; `terminate()` already runs + /// user exit handlers here, so this is not a new latency cliff. + pub fn wait_for_pending_work_pool_tasks(&self) { + while self.work_pool_pending.load(Ordering::Acquire) > 0 { + std::thread::yield_now(); + } + } + pub fn ref_concurrently(&self) { let _ = self.concurrent_ref.fetch_add(1, Ordering::SeqCst); self.wakeup(); diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index 718c8f01ff96..dac3a3c045e3 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -1312,6 +1312,14 @@ impl WebWorker { // or observes m_isShuttingDown under m_lock and drops. Idempotent; // teardownJSCVM sets it again. Bun__JSCTaskScheduler__markShuttingDown(vm.global()); + // Wait for in-flight WorkPool jobs scheduled from this VM + // (node:zlib async compression today; see `work_pool_pending`). + // The pool-thread callback reads this VM's `EventLoop` and the + // JSC-heap-backed input/output buffers; both are freed below + // (teardownJSCVM / step-5 dealloc). Runs before the drain so the + // completion each job posts is released by the matching + // `__bun_release_task_at_shutdown` arm while JSC is still live. + vm.event_loop_shared().wait_for_pending_work_pool_tasks(); // 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/dispatch.rs b/src/runtime/dispatch.rs index 685c1006bcc1..3b92d4a29441 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -1213,6 +1213,32 @@ pub(crate) fn __bun_release_task_at_shutdown(task: bun_event_loop::Task) -> bool unsafe { Bun__deleteDeferredWorkTask(task.ptr.cast::()) }; true } + // Async `node:zlib` write completion that reached the queue after the + // worker's last tick (the `work_pool_pending` barrier guarantees the + // post lands before this drain). 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 + } // 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/node/node_zlib_binding.rs b/src/runtime/node/node_zlib_binding.rs index 80019620ddd0..b52feed94d82 100644 --- a/src/runtime/node/node_zlib_binding.rs +++ b/src/runtime/node/node_zlib_binding.rs @@ -449,6 +449,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 + // `work_pool_task_unref()` at the end of `async_job_run`. + vm.event_loop_shared().work_pool_task_ref(); WorkPool::schedule(this.task().as_ptr()); Ok(JSValue::UNDEFINED) @@ -487,10 +491,18 @@ impl CompressionStream { // `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 + // `work_pool_task_ref()` 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 `work_pool_task_ref()` in `write()` and + // releases `WebWorker::shutdown`'s barrier. + vm.event_loop_shared().work_pool_task_unref(); } /// Dispatched from `dispatch.rs` when the worker-thread `do_work()` posts @@ -572,6 +584,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/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..b13499da0bf0 --- /dev/null +++ b/test/js/node/zlib/zlib-worker-terminate.test.ts @@ -0,0 +1,63 @@ +import { expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN } from "harness"; + +// 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 ? 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: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).not.toContain("heap-use-after-free"); + expect(stderr).not.toContain("ERROR: AddressSanitizer"); + 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);