Fix lost wakeup when a worker message arrives during a nested event-loop wait

src/jsc/bindings/webcore/MessagePort.cpp

@@ -427,13 +427,15 @@ 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 until the drain starts;
+    // Dispatching covers the dispatch loop itself (drainAndDispatch trades one
+    // for the other before running user JS so a racing send can post a fresh
+    // wakeup — see #37189).
     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

src/jsc/bindings/webcore/MessagePortPipe.cpp

@@ -115,20 +115,35 @@ void MessagePortPipe::drainAndDispatch(uint8_t side, ScriptExecutionContextIdent
             return;
         }
         limit = std::max<size_t>(s.inbox.size(), 1000);
+        // Hand DrainScheduled off to Dispatching before any user JS runs: a
+        // 'message' handler (or a promise continuation its microtask drain
+        // unblocks) can park this loop in a nested event-loop wait (e.g.
+        // bun:test's expect().rejects), and a send() arriving then must post
+        // a fresh drain task or that wait never wakes (#37189). Dispatching
+        // keeps hasPendingActivity() true across the dispatch window that
+        // DrainScheduled used to cover.
+        s.state.store((st & ~DrainScheduled) | Dispatching, std::memory_order_release);
     }
 
+    // Clears Dispatching only while this drain still owns the side: a detach
+    // mid-dispatch already cleared it, and by now it may belong to the next
+    // owner's drain.
+    auto finish = [&] {
+        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 +159,23 @@ 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;
+                s.state.store(st & ~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 to the rest of the event loop. If a
+                // racing send already posted a wakeup, let that task drain
+                // the rest; otherwise claim the flag and reschedule.
+                if (!(st & DrainScheduled)) {
+                    st |= DrainScheduled;
+                    rescheduleCtx = s.ctxId;
+                }
+                s.state.store(st & ~Dispatching, std::memory_order_release);
                 break;
             }
             message = s.inbox.takeFirst();
@@ -167,15 +187,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 +268,15 @@ 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. A drain mid-dispatch on this thread (detach happens on the
+    // owning thread, so only re-entrantly, from inside its handler) exits at
+    // that same check without touching the flags again. 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)

src/jsc/bindings/webcore/MessagePortPipe.h

@@ -42,10 +42,11 @@ class MessagePortPipe final : public ThreadSafeRefCounted<MessagePortPipe> {
     // 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,

src/jsc/bindings/webcore/Worker.cpp

@@ -273,60 +273,54 @@ 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 cleared before the first dispatch, so it is only set
+// while a posted drain task has not yet started draining. A handler (or a
+// promise continuation its microtask drain unblocks) can park this loop in a
+// nested event-loop wait (e.g. bun:test's expect().rejects); a send arriving
+// then must post a fresh wakeup task or that wait never wakes (#37189).
+// Messages are popped one at a time under the lock so such a nested drain
+// observes the shared queue and delivery stays FIFO.
 template<typename Dispatch>
 static inline bool drainInbox(Worker::MessageInbox& inbox, Zig::GlobalObject* globalObject, ScriptExecutionContext& context, Dispatch&& dispatch)
 {
     size_t limit;
-    Deque<MessageWithMessagePorts> 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<size_t>(inbox.queue.size(), 1000);
-        batch = std::exchange(inbox.queue, {});
     }
 
     while (true) {
-        while (!batch.isEmpty()) {
+        std::optional<MessageWithMessagePorts> 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 to the rest of the event loop. If a
+                // racing send already posted a wakeup, let that task drain
+                // the rest; otherwise 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 the rest — dispatch is a no-op
+            // once m_terminateRequested is set (drainToParent), and the
+            // worker thread is tearing down (drainToWorker).
             return false;
         }
-        if (limit == 0)
-            return true; // budget spent; caller reschedules
-        batch = std::exchange(inbox.queue, {});
     }
 }
 

test/regression/issue/37189.test.ts

@@ -0,0 +1,101 @@
+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, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(stdout).toBe("OK\n");
+  expect(exitCode).toBe(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);
+  },
+);