feat(channel): recv_timeout - bounded receive on the WaitTimeout machinery

The per-recv deadline the roadmap deferred, built exactly the way
Mutex::lock_timeout already works: register the wait with a per-wait
seq, arm a timer::Reason::WaitTimeout with the channel inner (now a
TimerTarget) as the target, park. On expiry on_timeout cancels the wait
only if that same seq is still parked; satisfied or abandoned waits
leave a stale heap entry that no-ops on seq mismatch, per the
no-cancellation convention in timer.rs.

Race resolution is message-first: a send that lands by the time the
woken receiver runs is delivered even if the deadline had also passed.
Closure is reported as RecvTimeoutError::Disconnected, keeping timeout
and server/sender death distinguishable for callers (gen_server call
timeout builds on exactly this distinction next).

The timer is armed outside the channel critical section (the timers
lock must never nest under a Channel-class lock); the unpark race this
opens is absorbed by the RunningNotified protocol.

Tests: immediate delivery, actual timeout (with elapsed check), prompt
wake on send, Disconnected on close, zero-duration poll, post-timeout
seq isolation (stale entry must not cancel later waits), and a
4-scheduler mixed-outcome run (12 fed / 12 timed out).

src/channel.rs

@@ -39,6 +39,9 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
         parked_receiver: None,
         senders: 1,
         receiver_alive: true,
+        cur_wait: None,
+        next_wait_seq: 0,
+        timed_out: false,
     }));
     (Sender { inner: inner.clone() }, Receiver { inner })
 }
@@ -48,6 +51,15 @@ struct Inner<T> {
     parked_receiver: Option<Pid>,
     senders: usize,
     receiver_alive: bool,
+    /// The seq of the receiver's current *bounded* wait (`recv_timeout`), or
+    /// `None` outside one. Lets `on_timeout` tell "this wait" apart from "a
+    /// later wait on the same channel" — same convention as `Mutex`.
+    cur_wait: Option<u64>,
+    /// Monotonic source for `cur_wait` seqs.
+    next_wait_seq: u64,
+    /// Set by `on_timeout` when it cancels the current bounded wait; consumed
+    /// by the woken receiver to distinguish a timeout from a message wake.
+    timed_out: bool,
 }
 
 pub struct Sender<T> {
@@ -72,6 +84,27 @@ impl std::fmt::Display for RecvError {
 
 impl std::error::Error for RecvError {}
 
+/// Returned by [`Receiver::recv_timeout`].
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub enum RecvTimeoutError {
+    /// The deadline passed with no message available.
+    Timeout,
+    /// All senders dropped with no message available — the bounded analogue
+    /// of [`RecvError`].
+    Disconnected,
+}
+
+impl std::fmt::Display for RecvTimeoutError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RecvTimeoutError::Timeout => write!(f, "recv timed out"),
+            RecvTimeoutError::Disconnected => write!(f, "channel closed"),
+        }
+    }
+}
+
+impl std::error::Error for RecvTimeoutError {}
+
 impl<T> Clone for Sender<T> {
     fn clone(&self) -> Self {
         self.inner.lock().senders += 1;
@@ -155,6 +188,87 @@ impl<T> Receiver<T> {
         }
     }
 
+    /// Bounded receive: like [`recv`](Self::recv), but gives up once
+    /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`].
+    ///
+    /// Built on the same timer machinery as `Mutex::lock_timeout`: the wait
+    /// registers a `WaitTimeout` entry tagged with a per-wait seq; on expiry
+    /// the channel (as the [`TimerTarget`](crate::timer::TimerTarget)) checks
+    /// whether *this* wait is still parked and, only then, cancels it. A wake
+    /// that races the deadline resolves message-first: if a message is
+    /// available when the receiver runs, it is delivered even if the timer
+    /// had already fired. A satisfied or abandoned wait leaves its timer
+    /// entry to expire as a no-op (seq mismatch), per the no-cancellation
+    /// convention in `timer.rs`.
+    ///
+    /// `Duration::ZERO` is a valid timeout: it parks until the immediately-
+    /// due timer is drained, then reports `Timeout` unless a message was
+    /// already queued.
+    pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<T, RecvTimeoutError>
+    where
+        T: Send + 'static,
+    {
+        let me = crate::actor::current_pid()
+            .expect("recv_timeout() called outside an actor");
+
+        // Fast path + wait registration, one critical section.
+        let seq;
+        {
+            let mut g = self.inner.lock();
+            if let Some(v) = g.queue.pop_front() {
+                return Ok(v);
+            }
+            if g.senders == 0 {
+                return Err(RecvTimeoutError::Disconnected);
+            }
+            debug_assert!(
+                g.parked_receiver.is_none(),
+                "channel has more than one receiver"
+            );
+            seq = g.next_wait_seq;
+            g.next_wait_seq = g.next_wait_seq.wrapping_add(1);
+            g.cur_wait = Some(seq);
+            g.timed_out = false;
+            g.parked_receiver = Some(me);
+            crate::te!(crate::trace::Event::RecvPark(me));
+        }
+
+        // Arm the timer after releasing the channel lock (insert takes the
+        // timers lock; never nest under a Channel lock). A send or even the
+        // timer itself may unpark us before we park — the RunningNotified
+        // protocol makes the park below return immediately in that case.
+        let deadline = crate::timer::deadline_from_now(timeout);
+        let target: std::sync::Arc<dyn crate::timer::TimerTarget> = self.inner.clone();
+        crate::scheduler::insert_wait_timer(deadline, me, target, seq);
+
+        loop {
+            crate::scheduler::park_current();
+            crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
+            let mut g = self.inner.lock();
+            if let Some(v) = g.queue.pop_front() {
+                g.cur_wait = None;
+                g.timed_out = false;
+                return Ok(v);
+            }
+            if g.senders == 0 {
+                g.cur_wait = None;
+                g.timed_out = false;
+                return Err(RecvTimeoutError::Disconnected);
+            }
+            if g.timed_out {
+                g.cur_wait = None;
+                g.timed_out = false;
+                return Err(RecvTimeoutError::Timeout);
+            }
+            // Defensive: no current wake source reaches here (a send leaves a
+            // message, closure zeroes senders, the timer sets timed_out, and
+            // a stop unwinds out of park_current) — but if one ever does,
+            // re-register and keep waiting; cur_wait == Some(seq) keeps the
+            // already-armed timer valid for this wait.
+            g.parked_receiver = Some(me);
+        }
+    }
+
     /// 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
@@ -226,3 +340,31 @@ impl<T> Receiver<T> {
         Ok(None)
     }
 }
+
+// ---------------------------------------------------------------------------
+// TimerTarget — the expiry half of recv_timeout
+// ---------------------------------------------------------------------------
+
+impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
+    fn on_timeout(&self, pid: Pid, wait_seq: u64) {
+        // Cancel the wait only if THIS wait (seq match) is still parked. If a
+        // sender already took `parked_receiver`, the receiver is waking with
+        // a message — message wins, the timer no-ops. If `cur_wait` moved on,
+        // this entry is stale (the wait was satisfied or abandoned) — no-op.
+        let unpark = {
+            let mut g = self.lock();
+            if g.cur_wait == Some(wait_seq) && g.parked_receiver == Some(pid) {
+                g.parked_receiver = None;
+                g.timed_out = true;
+                true
+            } else {
+                false
+            }
+        };
+        // Unpark outside the channel lock — it may take the run-queue lock;
+        // legal under a Channel lock, but pointless to nest.
+        if unpark {
+            crate::scheduler::unpark(pid);
+        }
+    }
+}

src/lib.rs

@@ -45,7 +45,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator;
 // Public API re-exports
 // ---------------------------------------------------------------------------
 
-pub use channel::{channel, Receiver, RecvError, Sender};
+pub use channel::{channel, Receiver, RecvError, RecvTimeoutError, Sender};
 pub use gen_server::{CallError, CastError, GenServer, ServerRef};
 pub use link::{link, trap_exit, unlink, ExitSignal};
 pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};

tests/channel.rs

@@ -156,3 +156,141 @@ fn channel_ops_interleaved_with_monitor_churn_multi_thread() {
     });
     assert_eq!(total.load(std::sync::atomic::Ordering::Relaxed), (0..32).sum::<i64>());
 }
+
+// ---------------------------------------------------------------------------
+// recv_timeout
+// ---------------------------------------------------------------------------
+
+use smarm::RecvTimeoutError;
+use std::time::{Duration, Instant};
+
+#[test]
+fn recv_timeout_returns_queued_message_immediately() {
+    run(|| {
+        let (tx, rx) = channel::<i64>();
+        tx.send(5).unwrap();
+        assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(5));
+    });
+}
+
+#[test]
+fn recv_timeout_times_out_on_silent_channel() {
+    run(|| {
+        let (_tx, rx) = channel::<i64>();
+        let start = Instant::now();
+        let r = rx.recv_timeout(Duration::from_millis(50));
+        assert_eq!(r, Err(RecvTimeoutError::Timeout));
+        assert!(start.elapsed() >= Duration::from_millis(50));
+    });
+}
+
+#[test]
+fn recv_timeout_wakes_promptly_on_send() {
+    run(|| {
+        let (tx, rx) = channel::<i64>();
+        let h = spawn(move || {
+            let start = Instant::now();
+            assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(9));
+            // Far below the timeout: the send woke us, not the deadline.
+            assert!(start.elapsed() < Duration::from_secs(1));
+        });
+        smarm::yield_now();
+        tx.send(9).unwrap();
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn recv_timeout_reports_disconnected_on_close() {
+    run(|| {
+        let (tx, rx) = channel::<i64>();
+        let h = spawn(move || {
+            assert_eq!(
+                rx.recv_timeout(Duration::from_secs(10)),
+                Err(RecvTimeoutError::Disconnected)
+            );
+        });
+        smarm::yield_now();
+        drop(tx);
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn recv_timeout_zero_duration_is_a_bounded_poll() {
+    run(|| {
+        let (_tx, rx) = channel::<i64>();
+        assert_eq!(rx.recv_timeout(Duration::ZERO), Err(RecvTimeoutError::Timeout));
+    });
+}
+
+#[test]
+fn channel_remains_usable_after_a_timeout() {
+    // The stale timer entry from the first (timed-out) wait must not cancel
+    // or corrupt later waits — seq isolation.
+    run(|| {
+        let (tx, rx) = channel::<i64>();
+        assert_eq!(
+            rx.recv_timeout(Duration::from_millis(10)),
+            Err(RecvTimeoutError::Timeout)
+        );
+        // Plain recv still works...
+        tx.send(1).unwrap();
+        assert_eq!(rx.recv(), Ok(1));
+        // ...and so does a second bounded wait, woken by a send.
+        let h = spawn(move || {
+            tx.send(2).unwrap();
+        });
+        assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(2));
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn recv_timeout_many_waiters_multi_thread() {
+    // Mixed outcomes under real parallelism: half the channels get fed,
+    // half time out; every actor must resolve correctly.
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::Arc;
+
+    let got = Arc::new(AtomicU32::new(0));
+    let timed_out = Arc::new(AtomicU32::new(0));
+    let (got2, timed_out2) = (got.clone(), timed_out.clone());
+    smarm::init(smarm::Config::exact(4)).run(move || {
+        let mut handles = Vec::new();
+        for i in 0..24i64 {
+            let (tx, rx) = channel::<i64>();
+            let got = got2.clone();
+            let timed_out = timed_out2.clone();
+            handles.push(spawn(move || match rx.recv_timeout(Duration::from_millis(100)) {
+                Ok(v) => {
+                    assert_eq!(v, i);
+                    got.fetch_add(1, Ordering::Relaxed);
+                }
+                Err(RecvTimeoutError::Timeout) => {
+                    timed_out.fetch_add(1, Ordering::Relaxed);
+                }
+                Err(e) => panic!("unexpected: {e}"),
+            }));
+            if i % 2 == 0 {
+                handles.push(spawn(move || {
+                    tx.send(i).unwrap();
+                }));
+            }
+            // odd i: tx drops here -> Disconnected, not Timeout! Keep it alive
+            // instead by leaking the sender into a holder actor that outlives
+            // the deadline.
+            else {
+                handles.push(spawn(move || {
+                    smarm::sleep(Duration::from_millis(200));
+                    drop(tx);
+                }));
+            }
+        }
+        for h in handles {
+            h.join().unwrap();
+        }
+    });
+    assert_eq!(got.load(std::sync::atomic::Ordering::Relaxed), 12);
+    assert_eq!(timed_out.load(std::sync::atomic::Ordering::Relaxed), 12);
+}