Worker teardown test gate: add a closed mode and pin each weak post's outcome

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", {});

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::<crate::web_worker::WebWorker>(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

src/jsc/VmHandle.rs

@@ -118,8 +118,11 @@ pub struct Shared {
 struct DebugState {
     js_thread: std::thread::ThreadId,
     live: Guarded<LiveTickets>,
-    /// 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<super::ConcurrentTaskItem>;
 
+    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<super::ConcurrentTaskItem>;
+    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 ──────────────────────────────────────────────────

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]