feat(monitor): demonitor + per-monitor MonitorId (roadmap #5)

Monitors could be installed but never taken down. That gap was about to
bite: the gen_server call timeout we want next is the Erlang dance —
monitor the server, wait for the reply or a Down or a deadline, then
demonitor — and without a way to remove a registration, every timed-out
call would leak a monitor on the server's slot and risk a stale Down
arriving later. So this lands the cleanup primitive before anything
depends on the old shape.

The decision flagged in the roadmap ("decide the monitor API NOW") is
resolved by giving each registration a process-unique MonitorId and
returning it to the caller. monitor() now hands back a
Monitor { id, target, rx } rather than a bare Receiver<Down>: read the
notice from rx as before, and pass &Monitor to demonitor to tear exactly
that registration down. The id comes from a monotonic counter in shared
state, bumped under the same lock that does the registration, so it's
deterministic and never reused — which is what lets demonitor name one of
several monitors on the same target unambiguously. target rode along on
the struct (over the sketched {id, rx}) purely so demonitor can go
straight to the slot instead of scanning every slot for the id.

demonitor returns Option<MonitorId> rather than a bool: Some(id) when a
live registration was found and removed, None when there was nothing left
to remove — it already fired (the registration is drained on finalize),
it was a NoProc, or the slot has been reclaimed. The generation half of
the pid quietly protects that last case: a recycled slot index fails
slot_mut's generation check, so a late demonitor is a clean no-op and can
never strip a different actor's monitor that happens to share the index.

The one piece of real care is reentrancy. Removing a registration drops
the slot's Sender, and Sender::drop can unpark a parked receiver, which
re-enters the shared mutex — which is not reentrant. So demonitor moves
the sender out of the Vec under the lock and lets it drop only after the
lock is released, the same discipline finalize_actor already follows for
its monitor and supervisor sends.

Flushing a Down the target already queued isn't a separate flag; it falls
out of dropping the Monitor. demonitor(&m); drop(m) stops future notices
and discards any queued one — the gen_server-call cleanup in one move.

Storage is now Vec<(MonitorId, Sender<Down>)>. The three slot-reset sites
were left alone on purpose: they clear/rebuild the Vec, which doesn't care
about the element type, so there's no fourth reset obligation. finalize
just destructures (_, m). Because chained_spawn and yield_many register no
monitors, that Vec stays empty on the hot path — taking an empty Vec costs
the same and the notify loop runs zero times — and a before/after general
probe confirmed both medians sit within noise.

Tests cover the three behaviours that matter: a demonitored watcher gets
no Down (its channel closes), demonitoring one of several leaves the
siblings firing normally, and demonitoring after the Down has already
fired reports None.

src/monitor.rs

@@ -1,8 +1,9 @@
 //! Process monitors.
 //!
 //! `monitor(target)` asks the runtime to deliver a single [`Down`] when
-//! `target` terminates, and hands back the [`Receiver`] to read it from. A
-//! monitor is:
+//! `target` terminates, and hands back a [`Monitor`] — the [`Receiver`] to read
+//! it from, plus the identity (`id`, `target`) needed to take the registration
+//! back down with [`demonitor`]. A monitor is:
 //!
 //!   - **unidirectional** — the watcher learns of the target's death, but the
 //!     target learns nothing of the watcher, and the watcher is unaffected by
@@ -24,14 +25,27 @@
 //! Monitoring a pid that is already gone (reclaimed, or never alive) yields
 //! [`DownReason::NoProc`] immediately, mirroring Erlang's `noproc`.
 //!
+//! ## Demonitoring
+//!
+//! Each `monitor()` registration is tagged with a process-unique [`MonitorId`].
+//! [`demonitor`] removes the registration named by a [`Monitor`] from its
+//! target's slot, returning `Some(id)` if a live registration was found or
+//! `None` if it had already fired (or the target is gone). Dropping the
+//! [`Monitor`] afterwards discards any `Down` that the target had *already*
+//! queued — the equivalent of Erlang's `demonitor(Ref, [flush])`.
+//!
 //! ## Races
 //!
 //! Registration (below) and `finalize_actor` (in `runtime`) both run under the
 //! shared-state mutex, so a target that is still alive when its monitor is
 //! registered is guaranteed to deliver a real `Down`; there is no window in
 //! which the death slips between the liveness check and the registration.
+//! `demonitor` is protected by the generation half of the pid: if the target
+//! has died and its slot index been recycled, `slot_mut(target)` fails the
+//! generation check and `demonitor` is a clean no-op — it can never strip a
+//! *different* actor's monitor that happens to share the slot index.
 
-use crate::channel::{channel, Receiver};
+use crate::channel::{channel, Receiver, Sender};
 use crate::pid::Pid;
 use crate::runtime::State;
 use crate::scheduler::with_runtime;
@@ -63,26 +77,56 @@ pub struct Down {
     pub reason: DownReason,
 }
 
-/// Monitor `target`. Returns a channel that will receive exactly one [`Down`].
+/// A process-unique identifier for one `monitor()` registration.
+///
+/// Opaque and `Copy`. Allocated from a monotonic counter in shared state, so
+/// it is never reused for the lifetime of the runtime — distinct `monitor()`
+/// calls on the same target get distinct ids, which is what lets [`demonitor`]
+/// tear down exactly one of several monitors on a target.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MonitorId(pub(crate) u64);
+
+/// A live monitor: the receiving end of the one-shot [`Down`] channel, plus the
+/// identity needed to [`demonitor`] it.
+///
+/// Read the notification from [`Monitor::rx`]. Not `Clone` (the receiver is a
+/// single consumer). Dropping it closes the receiving end; if a `Down` was
+/// already queued it is discarded with the channel.
+pub struct Monitor {
+    /// This registration's process-unique id.
+    pub id: MonitorId,
+    /// The pid being monitored.
+    pub target: Pid,
+    /// The one-shot channel the `Down` arrives on.
+    pub rx: Receiver<Down>,
+}
+
+/// Monitor `target`. Returns a [`Monitor`] whose `rx` receives exactly one
+/// [`Down`].
 ///
 /// If `target` is still live, the `Down` arrives when it terminates. If
 /// `target` is already gone, a [`DownReason::NoProc`] `Down` is queued
-/// immediately so the caller's `recv()` returns without parking.
-pub fn monitor(target: Pid) -> Receiver<Down> {
+/// immediately so the caller's `rx.recv()` returns without parking.
+pub fn monitor(target: Pid) -> Monitor {
     let (tx, rx) = channel::<Down>();
 
-    // Register under the shared lock. We clone the sender into the target's
-    // monitor list and keep the original for the NoProc fallback; we must not
-    // send while holding the lock, since `Sender::send` can call back into the
-    // runtime (to unpark a parked receiver) and the shared mutex is not
-    // reentrant.
-    let registered = with_runtime(|inner| {
-        inner.with_shared(|s| match s.slot_mut(target) {
-            Some(slot) if !matches!(slot.state, State::Done) => {
-                slot.monitors.push(tx.clone());
-                true
-            }
-            _ => false,
+    // Register under the shared lock. We allocate the id and (if the target is
+    // live) clone the sender into its monitor list, keeping the original `tx`
+    // for the NoProc fallback. `tx.clone()` only touches the channel's own
+    // mutex, never the shared runtime mutex, so it is safe under the lock — but
+    // we must not *send* here, as `Sender::send` can call back in to unpark a
+    // parked receiver and the shared mutex is not reentrant.
+    let (id, registered) = with_runtime(|inner| {
+        inner.with_shared(|s| {
+            let id = s.alloc_monitor_id();
+            let registered = match s.slot_mut(target) {
+                Some(slot) if !matches!(slot.state, State::Done) => {
+                    slot.monitors.push((id, tx.clone()));
+                    true
+                }
+                _ => false,
+            };
+            (id, registered)
         })
     });
 
@@ -90,5 +134,30 @@ pub fn monitor(target: Pid) -> Receiver<Down> {
         let _ = tx.send(Down { pid: target, reason: DownReason::NoProc });
     }
 
-    rx
+    Monitor { id, target, rx }
+}
+
+/// Cancel the monitor `m`. Returns `Some(id)` if a live registration was found
+/// on the target's slot and removed, or `None` if there was nothing to remove
+/// — the target already fired its `Down` (the registration is drained on
+/// finalize), was never alive (`NoProc`), or has been reclaimed.
+///
+/// This stops any *future* `Down`. To also discard a `Down` the target may have
+/// *already* queued (the finalize-races-demonitor case), drop `m` afterwards;
+/// dropping the [`Monitor`] closes its receiver and the queued notice goes with
+/// it — the analogue of Erlang's `demonitor(Ref, [flush])`.
+pub fn demonitor(m: &Monitor) -> Option<MonitorId> {
+    // Remove the registration under the lock, but move the `Sender` *out* and
+    // let it drop only after the lock is released: dropping the last sender
+    // runs `Sender::drop`, which may unpark a parked receiver → `with_shared`,
+    // and the shared mutex is not reentrant.
+    let removed: Option<(MonitorId, Sender<Down>)> = with_runtime(|inner| {
+        inner.with_shared(|s| {
+            let slot = s.slot_mut(m.target)?;
+            let pos = slot.monitors.iter().position(|(mid, _)| *mid == m.id)?;
+            Some(slot.monitors.remove(pos))
+        })
+    });
+    // `removed`'s sender drops here, outside the lock.
+    removed.map(|(id, _sender)| id)
 }