Worker: flush the worker→parent inbox before 'close' on natural exit

src/jsc/bindings/webcore/Worker.cpp

@@ -603,40 +603,71 @@ bool Worker::dispatchExit(int32_t exitCode)
         m_parentContextId,
         [this] { this->deref(); },
         [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext&) {
-            // Closing → dispatch 'close' → Closed. The split lets 'close'/'exit'
-            // handlers observe threadId == -1 and isOnline() == false while
-            // postMessage() (gated only on Closed) still accepts and drops the
-            // message, matching browser/Node and pre-refactor behaviour.
-            //
-            // Drop any tasks queued while the worker was Pending and never
-            // reached Running (m_pendingCrossVMRequests / rejectAllCrossVMRequests
-            // settles the callers' promises + frees the parent-VM Strong<>).
-            // Take the queue under the same lock that flips m_state so a racing
-            // postTaskToWorkerGlobalScope either lands in the cleared queue or
-            // sees Closing and returns false.
-            {
-                Locker lock(protectedThis->m_pendingTasksMutex);
-                protectedThis->m_state.store(State::Closing);
-                protectedThis->m_pendingTasks.clear();
-            }
-            // Reject any introspection promises whose round-trip never completed.
-            if (auto* ctx = protectedThis->scriptExecutionContext())
-                protectedThis->rejectAllCrossVMRequests(ctx->globalObject());
-
-            if (protectedThis->hasEventListeners(eventNames().closeEvent)) {
-                auto event = CloseEvent::create(exitCode == 0, static_cast<unsigned short>(exitCode), exitCode == 0 ? "Worker terminated normally"_s : "Worker exited abnormally"_s);
-                protectedThis->EventTargetWithInlineData::dispatchEvent(event);
-            }
-
-            protectedThis->m_state.store(State::Closed);
-            WebWorker__releaseParentPollRef(protectedThis->impl_);
+            protectedThis->closeTask(exitCode);
             // protectedThis (and the JSWorker GC cell, if still rooted) keep us
             // alive across the close-event dispatch; both deref on the parent
             // thread (lambda destruction here / GC sweep), so ~Worker never runs
             // on the worker thread.
         });
 }
 
+void Worker::closeTask(int32_t exitCode)
+{
+    // The worker VM is already torn down (this task is shutdown() step 4), so
+    // m_toParent cannot grow. If messages remain (a drainToParent hit its yield
+    // budget and re-posted itself behind us), re-post this task on the SAME
+    // lane the drain reschedule uses and return. In FIFO order each re-post
+    // lands behind one drain turn, so close re-checks once per turn and only
+    // proceeds once the inbox is empty and no drain is scheduled: every posted
+    // message is delivered before 'close', and the drain's per-turn budget
+    // still paces delivery instead of one unbounded synchronous flush.
+    //
+    // terminate() skips the chase: dispatchEvent() is already a no-op under
+    // m_terminateRequested, so any undrained tail is discarded and close fires
+    // on the next turn.
+    if (!m_terminateRequested.load()) {
+        bool pending;
+        {
+            Locker locker { m_toParent.lock };
+            pending = !m_toParent.queue.isEmpty() || m_toParent.drainScheduled.load(std::memory_order_relaxed);
+        }
+        if (pending) {
+            postTaskToParent([exitCode, protectedThis = Ref { *this }](ScriptExecutionContext&) {
+                protectedThis->closeTask(exitCode);
+            });
+            return;
+        }
+    }
+
+    // Closing → dispatch 'close' → Closed. The split lets 'close'/'exit'
+    // handlers observe threadId == -1 and isOnline() == false while
+    // postMessage() (gated only on Closed) still accepts and drops the
+    // message, matching browser/Node and pre-refactor behaviour.
+    //
+    // Drop any tasks queued while the worker was Pending and never
+    // reached Running (m_pendingCrossVMRequests / rejectAllCrossVMRequests
+    // settles the callers' promises + frees the parent-VM Strong<>).
+    // Take the queue under the same lock that flips m_state so a racing
+    // postTaskToWorkerGlobalScope either lands in the cleared queue or
+    // sees Closing and returns false.
+    {
+        Locker lock(m_pendingTasksMutex);
+        m_state.store(State::Closing);
+        m_pendingTasks.clear();
+    }
+    // Reject any introspection promises whose round-trip never completed.
+    if (auto* ctx = scriptExecutionContext())
+        rejectAllCrossVMRequests(ctx->globalObject());
+
+    if (hasEventListeners(eventNames().closeEvent)) {
+        auto event = CloseEvent::create(exitCode == 0, static_cast<unsigned short>(exitCode), exitCode == 0 ? "Worker terminated normally"_s : "Worker exited abnormally"_s);
+        EventTargetWithInlineData::dispatchEvent(event);
+    }
+
+    m_state.store(State::Closed);
+    WebWorker__releaseParentPollRef(impl_);
+}
+
 // ---- extern "C" shims (called from native code) ------------------------------
 
 extern "C" void WebWorker__teardownJSCVM(Zig::GlobalObject* globalObject)

src/jsc/bindings/webcore/Worker.h

@@ -173,6 +173,7 @@ class Worker final : public ThreadSafeRefCounted<Worker>, public EventTargetWith
 
     void enqueueToWorker(MessageWithMessagePorts&&);
     void drainToParent(ScriptExecutionContext&);
+    void closeTask(int32_t exitCode);
 
     WorkerOptions m_options;
 

test/js/node/worker_threads/worker_threads.test.ts

@@ -1652,6 +1652,73 @@ test("MessagePort: transferring a port from inside its own close()'s flush windo
   B2.close();
 });
 
+test("parentPort.postMessage queue is flushed to the parent before 'exit' fires", async () => {
+  // A worker that posts a burst and then falls off the end of its script must
+  // deliver every message (node: the worker's parentPort queue drains before
+  // teardown). The parent's drain loop yields after ~1000 messages and
+  // re-posts itself; without the pre-'close' flush that re-post lands behind
+  // the close task, which flips the Worker to Closed and makes the re-posted
+  // drain's dispatches silent no-ops.
+  //
+  // Subprocess because the handler busy-spins (would stall the test runner).
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+       const { Worker } = require("node:worker_threads");
+       const N = 5000;
+       const flag = new Int32Array(new SharedArrayBuffer(8));
+       const w = new Worker(
+         \`const { parentPort, workerData } = require("node:worker_threads");
+          const flag = new Int32Array(workerData);
+          parentPort.postMessage(-1);
+          Atomics.wait(flag, 0, 0);
+          for (let i = 0; i < \${N}; i++) parentPort.postMessage(i);
+          parentPort.postMessage("END");
+          process.on("exit", () => Atomics.store(flag, 1, 1));\`,
+         { eval: true, workerData: flag.buffer },
+       );
+       let got = 0, gotEnd = false, first = true;
+       w.on("message", m => {
+         if (first) {
+           first = false;
+           // Hold this handler (and so the current drain pass) open until the
+           // worker has enqueued the whole burst AND posted its close task, so
+           // this drain pass's reschedule lands behind it. flag[1] flips in
+           // process.on('exit'), which is shutdown() step 2; the close task is
+           // posted in step 4 after step 3's full sync GC (teardownJSCVM). No
+           // worker-side JS runs after step 2, so nothing observable marks the
+           // actual post: the busy-wait below covers steps 3+4. Wider on
+           // debug/ASAN where collectNow is slower; in practice the 999
+           // remaining dispatches after this handler returns buy additional
+           // margin on the same slow builds.
+           Atomics.store(flag, 0, 1);
+           Atomics.notify(flag, 0);
+           while (Atomics.load(flag, 1) === 0);
+           const t = Date.now(); while (Date.now() - t < ${isDebug ? 3000 : 500});
+           return;
+         }
+         if (m === "END") gotEnd = true; else got++;
+       });
+       w.on("error", e => { console.error(e); process.exit(1); });
+       w.on("exit", code => {
+         console.log(JSON.stringify({ got, gotEnd, code }));
+         process.exit(gotEnd && got === N && code === 0 ? 0 : 1);
+       });
+      `,
+    ],
+    env: bunEnv,
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ out: stdout.trim(), stderr, exitCode }).toEqual({
+    out: '{"got":5000,"gotEnd":true,"code":0}',
+    stderr: "",
+    exitCode: 0,
+  });
+});
+
 test("MessagePort: peer closing while a port is in transit still delivers 'close' and doesn't hang", async () => {
   await using proc = Bun.spawn({
     cmd: [

test/js/web/workers/worker.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, test } from "bun:test";
 import { once } from "events";
-import { bunEnv, bunExe } from "harness";
+import { bunEnv, bunExe, isDebug } from "harness";
 import path from "path";
 import wt from "worker_threads";
 
@@ -310,6 +310,69 @@ describe("web worker", () => {
     });
   });
 
+  test("messages posted before natural exit are all delivered before 'close'", async () => {
+    // A worker that posts a burst and then falls off the end of its script
+    // must deliver every message before 'close'. The parent's drain loop
+    // yields after ~1000 messages and re-posts itself; without the pre-close
+    // flush in dispatchExit that re-post lands behind the close task, which
+    // flips the Worker to Closed and makes the re-posted drain's dispatches
+    // silent no-ops.
+    //
+    // Subprocess because the handler busy-spins to pin the interleaving.
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+         const N = 5000;
+         const flag = new Int32Array(new SharedArrayBuffer(8));
+         const src = \`self.onmessage = e => {
+           const flag = new Int32Array(e.data);
+           self.onmessage = null;
+           postMessage(-1);
+           Atomics.wait(flag, 0, 0);
+           for (let i = 0; i < \${N}; i++) postMessage(i);
+           postMessage({ done: true });
+           process.on("exit", () => Atomics.store(flag, 1, 1));
+         };\`;
+         const w = new Worker(URL.createObjectURL(new Blob([src], { type: "text/javascript" })));
+         w.postMessage(flag.buffer);
+         let got = 0, sawDone = false, first = true;
+         w.onmessage = e => {
+           if (first) {
+             first = false;
+             // Hold this handler (and so the current drain pass) open until the
+             // worker has enqueued the whole burst AND posted its close task, so
+             // this drain pass's reschedule lands behind it. flag[1] flips in
+             // process.on('exit') (shutdown step 2); the close task is posted
+             // in step 4 after step 3's full sync GC. No worker-side JS runs
+             // after step 2, so the busy-wait below covers steps 3+4.
+             Atomics.store(flag, 0, 1);
+             Atomics.notify(flag, 0);
+             while (Atomics.load(flag, 1) === 0);
+             const t = Date.now(); while (Date.now() - t < ${isDebug ? 3000 : 500});
+             return;
+           }
+           if (e.data && e.data.done) sawDone = true; else got++;
+         };
+         w.onerror = e => { console.error(e.message); process.exit(1); };
+         w.addEventListener("close", () => {
+           console.log(JSON.stringify({ got, sawDone }));
+           process.exit(sawDone && got === N ? 0 : 1);
+         });
+        `,
+      ],
+      env: bunEnv,
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ out: stdout.trim(), stderr, exitCode }).toEqual({
+      out: '{"got":5000,"sawDone":true}',
+      stderr: "",
+      exitCode: 0,
+    });
+  }, 30_000);
+
   describe("worker event", () => {
     test("is fired with the right object", () => {
       const { promise, resolve } = Promise.withResolvers();