inspector: drain queued frontend messages to the debugger thread before exit

src/jsc/Debugger.rs

@@ -169,6 +169,7 @@ unsafe extern "C" {
         is_connect: bool,
         is_node_inspector: bool,
     );
+    safe fn Bun__debugger__drain();
 }
 
 static FUTEX_ATOMIC: AtomicU32 = AtomicU32::new(0);
@@ -354,6 +355,15 @@ impl Debugger {
         }
     }
 
+    /// Block (briefly, capped) until the debugger thread has delivered queued
+    /// inspector messages to the frontend socket. Call from the main thread
+    /// immediately before process exit so `exit()` doesn't kill the detached
+    /// debugger thread mid-delivery (e.g. `bun test`'s final `TestReporter.end`
+    /// events). See `Bun__debugger__drain` in `BunDebugger.cpp`.
+    pub fn drain() {
+        Bun__debugger__drain();
+    }
+
     /// `Debugger.create(vm, global)` — first-time debugger setup: create the
     /// JSC inspector context, spawn the debugger VM thread, and arm the
     /// keep-alive on the parent loop.

src/jsc/VirtualMachine.rs

@@ -1516,6 +1516,13 @@ impl VirtualMachine {
         // doing it causes like 50+ tests to break
         // self.event_loop().tick();
 
+        // Flush queued inspector messages so exit() doesn't kill the detached
+        // debugger thread mid-delivery. The debugger thread runs its own VM,
+        // so this wait never contends with the main VM's API lock.
+        if self.debugger.is_some() {
+            crate::debugger::Debugger::drain();
+        }
+
         if self.should_destruct_main_thread_on_exit() {
             #[cfg(windows)]
             if let Some(t) = self.event_loop_mut().forever_timer.take() {

src/jsc/bindings/BunDebugger.cpp

@@ -4,6 +4,8 @@
 #include "ZigGlobalObject.h"
 
 #include <JavaScriptCore/InspectorFrontendChannel.h>
+#include <JavaScriptCore/TopExceptionScope.h>
+#include <wtf/threads/BinarySemaphore.h>
 #include <JavaScriptCore/JSGlobalObjectDebuggable.h>
 #include <JavaScriptCore/JSGlobalObjectDebugger.h>
 #include <JavaScriptCore/Debugger.h>
@@ -51,6 +53,18 @@ static PausedWait& pausedWait()
     return instance;
 }
 
+// Messages queued via sendMessageToDebuggerThread() that haven't yet reached
+// onMessageFn on the debugger thread. Bun__debugger__drain() reads this only
+// as an entry gate; the FIFO sentinel's ordering is the actual guarantee.
+// Process-global (like debuggerScriptExecutionContext), not per-connection.
+static std::atomic<uint64_t> totalPendingDebuggerMessages { 0 };
+
+// Set once (never cleared) the first time anything is queued for the debugger
+// thread. Gates Bun__debugger__drain()'s socket-flush grace wait: the pending
+// counter above is zero once write() returns even if bytes are still buffered
+// below write(), so that counter can't gate the flush wait.
+static std::atomic<bool> hasQueuedAnyDebuggerMessage { false };
+
 static bool waitingForConnection = false;
 static bool bunControllerInstalled = false;
 // Tracks whether connectFrontend has ever been called on the Bun-installed
@@ -436,13 +450,21 @@ class BunInspectorConnection : public ThreadSafeRefCounted<BunInspectorConnectio
             this->debuggerThreadMessages.swap(messages);
         }
 
-        if (!jsBunDebuggerOnMessageFunction)
+        size_t messageCount = messages.size();
+
+        if (!jsBunDebuggerOnMessageFunction) {
+            // Disconnected: batch is dropped, so account for it here or
+            // Bun__debugger__drain()'s entry gate would count it pending forever.
+            if (messageCount > 0)
+                totalPendingDebuggerMessages.fetch_sub(messageCount, std::memory_order_release);
             return;
+        }
 
         JSFunction* onMessageFn = uncheckedDowncast<JSFunction>(jsBunDebuggerOnMessageFunction.get());
         MarkedArgumentBuffer arguments;
-        arguments.ensureCapacity(messages.size());
+        arguments.ensureCapacity(messageCount);
         auto& vm = debuggerGlobalObject->vm();
+        auto scope = DECLARE_TOP_EXCEPTION_SCOPE(vm);
 
         for (auto& message : messages) {
             arguments.append(jsString(vm, message));
@@ -451,13 +473,29 @@ class BunInspectorConnection : public ThreadSafeRefCounted<BunInspectorConnectio
         messages.clear();
 
         JSC::call(debuggerGlobalObject, onMessageFn, arguments, "BunInspectorConnection::receiveMessagesOnDebuggerThread - onMessageFn"_s);
+
+        // If onMessageFn threw partway through, we can't know how many reached
+        // write(), so leave the count pending (costs at most one capped wait in
+        // Bun__debugger__drain, never a stall).
+        bool delivered = true;
+        if (auto* exception = scope.exception()) [[unlikely]] {
+            delivered = false;
+            if (scope.clearExceptionExceptTermination())
+                debuggerGlobalObject->reportUncaughtExceptionAtEventLoop(debuggerGlobalObject, exception);
+        }
+        if (messageCount > 0 && delivered)
+            totalPendingDebuggerMessages.fetch_sub(messageCount, std::memory_order_release);
     }
 
     void sendMessageToDebuggerThread(WTF::String&& inputMessage)
     {
         {
             Locker<Lock> locker(debuggerThreadMessagesLock);
             debuggerThreadMessages.append(inputMessage);
+            // Inside the lock: incrementing after unlock could race the swap+
+            // decrement in receiveMessagesOnDebuggerThread and undercount.
+            totalPendingDebuggerMessages.fetch_add(1, std::memory_order_release);
+            hasQueuedAnyDebuggerMessage.store(true, std::memory_order_release);
         }
 
         if (this->debuggerThreadMessageScheduledCount++ == 0) {
@@ -668,6 +706,53 @@ extern "C" void Bun__ensureDebugger(ScriptExecutionContextIdentifier scriptId, b
     }
 }
 
+// Ref-counted so a timed-out wait doesn't leave a dangling pointer for the
+// already-posted task: both sides hold a Ref, last one out frees it.
+struct DebuggerDrainSignal : public ThreadSafeRefCounted<DebuggerDrainSignal> {
+public:
+    static Ref<DebuggerDrainSignal> create() { return adoptRef(*new DebuggerDrainSignal()); }
+    WTF::BinarySemaphore semaphore;
+
+private:
+    DebuggerDrainSignal() = default;
+};
+
+// Caps, not targets: a wedged debugger thread must never block exit forever.
+static constexpr double kDrainHandoffTimeoutMs = 250;
+static constexpr double kDrainFlushGraceMs = 150;
+
+// Called from VirtualMachine::global_exit immediately before process exit so
+// exit() doesn't kill the detached debugger thread mid-delivery of queued
+// inspector messages (most visibly `bun test`'s final TestReporter events).
+//
+// Two layers: (a) the main->debugger-thread handoff itself, drained by posting
+// a FIFO sentinel task and waiting for it (tasks on this context are FIFO, so
+// once the sentinel runs every earlier receiveMessagesOnDebuggerThread has
+// already called write()); and (b) a short further grace for the debugger
+// thread's own event loop to flush the socket send buffer. Both waits are
+// capped; on timeout we degrade to pre-fix behavior.
+extern "C" void Bun__debugger__drain()
+{
+    if (debuggerScriptExecutionContext == nullptr)
+        return;
+
+    if (totalPendingDebuggerMessages.load(std::memory_order_acquire) != 0) {
+        auto signal = DebuggerDrainSignal::create();
+        debuggerScriptExecutionContext->postTaskConcurrently([signal](ScriptExecutionContext&) {
+            signal->semaphore.signal();
+        });
+        signal->semaphore.waitFor(WTF::Seconds::fromMilliseconds(kDrainHandoffTimeoutMs));
+    }
+
+    // Gated on hasQueuedAnyDebuggerMessage, not totalPendingDebuggerMessages:
+    // the latter can be zero while bytes are still buffered below write().
+    // BinarySemaphore is used purely for its timeout (nothing signals it).
+    if (hasQueuedAnyDebuggerMessage.load(std::memory_order_acquire)) {
+        WTF::BinarySemaphore topUp;
+        topUp.waitFor(WTF::Seconds::fromMilliseconds(kDrainFlushGraceMs));
+    }
+}
+
 extern "C" void BunDebugger__willHotReload()
 {
     if (debuggerScriptExecutionContext == nullptr) {

test/cli/inspect/test-reporter.test.ts

@@ -1,6 +1,6 @@
 import { Subprocess, spawn, write } from "bun";
 import { afterEach, describe, expect, test } from "bun:test";
-import { bunEnv, bunExe, isPosix, tempDir } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, isPosix, tempDir } from "harness";
 import { join } from "node:path";
 import { InspectorSession, connect } from "./junit-reporter";
 import { SocketFramer } from "./socket-framer";
@@ -168,13 +168,18 @@ class TestReporterSession extends InspectorSession {
 describe.if(isPosix)("TestReporter inspector protocol", () => {
   let proc: Subprocess | undefined;
   let socket: ReturnType<typeof connect> extends Promise<infer T> ? T : never;
+  // The drain test below spawns several subprocesses in parallel; force-kill
+  // any that are still alive if the test times out or fails partway through.
+  const spawnedProcs = new Set<Subprocess>();
 
   afterEach(() => {
     proc?.kill();
     proc = undefined;
     // @ts-ignore - close the socket if it exists
     socket?.end?.();
     socket = undefined as any;
+    for (const p of spawnedProcs) p.kill("SIGKILL");
+    spawnedProcs.clear();
   });
 
   test("retroactively reports tests when TestReporter.enable is called after tests are discovered", async () => {
@@ -310,4 +315,115 @@ afterAll(async () => {
     const exitCode = await proc.exited;
     expect(exitCode).toBe(0);
   });
+
+  test(
+    "flushes pending inspector messages to the frontend before process exit",
+    async () => {
+      // Regression for the exit-race fixed by Bun__debugger__drain: the main
+      // thread queues the final TestReporter events for the detached debugger
+      // thread, then falls through to exit() which can kill that thread
+      // mid-delivery.
+      //
+      // Fast SYNCHRONOUS fixture tests on purpose: they let the runner reach
+      // exit() in the same tick as the last `end` event. A lone run on an idle
+      // machine may pass even without the fix; parallel spawns supply the CPU
+      // contention that makes the race bite reliably.
+      //
+      // Uses --inspect-wait=unix:... (SocketFramer path), not ws:, so this
+      // exercises the main->debugger-thread drain, not WebSocket backpressure.
+      const testCount = 5;
+      const body = Array.from(
+        { length: testCount },
+        (_, i) => `test("t${i}", () => { expect(${i}).toBe(${i}); });`,
+      ).join("\n");
+
+      using dir = tempDir("test-reporter-drain", {
+        "drain.test.ts": `
+import { test, expect } from "bun:test";
+${body}
+`,
+      });
+
+      async function once() {
+        const socketPath = join(String(dir), `inspector-${Math.random().toString(36).substring(2)}.sock`);
+
+        const session = new TestReporterSession();
+        const framer = new SocketFramer((message: string) => {
+          session.onMessage(message);
+        });
+
+        let localSocket: Awaited<ReturnType<typeof connect>>;
+        const socketClosed = Promise.withResolvers<void>();
+        const socketPromise = connect(`unix://${socketPath}`, () => socketClosed.resolve()).then(s => {
+          localSocket = s;
+          session.socket = s;
+          session.framer = framer;
+          s.data = {
+            onData: framer.onData.bind(framer),
+          };
+          return s;
+        });
+
+        const localProc = spawn({
+          cmd: [bunExe(), `--inspect-wait=unix:${socketPath}`, "test", "drain.test.ts"],
+          env: bunEnv,
+          cwd: String(dir),
+          // "ignore" (not "pipe") for stdout: we only read stderr below, and an
+          // undrained "pipe" can deadlock the child once the OS pipe buffer fills.
+          stdout: "ignore",
+          stderr: "pipe",
+        });
+        spawnedProcs.add(localProc);
+
+        await socketPromise;
+
+        // Enable TestReporter before Inspector.initialized so every test is
+        // reported via the normal live-collection path, not retroactively.
+        session.enableInspector();
+        session.enableTestReporter();
+        session.initialize();
+
+        const [stderr, exitCode] = await Promise.all([localProc.stderr.text(), localProc.exited]);
+        spawnedProcs.delete(localProc);
+
+        // Wait for the socket's FIN, not just process exit: FIN is ordered
+        // after data, so once `close` fires every byte the subprocess wrote has
+        // been delivered. Process exit alone wouldn't guarantee this.
+        // @ts-ignore - close the socket if it exists
+        localSocket?.end?.();
+        await socketClosed.promise;
+
+        return {
+          stderr,
+          exitCode,
+          found: session.getFoundTests(),
+          started: session.getStartedTests(),
+          ended: session.getEndedTests(),
+        };
+      }
+
+      // Parallel batch for contention (see above). ASAN/debug subprocess
+      // startup is slow enough that a smaller batch still fits the timeout.
+      const slow = isASAN || isDebug;
+      const width = slow ? 4 : 8;
+      const rounds = slow ? 1 : 2;
+
+      const results: Awaited<ReturnType<typeof once>>[] = [];
+      for (let round = 0; round < rounds; round++) {
+        results.push(...(await Promise.all(Array.from({ length: width }, () => once()))));
+      }
+
+      for (const { stderr, exitCode, found, started, ended } of results) {
+        expect(stderr).toContain(`${testCount} pass`);
+        // Zero trailing loss at any of the three reporting stages.
+        expect(found.size).toBe(testCount);
+        expect(started.size).toBe(testCount);
+        expect(ended.size).toBe(testCount);
+        expect([...ended.values()].map(e => e.status)).toEqual(Array(testCount).fill("pass"));
+        expect(exitCode).toBe(0);
+      }
+    },
+    // Spawns a batch of --inspect-wait subprocesses; ASAN/debug startup is slow.
+    (isASAN || isDebug ? 120 : 60) * 1000,
+  );
 });