From 8f23415148eff343e832119ed918f31bb95d83fb Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:35:55 +0000 Subject: [PATCH 1/6] worker_threads: flush parentPort queue to the parent before firing 'exit' A worker that posts a burst via parentPort.postMessage() and then exits (natural end-of-script or process.exit) was silently dropping the tail of that burst. enqueueToParent() coalesces: N posts schedule one drainToParent task, which dispatches up to max(queue.size(), 1000) messages and then re-posts itself to yield to the event loop. When the worker is still posting while the parent is draining, the re-post lands behind the close task that dispatchExit() queues, and once the close task flips m_state to Closed every subsequent Worker::dispatchEvent() is a no-op, so the re-posted drain quietly discards whatever is left. Node flushes the worker's parentPort queue before teardown and delivers every message ahead of 'exit'. Match that by calling drainToParent() at the top of the close task: the worker's VM is already torn down by the time the close task runs (it is posted from shutdown() step 4), so m_toParent cannot grow and a single pass empties it. terminate() keeps dropping pending messages because dispatchEvent() is separately gated on m_terminateRequested. --- src/jsc/bindings/webcore/Worker.cpp | 15 ++++- .../worker_threads/worker_threads.test.ts | 60 +++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 1a97f67c2571..73eb2c215a09 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -602,7 +602,20 @@ bool Worker::dispatchExit(int32_t exitCode) return ScriptExecutionContext::postTaskTo( m_parentContextId, [this] { this->deref(); }, - [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext&) { + [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext& context) { + // Flush the worker→parent inbox before 'close'. Node delivers every + // parentPort.postMessage() that ran before the worker exited, then + // fires 'exit'; without this, a drainToParent that hit its yield + // budget while the worker was still posting has re-posted itself + // BEHIND this close task, and once m_state == Closed dispatchEvent() + // drops the rest on the floor. The worker thread has already torn + // down its VM (this task is posted from shutdown() step 4), so + // m_toParent cannot grow and one drainToParent pass empties it + // (limit = max(queue.size(), 1000) ≥ queue.size()). terminate() + // still drops pending messages: dispatchEvent() is gated on + // m_terminateRequested, so the dispatches below are no-ops. + protectedThis->drainToParent(context); + // 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 diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 68d6c6f3f103..f2ad684fd18e 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -1652,6 +1652,66 @@ 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 entered shutdown, so the + // worker's close task is posted before this drain pass reschedules. + Atomics.store(flag, 0, 1); + Atomics.notify(flag, 0); + while (Atomics.load(flag, 1) === 0); + const t = Date.now(); while (Date.now() - t < 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: [ From 68efadcf21618e09677adb955d828ab552baaf9f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:51:27 +0000 Subject: [PATCH 2/6] test: widen the shutdown pad on debug/ASAN and document why The Atomics handshake can observe process.on('exit') (shutdown step 2) but nothing JS-visible runs between there and dispatchExit (step 4, after the full sync GC in step 3). The busy-wait covering that gap is a timing pad, not a synchronization point; widen it on debug/ASAN so the test keeps its regression-guard property on slow lanes. --- test/js/node/worker_threads/worker_threads.test.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index f2ad684fd18e..a043b558aa77 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -1684,12 +1684,19 @@ test("parentPort.postMessage queue is flushed to the parent before 'exit' fires" if (first) { first = false; // Hold this handler (and so the current drain pass) open until the - // worker has enqueued the whole burst AND entered shutdown, so the - // worker's close task is posted before this drain pass reschedules. + // 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 < 500); + const t = Date.now(); while (Date.now() - t < ${isDebug ? 3000 : 500}); return; } if (m === "END") gotEnd = true; else got++; From b45ee5661680c97850f6a3c7a0c28bd199b1623f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:34:48 +0000 Subject: [PATCH 3/6] test: cover the web Worker natural-exit flush path The close-task flush in dispatchExit serves both node's parentPort (fakeParentPort wraps jsFunctionPostMessage) and the web Worker global postMessage, since both feed m_toParent. Add the web-API variant of the SAB-pinned test so the fix is exercised through Worker.onmessage / 'close' as well as worker_threads 'message' / 'exit'. --- test/js/web/workers/worker.test.ts | 65 +++++++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/test/js/web/workers/worker.test.ts b/test/js/web/workers/worker.test.ts index 9e20bd0d8b4d..863bbfa1a5f3 100644 --- a/test/js/web/workers/worker.test.ts +++ b/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(); From d751c8ebf0705f6776766d071a7e1ca391b6a76f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 29 Jul 2026 03:45:24 +0000 Subject: [PATCH 4/6] trim dispatchExit flush comment --- src/jsc/bindings/webcore/Worker.cpp | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 73eb2c215a09..5523d6881c0f 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -603,17 +603,11 @@ bool Worker::dispatchExit(int32_t exitCode) m_parentContextId, [this] { this->deref(); }, [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext& context) { - // Flush the worker→parent inbox before 'close'. Node delivers every - // parentPort.postMessage() that ran before the worker exited, then - // fires 'exit'; without this, a drainToParent that hit its yield - // budget while the worker was still posting has re-posted itself - // BEHIND this close task, and once m_state == Closed dispatchEvent() - // drops the rest on the floor. The worker thread has already torn - // down its VM (this task is posted from shutdown() step 4), so - // m_toParent cannot grow and one drainToParent pass empties it - // (limit = max(queue.size(), 1000) ≥ queue.size()). terminate() - // still drops pending messages: dispatchEvent() is gated on - // m_terminateRequested, so the dispatches below are no-ops. + // Node delivers every parentPort message before 'exit'. The worker VM + // is already torn down (this task is shutdown() step 4), so m_toParent + // can't grow and one drain pass empties it; a drain that rescheduled + // behind this task would otherwise dispatch into Closed and drop. + // terminate() still discards: dispatchEvent gates on m_terminateRequested. protectedThis->drainToParent(context); // Closing → dispatch 'close' → Closed. The split lets 'close'/'exit' From b13a92b780a126302d8986356db78abe1cf43b18 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 29 Jul 2026 04:13:14 +0000 Subject: [PATCH 5/6] Worker: make the close task chase the drain instead of flushing synchronously The close task now re-posts itself on the same lane drainToParent's reschedule uses whenever m_toParent is non-empty (or a drain is still scheduled), and only proceeds to Closing/close/Closed once the inbox is observed empty. terminate() skips the chase so the undrained tail is still discarded. This replaces the earlier synchronous drainToParent() call in the close task: a synchronous drain-to-empty defeats the drain's per-turn yield budget, so a large backlog would be dispatched in one unbroken run on the parent. Chasing keeps the drain as the pacing mechanism (one cheap re-check per drain turn) and composes with any future change to the drain's reschedule lane or per-turn cap. --- src/jsc/bindings/webcore/Worker.cpp | 94 ++++++++++++++++++----------- src/jsc/bindings/webcore/Worker.h | 1 + 2 files changed, 60 insertions(+), 35 deletions(-) diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 5523d6881c0f..ce68ac85aa38 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -602,41 +602,8 @@ bool Worker::dispatchExit(int32_t exitCode) return ScriptExecutionContext::postTaskTo( m_parentContextId, [this] { this->deref(); }, - [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext& context) { - // Node delivers every parentPort message before 'exit'. The worker VM - // is already torn down (this task is shutdown() step 4), so m_toParent - // can't grow and one drain pass empties it; a drain that rescheduled - // behind this task would otherwise dispatch into Closed and drop. - // terminate() still discards: dispatchEvent gates on m_terminateRequested. - protectedThis->drainToParent(context); - - // 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(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_); + [exitCode, protectedThis = Ref { *this }](ScriptExecutionContext&) { + 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 @@ -644,6 +611,63 @@ bool Worker::dispatchExit(int32_t exitCode) }); } +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(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) diff --git a/src/jsc/bindings/webcore/Worker.h b/src/jsc/bindings/webcore/Worker.h index 657597d84ec5..53f9e0796ea5 100644 --- a/src/jsc/bindings/webcore/Worker.h +++ b/src/jsc/bindings/webcore/Worker.h @@ -173,6 +173,7 @@ class Worker final : public ThreadSafeRefCounted, public EventTargetWith void enqueueToWorker(MessageWithMessagePorts&&); void drainToParent(ScriptExecutionContext&); + void closeTask(int32_t exitCode); WorkerOptions m_options; From 461c7c406b3435fa7d4e3835e1cc9ffdeebbfd5e Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 29 Jul 2026 09:23:45 +0000 Subject: [PATCH 6/6] ci: retrigger