diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index a59c83b8c29b..64652fb962bd 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/src/jsc/bindings/webcore/MessagePort.cpp @@ -427,13 +427,13 @@ bool MessagePort::hasPendingActivity() const // Keep alive while a drain task is pending or mid-dispatch. drainAndDispatch // pops each message (queued -> 0) before invoking listeners, so the in-hand - // message is invisible to the queued count; without this bit a concurrent GC + // message is invisible to the queued count; without these bits a concurrent GC // running inside that window (queue empty, peer already closed) severs the // wrapper weak and the dispatch hits a dead JSEventListener wrapper (debug - // ASSERT m_wrapper). DrainScheduled is set from schedule until the inbox is - // observed empty, covering every dispatch. + // ASSERT m_wrapper). DrainScheduled covers schedule -> drain start; + // Dispatching covers the dispatch loop itself. uint64_t s = m_pipe->state(m_side); - if (s & MessagePortPipe::DrainScheduled) + if (s & (MessagePortPipe::DrainScheduled | MessagePortPipe::Dispatching)) return true; // Keep alive if the peer is still open and could send more, or messages are diff --git a/src/jsc/bindings/webcore/MessagePortPipe.cpp b/src/jsc/bindings/webcore/MessagePortPipe.cpp index 88f36aef6a18..7fa365b1c246 100644 --- a/src/jsc/bindings/webcore/MessagePortPipe.cpp +++ b/src/jsc/bindings/webcore/MessagePortPipe.cpp @@ -99,6 +99,7 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent RefPtr port; size_t limit; + bool ownsDispatching = false; { Locker locker { s.lock }; // This task was posted to `expectedCtx` (and is running there). If @@ -115,20 +116,36 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent return; } limit = std::max(s.inbox.size(), 1000); + // Trade DrainScheduled for Dispatching before user JS runs: a handler + // can park this loop in a nested event-loop wait, and a send() arriving + // then must post a fresh drain task (#37189). Dispatching keeps + // hasPendingActivity() true across the dispatch window DrainScheduled + // used to cover. A nested drain (posted by such a send) finds the bit + // already set and leaves clearing it to this outer invocation. + ownsDispatching = !(st & Dispatching); + s.state.store((st & ~DrainScheduled) | Dispatching, std::memory_order_release); } + // Clear Dispatching only if this invocation set it and still owns the + // side; after a detach the bit belongs to the next owner's drain. + auto finish = [&] { + if (!ownsDispatching) + return; + Locker locker { s.lock }; + if (s.ctxId == expectedCtx && s.port.get() == port) + s.state.fetch_and(~uint64_t(Dispatching), std::memory_order_acq_rel); + }; + // All 'message' listeners removed: the port is paused. Leave the inbox buffered // and stop draining; a later addEventListener re-schedules this drain. if (!port->hasMessageEventListener()) { - Locker locker { s.lock }; - s.state.fetch_and(~uint64_t(DrainScheduled), std::memory_order_acq_rel); + finish(); return; } auto* context = port->scriptExecutionContext(); if (!context || !context->globalObject()) { - Locker locker { s.lock }; - s.state.fetch_and(~uint64_t(DrainScheduled), std::memory_order_acq_rel); + finish(); return; } auto* globalObject = defaultGlobalObject(context->globalObject()); @@ -144,18 +161,26 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent // MessagePort, so compare port identity too — dispatching to // the stale (now m_isDetached) `port` would silently drop. // The new owner's attach() scheduled its own drain; leave the - // inbox for that. + // inbox (and the flags, which detach/close reset) for that. if (s.ctxId != expectedCtx || s.port.get() != port) - break; + return; uint64_t st = s.state.load(std::memory_order_relaxed); if (!(st & Attached) || s.inbox.isEmpty()) { - s.state.store(st & ~DrainScheduled, std::memory_order_release); - break; + if (ownsDispatching) + s.state.store(st & ~uint64_t(Dispatching), std::memory_order_release); + return; } if (limit-- == 0) { - // Yield to the rest of the event loop; DrainScheduled stays - // set so concurrent sends don't double-schedule. - rescheduleCtx = s.ctxId; + // Budget spent; yield. If a racing send already posted a + // wakeup let that task drain the rest, else claim the flag + // and reschedule. + if (!(st & DrainScheduled)) { + st |= DrainScheduled; + rescheduleCtx = s.ctxId; + } + if (ownsDispatching) + st &= ~uint64_t(Dispatching); + s.state.store(st, std::memory_order_release); break; } message = s.inbox.takeFirst(); @@ -167,15 +192,16 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent // Node's MakeCallback wraps each emit in an InternalCallbackScope, // which drains nextTick + microtasks on exit; match that so // queueMicrotask(cb) inside onmessage runs before the next message. - if (globalObject->drainMicrotasks()) - break; // termination pending + if (globalObject->drainMicrotasks()) { + finish(); + return; // termination pending + } // Listeners may have been removed mid-drain (port.off()); pause like the // pre-loop check instead of dispatching the rest to zero listeners. if (!port->hasMessageEventListener()) { - Locker locker { s.lock }; - s.state.fetch_and(~uint64_t(DrainScheduled), std::memory_order_acq_rel); - break; + finish(); + return; } } @@ -247,12 +273,12 @@ void MessagePortPipe::detach(uint8_t side) Locker locker { s.lock }; s.ctxId = 0; s.port = nullptr; - // Drop Attached and DrainScheduled. A drain task already in flight on - // the old context can't be recalled, but it captured the old ctxId and - // drainAndDispatch()'s s.ctxId != expectedCtx check makes it a no-op — - // even if a new owner attach()es to a different context before it runs. - // Messages remain queued for the next owner. - s.state.fetch_and(~uint64_t(Attached | ContextKnown | DrainScheduled), std::memory_order_acq_rel); + // Drop Attached, DrainScheduled and Dispatching. A drain task already in + // flight on the old context can't be recalled, but it captured the old + // ctxId and drainAndDispatch()'s s.ctxId != expectedCtx check makes it a + // no-op — even if a new owner attach()es to a different context before + // it runs. Messages remain queued for the next owner. + s.state.fetch_and(~uint64_t(Attached | ContextKnown | DrainScheduled | Dispatching), std::memory_order_acq_rel); } void MessagePortPipe::close(uint8_t side, CloseKind kind) diff --git a/src/jsc/bindings/webcore/MessagePortPipe.h b/src/jsc/bindings/webcore/MessagePortPipe.h index e58fd648e3db..3149dc6188d6 100644 --- a/src/jsc/bindings/webcore/MessagePortPipe.h +++ b/src/jsc/bindings/webcore/MessagePortPipe.h @@ -42,10 +42,11 @@ class MessagePortPipe final : public ThreadSafeRefCounted { // lives in the upper bits so it can be bumped with fetch_add(QueuedOne). enum State : uint64_t { Closed = 1ull << 0, // close() was called on this side; drops further deliveries. - DrainScheduled = 1ull << 1, // a drain task for this side is in flight. + DrainScheduled = 1ull << 1, // a posted drain task for this side has not yet started draining. Attached = 1ull << 2, // ctxId/port are valid; ok to schedule drains. ContextKnown = 1ull << 3, // ctxId/port are valid for close-notification only (no drains). ClosedByRequest = 1ull << 4, // the Closed above came from close(), not from the port being collected. + Dispatching = 1ull << 5, // drainAndDispatch is mid-dispatch on this side (GC liveness; see hasPendingActivity). QueuedShift = 8, QueuedOne = 1ull << QueuedShift, diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 1a97f67c2571..4a3f30d61650 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -273,60 +273,60 @@ void Worker::enqueueToParent(MessageWithMessagePorts&& message) // queueMicrotask/Promise callbacks observe messages one at a time, then // yields and reschedules if more remain. // -// Unlike MessagePortPipe, Worker sides never transfer, so we don't need to -// re-check port identity each iteration — which lets us swap the whole inbox -// into a local deque under the lock and dispatch without contending with the -// sender. A sustained producer (e.g. a tight postMessage loop) would otherwise -// make every per-message pop a contended acquire. +// drainScheduled is only set while a posted drain task has not yet started +// draining. It must be clear while user JS runs: a handler can park this loop +// in a nested event-loop wait, and a send arriving then has to post a fresh +// wakeup or that wait never wakes (#37189). Per-message pops under the lock +// (rather than swapping the queue out) keep delivery FIFO when such a nested +// drain runs. template static inline bool drainInbox(Worker::MessageInbox& inbox, Zig::GlobalObject* globalObject, ScriptExecutionContext& context, Dispatch&& dispatch) { size_t limit; - Deque batch; { Locker locker { inbox.lock }; - if (inbox.queue.isEmpty()) { - inbox.drainScheduled.store(false, std::memory_order_relaxed); + inbox.drainScheduled.store(false, std::memory_order_relaxed); + if (inbox.queue.isEmpty()) return false; - } limit = std::max(inbox.queue.size(), 1000); - batch = std::exchange(inbox.queue, {}); } while (true) { - while (!batch.isEmpty()) { + std::optional message; + { + Locker locker { inbox.lock }; + if (inbox.queue.isEmpty()) + return false; if (limit-- == 0) { - // Yield to the rest of the event loop. Return the undrained - // tail to the front of the inbox so it stays ahead of - // anything enqueued concurrently; caller reschedules. - Locker locker { inbox.lock }; - while (!batch.isEmpty()) - inbox.queue.prepend(batch.takeLast()); + // Budget spent; yield. If a racing send already posted a + // wakeup let that task drain the rest, else claim the flag + // and have the caller reschedule. + if (inbox.drainScheduled.load(std::memory_order_relaxed)) + return false; + inbox.drainScheduled.store(true, std::memory_order_relaxed); return true; } - auto message = batch.takeFirst(); - - auto ports = MessagePort::entanglePorts(context, WTF::move(message.transferredPorts)); - auto event = MessageEvent::create(*context.jsGlobalObject(), message.message.releaseNonNull(), nullptr, WTF::move(ports)); - dispatch(event.event); - - if (globalObject->drainMicrotasks()) { - // Termination pending. Drop the rest — dispatch is a no-op - // once m_terminateRequested is set (drainToParent), and the - // worker thread is tearing down (drainToWorker). - return false; - } + message = inbox.queue.takeFirst(); } - // Batch exhausted — see if more arrived while we were dispatching. - Locker locker { inbox.lock }; - if (inbox.queue.isEmpty()) { - inbox.drainScheduled.store(false, std::memory_order_relaxed); + auto ports = MessagePort::entanglePorts(context, WTF::move(message->transferredPorts)); + auto event = MessageEvent::create(*context.jsGlobalObject(), message->message.releaseNonNull(), nullptr, WTF::move(ports)); + dispatch(event.event); + + if (globalObject->drainMicrotasks()) { + // Termination pending. Drop everything still queued — dispatch + // is a no-op once m_terminateRequested is set (drainToParent), + // and the worker thread is tearing down (drainToWorker). + // Destructing the messages now (outside the lock) closes any + // transferred ports so their peers see 'close' promptly instead + // of waiting for ~Worker. + Deque dropped; + { + Locker locker { inbox.lock }; + dropped = std::exchange(inbox.queue, {}); + } return false; } - if (limit == 0) - return true; // budget spent; caller reschedules - batch = std::exchange(inbox.queue, {}); } } diff --git a/test/js/web/workers/message-port-pipe.test.ts b/test/js/web/workers/message-port-pipe.test.ts index 9ff401768d06..cf5aacdd977a 100644 --- a/test/js/web/workers/message-port-pipe.test.ts +++ b/test/js/web/workers/message-port-pipe.test.ts @@ -363,6 +363,39 @@ describe("MessagePort pipe", () => { expect(stdout.trim()).toBe("OK"); expect(exitCode).toBe(0); }); + + // Each onmessage enqueues the next message mid-drain, so the drain's budget + // (1000 when the inbox starts with one message) runs out with the inbox + // non-empty while the in-handler send has already posted the continuation + // drain task. Exercises the budget-yield handoff; out-of-order or missing + // delivery fails. + test("self-feeding chain outlives the drain budget and stays in order", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { port1, port2 } = new MessageChannel(); + const N = 2500; + let next = 0; + port1.onmessage = e => { + if (e.data !== next) { console.error("out of order", e.data, next); process.exit(1); } + next++; + if (next === N) { console.log("OK"); port1.close(); port2.close(); return; } + port2.postMessage(next); + }; + port2.postMessage(0); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe("OK"); + expect(exitCode).toBe(0); + }); }); // worker.postMessage / parentPort.postMessage go through the same coalesced diff --git a/test/regression/issue/37189.test.ts b/test/regression/issue/37189.test.ts new file mode 100644 index 000000000000..341712bcacf1 --- /dev/null +++ b/test/regression/issue/37189.test.ts @@ -0,0 +1,100 @@ +import { expect, test } from "bun:test"; +import { bunEnv, bunExe, tempDir } from "harness"; + +// https://github.com/oven-sh/bun/issues/37189 +// Regression in 1.3.14: a Worker/MessagePort message arriving while the +// receiver was parked in a nested event-loop wait (expect().rejects) reached +// from a previous message's continuation was enqueued without posting a +// wakeup, deadlocking the process. The first awaited call below makes the +// second call's assertion run as a continuation of the message dispatch, +// with the drain loop still on the native stack. + +const channelMain = `import { expect } from "bun:test"; +const { port1, port2 } = new MessageChannel(); +port2.onmessage = e => { + port2.postMessage({ type: "reply", id: e.data.id, error: "boom" }); +}; +const pending = new Map(); +port1.onmessage = e => { + const msg = e.data; + if (msg.type !== "reply") return; + const reject = pending.get(msg.id); + pending.delete(msg.id); + reject?.(new Error(msg.error)); +}; +let seq = 0; +const call = () => new Promise((_resolve, reject) => { + const id = ++seq; + pending.set(id, reject); + port1.postMessage({ id }); +}); + +await call().catch(() => {}); +await expect(call()).rejects.toThrow("boom"); +port1.close(); +port2.close(); +console.log("OK");`; + +const workerMain = `import { expect } from "bun:test"; +const worker = new Worker(new URL("./worker.js", import.meta.url).href); +const pending = new Map(); +worker.onmessage = e => { + const msg = e.data; + if (msg.type !== "reply") return; + const reject = pending.get(msg.id); + pending.delete(msg.id); + reject?.(new Error(msg.error)); +}; +let seq = 0; +const call = () => new Promise((_resolve, reject) => { + const id = ++seq; + pending.set(id, reject); + worker.postMessage({ id }); +}); + +await call().catch(() => {}); +await expect(call()).rejects.toThrow("boom"); +worker.terminate(); +console.log("OK");`; + +async function expectExitsCleanly(proc: Bun.Subprocess<"ignore", "pipe", "pipe">) { + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "OK\n", stderr: "", exitCode: 0 }); +} + +test.concurrent( + "expect().rejects settles when the rejection arrives from a Worker message during a nested wait", + async () => { + using dir = tempDir("issue-37189-worker", { + "worker.js": `self.onmessage = e => { + self.postMessage({ type: "reply", id: e.data.id, error: "boom" }); + };`, + "main.js": workerMain, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "main.js"], + env: bunEnv, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + timeout: 15_000, + killSignal: "SIGKILL", + }); + await expectExitsCleanly(proc); + }, +); + +test.concurrent( + "expect().rejects settles when the rejection arrives from a MessageChannel message during a nested wait", + async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", channelMain], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + timeout: 15_000, + killSignal: "SIGKILL", + }); + await expectExitsCleanly(proc); + }, +);