feat(channel): selective receive — recv_match + try_recv_match (roadmap #4)

recv_match(pred) scans the queue front-to-back, removes and returns the
first match (rest preserved in arrival order), and parks/re-scans on every
send when nothing matches — a selective receiver may park on a non-empty
queue. Returns Err only once the channel is closed with no queued match.
try_recv_match is the non-blocking variant, mirroring try_recv.

Sender::drop now wakes the parked receiver on the last-sender drop
regardless of queue emptiness, so a selective receiver parked on a
non-empty no-match queue observes closure instead of sleeping forever.
No-op for plain recv (which only ever parks on an empty queue).

src/channel.rs

@@ -69,7 +69,13 @@ impl<T> Drop for Sender<T> {
         let unpark = {
             let mut g = self.inner.lock().unwrap();
             g.senders -= 1;
-            if g.senders == 0 && g.queue.is_empty() {
+            // Wake the parked receiver on the last sender drop regardless of
+            // whether the queue is empty. A plain `recv` only ever parks on an
+            // empty queue (so this is unchanged for it), but a selective
+            // `recv_match` may be parked on a *non-empty* queue holding only
+            // non-matching messages — it must wake to observe closure and
+            // return Err rather than sleep forever.
+            if g.senders == 0 {
                 g.parked_receiver.take()
             } else {
                 None
@@ -134,6 +140,64 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// Selective receive: remove and return the first queued message for which
+    /// `pred` holds, leaving the rest in arrival order. If no queued message
+    /// matches, parks and re-scans on every send (a selective receiver may park
+    /// on a *non-empty* queue). Returns `Err(RecvError)` only once the channel
+    /// is closed and no queued message matches.
+    ///
+    /// `pred` is run while the channel lock is held: keep it cheap and pure,
+    /// and do not call back into this channel from inside it. It is modelled as
+    /// `Fn` (not `FnMut`) deliberately — it is re-run from scratch on every
+    /// scan, so a stateful predicate would observe surprising re-counting.
+    pub fn recv_match<F>(&self, pred: F) -> Result<T, RecvError>
+    where
+        F: Fn(&T) -> bool,
+    {
+        loop {
+            {
+                let mut g = self.inner.lock().unwrap();
+                if let Some(i) = g.queue.iter().position(|v| pred(v)) {
+                    // position() found it, so remove() returns Some.
+                    return Ok(g.queue.remove(i).unwrap());
+                }
+                if g.senders == 0 {
+                    // Closed and nothing queued can ever match.
+                    return Err(RecvError);
+                }
+                let me = crate::actor::current_pid()
+                    .expect("recv_match() called outside an actor");
+                debug_assert!(
+                    g.parked_receiver.is_none(),
+                    "channel has more than one receiver"
+                );
+                g.parked_receiver = Some(me);
+                crate::te!(crate::trace::Event::RecvPark(me));
+            }
+            // Release the lock before parking — the unparker will need it.
+            crate::scheduler::park_current();
+            crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
+        }
+    }
+
+    /// Non-blocking selective receive. `Ok(Some(v))` if a queued message
+    /// matched `pred` (removed, rest left in order), `Ok(None)` if the channel
+    /// is open but nothing matched, `Err(RecvError)` if closed and nothing
+    /// matched. Same predicate contract as [`recv_match`](Self::recv_match).
+    pub fn try_recv_match<F>(&self, pred: F) -> Result<Option<T>, RecvError>
+    where
+        F: Fn(&T) -> bool,
+    {
+        let mut g = self.inner.lock().unwrap();
+        if let Some(i) = g.queue.iter().position(|v| pred(v)) {
+            return Ok(Some(g.queue.remove(i).unwrap()));
+        }
+        if g.senders == 0 {
+            return Err(RecvError);
+        }
+        Ok(None)
+    }
+
     /// Non-blocking. `Ok(Some(v))` if a message was available, `Ok(None)` if
     /// the channel is empty but open, `Err(RecvError)` if closed and drained.
     pub fn try_recv(&self) -> Result<Option<T>, RecvError> {

tests/selective_recv.rs

@@ -0,0 +1,193 @@
+//! Selective-receive tests: `Receiver::recv_match` and `try_recv_match`.
+//!
+//! `recv_match(pred)` scans the queued messages, removes and returns the first
+//! one for which `pred` holds, and leaves the rest in arrival order. With no
+//! match it parks and re-scans on every send (a selective receiver may park on
+//! a *non-empty* queue), returning `Err` only once the channel is closed and no
+//! queued message matches. `try_recv_match` is the non-blocking variant.
+
+use smarm::{channel, run, spawn};
+use std::sync::{Arc, Mutex};
+
+// ---------------------------------------------------------------------------
+// recv_match — same-actor scan semantics (no parking)
+// ---------------------------------------------------------------------------
+
+// First matching message is pulled out of arrival order; non-matches stay
+// queued in their original order for later plain recv().
+#[test]
+fn match_pulled_first_non_matches_remain_in_order() {
+    let got = Arc::new(Mutex::new(Vec::<i64>::new()));
+    let got2 = got.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        for v in [1, 2, 3] {
+            tx.send(v).unwrap();
+        }
+        // 2 is the only even; it's pulled despite arriving second.
+        let m = rx.recv_match(|&v| v % 2 == 0).unwrap();
+        // The non-matches remain, in order: 1 then 3.
+        let a = rx.recv().unwrap();
+        let b = rx.recv().unwrap();
+        let mut g = got2.lock().unwrap();
+        g.push(m);
+        g.push(a);
+        g.push(b);
+    });
+
+    assert_eq!(*got.lock().unwrap(), vec![2, 1, 3]);
+}
+
+// A read-only captured value (request-id style) is a plain `Fn` predicate.
+#[test]
+fn matches_on_captured_value() {
+    let got = Arc::new(Mutex::new(0i64));
+    let got2 = got.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        for v in [10, 20, 30] {
+            tx.send(v).unwrap();
+        }
+        let target = 20;
+        let m = rx.recv_match(move |&v| v == target).unwrap();
+        *got2.lock().unwrap() = m;
+    });
+
+    assert_eq!(*got.lock().unwrap(), 20);
+}
+
+// ---------------------------------------------------------------------------
+// recv_match — parking on a non-empty, no-match queue
+// ---------------------------------------------------------------------------
+
+// The receiver parks while the queue holds only non-matching messages, must
+// wake on a send that does NOT empty->fill the queue, re-scan, re-park on a
+// still-non-matching send, and finally return when a match arrives.
+#[test]
+fn parks_on_non_matching_queue_then_wakes_on_match() {
+    let got = Arc::new(Mutex::new(0i64));
+    let got2 = got.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        tx.send(1).unwrap(); // odd: queued, will not match
+        let h = spawn(move || {
+            // Scans [1], no even -> parks on a non-empty queue.
+            let v = rx.recv_match(|&v| v % 2 == 0).unwrap();
+            *got2.lock().unwrap() = v;
+        });
+        smarm::yield_now(); // let the child park
+        tx.send(3).unwrap(); // odd: child wakes, re-scans [1,3], re-parks
+        tx.send(4).unwrap(); // even: child wakes, scans [1,3,4], returns 4
+        h.join().unwrap();
+    });
+
+    assert_eq!(*got.lock().unwrap(), 4);
+}
+
+// ---------------------------------------------------------------------------
+// recv_match — close semantics
+// ---------------------------------------------------------------------------
+
+// Channel closes (all senders dropped) while only non-matching messages are
+// queued: recv_match must return Err rather than block forever.
+#[test]
+fn closed_with_only_non_matches_returns_err() {
+    let saw_err = Arc::new(Mutex::new(false));
+    let saw_err2 = saw_err.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        let h = spawn(move || {
+            // Wants an even; only odds will ever exist.
+            let r = rx.recv_match(|&v| v % 2 == 0);
+            *saw_err2.lock().unwrap() = r.is_err();
+        });
+        smarm::yield_now(); // child parks on empty queue
+        tx.send(1).unwrap(); // odd: wakes, re-scans, re-parks (no match)
+        drop(tx); // last sender gone, still no match -> Err
+        h.join().unwrap();
+    });
+
+    assert!(*saw_err.lock().unwrap());
+}
+
+// A match present on a closed channel is still returned (closure doesn't
+// pre-empt an available match).
+#[test]
+fn closed_but_match_present_returns_match() {
+    let got = Arc::new(Mutex::new(0i64));
+    let got2 = got.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        tx.send(1).unwrap();
+        tx.send(2).unwrap();
+        drop(tx); // closed, but a match (2) is queued
+        let m = rx.recv_match(|&v| v % 2 == 0).unwrap();
+        *got2.lock().unwrap() = m;
+    });
+
+    assert_eq!(*got.lock().unwrap(), 2);
+}
+
+// ---------------------------------------------------------------------------
+// try_recv_match — non-blocking
+// ---------------------------------------------------------------------------
+
+#[test]
+fn try_recv_match_states() {
+    let out = Arc::new(Mutex::new(Vec::<String>::new()));
+    let out2 = out.clone();
+
+    run(move || {
+        let (tx, rx) = channel::<i64>();
+        let log = |s: &str| out2.lock().unwrap().push(s.to_string());
+
+        // Empty + open -> Ok(None).
+        log(match rx.try_recv_match(|&v| v % 2 == 0) {
+            Ok(None) => "open_empty_none",
+            _ => "WRONG_1",
+        });
+
+        // Non-matching present + open -> Ok(None), message left in place.
+        tx.send(1).unwrap();
+        log(match rx.try_recv_match(|&v| v % 2 == 0) {
+            Ok(None) => "open_nomatch_none",
+            _ => "WRONG_2",
+        });
+
+        // Match present -> Ok(Some(v)).
+        tx.send(2).unwrap();
+        log(match rx.try_recv_match(|&v| v % 2 == 0) {
+            Ok(Some(2)) => "got_match",
+            _ => "WRONG_3",
+        });
+
+        // Drain the leftover non-match with a plain recv to confirm it stayed.
+        log(match rx.recv() {
+            Ok(1) => "leftover_one",
+            _ => "WRONG_4",
+        });
+
+        // Closed + no match -> Err.
+        drop(tx);
+        log(match rx.try_recv_match(|&v| v % 2 == 0) {
+            Err(_) => "closed_err",
+            _ => "WRONG_5",
+        });
+    });
+
+    assert_eq!(
+        *out.lock().unwrap(),
+        vec![
+            "open_empty_none",
+            "open_nomatch_none",
+            "got_match",
+            "leftover_one",
+            "closed_err",
+        ]
+    );
+}