feat(gen_server): handle_down — runtime monitor forwarding via ServerCtx/Watcher

Monitors are created at runtime (you watch a worker you just spawned in a
handler), so down arms can't ride the static info list. init grows a
&ServerCtx parameter (breaking; no-op default) whose clonable Watcher hands
Monitors to the loop over a control arm; the loop selects the live down
arms ahead of everything else. Arm priority: downs → control → infos →
inbox. A delivered Down retires its arm (monitors are one-shot); a state
that never clones the Watcher closes the control arm after init and the
loop falls back to the plain-inbox park.

src/gen_server.rs

@@ -50,6 +50,7 @@
 //! see `handle_down` — because monitors are inherently created at runtime.)
 
 use crate::channel::{channel, select, Receiver, Selectable, Sender};
+use crate::monitor::{Down, Monitor};
 use crate::pid::Pid;
 use crate::scheduler::{spawn, spawn_under};
 
@@ -71,8 +72,11 @@ pub trait GenServer: Send + 'static {
     /// sources enum them up into one `Info`. Use `()` if unused.
     type Info: Send + 'static;
 
-    /// Runs once inside the server actor before any message is handled.
-    fn init(&mut self) {}
+    /// Runs once inside the server actor before any message is handled. The
+    /// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`]
+    /// into the state here to be able to [`watch`](Watcher::watch) monitors
+    /// from any later handler.
+    fn init(&mut self, _ctx: &ServerCtx) {}
 
     /// Handle a synchronous call and produce the reply sent back to the caller.
     fn handle_call(&mut self, request: Self::Call) -> Self::Reply;
@@ -84,6 +88,10 @@ pub trait GenServer: Send + 'static {
     /// drop it.
     fn handle_info(&mut self, _info: Self::Info) {}
 
+    /// Handle a [`Down`] from a monitor handed to the loop via
+    /// [`Watcher::watch`]. Default: drop it.
+    fn handle_down(&mut self, _down: Down) {}
+
     /// Runs as the server actor exits, on any exit path (see module docs).
     fn terminate(&mut self) {}
 }
@@ -190,6 +198,43 @@ impl<G: GenServer> ServerRef<G> {
     }
 }
 
+/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently
+/// carries only the [`Watcher`]; opaque so fields can grow without breaking.
+pub struct ServerCtx {
+    watcher: Watcher,
+}
+
+impl ServerCtx {
+    /// A clonable handle to the loop's monitor intake. Store it in the state
+    /// during `init` to watch monitors from later handlers.
+    pub fn watcher(&self) -> Watcher {
+        self.watcher.clone()
+    }
+
+    /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`.
+    pub fn watch(&self, m: Monitor) {
+        self.watcher.watch(m)
+    }
+}
+
+/// Hands [`Monitor`]s to a server loop; their [`Down`]s are dispatched to
+/// [`GenServer::handle_down`]. Down arms outrank every other arm (a death
+/// notice cannot be starved), and a delivered `Down` retires its arm —
+/// monitors are one-shot.
+#[derive(Clone)]
+pub struct Watcher {
+    tx: Sender<Monitor>,
+}
+
+impl Watcher {
+    /// Transfer `m` to the server loop. If the server is already gone the
+    /// monitor is silently dropped (its queued `Down`, if any, with it) —
+    /// there is no loop left to care.
+    pub fn watch(&self, m: Monitor) {
+        let _ = self.tx.send(m);
+    }
+}
+
 /// Configure-then-start construction for servers that need more than a bare
 /// [`start`]: info channels, a supervisor. Consumed by [`start`](Self::start).
 ///
@@ -278,10 +323,19 @@ fn server_loop<G: GenServer>(
     }
 
     let mut guard = Terminate(state);
-    guard.0.init();
+
+    // The control arm: Watchers feed Monitors to the loop through it. The
+    // ctx (and with it the loop's own sender) drops right after init — a
+    // state that didn't clone the Watcher closes the arm, the first select
+    // observes the closure, and the loop falls back to the plain-inbox park.
+    let (watch_tx, watch_rx) = channel::<Monitor>();
+    guard.0.init(&ServerCtx { watcher: Watcher { tx: watch_tx } });
+
+    let mut monitors: Vec<Monitor> = Vec::new();
+    let mut watch_open = true;
 
     loop {
-        if infos.is_empty() {
+        if monitors.is_empty() && !watch_open && infos.is_empty() {
             // Nothing to select over: park on the inbox alone, exactly the
             // pre-v0.8 loop.
             match rx.recv() {
@@ -290,27 +344,51 @@ fn server_loop<G: GenServer>(
                 Err(_) => break,
             }
         } else {
-            // Arm priority: infos (declaration order), then the inbox. The
-            // arm slice is rebuilt per iteration because `infos` shrinks as
-            // arms close.
+            // Arm priority: downs, then the control arm, then infos (each in
+            // declaration order), then the inbox. The arm slice is rebuilt
+            // per iteration because every set but the inbox shrinks or grows.
+            let nd = monitors.len();
+            let nw = watch_open as usize;
             let i = {
-                let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(infos.len() + 1);
+                let mut arms: Vec<&dyn Selectable> =
+                    Vec::with_capacity(nd + nw + infos.len() + 1);
+                for m in &monitors {
+                    arms.push(&m.rx);
+                }
+                if watch_open {
+                    arms.push(&watch_rx);
+                }
                 for r in &infos {
                     arms.push(r);
                 }
                 arms.push(&rx);
                 select(&arms)
             };
-            if i < infos.len() {
-                match infos[i].try_recv() {
-                    Ok(Some(info)) => guard.0.handle_info(info),
+            if i < nd {
+                // A Down retires its arm either way: delivered (one-shot) or
+                // closed without delivering (defensive; shouldn't happen).
+                let m = monitors.remove(i);
+                if let Ok(Some(down)) = m.rx.try_recv() {
+                    guard.0.handle_down(down);
+                }
+            } else if i < nd + nw {
+                match watch_rx.try_recv() {
+                    Ok(Some(m)) => monitors.push(m),
                     // Single-receiver: nothing can drain the arm between
                     // select's ready and our try_recv.
+                    Ok(None) => debug_assert!(false, "ready control arm was empty"),
+                    // Every Watcher gone: stop selecting on the arm.
+                    Err(_) => watch_open = false,
+                }
+            } else if i < nd + nw + infos.len() {
+                let j = i - nd - nw;
+                match infos[j].try_recv() {
+                    Ok(Some(info)) => guard.0.handle_info(info),
                     Ok(None) => debug_assert!(false, "ready info arm was empty"),
                     // Senders all gone: drop the arm, keep serving. `remove`
                     // (not swap_remove) — order is priority.
                     Err(_) => {
-                        infos.remove(i);
+                        infos.remove(j);
                     }
                 }
             } else {

src/lib.rs

@@ -46,7 +46,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator;
 // ---------------------------------------------------------------------------
 
 pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender};
-pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerRef};
+pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerCtx, ServerRef, Watcher};
 pub use link::{link, trap_exit, unlink, ExitSignal};
 pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};
 pub use mutex::{LockTimeout, Mutex, MutexGuard};

tests/gen_server.rs

@@ -71,7 +71,7 @@ impl GenServer for Lifecycle {
     type Cast = ();
     type Info = ();
 
-    fn init(&mut self) {
+    fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx) {
         self.log.lock().unwrap().push("init");
     }
 
@@ -339,3 +339,101 @@ fn closed_info_arm_is_dropped_silently() {
     });
     assert_eq!(*got.lock().unwrap(), vec!["cast"]);
 }
+
+// ---------------------------------------------------------------------------
+// handle_down: monitors handed to the loop via Watcher (v0.8)
+// ---------------------------------------------------------------------------
+
+use smarm::gen_server::Watcher;
+use smarm::{monitor, spawn, DownReason, Pid};
+
+/// The motivating pattern: a server that spawns workers from a handler,
+/// watches them, and logs their deaths.
+struct Pool {
+    watcher: Option<Watcher>,
+    log: Vec<DownReason>,
+}
+
+enum PoolCast {
+    SpawnDoomedWorker,
+    Watch(Pid),
+}
+
+impl GenServer for Pool {
+    type Call = ();
+    type Reply = Vec<DownReason>;
+    type Cast = PoolCast;
+    type Info = ();
+
+    fn init(&mut self, ctx: &smarm::gen_server::ServerCtx) {
+        self.watcher = Some(ctx.watcher());
+    }
+
+    fn handle_call(&mut self, _: ()) -> Vec<DownReason> {
+        self.log.clone()
+    }
+
+    fn handle_cast(&mut self, cast: PoolCast) {
+        let watcher = self.watcher.as_ref().expect("init ran first");
+        match cast {
+            PoolCast::SpawnDoomedWorker => {
+                let h = spawn(|| panic!("worker died"));
+                watcher.watch(monitor(h.pid()));
+            }
+            PoolCast::Watch(pid) => watcher.watch(monitor(pid)),
+        }
+    }
+
+    fn handle_down(&mut self, down: smarm::Down) {
+        self.log.push(down.reason);
+    }
+}
+
+// A worker spawned and watched from inside a handler delivers its Down to
+// handle_down. Down arms outrank the inbox, so the death is in the log by
+// the time the follow-up call is answered.
+#[test]
+fn worker_pool_down_reaches_handle_down() {
+    let got = Arc::new(Mutex::new(Vec::new()));
+    let got2 = got.clone();
+    run(move || {
+        let server = start(Pool { watcher: None, log: Vec::new() });
+        server.cast(PoolCast::SpawnDoomedWorker).unwrap();
+        let _ = server.call(()).unwrap(); // sync point: cast handled, worker live
+        *got2.lock().unwrap() = server.call(()).unwrap();
+    });
+    assert_eq!(*got.lock().unwrap(), vec![DownReason::Panic]);
+}
+
+// Watching an already-dead pid yields an immediate NoProc Down, and the down
+// arm outranks the inbox: the call cast *after* the watch still observes it.
+#[test]
+fn watch_dead_pid_is_noproc_down() {
+    let got = Arc::new(Mutex::new(Vec::new()));
+    let got2 = got.clone();
+    run(move || {
+        let h = spawn(|| {});
+        let dead = h.pid();
+        h.join().unwrap();
+        let server = start(Pool { watcher: None, log: Vec::new() });
+        server.cast(PoolCast::Watch(dead)).unwrap();
+        *got2.lock().unwrap() = server.call(()).unwrap();
+    });
+    assert_eq!(*got.lock().unwrap(), vec![DownReason::NoProc]);
+}
+
+// A state that never clones the Watcher closes the control arm; the loop
+// falls back to the plain-inbox park and keeps serving. (Every pre-v0.8 test
+// in this file also exercises this path.)
+#[test]
+fn unused_ctx_closes_control_arm_silently() {
+    let got = Arc::new(Mutex::new(0i64));
+    let got2 = got.clone();
+    run(move || {
+        let server = start(Counter { n: 0 });
+        server.cast(Op::Add(2)).unwrap();
+        server.cast(Op::Add(40)).unwrap();
+        *got2.lock().unwrap() = server.call(Req::Get).unwrap();
+    });
+    assert_eq!(*got.lock().unwrap(), 42);
+}