bun test --parallel: act on SIGINT/SIGTERM immediately instead of at the next housekeeping timer

src/runtime/cli/test/parallel/Coordinator.rs

@@ -7,7 +7,7 @@
 use core::ffi::c_void;
 #[cfg(unix)]
 use core::mem::MaybeUninit;
-use core::sync::atomic::{AtomicBool, Ordering};
+use core::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
 use std::io::Write as _;
 
 use bun_core::strings;
@@ -921,16 +921,29 @@ fn describe_status<'b>(buf: &'b mut [u8; 32], status: &SpawnStatus) -> &'b [u8]
     }
 }
 
-/// Coordinator-side SIGINT/SIGTERM handling. The signal handler only sets a
-/// flag; `Coordinator::drive` checks it and tears down workers itself so we
-/// don't do non-signal-safe work in the handler. Linux PDEATHSIG and the
-/// Windows Job Object are the safety net for when the coordinator can't run
-/// this (SIGKILL).
+/// Coordinator-side SIGINT/SIGTERM handling. The handler sets a flag and wakes
+/// the loop (the poll retries on EINTR, so the flag alone would wait for the
+/// next ~1s housekeeping timer); `Coordinator::drive` does the teardown. Linux
+/// PDEATHSIG and the Windows Job Object cover the case where it can't (SIGKILL).
 pub(crate) mod abort_handler {
     use super::*;
 
     pub(crate) static SHOULD_ABORT: AtomicBool = AtomicBool::new(false);
 
+    /// Set by `install`, cleared by `uninstall`. The uws loop is process-lifetime.
+    static LOOP: AtomicPtr<bun_uws::Loop> = AtomicPtr::new(core::ptr::null_mut());
+
+    /// Signal context on POSIX; the console control thread on Windows.
+    fn request_abort() {
+        SHOULD_ABORT.store(true, Ordering::Release);
+        let loop_ = LOOP.load(Ordering::Acquire);
+        if !loop_.is_null() {
+            // SAFETY: the loop is process-lifetime, and the raw extern is the
+            // thread-safe entry point (no `&mut Loop` formed off the main thread).
+            unsafe { bun_uws::us_wakeup_loop(loop_) };
+        }
+    }
+
     // PORTING.md §Global mutable state: written once in `install()` (single
     // call site), read once in `uninstall()`. RacyCell — `sigaction` is POD,
     // no concurrent access.
@@ -943,7 +956,7 @@ pub(crate) mod abort_handler {
 
     #[cfg(unix)]
     extern "C" fn posix_handler(_: i32, _: *const libc::siginfo_t, _: *const c_void) {
-        SHOULD_ABORT.store(true, Ordering::Release);
+        request_abort();
     }
 
     #[cfg(windows)]
@@ -953,7 +966,7 @@ pub(crate) mod abort_handler {
         use bun_sys::windows;
         match ctrl {
             windows::CTRL_C_EVENT | windows::CTRL_BREAK_EVENT | windows::CTRL_CLOSE_EVENT => {
-                SHOULD_ABORT.store(true, Ordering::Release);
+                request_abort();
                 windows::TRUE
             }
             _ => windows::FALSE,
@@ -971,7 +984,9 @@ pub(crate) mod abort_handler {
         }
     }
 
-    pub(crate) fn install() -> Guard {
+    /// `loop_` is the loop `Coordinator::drive` ticks (`ensure_waker` already run).
+    pub(crate) fn install(loop_: *mut bun_uws::Loop) -> Guard {
+        LOOP.store(loop_, Ordering::Release);
         #[cfg(unix)]
         {
             // SAFETY: signal handler installation; PREV_* are written before
@@ -1031,5 +1046,6 @@ pub(crate) mod abort_handler {
                 bun_sys::windows::FALSE,
             );
         }
+        LOOP.store(core::ptr::null_mut(), Ordering::Release);
     }
 }

src/runtime/cli/test/parallel/runner.rs

@@ -191,7 +191,13 @@ pub(crate) fn run_as_coordinator(
         windows_job: Coordinator::create_windows_kill_on_close_job(),
     };
 
-    let _abort_guard = abort_handler::install();
+    // SAFETY: event_loop pointer is valid while vm lives.
+    let uws_loop = unsafe {
+        let event_loop = (*vm_ptr).event_loop();
+        (*event_loop).ensure_waker();
+        (*event_loop).usockets_loop()
+    };
+    let _abort_guard = abort_handler::install(uws_loop);
 
     // Patch the Worker→Coordinator backref now that `coord`'s address is fixed.
     // Access workers through `coord.workers` to avoid a second &mut on the Vec.
@@ -209,8 +215,6 @@ pub(crate) fn run_as_coordinator(
         }
     }
 
-    // SAFETY: event_loop pointer is valid while vm lives.
-    unsafe { (*(*vm_ptr).event_loop()).ensure_waker() };
     // SAFETY: see vm_ptr note above.
     unsafe { &*vm_ptr }.run_with_api_lock(|| coord.drive());
     coord.end_group();

test/cli/test/parallel.test.ts

@@ -1,5 +1,6 @@
 import { expect, test } from "bun:test";
-import { bunEnv, bunExe, isASAN, isDebug, isWindows, normalizeBunSnapshot, tempDir, tls } from "harness";
+import { readFileSync } from "fs";
+import { bunEnv, bunExe, isASAN, isDebug, isLinux, isWindows, normalizeBunSnapshot, tempDir, tls } from "harness";
 
 test("--parallel: each worker has a unique JEST_WORKER_ID and BUN_TEST_WORKER_ID", async () => {
   // Sleep so worker 0 is busy when workers 1/2 come online and pick up the
@@ -1215,6 +1216,86 @@ test("--parallel: SIGTERM on coordinator kills workers and their grandchildren",
   expect(outstanding).toEqual([]);
 }, 15000);
 
+// With every worker mid-file, the only thing that wakes the coordinator's poll is
+// its own housekeeping (a burst of timers about once a second). A signal handler
+// that does not wake the poll is therefore acted on at the next burst, 0 to ~1s
+// later, depending on when the signal lands. On Linux, wait for a burst (the main
+// thread's voluntary context switch count moves) and for it to settle, so the
+// signal is sent at the start of the ~1s quiet period and such a coordinator
+// reliably takes ~900ms; one that wakes its poll reacts at once either way.
+async function waitForCoordinatorToPark(pid: number) {
+  const switches = () =>
+    Number(/voluntary_ctxt_switches:\s+(\d+)/.exec(readFileSync(`/proc/${pid}/status`, "utf8"))![1]);
+  const deadline = Date.now() + 3000;
+  const before = switches();
+  while (switches() === before && Date.now() < deadline) await Bun.sleep(1);
+  let last = switches();
+  let quietSince = Date.now();
+  while (Date.now() - quietSince < 100 && Date.now() < deadline) {
+    await Bun.sleep(1);
+    const now = switches();
+    if (now !== last) {
+      last = now;
+      quietSince = Date.now();
+    }
+  }
+}
+
+test.skipIf(isWindows).concurrent.each(["SIGTERM", "SIGINT"] as const)(
+  "--parallel: coordinator acts on %s immediately while every worker is mid-file",
+  async signal => {
+    const fixture = `
+      import { test } from "bun:test";
+      import { appendFileSync } from "fs";
+      test("slow", async () => {
+        appendFileSync(process.env.PIDS, process.pid + "\\n");
+        await Bun.sleep(60000);
+      });
+    `;
+    using dir = tempDir("parallel-abort-latency", { "a.test.ts": fixture, "b.test.ts": fixture });
+    const pids = String(dir) + "/pids.txt";
+
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "test", "--parallel=2", "--parallel-delay=0"],
+      env: { ...bunEnv, PIDS: pids },
+      cwd: String(dir),
+      stdout: "ignore",
+      stderr: "pipe",
+    });
+
+    const workersRunning = () => {
+      try {
+        return readFileSync(pids, "utf8").split("\n").filter(Boolean).length;
+      } catch {
+        return 0;
+      }
+    };
+    while (workersRunning() < 2 && proc.exitCode === null) await Bun.sleep(10);
+    expect(workersRunning()).toBe(2);
+    if (isLinux) await waitForCoordinatorToPark(proc.pid);
+
+    // The "Interrupted" report is the first thing the coordinator does once it
+    // has seen the signal; measure up to that rather than up to exit so process
+    // teardown (slow under ASAN) stays out of the number.
+    const stderr = proc.stderr.getReader();
+    const decoder = new TextDecoder();
+    let seen = "";
+    const start = performance.now();
+    proc.kill(signal);
+    while (!seen.includes("Interrupted")) {
+      const { value, done } = await stderr.read();
+      if (done) break;
+      seen += decoder.decode(value, { stream: true });
+    }
+    const reactedAfterMs = performance.now() - start;
+
+    expect(seen).toContain("Interrupted");
+    expect(reactedAfterMs).toBeLessThan(500);
+    expect(await proc.exited).toBe(130);
+  },
+  15000,
+);
+
 test("--parallel --no-isolate: a worker keeps one global and module registry across its files", async () => {
   const files: Record<string, string> = {
     "shared.ts": `export let count = 0; export const bump = () => ++count;`,