diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 1a97f67c2571..ce68ac85aa38 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -603,33 +603,7 @@ 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(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 @@ -637,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; diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index 68d6c6f3f103..a043b558aa77 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/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: [ 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();