diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index c7cd6c5..5c28cfe 100644 --- a/CLAUDE/notes/progress.md +++ b/CLAUDE/notes/progress.md @@ -2,7 +2,7 @@ ## Current Stage: Phase 1 — Research + First Improvement Cycle -### Status: Cycle 12 COMPLETE +### Status: Cycle 13 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -124,6 +124,36 @@ - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err - **Result**: 82 tests pass, all workspace compiles +### Cycle 13: Actor Monitoring / Death Watch +- **Research**: Studied monitoring across Erlang (monitor/2, DOWN messages), Akka (watch/Terminated), + Ractor (link, SupervisionEvent), Actix (none), Kameo (link, on_link_died callback) + - Key finding: Erlang's unidirectional monitor + message delivery is the best fit for swactor + (reuses existing type-erased handler, zero trait changes, composable) + - Callbacks (Ractor/Kameo style) rejected: would require adding to AnyActor/ActorInterface traits + - Bidirectional links deferred: can layer on top of monitors later +- **Implementation**: `MonitorRegistry` in delivery.rs + `Down`/`StopReason`/`MonitorRef` in actor.rs + - `MonitorRegistry`: `RwLock>>` (watched→watchers) + + reverse `RwLock>` for O(1) demonitor + - `MonitorRef(u64)`: unique token from `AtomicU64` counter + - `Down { addr: ActorAddress, reason: StopReason }`: delivered as normal mailbox message + - `StopReason`: `Normal` (graceful stop) | `Panicked` (panic, not restartable) + - `ctx.monitor(target)` → `MonitorRef` — subscribe to death notifications + - `ctx.demonitor(mref)` — cancel a subscription + - `cleanup_dead` now returns `Vec<(ActorAddress, StopReason)>` instead of `Vec` + - After cleanup_dead: iterate dead actors, take_monitors from registry, route Down through normal + delivery (pool.deliver for same-worker, transfer_txs for cross-worker, inbox_registry for inboxes) + - Dead watcher cleanup: `remove_watcher()` strips monitor subscriptions for dead watchers + - Multiple monitors of same target produce independent notifications (stacking, like Erlang) +- **Tests**: 7 new behavioral tests + - `monitor_notifies_on_graceful_stop` — Down{reason: Normal} on stop + - `monitor_notifies_on_panic` — Down{reason: Panicked} on panic + - `multiple_watchers_all_notified` — two watchers both get Down + - `demonitor_cancels_notification` — demonitor → no Down delivered + - `dead_watcher_does_not_receive_down` — dead watcher's monitors cleaned up + - `down_delivered_to_external_inbox` — Down forwarded through inbox + - `stacked_monitors_produce_multiple_notifications` — two monitors on same target → two Downs +- **Result**: 113 tests pass (106 behavioral + 7 proptest), all workspace compiles, zero warnings + ### Cycle 12: Named Actor Registry - **Research**: Studied named actor/service discovery across Erlang (register/2, whereis/1, global, pg), Actix (Registry, SystemRegistry — TypeId keys), Bastion (hierarchy-based), Ractor (String keys, DashMap, diff --git a/src/actor.rs b/src/actor.rs index aa5694c..3ba2008 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -137,6 +137,33 @@ where } } +/// Unique token identifying a monitor subscription. +/// +/// Returned by [`Ctx::monitor`] and used with [`Ctx::demonitor`] to cancel. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MonitorRef(pub(crate) u64); + +/// Reason an actor was removed from the runtime. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StopReason { + /// Graceful stop (via `ctx.stop_self()` or `Runtime::stop_actor()`). + Normal, + /// Actor panicked and could not be restarted. + Panicked, +} + +/// Death notification delivered as a normal message when a monitored actor dies. +/// +/// Subscribe via [`Ctx::monitor`]. The `Down` message arrives in the watcher's +/// regular `handle()` method — no special callback needed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Down { + /// Address of the dead actor. + pub addr: ActorAddress, + /// Why it died. + pub reason: StopReason, +} + /// Internal sentinel message for graceful actor stop. /// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`. pub(crate) struct StopSignal; @@ -182,6 +209,10 @@ pub trait ContextInner { fn where_is(&self, name: &str) -> Option; /// Register a name → address mapping. Returns `Err` if the name is taken. fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error>; + /// Subscribe to death notifications for `target`. Returns a MonitorRef for cancellation. + fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef; + /// Cancel a monitor subscription. + fn demonitor(&self, mref: MonitorRef); } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -276,6 +307,20 @@ impl<'a> Ctx<'a> { Ok(addr) } + /// Subscribe to death notifications for `target`. + /// + /// When `target` dies (stop or panic), a [`Down`] message is delivered to + /// this actor's mailbox as a normal message. Multiple monitors of the same + /// target create independent subscriptions. + pub fn monitor(&self, target: ActorAddress) -> MonitorRef { + self.inner.monitor(self.self_addr, target) + } + + /// Cancel a previously created monitor subscription. + pub fn demonitor(&self, mref: MonitorRef) { + self.inner.demonitor(mref); + } + /// Spawn a restartable actor. On panic, recreated via `factory` up to /// `max_restarts` times before permanent poisoning. pub fn spawn_restartable( diff --git a/src/delivery.rs b/src/delivery.rs index 4b49a0d..027f72b 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -1,10 +1,10 @@ use std::any::Any; use std::collections::HashMap; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock, RwLock}; use std::thread::Thread; -use crate::actor::{ActorAddress, AnyActor, Message}; +use crate::actor::{ActorAddress, AnyActor, Message, MonitorRef}; use crate::channel::Sender; use crate::config::RuntimeConfig; use crate::stats::WorkerStats; @@ -190,6 +190,7 @@ pub(crate) struct TickContext<'a> { pub(crate) inbox_registry: &'a InboxRegistry, pub(crate) config: &'a RuntimeConfig, pub(crate) name_registry: &'a NameRegistry, + pub(crate) monitor_registry: &'a MonitorRegistry, pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>, /// Thread handles for waking parked workers on cross-worker sends. pub(crate) worker_threads: &'a [OnceLock], @@ -255,6 +256,82 @@ impl NameRegistry { } } +// ─── Monitor Registry ──────────────────────────────────────────────────────── + +/// Tracks monitor subscriptions: watched actor → list of (MonitorRef, watcher address). +/// +/// Write-rare (monitor/demonitor/death), read at cleanup time. +pub(crate) struct MonitorRegistry { + /// watched_addr → [(mref, watcher_addr)] + monitors: RwLock>>, + /// mref → watched_addr (for O(1) demonitor) + ref_to_target: RwLock>, + next_ref: AtomicU64, +} + +impl MonitorRegistry { + pub fn new() -> Self { + Self { + monitors: RwLock::new(HashMap::new()), + ref_to_target: RwLock::new(HashMap::new()), + next_ref: AtomicU64::new(1), + } + } + + /// Register a monitor: `watcher` wants to know when `target` dies. + pub fn register(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef { + let id = self.next_ref.fetch_add(1, Ordering::Relaxed); + let mref = MonitorRef(id); + self.monitors.write().unwrap() + .entry(target) + .or_default() + .push((mref, watcher)); + self.ref_to_target.write().unwrap().insert(mref, target); + mref + } + + /// Cancel a monitor by its ref. + pub fn deregister(&self, mref: MonitorRef) { + if let Some(target) = self.ref_to_target.write().unwrap().remove(&mref) { + let mut monitors = self.monitors.write().unwrap(); + if let Some(watchers) = monitors.get_mut(&target) { + watchers.retain(|(r, _)| *r != mref); + if watchers.is_empty() { + monitors.remove(&target); + } + } + } + } + + /// Remove and return all monitors for a dead actor. + pub fn take_monitors(&self, target: &ActorAddress) -> Vec<(MonitorRef, ActorAddress)> { + let watchers = self.monitors.write().unwrap().remove(target).unwrap_or_default(); + let mut ref_map = self.ref_to_target.write().unwrap(); + for (mref, _) in &watchers { + ref_map.remove(mref); + } + watchers + } + + /// Remove all monitor subscriptions where `addr` is the watcher (dead watcher cleanup). + pub fn remove_watcher(&self, addr: &ActorAddress) { + let mut monitors = self.monitors.write().unwrap(); + let mut ref_map = self.ref_to_target.write().unwrap(); + // Iterate all targets and remove entries where this addr is the watcher + monitors.retain(|_target, watchers| { + watchers.retain(|(mref, watcher)| { + if watcher == addr { + ref_map.remove(mref); + false + } else { + true + } + }); + !watchers.is_empty() + }); + } +} + impl<'a> TickContext<'a> { /// Route a message whose destination is not in the local address map. /// Tries inbox registry, then remote transport, then falls back to inbox error. diff --git a/src/runtime.rs b/src/runtime.rs index bc330bd..d1330b2 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -9,7 +9,7 @@ use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopS use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; -use crate::delivery::{AddressMap, Envelope, InboxRegistry, NameRegistry, Placement, TickContext, WorkerId}; +use crate::delivery::{AddressMap, Envelope, InboxRegistry, MonitorRegistry, NameRegistry, Placement, TickContext, WorkerId}; use crate::stats::{StatsHook, WorkerStats}; // Re-export stats types so existing code using `runtime::*` still works pub use crate::stats::{RuntimeStats, WorkerInfo}; @@ -63,6 +63,7 @@ pub struct Runtime { address_map: Arc, inbox_registry: Arc, name_registry: Arc, + monitor_registry: Arc, transfer_txs: Vec>, spawn_txs: Vec)>>, placement: Placement, @@ -121,6 +122,7 @@ impl Runtime { let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); let inbox_registry = Arc::new(InboxRegistry::new()); let name_registry = Arc::new(NameRegistry::new()); + let monitor_registry = Arc::new(MonitorRegistry::new()); let mut transfer_txs = Vec::with_capacity(num_workers); let mut spawn_txs = Vec::with_capacity(num_workers); @@ -159,6 +161,7 @@ impl Runtime { address_map, inbox_registry, name_registry, + monitor_registry, transfer_txs, spawn_txs, placement, @@ -292,6 +295,7 @@ impl Runtime { inbox_registry: &self.inbox_registry, config: &self.config, name_registry: &self.name_registry, + monitor_registry: &self.monitor_registry, stats_hook: self.stats_hook.as_deref(), worker_threads: &self.worker_threads, #[cfg(feature = "transport")] @@ -500,4 +504,12 @@ impl ContextInner for Runtime { fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error> { self.name_registry.register(name, addr) } + + fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef { + self.monitor_registry.register(watcher, target) + } + + fn demonitor(&self, mref: crate::actor::MonitorRef) { + self.monitor_registry.deregister(mref); + } } diff --git a/src/worker.rs b/src/worker.rs index 3583fc5..bacbc0d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::thread; use std::time::Instant; -use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopSignal, TimerRequest}; +use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest}; use crate::channel::Receiver; use crate::config::MailboxOverflow; use crate::delivery::{Envelope, TickContext, WorkerId}; @@ -319,9 +319,9 @@ impl Worker { }; let dead = self.pool.cleanup_dead(&cleanup_ctx); if !dead.is_empty() { - for addr in &dead { - tc.address_map.remove(addr); - tc.name_registry.unregister_by_addr(addr); + for &(addr, _) in &dead { + tc.address_map.remove(&addr); + tc.name_registry.unregister_by_addr(&addr); } // Re-publish num_actors after cleanup so stats reflect removal self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed); @@ -334,8 +334,34 @@ impl Worker { self.pool.deliver(&addr, msg); } + // Emit Down notifications for monitored dead actors + for &(addr, reason) in &dead { + let watchers = tc.monitor_registry.take_monitors(&addr); + for (_mref, watcher) in watchers { + let down = crate::actor::Down { addr, reason }; + // Route through normal delivery path + if self.pool.contains(&watcher) { + self.pool.deliver(&watcher, Box::new(down)); + } else { + match tc.address_map.lookup(&watcher) { + Some(wid) => { + tc.transfer_txs[wid.as_usize()] + .send(Envelope::new(watcher, Box::new(down))); + crate::runtime::notify_worker(tc.worker_threads, wid.as_usize()); + } + None => { + let _ = tc.inbox_registry.try_deliver(watcher, Box::new(down)); + } + } + } + } + // Clean up any monitors the dead actor had placed on others + tc.monitor_registry.remove_watcher(&addr); + } + // GC orphaned interval timers for actors that were just removed - self.timers.gc_dead_intervals(&dead); + let dead_addrs: Vec = dead.iter().map(|(a, _)| *a).collect(); + self.timers.gc_dead_intervals(&dead_addrs); did_work } @@ -427,6 +453,14 @@ impl ContextInner for WorkerContext<'_> { fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> { self.tc.name_registry.register(name, addr) } + + fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef { + self.tc.monitor_registry.register(watcher, target) + } + + fn demonitor(&self, mref: crate::actor::MonitorRef) { + self.tc.monitor_registry.deregister(mref); + } } struct ActorSlot { @@ -626,19 +660,22 @@ impl ActorPool { self.actors.values().map(|slot| slot.mailbox.len()).sum() } - /// Remove poisoned and stopping actors, returning their addresses. + /// Remove poisoned and stopping actors, returning their addresses and stop reasons. /// Called after tick_all so the caller can clean up the address map. /// /// For stopping actors: calls `on_stop()` before removal (wrapped in catch_unwind). /// For poisoned actors: `on_stop()` is NOT called (state may be corrupt). - pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec { - let dead: Vec = self + pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec<(ActorAddress, StopReason)> { + let dead: Vec<(ActorAddress, StopReason)> = self .actors .iter() .filter(|(_, slot)| slot.poisoned || slot.stopping) - .map(|(&addr, _)| addr) + .map(|(&addr, slot)| { + let reason = if slot.poisoned { StopReason::Panicked } else { StopReason::Normal }; + (addr, reason) + }) .collect(); - for &addr in &dead { + for &(addr, _) in &dead { if let Some(mut slot) = self.actors.remove(&addr) { // Call on_stop for gracefully stopping actors only if slot.stopping && !slot.poisoned { diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index c31ba4f..3c2fed2 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -1,7 +1,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; -use swactor::actor::{ActorAddress, ActorInterface}; +use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason}; use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig}; // ── Messages ──────────────────────────────────────────────────────────────── @@ -3086,3 +3086,246 @@ fn ctx_spawn_named_registers_from_handler() { let child_addr = inbox.try_recv().expect("should receive child address"); assert_eq!(rt.where_is("child"), Some(child_addr.0), "name registered from handler"); } + +// ── Actor Monitoring / Death Watch ────────────────────────────────────────── + +/// An actor that monitors a target and forwards Down notifications to a reply address. +struct WatcherActor { + watch_target: ActorAddress, + reply_to: ActorAddress, + mref: Option, +} + +impl ActorInterface for WatcherActor { + type Incoming = Down; + type Response = (); + fn on_start(&mut self, ctx: &Ctx) { + self.mref = Some(ctx.monitor(self.watch_target)); + } + fn handle(&mut self, ctx: &Ctx, msg: Down) { + // Forward the Down notification to the test inbox + ctx.send(self.reply_to, msg).unwrap(); + } +} + +/// Given actor A monitors actor B, +/// when B is gracefully stopped, +/// then A receives a Down { reason: Normal } message. +#[test] +fn monitor_notifies_on_graceful_stop() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PingPongActor).unwrap(); + let _watcher = rt.spawn(WatcherActor { + watch_target: target, + reply_to: *inbox.addr(), + mref: None, + }).unwrap(); + + rt.tick(); // on_start → watcher sets up monitor + rt.stop_actor(target).unwrap(); + rt.tick(); // target receives StopSignal → cleanup_dead emits Down + rt.tick(); // watcher receives Down → forwards to inbox + + let down = inbox.try_recv().expect("should receive Down notification"); + assert_eq!(down.addr, target); + assert_eq!(down.reason, StopReason::Normal); +} + +/// Given actor A monitors actor B, +/// when B panics, +/// then A receives a Down { reason: Panicked } message. +#[test] +fn monitor_notifies_on_panic() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PanicActor).unwrap(); + let _watcher = rt.spawn(WatcherActor { + watch_target: target, + reply_to: *inbox.addr(), + mref: None, + }).unwrap(); + + rt.tick(); // on_start + rt.send_to(target, PanicMsg).unwrap(); + rt.tick(); // target panics → cleanup_dead emits Down + rt.tick(); // watcher receives Down → forwards to inbox + + let down = inbox.try_recv().expect("should receive Down on panic"); + assert_eq!(down.addr, target); + assert_eq!(down.reason, StopReason::Panicked); +} + +/// Given two actors both monitor the same target, +/// when the target dies, +/// then both watchers receive independent Down notifications. +#[test] +fn multiple_watchers_all_notified() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox1 = rt.new_inbox::().unwrap(); + let inbox2 = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PingPongActor).unwrap(); + rt.spawn(WatcherActor { + watch_target: target, + reply_to: *inbox1.addr(), + mref: None, + }).unwrap(); + rt.spawn(WatcherActor { + watch_target: target, + reply_to: *inbox2.addr(), + mref: None, + }).unwrap(); + + rt.tick(); // on_start for all + rt.stop_actor(target).unwrap(); + rt.tick(); // cleanup → Down emitted to both watchers + rt.tick(); // watchers forward Down to inboxes + + assert!(inbox1.try_recv().is_some(), "watcher 1 should receive Down"); + assert!(inbox2.try_recv().is_some(), "watcher 2 should receive Down"); +} + +/// An actor that demonitors in response to a Ping message. +struct DemonitorActor { + watch_target: ActorAddress, + mref: Option, +} + +impl ActorInterface for DemonitorActor { + type Incoming = Ping; + type Response = (); + fn on_start(&mut self, ctx: &Ctx) { + self.mref = Some(ctx.monitor(self.watch_target)); + } + fn handle(&mut self, ctx: &Ctx, _msg: Ping) { + // Cancel the monitor + if let Some(mref) = self.mref.take() { + ctx.demonitor(mref); + } + } +} + +/// Given actor A monitors actor B then demonitors, +/// when B dies, +/// then A does NOT receive a Down notification. +#[test] +fn demonitor_cancels_notification() { + let rt = Runtime::new(RuntimeConfig::default()); + let down_inbox = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PingPongActor).unwrap(); + let watcher = rt.spawn(DemonitorActor { + watch_target: target, + mref: None, + }).unwrap(); + + rt.tick(); // on_start → monitor set up + + // Trigger demonitor + rt.send_to(watcher, Ping { reply_to: ActorAddress::default() }).unwrap(); + rt.tick(); // handle → demonitor + + // Now kill the target + rt.stop_actor(target).unwrap(); + rt.tick(); // cleanup — no Down should be emitted + rt.tick(); // extra tick to be sure + + assert!(down_inbox.try_recv().is_none(), "demonitored — should NOT receive Down"); +} + +/// Given actor A monitors B, and A dies before B, +/// when B dies, +/// then no Down is delivered (dead watcher cleaned up). +#[test] +fn dead_watcher_does_not_receive_down() { + let rt = Runtime::new(RuntimeConfig::default()); + + let target = rt.spawn(PingPongActor).unwrap(); + let watcher = rt.spawn(WatcherActor { + watch_target: target, + reply_to: ActorAddress::default(), // won't matter, watcher dies first + mref: None, + }).unwrap(); + + rt.tick(); // on_start → monitor set up + rt.stop_actor(watcher).unwrap(); + rt.tick(); // watcher dies → its monitors are cleaned up + + // Now kill the target — the dead watcher's subscription should be gone + rt.stop_actor(target).unwrap(); + rt.tick(); // cleanup — should not panic or try to deliver to dead watcher + // If we get here without panic, the test passes +} + +/// Given an external inbox monitors via the runtime, +/// when the target dies, +/// then the inbox receives a Down message. +#[test] +fn down_delivered_to_external_inbox() { + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PingPongActor).unwrap(); + + // Set up a monitor from an actor that forwards Down to the inbox. + // The watcher is an actor, but the final recipient is the inbox. + let _watcher = rt.spawn(WatcherActor { + watch_target: target, + reply_to: *inbox.addr(), + mref: None, + }).unwrap(); + + rt.tick(); // on_start + rt.stop_actor(target).unwrap(); + rt.tick(); // cleanup → Down to watcher + rt.tick(); // watcher forwards to inbox + + let down = inbox.try_recv().expect("inbox should receive forwarded Down"); + assert_eq!(down.addr, target); + assert_eq!(down.reason, StopReason::Normal); +} + +/// Given actor A monitors B with two independent monitors, +/// when B dies, +/// then A receives two Down messages (one per monitor). +#[test] +fn stacked_monitors_produce_multiple_notifications() { + /// An actor that creates two monitors on the same target. + struct DoubleWatcherActor { + target: ActorAddress, + reply_to: ActorAddress, + } + + impl ActorInterface for DoubleWatcherActor { + type Incoming = Down; + type Response = (); + fn on_start(&mut self, ctx: &Ctx) { + ctx.monitor(self.target); + ctx.monitor(self.target); + } + fn handle(&mut self, ctx: &Ctx, msg: Down) { + ctx.send(self.reply_to, msg).unwrap(); + } + } + + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + + let target = rt.spawn(PingPongActor).unwrap(); + rt.spawn(DoubleWatcherActor { + target, + reply_to: *inbox.addr(), + }).unwrap(); + + rt.tick(); // on_start → 2 monitors + rt.stop_actor(target).unwrap(); + rt.tick(); // cleanup → 2 Down messages to watcher + rt.tick(); // watcher forwards both to inbox + + assert!(inbox.try_recv().is_some(), "first Down"); + assert!(inbox.try_recv().is_some(), "second Down"); + assert!(inbox.try_recv().is_none(), "no more"); +}