gen_server: tick_every periodic sugar — loop-driven re-arm via factory + Sys::Tick (RFC 015 §4.3)

src/gen_server.rs

@@ -266,6 +266,10 @@ enum Sys<G: GenServer> {
     /// it from the live set) and the server's payload, dispatched to
     /// [`GenServer::handle_timer`].
     Timer(crate::timer::TimerId, G::Timer),
+    /// A fired periodic tick carrying its stable local id. The loop resolves the
+    /// payload from the registry's factory, dispatches it to
+    /// [`GenServer::handle_timer`], and re-arms the next tick (RFC 015 §4.3).
+    Tick(crate::timer::TimerId),
 }
 
 /// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the
@@ -274,7 +278,7 @@ enum Sys<G: GenServer> {
 /// fields can grow without breaking.
 pub struct ServerCtx<G: GenServer> {
     sys_tx: Sender<Sys<G>>,
-    reg: Arc<Mutex<TimerReg>>,
+    reg: Arc<Mutex<TimerReg<G>>>,
 }
 
 impl<G: GenServer> ServerCtx<G> {
@@ -305,8 +309,7 @@ impl<G: GenServer> ServerCtx<G> {
 /// `Mutex` is always uncontended; it is `Arc<Mutex>` rather than `Rc<RefCell>`
 /// only because the handle is stored on `self` and `GenServer: Send` forces the
 /// handle (hence its shared state) to be `Send`.
-#[derive(Default)]
-struct TimerReg {
+struct TimerReg<G: GenServer> {
     /// Monotonic minter for loop-local [`TimerId`]s — the ids handed to users,
     /// kept distinct from the substrate `seq` (which changes on every periodic
     /// re-arm). A local id is resolved only through this registry, never passed
@@ -316,9 +319,47 @@ struct TimerReg {
     /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire
     /// or [`cancel`](TimerHandle::cancel) removes it.
     oneshots: HashMap<TimerId, crate::timer::TimerId>,
+    /// Live periodic timers: stable local id → its bookkeeping. The stable id
+    /// is minted once by [`tick_every`](TimerHandle::tick_every) and survives
+    /// every re-arm; the entry lives until [`cancel`](TimerHandle::cancel)
+    /// removes it (steady re-arm never self-removes).
+    periodics: HashMap<TimerId, Periodic<G>>,
+    /// A `Sys` sender held *only while ≥1 periodic is live*, used by the loop to
+    /// re-arm the next tick. `Some` exactly when `periodics` is non-empty: this
+    /// keeps the system arm open for a server whose only system use is a
+    /// periodic, while leaving a non-timer server's arm to auto-close (the
+    /// `unused_ctx` behaviour) since this stays `None` for it.
+    rearm_tx: Option<Sender<Sys<G>>>,
 }
 
-impl TimerReg {
+/// One periodic timer's loop-side bookkeeping.
+struct Periodic<G: GenServer> {
+    /// Re-arm interval; each fire schedules the next at `now + every` (§4.3).
+    every: Duration,
+    /// The currently-armed substrate timer for this periodic (changes on every
+    /// re-arm); cancelled by [`cancel`](TimerHandle::cancel).
+    live: crate::timer::TimerId,
+    /// Produces a fresh payload for each tick. Built in `tick_every` as
+    /// `move || msg.clone()` — so the `Clone` bound lives there, on the one
+    /// method that needs it, and never leaks onto `type Timer` or the loop
+    /// (which is monomorphised per server and only *calls* this box).
+    make: Box<dyn FnMut() -> G::Timer + Send>,
+}
+
+// Manual Default: deriving would demand `G: Default`, but a fresh registry is
+// independent of the server type.
+impl<G: GenServer> Default for TimerReg<G> {
+    fn default() -> Self {
+        TimerReg {
+            next_local: 0,
+            oneshots: HashMap::new(),
+            periodics: HashMap::new(),
+            rearm_tx: None,
+        }
+    }
+}
+
+impl<G: GenServer> TimerReg<G> {
     /// Mint the next loop-local id.
     fn mint(&mut self) -> TimerId {
         let id = TimerId::from_raw(self.next_local);
@@ -334,7 +375,7 @@ impl TimerReg {
 /// in-flight fire) lives.
 pub struct TimerHandle<G: GenServer> {
     sys_tx: Sender<Sys<G>>,
-    reg: Arc<Mutex<TimerReg>>,
+    reg: Arc<Mutex<TimerReg<G>>>,
 }
 
 // Manual Clone for the same reason as `Watcher`: no `G: Clone` needed.
@@ -360,14 +401,54 @@ impl<G: GenServer> TimerHandle<G> {
         local
     }
 
-    /// Cancel an armed timer. Returns the substrate's race signal — `true` if
-    /// the cancel beat the fire (delivery is now prevented), `false` if the
-    /// timer had already fired or been cancelled / is unknown.
+    /// Arm a periodic timer: deliver a fresh `msg` to
+    /// [`GenServer::handle_timer`] every `every`, until
+    /// [`cancel`](Self::cancel)led. Loop-managed sugar over the one-shot
+    /// substrate (RFC 015 §4.3): the substrate stays one-shot; the loop re-arms
+    /// at `now + every` after each fire and exposes one stable [`TimerId`], so a
+    /// single `cancel` stops the re-arm *and* the pending instance.
+    ///
+    /// Requires `Self::Timer: Clone` — a periodic re-delivers the same logical
+    /// message each tick, so the loop needs a fresh copy per period. The bound
+    /// sits on this method alone; one-shot [`arm_after`](Self::arm_after) and
+    /// the `type Timer` declaration stay unconstrained.
+    pub fn tick_every(&self, every: Duration, msg: G::Timer) -> TimerId
+    where
+        G::Timer: Clone,
+    {
+        let mut reg = self.reg.lock().unwrap();
+        let local = reg.mint();
+        // A live periodic must keep the system arm open so the next tick can be
+        // re-armed; the first periodic installs the loop's re-arm sender.
+        if reg.periodics.is_empty() {
+            reg.rearm_tx = Some(self.sys_tx.clone());
+        }
+        let make: Box<dyn FnMut() -> G::Timer + Send> = Box::new(move || msg.clone());
+        // First instance fires after `every`; the payload is produced loop-side
+        // from `make` on fire, so the tick carries only the stable id.
+        let sub = send_after_to(every, self.sys_tx.clone(), Sys::Tick(local));
+        reg.periodics.insert(local, Periodic { every, live: sub, make });
+        local
+    }
+
+    /// Cancel an armed timer (one-shot or periodic). Returns the substrate's
+    /// race signal — `true` if the cancel beat the (pending) fire, `false` if
+    /// it had already fired / been cancelled / is unknown. For a periodic this
+    /// also stops the re-arm: the entry is dropped, so the loop will not
+    /// schedule another tick even if the pending one already escaped onto the
+    /// channel (that in-flight tick is discarded on dispatch).
     pub fn cancel(&self, id: TimerId) -> bool {
         let mut reg = self.reg.lock().unwrap();
         if let Some(sub) = reg.oneshots.remove(&id) {
             return cancel_timer(sub);
         }
+        if let Some(p) = reg.periodics.remove(&id) {
+            let beat = cancel_timer(p.live);
+            if reg.periodics.is_empty() {
+                reg.rearm_tx = None;
+            }
+            return beat;
+        }
         false
     }
 }
@@ -691,6 +772,34 @@ fn server_loop<G: GenServer>(
                         reg.lock().unwrap().oneshots.remove(&id);
                         guard.0.handle_timer(msg);
                     }
+                    Ok(Some(Sys::Tick(id))) => {
+                        // A periodic fired. Under the lock: if it is still live
+                        // (cancel didn't beat us here), produce its payload and
+                        // re-arm the next tick at now + every *before* dispatch,
+                        // so the period is measured from fire-handling and a
+                        // handler that cancels stops the just-armed instance.
+                        let msg = {
+                            let mut g = reg.lock().unwrap();
+                            let r = &mut *g;
+                            if let Some(p) = r.periodics.get_mut(&id) {
+                                let every = p.every;
+                                let msg = (p.make)();
+                                let tx = r
+                                    .rearm_tx
+                                    .as_ref()
+                                    .expect("a live periodic keeps rearm_tx Some")
+                                    .clone();
+                                p.live = send_after_to(every, tx, Sys::Tick(id));
+                                Some(msg)
+                            } else {
+                                // Cancelled between fire and dispatch: discard.
+                                None
+                            }
+                        };
+                        if let Some(msg) = msg {
+                            guard.0.handle_timer(msg);
+                        }
+                    }
                     // Single-receiver: nothing can drain the arm between
                     // select's ready and our try_recv.
                     Ok(None) => debug_assert!(false, "ready system arm was empty"),

tests/gen_server.rs

@@ -453,6 +453,7 @@ use smarm::TimerId;
 
 enum TkCast {
     Arm(Duration),
+    Tick(Duration),
     CancelLast,
 }
 
@@ -482,6 +483,7 @@ impl GenServer for Timed {
         let t = self.timer.as_ref().expect("init ran first");
         match c {
             TkCast::Arm(d) => self.last = Some(t.arm_after(d, 7)),
+            TkCast::Tick(d) => self.last = Some(t.tick_every(d, 9)),
             TkCast::CancelLast => {
                 let id = self.last.take().expect("a timer was armed");
                 *self.cancel_won.lock().unwrap() = Some(t.cancel(id));
@@ -535,3 +537,46 @@ fn cancel_before_fire_suppresses_it() {
     assert_eq!(*cancel_won.lock().unwrap(), Some(true), "cancel beat the fire");
     assert!(fired.lock().unwrap().is_empty());
 }
+
+// tick_every re-arms: a periodic fires repeatedly off one arm, each tick
+// carrying a fresh payload. (Timer fires outrank the inbox by arm position, so
+// a periodic cannot be starved by userspace traffic — the same property the
+// info_outranks_inbox test pins for infos.)
+#[test]
+fn tick_every_rearms_repeatedly() {
+    let fired = Arc::new(Mutex::new(Vec::new()));
+    let f2 = fired.clone();
+    run(move || {
+        let cw = Arc::new(Mutex::new(None));
+        let server = start(timed(f2, cw));
+        server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap();
+        let _ = server.call(()).unwrap(); // sync: periodic armed
+        smarm::sleep(Duration::from_millis(130)); // ~6 periods
+        let count = server.call(()).unwrap();
+        assert!(count >= 3, "periodic should have re-armed several times, got {count}");
+    });
+    // Every tick delivered the same payload.
+    assert!(fired.lock().unwrap().iter().all(|&v| v == 9));
+}
+
+// Cancelling a periodic stops the re-arm: no further ticks land after cancel.
+#[test]
+fn cancel_stops_a_periodic() {
+    let fired = Arc::new(Mutex::new(Vec::new()));
+    let cancel_won = Arc::new(Mutex::new(None));
+    let f2 = fired.clone();
+    let c2 = cancel_won.clone();
+    run(move || {
+        let server = start(timed(f2, c2));
+        server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap();
+        let _ = server.call(()).unwrap();
+        smarm::sleep(Duration::from_millis(70)); // a few ticks
+        server.cast(TkCast::CancelLast).unwrap();
+        let after_cancel = server.call(()).unwrap(); // sync: cancel handled
+        smarm::sleep(Duration::from_millis(80)); // would be several more ticks
+        let later = server.call(()).unwrap();
+        assert_eq!(later, after_cancel, "no ticks may land after cancel");
+    });
+    // The periodic had fired at least once before being cancelled.
+    assert!(!fired.lock().unwrap().is_empty());
+}