diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index e64145537196..b67df023b784 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -69,6 +69,81 @@ pub type ExceptionList = Vec; // VirtualMachine struct (file-level @This()) // ────────────────────────────────────────────────────────────────────────── +/// Arc'd per-VM shutdown flag + reader fence for threads that may outlive the +/// VM. A worker's [`WebWorker::shutdown`] raw-`dealloc`s the `VirtualMachine` +/// while the shared HTTP client thread can still be delivering `FetchTasklet` +/// callbacks that dereference it; those callbacks hold a clone of this Arc +/// and bracket every VM access with [`Self::try_begin_vm_read`] / +/// [`Self::end_vm_read`] so shutdown can spin the readers out first. +/// +/// [`WebWorker::shutdown`]: crate::web_worker::WebWorker::shutdown +pub struct CrossThreadShutdownSignal { + shutting_down: core::sync::atomic::AtomicBool, + readers: core::sync::atomic::AtomicUsize, + is_main_thread: bool, +} + +impl CrossThreadShutdownSignal { + fn new(is_main_thread: bool) -> std::sync::Arc { + std::sync::Arc::new(Self { + shutting_down: core::sync::atomic::AtomicBool::new(false), + readers: core::sync::atomic::AtomicUsize::new(0), + is_main_thread, + }) + } + + #[inline] + pub fn is_shutting_down(&self) -> bool { + self.shutting_down + .load(core::sync::atomic::Ordering::Acquire) + } + + #[inline] + pub fn is_main_thread(&self) -> bool { + self.is_main_thread + } + + pub fn mark_shutting_down(&self) { + self.shutting_down + .store(true, core::sync::atomic::Ordering::SeqCst); + } + + pub fn wait_for_readers(&self) { + debug_assert!(self.is_shutting_down()); + while self.readers.load(core::sync::atomic::Ordering::SeqCst) > 0 { + std::hint::spin_loop(); + } + } + + /// On `true` the caller may dereference the owning VM until the paired + /// [`Self::end_vm_read`]; on `false` the VM is (or is about to be) freed + /// and no `end_vm_read` is owed. SeqCst on all four ops (this increment + + /// flag load, and the writer's flag store + reader-count load) is the + /// Dekker-style fence that keeps "reader saw `false`" ordered before + /// "writer saw `readers == 0`". + #[inline] + #[must_use] + pub fn try_begin_vm_read(&self) -> bool { + self.readers + .fetch_add(1, core::sync::atomic::Ordering::SeqCst); + if self + .shutting_down + .load(core::sync::atomic::Ordering::SeqCst) + { + self.readers + .fetch_sub(1, core::sync::atomic::Ordering::SeqCst); + return false; + } + true + } + + #[inline] + pub fn end_vm_read(&self) { + self.readers + .fetch_sub(1, core::sync::atomic::Ordering::SeqCst); + } +} + #[derive(Default)] pub struct EntryPointResult { pub value: crate::strong::Optional, // jsc.Strong.Optional @@ -207,6 +282,9 @@ pub struct VirtualMachine { pub(crate) hide_bun_stackframes: bool, pub is_shutting_down: bool, + /// See [`CrossThreadShutdownSignal`]. `Option` only so `destroy()` can + /// release the strong ref (the box is raw-`dealloc`'d, no field `Drop`s). + pub cross_thread_shutdown: Option>, /// 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 @@ -979,6 +1057,13 @@ impl VirtualMachine { self.is_shutting_down } + #[inline] + pub fn cross_thread_shutdown(&self) -> &std::sync::Arc { + self.cross_thread_shutdown + .as_ref() + .expect("cross_thread_shutdown is Some from init() to destroy()") + } + pub fn has_run_cleanup_hooks(&self) -> bool { self.has_run_cleanup_hooks } @@ -1488,6 +1573,7 @@ impl VirtualMachine { } self.is_shutting_down = true; + self.cross_thread_shutdown().mark_shutting_down(); // Make sure we run new cleanup hooks introduced by running cleanup // hooks. @@ -2110,6 +2196,8 @@ impl VirtualMachine { addr_of_mut!((*vm).resolved_path_dups).write(Vec::new()); addr_of_mut!((*vm).macros).write(Default::default()); addr_of_mut!((*vm).macro_entry_points).write(Default::default()); + addr_of_mut!((*vm).cross_thread_shutdown) + .write(Some(CrossThreadShutdownSignal::new(opts.is_main_thread))); addr_of_mut!((*vm).auto_killer).write(Default::default()); addr_of_mut!((*vm).commonjs_custom_extensions).write(Default::default()); addr_of_mut!((*vm).entry_point).write(Default::default()); @@ -4463,6 +4551,8 @@ impl VirtualMachine { // proxy strings; `ProxyEnvStorage: Default` so take()+drop suffices. drop(core::mem::take(&mut self.proxy_env_storage)); + drop(self.cross_thread_shutdown.take()); + // 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/web_worker.rs b/src/jsc/web_worker.rs index 6e67c931d47c..6960878d512b 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -1248,6 +1248,7 @@ impl WebWorker { // re-sets it for the JSC VM teardown. vm.jsc_vm().clear_has_termination_request(); vm.is_shutting_down = true; + vm.cross_thread_shutdown().mark_shutting_down(); vm.on_exit(); if let Some(hooks) = runtime_hooks() { (hooks.cron_clear_all_teardown)(vm); @@ -1302,6 +1303,10 @@ impl WebWorker { // or observes m_isShuttingDown under m_lock and drops. Idempotent; // teardownJSCVM sets it again. Bun__JSCTaskScheduler__markShuttingDown(vm.global()); + // Per-worker fence for HTTP-thread FetchTasklet callbacks (the + // main-thread equivalent is `bun_http::shutdown_for_exit()`, a + // process-global one-shot). See `CrossThreadShutdownSignal`. + vm.cross_thread_shutdown().wait_for_readers(); // 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/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index 93c274dc1369..3cd580f27b4b 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -17,7 +17,7 @@ use bun_http::{ }; use bun_io::KeepAlive; use bun_jsc::debugger::AsyncTaskTracker; -use bun_jsc::virtual_machine::VirtualMachine; +use bun_jsc::virtual_machine::{CrossThreadShutdownSignal, VirtualMachine}; use bun_jsc::{ self as jsc, GlobalRef, JSGlobalObject, JSValue, JsResult, StringJsc, StrongOptional, }; @@ -68,6 +68,9 @@ pub struct FetchTasklet { pub(crate) result: HTTPClientResult<'static>, pub(crate) metadata: Option, pub(crate) javascript_vm: &'static VirtualMachine, + /// `javascript_vm` dangles once a worker VM is `dealloc`'d; HTTP-thread + /// callbacks fence every VM deref through this (see the struct doc). + pub(crate) vm_shutdown_signal: std::sync::Arc, pub global_this: GlobalRef, pub(crate) request_body: HTTPRequestBody, // ThreadSafeStreamBuffer is intrusively refcounted (`ref_count: AtomicU32`, @@ -397,7 +400,11 @@ impl FetchTasklet { return; } let self_ = Self::from_raw_ref(this); - if self_.javascript_vm.is_shutting_down() { + // The enqueued `deinit_callback` may free `this` before we return, so + // snapshot what's needed past the enqueue into locals now. + let signal = std::sync::Arc::clone(&self_.vm_shutdown_signal); + let vm = self_.javascript_vm; + if !signal.try_begin_vm_read() { // SAFETY: last ref; exclusive access. `deinit()` would run // `clear_data()` + `Drop` for the JSC `Strong`/`Weak` fields, which // reach into the VM's StrongRootBlock list / WeakSet from this @@ -411,9 +418,10 @@ impl FetchTasklet { // `from_callback` heap-allocates a fresh `ConcurrentTaskItem`; the queue // takes ownership of it. Self::enqueue_concurrent( - self_.javascript_vm, + vm, ConcurrentTask::from_callback(this, FetchTasklet::deinit_callback), ); + signal.end_vm_read(); } // ConcurrentTask::from_callback takes `fn(*mut T) -> bun_event_loop::JsResult<()>` @@ -518,19 +526,26 @@ 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: no such drain exists and the JSC heap is (about to be) + /// freed, so a parked `deinit()` would dereference dead handles. Leak the + /// box; the large buffers were already released by the caller. + /// /// 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 — `this` is live with ref_count == 0. + if unsafe { (*this).vm_shutdown_signal.is_main_thread() } { + http::defer_shutdown_reclaim(this.cast(), FetchTasklet::deinit_erased); + } } unsafe fn deinit_erased(this: *mut c_void) { @@ -2027,6 +2042,7 @@ impl FetchTasklet { result: HTTPClientResult::default(), metadata: None, javascript_vm: jsc_vm, + vm_shutdown_signal: std::sync::Arc::clone(jsc_vm.cross_thread_shutdown()), global_this: GlobalRef::from(global_this), request_body: fetch_options.body, request_body_streaming_buffer: None, @@ -2271,7 +2287,7 @@ impl FetchTasklet { /// This is ALWAYS called from the http thread and we cannot touch the buffer here because is locked fn on_write_request_data_drain(this: *mut FetchTasklet) { let this_ref = Self::from_raw_ref(this); - if this_ref.javascript_vm.is_shutting_down() { + if !this_ref.vm_shutdown_signal.try_begin_vm_read() { return; } // ref until the main thread callback is called @@ -2282,6 +2298,7 @@ impl FetchTasklet { this_ref.javascript_vm, ConcurrentTask::from_callback(this, FetchTasklet::resume_request_data_stream), ); + this_ref.vm_shutdown_signal.end_vm_read(); } /// This is ALWAYS called from the main thread @@ -2649,7 +2666,7 @@ impl FetchTasklet { } } // will deinit when done with the http client (when is_done = true) - if task_ref.javascript_vm.is_shutting_down() { + if !task_ref.vm_shutdown_signal.try_begin_vm_read() { // VM teardown: the JS-thread side will never drain this buffer (its // on_progress_update bails the same way), so free the body bytes now. task_ref.scheduled_response_buffer = MutableString::default(); @@ -2689,6 +2706,7 @@ impl FetchTasklet { // `ct` is the inline `concurrent_task` field of the heap tasklet; the // queue takes ownership of its `next` link. Self::enqueue_concurrent(task_ref.javascript_vm, ct); + task_ref.vm_shutdown_signal.end_vm_read(); task_ref.mutex.unlock(); // we are done with the http client so we can deref our side diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 6b9f7c1b3366..d74fb937a183 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -455,3 +455,103 @@ test.skipIf(!isDebug)( }, 120_000, ); + +// Regression: FetchTasklet holds a lifetime-erased &'static VirtualMachine and +// the shared HTTP client thread read it (is_shutting_down / +// enqueue_task_concurrent) after WebWorker::shutdown had 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", + () => { + // 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 * 2}; 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: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // Check stderr first: on failure the sanitizer report is the useful part. + expect(stderr).toBe(""); + expect(stdout).toBe("survived\n"); + expect(exitCode).toBe(0); + }, + timeout, + ); + } + }, +);