worker: mark the context terminating before the final concurrent-queue drain

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<JSDOMGlobalObject>())
@@ -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

src/jsc/bindings/ScriptExecutionContext.h

@@ -130,7 +130,10 @@ class ScriptExecutionContext : public CanMakeWeakPtr<ScriptExecutionContext>, 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

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

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