diff --git a/src/bun_core/env_var.rs b/src/bun_core/env_var.rs index 1c5082acfb2..fb902d80c77 100644 --- a/src/bun_core/env_var.rs +++ b/src/bun_core/env_var.rs @@ -72,6 +72,11 @@ new!(pub BUN_INTERNAL_NAPI_FORCE_MUSL_CHECK: boolean, "BUN_INTERNAL_NAPI_FORCE_M new!(pub BUN_DEBUG_HASH_RANDOM_SEED: unsigned, "BUN_DEBUG_HASH_RANDOM_SEED", { deser: { error_handling: NotSet } }); new!(pub BUN_DEBUG_QUIET_LOGS: boolean, "BUN_DEBUG_QUIET_LOGS", {}); new!(pub BUN_DEBUG_TEST_TEXT_LOCKFILE: boolean, "BUN_DEBUG_TEST_TEXT_LOCKFILE", { default: false }); +// Test suite only, builds with debug assertions: `draining` or `closed`. A +// worker VM holds each cross-thread post until its teardown has reached that +// state, so the "arrived during teardown" paths run deterministically +// (bun_jsc::vm_handle::test_gate). +new!(pub BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE: string, "BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE", {}); new!(pub BUN_DEV_SERVER_TEST_RUNNER: string, "BUN_DEV_SERVER_TEST_RUNNER", {}); // Debug-only: when set, `NumberRenamer` dumps the symbol table before // renaming (`src/js_printer/renamer.rs`). Presence-checked, value ignored. @@ -206,10 +211,6 @@ pub mod feature_flag { // Run the full VM teardown when the main thread exits (workers always do). // The CI runner turns it on for LeakSanitizer-validated files on ASAN. new_feature_flag!(pub BUN_DESTRUCT_VM_ON_EXIT, "BUN_DESTRUCT_VM_ON_EXIT", {}); - // Test suite only, builds with debug assertions: a worker VM holds every - // cross-thread completion until its teardown is waiting, so the "arrived - // during teardown" paths run deterministically (bun_jsc::vm_handle::test_gate). - new_feature_flag!(pub BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE, "BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE", {}); // Disable "nativeDependencies" new_feature_flag!(pub BUN_FEATURE_FLAG_DISABLE_NATIVE_DEPENDENCY_LINKER, "BUN_FEATURE_FLAG_DISABLE_NATIVE_DEPENDENCY_LINKER", {}); diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index ca925367b68..83a739c682d 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -3968,9 +3968,7 @@ impl VirtualMachine { // SAFETY: `vm` is the unique live VM on this thread. let vm_ref = unsafe { &mut *vm }; vm_ref.worker = Some(std::ptr::from_ref::(worker).cast()); - if worker.arm_test_gate() { - vm_ref.handle.arm_test_gate(); - } + vm_ref.handle.arm_test_gate(worker.test_gate()); // The worker's resolver also // needs the standalone graph, otherwise embedded `/$bunfs/...` specifiers // (e.g. a `new Worker("./worker.ts")` entry point inside a compiled diff --git a/src/jsc/VmHandle.rs b/src/jsc/VmHandle.rs index 0e4d5cce332..7fba62ec8c3 100644 --- a/src/jsc/VmHandle.rs +++ b/src/jsc/VmHandle.rs @@ -118,8 +118,11 @@ pub struct Shared { struct DebugState { js_thread: std::thread::ThreadId, live: Guarded, - /// Test suite only — see [`test_gate`]. - gate: core::sync::atomic::AtomicBool, + /// Test suite only (see [`test_gate`]): the [`TestGate`] mode armed, and + /// the posts the gate is holding for `Draining`, which the wait does not + /// close past. + gate: AtomicU8, + parked: AtomicU32, } #[cfg(debug_assertions)] @@ -314,7 +317,8 @@ impl VmHandle { debug: DebugState { js_thread: std::thread::current().id(), live: Default::default(), - gate: core::sync::atomic::AtomicBool::new(false), + gate: AtomicU8::new(TestGate::Off as u8), + parked: AtomicU32::new(0), }, })) } @@ -437,7 +441,7 @@ impl VmHandle { self.assert_js_thread(); let s = &*self.0; s.hot.state.store(State::Draining as u8, Ordering::SeqCst); - test_gate::draining(self); + test_gate::state_published(self); #[cfg(debug_assertions)] let started = std::time::Instant::now(); #[cfg(debug_assertions)] @@ -450,7 +454,7 @@ impl VmHandle { { continue; } - if s.tickets.load(Ordering::SeqCst) == 0 { + if s.tickets.load(Ordering::SeqCst) == 0 && test_gate::nothing_parked(s) { s.hot.state.store(State::Closed as u8, Ordering::SeqCst); break; } @@ -465,6 +469,7 @@ impl VmHandle { } } } + test_gate::state_published(self); if s.active.load(Ordering::SeqCst) != 0 { let mut g = s.drained.0.lock(); while s.active.load(Ordering::SeqCst) != 0 { @@ -522,33 +527,68 @@ impl VirtualMachine { // // `BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE` (first-level worker VMs; builds with // debug assertions): a post from another thread is held until the worker's -// teardown has begun waiting, so it always arrives *during* the wait — the -// ticketed path (queued, released on the JS thread, then the wait ends) and -// the weak path (queued-and-released while draining, or refused once closed) -// run with their real preconditions every time instead of only when they lose -// the race. Each is named on stderr. The parked thread keeps whatever locks it -// holds (the fetch tasklet's mutex, a streaming body's buffer lock), so a row -// whose worker then blocks on that same lock never reaches teardown: a hang -// under the gate, not in production. +// teardown has reached the state the mode names, so the path under test runs +// with its real preconditions every time instead of only when the post loses +// the race with teardown. `draining` holds every post until the wait has +// begun, and the wait does not close until a post it let through has been +// made: a ticket's completion and a weak post alike are queued and released +// by the wait. `closed` holds a weak post until the wait has ended, and the +// teardown does not go on until that post has been made: it is refused and +// the poster frees its own payload. A ticket's completion cannot be held past +// the wait (the wait is for it), and neither can a weak post made while a +// ticket is outstanding, which may be what that ticket's holder does before +// handing the ticket back (WebCrypto posts its result from the pool task that +// carries the ticket); both park only until `Draining` in either mode. Each +// post is named on stderr with its outcome. The parked thread keeps whatever +// locks it holds (the fetch tasklet's mutex, a streaming body's buffer lock), +// so a row whose worker then blocks on that same lock never reaches teardown: +// a hang under the gate, not in production. + +/// The mode a worker VM arms (see above); `Off` everywhere else. +#[repr(u8)] +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum TestGate { + Off = 0, + Draining = 1, + Closed = 2, +} + #[cfg(debug_assertions)] mod test_gate { - use super::{Ordering, Posted, Shared, State, Ticket, VmHandle}; + use super::{Access, Ordering, Posted, Shared, State, TestGate, Ticket, VmHandle}; type Task = core::ptr::NonNull; + impl TestGate { + /// `BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE=draining|closed`. + pub fn from_env() -> TestGate { + match bun_core::env_var::BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE::get() { + Some(b"draining") => TestGate::Draining, + Some(b"closed") => TestGate::Closed, + _ => TestGate::Off, + } + } + } impl VmHandle { - pub(crate) fn arm_test_gate(&self) { - self.0.debug.gate.store(true, Ordering::Relaxed); + pub(crate) fn arm_test_gate(&self, mode: TestGate) { + self.0.debug.gate.store(mode as u8, Ordering::Relaxed); + } + } + fn mode(s: &Shared) -> TestGate { + match s.debug.gate.load(Ordering::Relaxed) { + 1 => TestGate::Draining, + 2 => TestGate::Closed, + _ => TestGate::Off, } } fn on(s: &Shared) -> bool { - s.debug.gate.load(Ordering::Relaxed) + mode(s) != TestGate::Off } fn armed(s: &Shared) -> bool { on(s) && std::thread::current().id() != s.debug.js_thread } - fn park_until_draining(s: &Shared) { + fn park_until(s: &Shared, state: State) { let mut g = s.drained.0.lock(); - while s.state() < State::Draining { + while s.state() < state { s.drained.1.wait_guarded(&mut g); } } @@ -563,7 +603,7 @@ mod test_gate { pub(super) fn before_ticket_post(t: &Ticket) { if armed(&t.shared) { - park_until_draining(&t.shared); + park_until(&t.shared, State::Draining); let l = *t .shared .debug @@ -583,19 +623,40 @@ mod test_gate { if !armed(s) { return post(task); } - park_until_draining(s); // SAFETY: handed over by the caller and not yet queued anywhere. let tag = unsafe { task.as_ref() }.task.tag; - let r = post(task); - let outcome = match r { - Posted::Queued => "released by the wait", - Posted::Refused(_) => "refused", + let report = |r: &Posted| { + let outcome = match r { + Posted::Queued => "released by the wait", + Posted::Refused(_) => "refused", + }; + say(format_args!("late post: {} ({outcome})", tag.name())); }; - say(format_args!("late post: {} ({outcome})", tag.name())); + if mode(s) == TestGate::Closed && s.tickets.load(Ordering::SeqCst) == 0 { + // Counted as a weak access in progress: `close_and_wait` returns + // only once this thread has posted and reported. + s.active.fetch_add(1, Ordering::SeqCst); + let _in_progress = Access(s); + park_until(s, State::Closed); + let r = post(task); + report(&r); + return r; + } + s.debug.parked.fetch_add(1, Ordering::SeqCst); + park_until(s, State::Draining); + let r = post(task); + report(&r); + s.debug.parked.fetch_sub(1, Ordering::SeqCst); + s.notify(); r } - /// The wait began: parked posts go now. - pub(super) fn draining(h: &VmHandle) { + /// `close_and_wait`: may `Draining` become `Closed`? Not while a post the + /// gate is holding for `Draining` has yet to be made. + pub(super) fn nothing_parked(s: &Shared) -> bool { + s.debug.parked.load(Ordering::SeqCst) == 0 + } + /// `Draining` / `Closed` was just published: posts parked for it go now. + pub(super) fn state_published(h: &VmHandle) { if on(&h.0) { h.0.notify(); } @@ -608,11 +669,17 @@ mod test_gate { } #[cfg(not(debug_assertions))] mod test_gate { - use super::{Posted, Shared, Ticket, VmHandle}; + use super::{Posted, Shared, TestGate, Ticket, VmHandle}; type Task = core::ptr::NonNull; + impl TestGate { + #[inline(always)] + pub fn from_env() -> TestGate { + TestGate::Off + } + } impl VmHandle { #[inline(always)] - pub(crate) fn arm_test_gate(&self) {} + pub(crate) fn arm_test_gate(&self, _: TestGate) {} } #[inline(always)] pub(super) fn before_ticket_post(_: &Ticket) {} @@ -621,7 +688,11 @@ mod test_gate { post(task) } #[inline(always)] - pub(super) fn draining(_: &VmHandle) {} + pub(super) fn nothing_parked(_: &Shared) -> bool { + true + } + #[inline(always)] + pub(super) fn state_published(_: &VmHandle) {} } // ── C++ holds references ────────────────────────────────────────────────── diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index cf2c637b0e5..554ee065ee4 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -64,11 +64,11 @@ pub struct WebWorker { parent: *mut VirtualMachine, /// The parent's `--hot` / `--watch` mode, inherited by the worker VM. hot_reload: crate::virtual_machine::HotReload, - /// Whether the worker VM arms `bun_jsc::vm_handle`'s test gate (debug + /// The `bun_jsc::vm_handle` test gate mode the worker VM arms (debug /// builds, `BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE`, first-level workers only: /// a nested worker parked on a post to its worker parent would keep that /// parent from ever reaching its own wait). - arm_test_gate: bool, + test_gate: crate::vm_handle::TestGate, execution_context_id: u32, mini: bool, eval_mode: bool, @@ -389,10 +389,11 @@ impl WebWorker { messaging_proxy: proxy, parent, hot_reload: parent_ref.hot_reload, - arm_test_gate: cfg!(debug_assertions) - && parent_ref.is_main_thread() - && bun_core::env_var::feature_flag::BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE::get() - .unwrap_or(false), + test_gate: if parent_ref.is_main_thread() { + crate::vm_handle::TestGate::from_env() + } else { + crate::vm_handle::TestGate::Off + }, execution_context_id: this_context_id, mini, eval_mode, @@ -583,8 +584,8 @@ impl WebWorker { } #[inline] - pub(crate) fn arm_test_gate(&self) -> bool { - self.arm_test_gate + pub(crate) fn test_gate(&self) -> crate::vm_handle::TestGate { + self.test_gate } #[inline] diff --git a/test/js/web/workers/worker-late-completion.test.ts b/test/js/web/workers/worker-late-completion.test.ts index 7d27ea293c9..0f00d8c883f 100644 --- a/test/js/web/workers/worker-late-completion.test.ts +++ b/test/js/web/workers/worker-late-completion.test.ts @@ -7,15 +7,19 @@ // waiter thread) holds no ticket; its post is delivered-and-released while the // worker drains, or refused once it has closed, and it frees its own payload. // -// Here each of those paths runs deterministically for one producer at a time: -// with BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE the other thread's post is held -// until the worker's teardown is already waiting, so it always lands *during* -// the wait, and the runtime names it on stderr. A row passes only if the named -// line appeared (the work really was on another thread, really came back -// during teardown, and — for ticketed work — was taken through the door at -// the expected site) and the process exited cleanly; on the ASAN build the -// release paths are also checked for use-after-free and leaks. Builds with -// debug assertions only (debug, ASAN): the gate does not exist in release. +// Here each of those paths runs deterministically for one producer at a time. +// With BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE=draining the other thread's post is +// held until the worker's teardown is already waiting, so it always lands +// *during* the wait and is released by it; with =closed a weak post is held +// until the wait has ended, so it is always refused (a post made while a +// ticket is outstanding is never held past the start of the wait, since the +// wait may be for that very poster). Either way the runtime names the post and +// its outcome on stderr. A row passes only if the expected line appeared (the +// work really was on another thread, really came back at the stage under test +// and, for ticketed work, was taken through the door at the expected site) and +// the process exited cleanly; on the ASAN build the release and refusal paths +// are also checked for use-after-free and leaks. Builds with debug assertions +// only (debug, ASAN): the gate does not exist in release. import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, isAndroid, isASAN, isDebug, isLinux, isWindows, tempDir } from "harness"; import fs from "node:fs"; @@ -23,8 +27,11 @@ import path from "node:path"; type Row = { name: string; - // Runs in the worker before it exits; starts exactly one piece of off-thread work. + // Runs in the worker; starts exactly one piece of off-thread work. The host + // then calls `armed()` (report to the parent, exit two turns later), unless + // the row does so itself once the work is really in flight. worker: string; + armsItself?: true; // Runs in the parent before the worker is created; may add to `data` (workerData). prelude?: string; // Runs in the parent once the worker says it is set up (for producers on the parent's side). @@ -36,10 +43,19 @@ type Row = { skip?: boolean; } & (Ticketed | Weak); // Ticketed work: substring of the site (file) the ticket was taken at, as -// logged by "[vm] late completion from :". -type Ticketed = { ticket: string; weak?: never }; -// Weak posters: the task tag logged by "[vm] late post: (...)". -type Weak = { weak: string; ticket?: never }; +// logged by "[vm] late completion from :". Runs under the draining +// gate; the wait is what releases it, so there is no other outcome to test. +type Ticketed = { ticket: string; weak?: never; underTicket?: never }; +// Weak posters: the task tag logged by "[vm] late post: ()". +// Each runs under both gates: released by the wait under `draining`, refused +// under `closed`. The exception is a post made while a ticket is outstanding +// (`underTicket`): the gate holds it no further than draining under either, so +// it is released by the wait under both. +type Weak = { weak: string; underTicket?: true; ticket?: never }; + +type Gate = "draining" | "closed"; +const RELEASED = "released by the wait"; +const REFUSED = "refused"; const ROWS: Row[] = [ // ── thread pool: bun_jsc::Job ──────────────────────────────────────────── @@ -89,13 +105,16 @@ const ROWS: Row[] = [ }, { // WebCrypto's work queue: a C++ closure on the pool, carried (with a - // ticket) by ConcurrentCppTask; its *result* comes back by context id — - // WebCore's postTaskTo(), a weak post — and because the ticket kept the - // worker draining rather than closed, that post is delivered and its - // promise/callback refs are released on the worker's thread. + // ticket) by ConcurrentCppTask. Its *result* comes back by context id + // (WebCore's postTaskTo(), a weak post), made before the task drops its + // ticket; because that ticket keeps the worker draining rather than + // closed, the post is delivered and its promise/callback refs are released + // on the worker's thread. Under the closed gate this is the row that shows + // the gate not holding such a post past the wait (which would never end). name: "crypto.subtle.digest", worker: `crypto.subtle.digest("SHA-256", Buffer.alloc(65536));`, weak: "CppTask", + underTicket: true, }, { name: "Bun.password.hash", @@ -194,8 +213,15 @@ const ROWS: Row[] = [ }, // ── weak posters (no ticket): delivered while draining, or refused ─────── { + // The waiter thread posts as soon as it has reaped the child. The worker + // cannot see that happen (the post is what the gate holds), so it leaves + // once the child's stdout has hit EOF instead: the child closes it on its + // way out, just before it becomes reapable, and the waiter has microseconds + // of work to do before its post reaches the gate, against the milliseconds + // the worker takes to exit and tear down to its wait. name: "child exit reported by the waiter thread", - worker: `require("node:child_process").execFile(process.execPath, ["-e", "0"], () => {});`, + worker: `Bun.spawn([process.execPath, "-e", "0"], { stdin: "ignore", stdout: "pipe", stderr: "ignore" }).stdout.text().then(armed);`, + armsItself: true, weak: "ProcessWaiterThreadTask", // The waiter thread is a POSIX fallback path, opted into here the way the runtime's own tests do. env: { BUN_GARBAGE_COLLECTOR_LEVEL: "0", BUN_FEATURE_FLAG_FORCE_WAITER_THREAD: "1" }, @@ -226,9 +252,12 @@ function host(row: Row, dir: string) { const worker = ` const { parentPort, workerData } = require("node:worker_threads"); parentPort.on("message", () => {}); + const armed = () => { + parentPort.postMessage("armed"); + setImmediate(() => setImmediate(() => process.exit(0))); + }; ${row.worker} - parentPort.postMessage("armed"); - setImmediate(() => setImmediate(() => process.exit(0))); + ${row.armsItself ? "" : "armed();"} `; return ` const { Worker, MessageChannel } = require("node:worker_threads"); @@ -242,29 +271,41 @@ function host(row: Row, dir: string) { `; } +// Runs the row's host under `gate`; passes iff the host exited cleanly and one +// of the "[vm] " lines it printed satisfies `expected`. +async function expectLateLine(row: Row, gate: Gate, expected: (line: string) => boolean) { + using dir = tempDir("worker-late-completion", row.files ?? {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", host(row, String(dir))], + env: { ...bunEnv, ...row.env, BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE: gate }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const seen = stderr.split("\n").some(l => l.startsWith("[vm] ") && expected(l)); + expect({ + exitCode, + seen, + // On a failure, everything the host printed. + detail: exitCode === 0 && seen ? "" : stdout + stderr, + }).toEqual({ exitCode: 0, seen: true, detail: "" }); +} + describe.skipIf(!isDebug && !isASAN)("work that comes back after its worker began tearing down", () => { for (const row of ROWS) { - test.concurrent.skipIf(!!row.skip)(row.name, async () => { - using dir = tempDir("worker-late-completion", row.files ?? {}); - await using proc = Bun.spawn({ - cmd: [bunExe(), "-e", host(row, String(dir))], - env: { ...bunEnv, ...row.env, BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE: "1" }, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - const lines = stderr.split("\n").filter(l => l.startsWith("[vm] ")); - const seen = - "ticket" in row && row.ticket - ? lines.some(l => l.startsWith("[vm] late completion from ") && l.includes(row.ticket)) - : lines.some(l => l.startsWith(`[vm] late post: ${row.weak} (`)); - expect({ - exitCode, - seen, - // On a failure, everything the host printed. - detail: exitCode === 0 && seen ? "" : stdout + stderr, - }).toEqual({ exitCode: 0, seen: true, detail: "" }); - }); + if (row.ticket) { + const ticket = row.ticket; + test.concurrent.skipIf(!!row.skip)(row.name, () => + expectLateLine(row, "draining", l => l.startsWith("[vm] late completion from ") && l.includes(ticket)), + ); + continue; + } + for (const gate of ["draining", "closed"] as const) { + const outcome = gate === "draining" || row.underTicket ? RELEASED : REFUSED; + test.concurrent.skipIf(!!row.skip)(`${row.name} (${gate} gate: ${outcome})`, () => + expectLateLine(row, gate, l => l === `[vm] late post: ${row.weak} (${outcome})`), + ); + } } }); @@ -357,7 +398,7 @@ describe.skipIf(isWindows)("terminate() waits for work that cannot be cancelled" ], // The gate env also brings the debug build's outstanding-ticket report // forward (2s instead of 10s). - env: { ...bunEnv, BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE: "1" }, + env: { ...bunEnv, BUN_DEBUG_TEST_WORKER_TEARDOWN_GATE: "draining" }, stdout: "pipe", stderr: "pipe", });