diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index 204affdd8d02..e9cf5b56f3f5 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; @@ -88,6 +88,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 callback (and its completion post) are + /// 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 @@ -128,6 +137,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, @@ -989,6 +999,49 @@ 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. This must be the + /// last access to `self`: once the waiter observes zero it may free the + /// `VirtualMachine` box this `EventLoop` lives in. + #[inline] + pub fn work_pool_task_unref(&self) { + self.work_pool_pending.fetch_sub(1, Ordering::Release); + } + + /// Worker-thread shutdown barrier. Blocks until every outstanding + /// [`Self::work_pool_task_ref`] has been matched by + /// [`Self::work_pool_task_unref`]. The wait is bounded only by the + /// slowest in-flight pool callback (for `napi_async_work` that is + /// arbitrary addon code); Node.js's env-close `uv_run` drain has the + /// same `terminate()` latency model. + pub fn wait_for_pending_work_pool_tasks(&self) { + loop { + let n = self.work_pool_pending.load(Ordering::Acquire); + if n == 0 { + return; + } + // Timed wait: the pool thread cannot `Futex::wake` here because + // its last safe access to `self` is the `fetch_sub` above, after + // which this `EventLoop` may be freed. 1ms re-check adds at most + // 1ms to `terminate()`. + let _ = bun_threading::Futex::wait(&self.work_pool_pending, n, Some(1_000_000)); + } + } + 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 7f30e0f913e2..24913b51ab2a 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -1303,6 +1303,10 @@ impl WebWorker { // or observes m_isShuttingDown under m_lock and drops. Idempotent; // teardownJSCVM sets it again. Bun__JSCTaskScheduler__markShuttingDown(vm.global()); + // Join in-flight WorkPool jobs (napi_async_work; see + // `work_pool_pending`) whose pool-thread callback reads this VM's + // EventLoop / JSC heap, both freed below. + 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 fb6f5506de64..c164ab6aeb9c 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -1331,6 +1331,12 @@ 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 (`work_pool_pending` barrier). + unsafe { napi_async_work::release_for_shutdown(task.ptr.cast::()) }; + 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 bb0506b0b16d..18b90aba636d 100644 --- a/src/runtime/napi/napi_body.rs +++ b/src/runtime/napi/napi_body.rs @@ -1758,7 +1758,10 @@ 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). Liveness across a worker `terminate()` is + // guaranteed by `schedule()`'s `work_pool_task_ref()`: the worker + // shutdown barrier waits for `run()` to `work_pool_task_unref()` + // before the VM box (and this `EventLoop`) are freed. event_loop: unsafe { bun_ptr::BackRef::from_raw(global.bun_vm().event_loop()) }, complete, data, @@ -1777,12 +1780,28 @@ impl napi_async_work { drop(unsafe { bun_core::heap::take(this) }); } + /// Shutdown-drain release: unref the loop `KeepAlive` taken in + /// `schedule()` and free the box. Does not call `complete` (it would run + /// after `NapiEnv::cleanup()`); the addon's `data` is left for the + /// process to reclaim. + /// + /// # Safety + /// `this` must be the heap work popped from the shutdown drain; the pool + /// thread no longer holds it (`work_pool_pending` barrier). + 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()); + Self::destroy(this); + } + pub(crate) fn schedule(&mut self) { if self.scheduled { return; } self.scheduled = true; self.poll_ref.ref_(bun_io::js_vm_ctx()); + // Matched by `work_pool_task_unref()` at the end of `run()`. + self.event_loop.work_pool_task_ref(); WorkPool::schedule(&raw mut self.task); } @@ -1794,6 +1813,11 @@ impl napi_async_work { fn run(&mut self) { let self_ptr: *mut Self = self; + // After `enqueue_task_concurrent` the JS thread may pick this work up, + // run `complete`, and `napi_delete_async_work` it before we reach the + // `work_pool_task_unref()` below; copy the handle out so that last + // access 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 +1827,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.work_pool_task_unref(); return; } } @@ -1817,11 +1841,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.work_pool_task_unref(); } pub(crate) fn cancel(&mut self) -> bool { diff --git a/test/napi/napi-app/binding.gyp b/test/napi/napi-app/binding.gyp index 94f07ed09603..be26636aa0c0 100644 --- a/test/napi/napi-app/binding.gyp +++ b/test/napi/napi-app/binding.gyp @@ -286,5 +286,16 @@ "NODE_API_EXPERIMENTAL_NOGC_ENV_OPT_OUT=1", ], }, + { + "target_name": "test_async_work_worker_terminate", + "sources": ["test_async_work_worker_terminate.c"], + "include_dirs": [" +#include +#include +#include + +#ifdef _WIN32 +#include +static void sleep_ms(unsigned ms) { Sleep(ms); } +#else +#include +static void sleep_ms(unsigned ms) { usleep(ms * 1000); } +#endif + +#define CHECK(env, call) \ + do { \ + napi_status s_ = (call); \ + if (s_ != napi_ok) { \ + napi_throw_error((env), NULL, #call " failed"); \ + return NULL; \ + } \ + } while (0) + +typedef struct { + napi_async_work work; + napi_ref buf_ref; + napi_ref cb_ref; + unsigned char *data; + size_t len; + unsigned sleep_ms; +} work_t; + +static void exec_cb(napi_env env, void *arg) { + work_t *w = (work_t *)arg; + sleep_ms(w->sleep_ms); + // Touch the ArrayBuffer backing store. Before the fix, worker.terminate() + // could free it (via JSC VM teardown running the ArrayBuffer finalizer) + // while this callback is still running on the pool thread. + if (w->len > 0) { + volatile unsigned char first = w->data[0]; + (void)first; + memset(w->data, 0xab, w->len); + } +} + +static void done_cb(napi_env env, napi_status status, void *arg) { + work_t *w = (work_t *)arg; + napi_value cb = NULL, undef = NULL, argv[1]; + if (w->cb_ref != NULL) { + napi_get_reference_value(env, w->cb_ref, &cb); + } + napi_get_undefined(env, &undef); + napi_create_int32(env, (int)status, &argv[0]); + if (cb != NULL) { + napi_call_function(env, undef, cb, 1, argv, NULL); + } + napi_delete_reference(env, w->buf_ref); + if (w->cb_ref != NULL) napi_delete_reference(env, w->cb_ref); + napi_delete_async_work(env, w->work); + free(w); +} + +// queueWork(arrayBuffer, ms[, cb]) -> undefined +static napi_value queue_work(napi_env env, napi_callback_info info) { + size_t argc = 3; + napi_value argv[3]; + CHECK(env, napi_get_cb_info(env, info, &argc, argv, NULL, NULL)); + if (argc < 2) { + napi_throw_error(env, NULL, "expected (arrayBuffer, ms[, cb])"); + return NULL; + } + + work_t *w = (work_t *)calloc(1, sizeof(*w)); + + void *data = NULL; + size_t len = 0; + CHECK(env, napi_get_arraybuffer_info(env, argv[0], &data, &len)); + w->data = (unsigned char *)data; + w->len = len; + + int32_t ms = 0; + CHECK(env, napi_get_value_int32(env, argv[1], &ms)); + w->sleep_ms = (unsigned)(ms < 0 ? 0 : ms); + + CHECK(env, napi_create_reference(env, argv[0], 1, &w->buf_ref)); + if (argc >= 3) { + napi_valuetype t; + CHECK(env, napi_typeof(env, argv[2], &t)); + if (t == napi_function) { + CHECK(env, napi_create_reference(env, argv[2], 1, &w->cb_ref)); + } + } + + napi_value name; + CHECK(env, napi_create_string_utf8(env, "test_async_work_worker_terminate", + NAPI_AUTO_LENGTH, &name)); + CHECK(env, napi_create_async_work(env, NULL, name, exec_cb, done_cb, w, + &w->work)); + CHECK(env, napi_queue_async_work(env, w->work)); + + napi_value undef; + CHECK(env, napi_get_undefined(env, &undef)); + return undef; +} + +NAPI_MODULE_INIT() { + napi_value fn; + CHECK(env, napi_create_function(env, "queueWork", NAPI_AUTO_LENGTH, + queue_work, NULL, &fn)); + CHECK(env, napi_set_named_property(env, exports, "queueWork", fn)); + return exports; +} diff --git a/test/napi/napi.test.ts b/test/napi/napi.test.ts index 63902deafb10..3c3267ca84c5 100644 --- a/test/napi/napi.test.ts +++ b/test/napi/napi.test.ts @@ -518,6 +518,65 @@ describe.concurrent.skipIf(!canBuildNodeAddons())("napi", () => { expect(output).toContain("success!"); expect(output).not.toContain("failure!"); }); + // worker.terminate() while execute callbacks are still running on the + // thread pool must not free the worker's VirtualMachine (the pool-thread + // completion would enqueue into a freed EventLoop) or its JSC heap (the + // ArrayBuffer backing store would be finalized while the addon is still + // writing it). WebWorker::shutdown now waits for every queued + // napi_async_work's pool-thread callback to finish before teardown. On an + // unfixed build the subprocess aborts before printing PASS. + it("worker.terminate() with execute callbacks in flight waits for them and does not UAF", async () => { + const addon = join(__dirname, "napi-app/build/Debug/test_async_work_worker_terminate.node"); + const workerSrc = /* js */ ` + const { parentPort, workerData } = require("node:worker_threads"); + const addon = require(workerData.addon); + const keep = []; + for (let i = 0; i < 4; i++) { + const ab = new ArrayBuffer(16 << 20); + keep.push(ab); + addon.queueWork(ab, 300 + i * 50, () => {}); + } + parentPort.postMessage("up"); + setInterval(() => {}, 1000); + `; + const script = /* js */ ` + const { Worker } = require("node:worker_threads"); + (async () => { + for (let r = 0; r < ${isASAN ? 3 : 5}; r++) { + const w = new Worker(process.env.WORKER_SRC, { + eval: true, + workerData: { addon: process.env.ADDON }, + }); + await new Promise((resolve, reject) => { + w.once("message", resolve); + w.once("error", reject); + w.once("exit", code => reject(new Error("worker exited before queueing work, code " + code))); + }); + await w.terminate(); + } + console.log("PASS"); + })().catch(e => { + console.error(String(e)); + process.exit(1); + }); + `; + await using proc = spawn({ + cmd: [bunExe(), "-e", script], + // complete() is not invoked on the terminate path (see PR body), so the + // addon's per-work calloc leaks by design. LSan stays off so the crash + // assertion below is what decides pass/fail. + env: { + ...bunEnv, + ADDON: addon, + WORKER_SRC: workerSrc, + ASAN_OPTIONS: "detect_leaks=0:allow_user_segv_handler=1", + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ stdout: "PASS", stderr: "", exitCode: 0 }); + }, 30_000); }); describe("napi_threadsafe_function", () => {