diff --git a/src/jsc/Debugger.rs b/src/jsc/Debugger.rs index da6c7a0aa0a1..919458e13501 100644 --- a/src/jsc/Debugger.rs +++ b/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,11 @@ impl Debugger { } } + /// Bounded wait for the debugger thread to deliver queued inspector messages before `exit()` kills it mid-delivery; 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. diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index ca5a75f82164..5f3a1cf46b28 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -1681,6 +1681,11 @@ impl VirtualMachine { // doing it causes like 50+ tests to break // self.event_loop().tick(); + if self.debugger.is_some() { + // Flush queued inspector messages so exit() doesn't kill the detached debugger thread mid-delivery; that thread runs its own VM so this never contends with the main VM's API lock. + crate::debugger::Debugger::drain(); + } + if self.should_destruct_main_thread_on_exit() { // SAFETY: main-thread VM on the main thread; exit handlers have run. unsafe { Self::teardown(core::ptr::from_mut(self), Teardown::MainThreadExit) }; diff --git a/src/jsc/bindings/BunDebugger.cpp b/src/jsc/bindings/BunDebugger.cpp index 4dc13cd2a11a..f8158cf81227 100644 --- a/src/jsc/bindings/BunDebugger.cpp +++ b/src/jsc/bindings/BunDebugger.cpp @@ -4,6 +4,8 @@ #include "ZigGlobalObject.h" #include +#include +#include #include #include #include @@ -50,6 +52,9 @@ static PausedWait& pausedWait() return instance; } +// Entry gate for Bun__debugger__drain (process-global, not per-connection). +static std::atomic totalPendingDebuggerMessages { 0 }; + static bool waitingForConnection = false; static bool bunControllerInstalled = false; // Tracks whether connectFrontend has ever been called on the Bun-installed @@ -435,13 +440,20 @@ class BunInspectorConnection : public ThreadSafeRefCounteddebuggerThreadMessages.swap(messages); } - if (!jsBunDebuggerOnMessageFunction) + size_t messageCount = messages.size(); + + if (!jsBunDebuggerOnMessageFunction) { + // Disconnected: batch dropped, account for it so Bun__debugger__drain's gate doesn't count it pending forever. + if (messageCount > 0) + totalPendingDebuggerMessages.fetch_sub(messageCount, std::memory_order_release); return; + } JSFunction* onMessageFn = uncheckedDowncast(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)); @@ -450,6 +462,16 @@ class BunInspectorConnection : public ThreadSafeRefCountedreportUncaughtExceptionAtEventLoop(debuggerGlobalObject, exception); + } + if (messageCount > 0 && delivered) + totalPendingDebuggerMessages.fetch_sub(messageCount, std::memory_order_release); } void sendMessageToDebuggerThread(WTF::String&& inputMessage) @@ -457,6 +479,8 @@ class BunInspectorConnection : public ThreadSafeRefCounted locker(debuggerThreadMessagesLock); debuggerThreadMessages.append(inputMessage); + // Inside the lock so receiveMessagesOnDebuggerThread's swap+decrement can't undercount. + totalPendingDebuggerMessages.fetch_add(1, std::memory_order_release); } if (this->debuggerThreadMessageScheduledCount++ == 0) { @@ -667,6 +691,35 @@ 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. +struct DebuggerDrainSignal : public ThreadSafeRefCounted { +public: + static Ref create() { return adoptRef(*new DebuggerDrainSignal()); } + WTF::BinarySemaphore semaphore; + +private: + DebuggerDrainSignal() = default; +}; + +static constexpr double kDrainHandoffTimeoutMs = 250; + +// Called from VirtualMachine::global_exit so exit() doesn't kill the detached debugger thread mid-delivery of queued inspector messages (e.g. `bun test`'s final TestReporter events). +extern "C" void Bun__debugger__drain() +{ + if (debuggerScriptExecutionContext == nullptr) + return; + + if (totalPendingDebuggerMessages.load(std::memory_order_acquire) == 0) + return; + + // FIFO sentinel: tasks on this context are FIFO, so once it runs every earlier receiveMessagesOnDebuggerThread has already called write() on the socket. + auto signal = DebuggerDrainSignal::create(); + debuggerScriptExecutionContext->postTaskConcurrently([signal](ScriptExecutionContext&) { + signal->semaphore.signal(); + }); + signal->semaphore.waitFor(WTF::Seconds::fromMilliseconds(kDrainHandoffTimeoutMs)); +} + extern "C" void BunDebugger__willHotReload() { if (debuggerScriptExecutionContext == nullptr) { diff --git a/test/cli/inspect/test-reporter.test.ts b/test/cli/inspect/test-reporter.test.ts index 45dd1d10d451..ea660109bc3c 100644 --- a/test/cli/inspect/test-reporter.test.ts +++ b/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,6 +168,9 @@ class TestReporterSession extends InspectorSession { describe.if(isPosix)("TestReporter inspector protocol", () => { let proc: Subprocess | undefined; let socket: ReturnType extends Promise ? 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(); afterEach(() => { proc?.kill(); @@ -175,6 +178,8 @@ describe.if(isPosix)("TestReporter inspector protocol", () => { // @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 () => { @@ -455,4 +460,115 @@ afterAll(async () => { } 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>; + const socketClosed = Promise.withResolvers(); + 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>[] = []; + 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, + ); });