feat(channel): select_timeout — bounded select on a stateless stamped timer arm

The deadline is one more stamped waker on the select's wait epoch: a unit
TimerTarget whose on_timeout is a bare unpark_at. Nothing registers in any
arm for it, so there is nothing to cancel or leak — an arm winning leaves
the timer entry to die at its epoch CAS; the timer winning leaves the arms'
registrations to self-clean like any select loser's. Classification is pure
channel state (wakes are precise): some arm ready -> Some(first, priority
order); none -> the timer was the only remaining stamped waker -> None.
Message-first on a raced deadline, same as recv_timeout.

Registration pass factored into register_arms, which owns the retire_wait
obligation on the ready-now exit for both entry points.

src/channel.rs

@@ -435,24 +435,7 @@ pub fn select(arms: &[&dyn Selectable]) -> usize {
     let me = crate::actor::current_pid().expect("select() called outside an actor");
     loop {
         let epoch = crate::scheduler::begin_wait();
-
-        // Check-or-register each arm, in priority order, each atomically
-        // under its own lock. Cross-arm atomicity is unnecessary: an arm
-        // becoming ready right after its registration wakes us through the
-        // protocol (the prep-to-park window is closed by RunningNotified).
-        let mut ready = None;
-        for (i, arm) in arms.iter().enumerate() {
-            if !arm.sel_register(me, epoch) {
-                ready = Some(i);
-                break;
-            }
-        }
-        if let Some(i) = ready {
-            // Returning WITHOUT parking while earlier arms hold live-epoch
-            // registrations: retire the wait (bump the epoch, eat a
-            // notification that already landed, re-observe a pending stop)
-            // so no stale arm wake can fault a later one-shot park.
-            crate::scheduler::retire_wait();
+        if let Some(i) = register_arms(me, epoch, arms) {
             return i;
         }
 
@@ -472,3 +455,76 @@ pub fn select(arms: &[&dyn Selectable]) -> usize {
         // stale own-registrations are overwritten.
     }
 }
+
+/// The registration pass shared by [`select`] and [`select_timeout`]:
+/// check-or-register each arm, in priority order, each atomically under its
+/// own lock. Cross-arm atomicity is unnecessary: an arm becoming ready
+/// right after its registration wakes the caller through the protocol (the
+/// prep-to-park window is closed by RunningNotified).
+///
+/// `Some(i)` = arm `i` was ready, the pass stopped, and the wait has been
+/// RETIRED (no park may follow): earlier arms hold live-epoch
+/// registrations, so the epoch is bumped, a landed notification eaten, and
+/// a pending stop re-observed — without which a stale arm wake could fault
+/// a later one-shot park. `None` = every arm registered; the caller parks.
+fn register_arms(me: Pid, epoch: u32, arms: &[&dyn Selectable]) -> Option<usize> {
+    for (i, arm) in arms.iter().enumerate() {
+        if !arm.sel_register(me, epoch) {
+            crate::scheduler::retire_wait();
+            return Some(i);
+        }
+    }
+    None
+}
+
+/// The [`select_timeout`] timer target: stateless, because precise wakes
+/// make classification a pure function of channel state. The entry is
+/// stamped with the select's epoch; if an arm already won, this unpark dies
+/// at the word's epoch CAS (the no-cancellation convention in `timer.rs`).
+struct SelectTimeout;
+impl crate::timer::TimerTarget for SelectTimeout {
+    fn on_timeout(&self, pid: Pid, epoch: u32) {
+        crate::scheduler::unpark_at(pid, epoch);
+    }
+}
+
+/// [`select`] with a deadline: returns `Some(index)` like `select`, or
+/// `None` once `timeout` elapses with no arm ready.
+///
+/// All of `select`'s semantics carry over (priority order, closed arms
+/// permanently ready, no fairness promise). The timeout is one more stamped
+/// waker on the same wait epoch — nothing is registered in any arm for it,
+/// so there is nothing to cancel or leak: an arm winning leaves the timer
+/// entry to expire as a stale-epoch no-op; the timer winning leaves the
+/// arms' registrations to self-clean exactly as a `select` loser's would.
+///
+/// The wake is classified from state alone (wakes are precise): some arm
+/// ready → `Some` of the first, in priority order; none ready → the timer
+/// was the only remaining stamped waker → `None`. A message that races the
+/// deadline resolves message-first, as `recv_timeout` does.
+///
+/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due
+/// timer is drained, then reports `None` unless an arm was already ready.
+///
+/// Panics if `arms` is empty, or when called outside an actor.
+pub fn select_timeout(
+    arms: &[&dyn Selectable],
+    timeout: std::time::Duration,
+) -> Option<usize> {
+    assert!(!arms.is_empty(), "select_timeout() on an empty arm list");
+    let me = crate::actor::current_pid().expect("select_timeout() called outside an actor");
+    let epoch = crate::scheduler::begin_wait();
+    if let Some(i) = register_arms(me, epoch, arms) {
+        return Some(i); // ready now: the timer was never armed
+    }
+
+    // Arm the timer after the registration pass, outside every Channel
+    // lock (insert takes the timers lock).
+    let deadline = crate::timer::deadline_from_now(timeout);
+    let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout);
+    crate::scheduler::insert_wait_timer(deadline, me, target, epoch);
+
+    crate::scheduler::park_current();
+    // Woken precisely: an arm (ready below) or the timer (nothing ready).
+    arms.iter().position(|arm| arm.sel_ready())
+}

src/lib.rs

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

tests/select.rs

@@ -243,3 +243,98 @@ fn motivating_pattern_inbox_plus_monitor_down() {
         worker.join().unwrap();
     });
 }
+
+// ---------------------------------------------------------------------------
+// select_timeout
+// ---------------------------------------------------------------------------
+
+use smarm::select_timeout;
+
+#[test]
+fn select_timeout_returns_none_after_the_deadline() {
+    run(|| {
+        let (_keep_a, rxa) = channel::<i64>();
+        let (_keep_b, rxb) = channel::<i64>();
+        let t0 = Instant::now();
+        let r = select_timeout(&[&rxa, &rxb], Duration::from_millis(30));
+        assert_eq!(r, None);
+        assert!(t0.elapsed() >= Duration::from_millis(30));
+    });
+}
+
+#[test]
+fn select_timeout_ready_arm_wins_without_arming_a_timer() {
+    run(|| {
+        let (txa, rxa) = channel::<i64>();
+        let (_keep_b, rxb) = channel::<i64>();
+        txa.send(5).unwrap();
+        assert_eq!(select_timeout(&[&rxb, &rxa], Duration::from_millis(500)), Some(1));
+        assert_eq!(rxa.try_recv().unwrap(), Some(5));
+    });
+}
+
+#[test]
+fn select_timeout_arm_beats_timer_and_stale_entry_stays_inert() {
+    run(|| {
+        let (txa, rxa) = channel::<i64>();
+        let (_keep_b, rxb) = channel::<i64>();
+        let h = spawn(move || {
+            smarm::sleep(Duration::from_millis(5));
+            txa.send(1).unwrap();
+        });
+        let t0 = Instant::now();
+        let r = select_timeout(&[&rxa, &rxb], Duration::from_millis(200));
+        assert_eq!(r, Some(0));
+        assert!(t0.elapsed() < Duration::from_millis(200));
+        assert_eq!(rxa.try_recv().unwrap(), Some(1));
+        h.join().unwrap();
+        // The abandoned timer entry expires mid-sleep; a stale-epoch wake
+        // landing would return this one-shot park early.
+        let t1 = Instant::now();
+        smarm::sleep(Duration::from_millis(250));
+        assert!(
+            t1.elapsed() >= Duration::from_millis(250),
+            "one-shot park returned early: the stale select timer landed"
+        );
+    });
+}
+
+#[test]
+fn select_timeout_timer_first_message_still_delivered_later() {
+    run(|| {
+        let (txa, rxa) = channel::<i64>();
+        let h = spawn(move || {
+            smarm::sleep(Duration::from_millis(60));
+            txa.send(9).unwrap();
+        });
+        assert_eq!(select_timeout(&[&rxa], Duration::from_millis(10)), None);
+        // The wait is over but the channel is intact: the late message
+        // arrives and a fresh select sees it.
+        assert_eq!(select(&[&rxa]), 0);
+        assert_eq!(rxa.try_recv().unwrap(), Some(9));
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn select_timeout_zero_duration_polls() {
+    run(|| {
+        let (txa, rxa) = channel::<i64>();
+        let (_keep_b, rxb) = channel::<i64>();
+        assert_eq!(select_timeout(&[&rxa, &rxb], Duration::ZERO), None);
+        txa.send(3).unwrap();
+        assert_eq!(select_timeout(&[&rxa, &rxb], Duration::ZERO), Some(0));
+        assert_eq!(rxa.try_recv().unwrap(), Some(3));
+    });
+}
+
+#[test]
+fn select_timeout_closed_arm_is_ready_not_a_timeout() {
+    run(|| {
+        let (_keep_a, rxa) = channel::<i64>();
+        let (txb, rxb) = channel::<i64>();
+        drop(txb);
+        assert_eq!(select_timeout(&[&rxa, &rxb], Duration::from_millis(200)), Some(1));
+        assert!(rxb.try_recv().is_err());
+    });
+}