diff --git a/src/jsc/bindings/ScriptExecutionContext.cpp b/src/jsc/bindings/ScriptExecutionContext.cpp index d5ada4518581..60f4ca58b9ef 100644 --- a/src/jsc/bindings/ScriptExecutionContext.cpp +++ b/src/jsc/bindings/ScriptExecutionContext.cpp @@ -253,6 +253,18 @@ void ScriptExecutionContext::removeFromContextsMap() allScriptExecutionContextsMap().remove(m_identifier); } +void ScriptExecutionContext::markTerminating() +{ + // postTaskTo() holds this lock across its isTerminating() check and + // postTaskConcurrently() enqueue. Taking it here establishes an ordering + // with every concurrent poster: either its whole critical section ran + // before ours (task enqueued, and the caller's subsequent concurrent-queue + // drain will see it), or ours ran first (poster observes true and drops + // the task instead of enqueueing onto a queue that will never drain). + Locker locker { allScriptExecutionContextsMapLock }; + m_isTerminating.store(true, std::memory_order_release); +} + ScriptExecutionContext* executionContext(JSC::JSGlobalObject* globalObject) { if (!globalObject || !globalObject->inherits()) @@ -290,4 +302,10 @@ extern "C" JSC::JSGlobalObject* ScriptExecutionContextIdentifier__getGlobalObjec return context->globalObject(); } +extern "C" void ScriptExecutionContext__markTerminating(JSC::JSGlobalObject* globalObject) +{ + if (auto* context = defaultGlobalObject(globalObject)->scriptExecutionContext()) + context->markTerminating(); +} + } // namespace WebCore diff --git a/src/jsc/bindings/ScriptExecutionContext.h b/src/jsc/bindings/ScriptExecutionContext.h index e7a68ce4fd13..bc0ec884eb46 100644 --- a/src/jsc/bindings/ScriptExecutionContext.h +++ b/src/jsc/bindings/ScriptExecutionContext.h @@ -130,7 +130,10 @@ class ScriptExecutionContext : public CanMakeWeakPtr, pu // Set once when the context is permanently shutting down (WebWorker__teardownJSCVM). // Unlike VM::hasTerminationRequest(), never set transiently (node:vm {timeout}). - void markTerminating() { m_isTerminating.store(true, std::memory_order_release); } + // Takes allScriptExecutionContextsMapLock so it serializes with postTaskTo's + // check-then-enqueue; a caller that drains the concurrent queue after this + // returns will observe every task enqueued before the flag flipped. + void markTerminating(); bool isTerminating() const { return m_isTerminating.load(std::memory_order_acquire); } // Non-null once this thread joins a `worker_threads` SHARE_ENV tree; every diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index e94c6266d17b..28301a11f122 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -204,6 +204,9 @@ unsafe extern "C" { // ABI-identical to non-null `*const`); C++ mutating VM state through it is // interior to the cell. safe fn WebWorker__teardownJSCVM(global: &JSGlobalObject); + // safe: opaque `&JSGlobalObject` handle (see above); takes the contexts-map + // lock and flips an atomic flag, no Rust-visible state touched. + safe fn ScriptExecutionContext__markTerminating(global: &JSGlobalObject); // safe: `cpp_worker` is an opaque round-trip pointer owned by C++ (allocated // there, stored in `WebWorker.cpp_worker`, and only ever passed back to C++ // — never dereferenced as Rust data); same contract as `JSC__VM__holdAPILock`'s @@ -1277,6 +1280,15 @@ impl WebWorker { // is step 3 below). rare.close_all_socket_groups(unsafe { &*vm_ptr }); } + // Stop cross-thread posters first: markTerminating() serializes + // with postTaskTo() on the contexts-map lock, so after this call + // every task another thread has already enqueued is visible to the + // drain below and no new one can land. teardownJSCVM() will call + // it again (redundantly) after the drain; without this earlier + // call a parent-side MessagePort ack (worker stdio backpressure) + // posted in the gap would sit in concurrent_tasks past the raw VM + // dealloc and leak under LSan. + ScriptExecutionContext__markTerminating(vm.global()); // Reclaim queued CppTasks (the per-worker stdio/messaging // MessagePort drain tasks that can be in self.tasks mid-tick when // terminate() lands, and any Worker dispatchExit close task from a diff --git a/test/js/node/worker_threads/worker-shutdown-post-leak.test.ts b/test/js/node/worker_threads/worker-shutdown-post-leak.test.ts new file mode 100644 index 000000000000..b6b03722c6be --- /dev/null +++ b/test/js/node/worker_threads/worker-shutdown-post-leak.test.ts @@ -0,0 +1,49 @@ +import { expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, isWindows } from "harness"; +import { join } from "path"; + +// A worker's shutdown used to drain its concurrent queue and only then mark +// the context terminating. A cross-thread postTaskTo landing in between (the +// parent's stdio-backpressure ack, any MessagePort scheduleDrain) was enqueued +// onto a queue that is never drained again, leaking the ConcurrentTask + +// EventLoopTask. The window is a handful of instructions so debug builds +// essentially never hit it; CI's release-asan lane does (see +// test/js/node/test/parallel/test-worker-stdio-flush.js). This runs the +// worker-stdio-on-exit scenario under LSan as a guard on the asan lane. +test.skipIf(!isASAN || isWindows)( + "cross-thread MessagePort post during worker shutdown does not leak a ConcurrentTask", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("worker_threads"); + const assert = require("assert"); + const w = new Worker( + 'process.on("exit", () => {' + + ' process.stdout.write(" ");' + + ' process.stdout.write("world");' + + '});' + + 'process.stdout.write("hello");', + { eval: true, stdout: true }, + ); + let data = ""; + w.stdout.setEncoding("utf8"); + w.stdout.on("data", chunk => { data += chunk; }); + w.on("exit", () => assert.strictEqual(data, "hello world")); + `, + ], + env: { + ...bunEnv, + BUN_DESTRUCT_VM_ON_EXIT: "1", + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=1"].filter(Boolean).join(":"), + LSAN_OPTIONS: `print_suppressions=0:suppressions=${join(import.meta.dirname, "../../../leaksan.supp")}`, + }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "", stderr: "", exitCode: 0 }); + }, +);