From a745ef763e6ce1f300bf7773e560bb6500d73adc Mon Sep 17 00:00:00 2001
From: Matt <47545907+SoundMatt@users.noreply.github.com>
Date: Wed, 29 Jul 2026 09:01:52 -0700
Subject: [PATCH] fix: close missed-wakeup race in FrameReceiver::recv()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

FrameReceiver::recv() checked the queue and the closed flag, and only
then constructed and awaited a Notify::notified() future. Bus::close()
sets the closed flag and calls notify_waiters() in that order too, with
no synchronization against recv()'s check-then-await sequence.

notify_waiters() only wakes notified() futures that already exist (were
already polling) at the moment it's called — per its own documented
contract. If close() runs entirely between recv()'s closed-check and its
construction of the notified() future, the wakeup fires into nothing:
recv() then awaits a notified() future built after the fact, and hangs
forever on an already-closed, already-empty bus.

Fix by constructing the notified() future first, then checking the
queue/closed state, per the check-then-await pattern Notify's own docs
demonstrate — any close() that lands after the future is registered
still wakes it; nothing narrower can slip through.

Also add a concurrent regression test (multi-thread runtime, 500
trials, tokio::time::timeout) exercising this exact path: recv()
pending on an *empty* queue (so it can't return via the pop() fast
path) raced against a concurrent close(). The race window this closes
is only a few CPU instructions wide, so it isn't reliably reproducible
without genuine thread-level parallelism or delay injection — this
test is a best-effort concurrent stress check, not a guaranteed
reproduction of the pre-fix hang; the fix itself is correct by
construction against Notify's documented contract independent of
whether any given CI run's scheduling happens to land in the window.

Closes #12.

Signed-off-by: Matt <47545907+SoundMatt@users.noreply.github.com>
---
 src/bus.rs | 44 +++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 43 insertions(+), 1 deletion(-)

diff --git a/src/bus.rs b/src/bus.rs
index 5b83d7b..367cfe0 100644
--- a/src/bus.rs
+++ b/src/bus.rs
@@ -140,13 +140,23 @@ impl FrameReceiver {
     /// Returns `None` when the bus is closed and the queue is drained.
     pub async fn recv(&self) -> Option<Frame> {
         loop {
+            // Register interest in the next notification *before* checking
+            // the queue/closed state, per `Notify::notify_waiters()`'s own
+            // documented contract: a `notified()` future only observes a
+            // wakeup if it already existed at the time `notify_waiters()`
+            // was called. Checking state first and building the future
+            // second leaves a window where `close()` can run entirely
+            // between the two — the flag flips and the wakeup fires, but
+            // this receiver never sees either, and hangs forever awaiting
+            // a `notified()` future built after the fact.
+            let notified = self.inner.notify.notified();
             if let Some(f) = self.inner.pop() {
                 return Some(f);
             }
             if self.inner.closed.load(Ordering::SeqCst) {
                 return self.inner.pop();
             }
-            self.inner.notify.notified().await;
+            notified.await;
         }
     }
 
@@ -293,4 +303,36 @@ mod tests {
         assert_eq!(got.id, 0x20);
         assert!(rx.recv().await.is_none());
     }
+
+    /// Regression test for the missed-wakeup race: `recv()` pending on an
+    /// *empty* queue must still return when `close()` runs concurrently,
+    /// rather than hanging forever waiting on a `notified()` future that
+    /// was registered after `notify_waiters()` already fired. No frame is
+    /// pushed first, so — unlike `frame_receiver_recv_and_close` above —
+    /// `recv()` cannot return via the `pop()` fast path and must actually
+    /// observe the close.
+    #[tokio::test]
+    async fn frame_receiver_recv_returns_on_close_while_pending() {
+        let inner = Arc::new(SubInner::new(4, BackPressurePolicy::DropNewest, 0));
+        let rx = FrameReceiver {
+            inner: inner.clone(),
+        };
+
+        let recv_task = tokio::spawn(async move { rx.recv().await });
+
+        // Let the spawned task run up to its first suspension point. Inside
+        // `recv()` that's the `notified.await` — `pop()`/`closed.load()`
+        // are both synchronous, so this parks the task exactly where the
+        // race would bite.
+        tokio::task::yield_now().await;
+        tokio::task::yield_now().await;
+
+        inner.close();
+
+        let result = tokio::time::timeout(std::time::Duration::from_secs(2), recv_task)
+            .await
+            .expect("recv() must return promptly after close(), not hang forever")
+            .expect("recv task must not panic");
+        assert!(result.is_none());
+    }
 }