diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md index 3630b10..287167f 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 16 COMPLETE +### Status: Cycle 17 COMPLETE ## Plan Overview 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ @@ -124,6 +124,40 @@ - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err - **Result**: 82 tests pass, all workspace compiles +### Cycle 17: Supervision Trees (handle_down + Supervisor Actor) +- **Research**: Cross-framework supervision analysis — Erlang (one_for_one/all/rest, child specs, intensity/period), + Akka (SupervisorStrategy, Resume/Restart/Stop/Escalate, BackoffSupervisor), Ractor (SupervisionEvent, + ractor-supervisor crate), Bastion (hierarchy, redundancy groups), CAF (no built-in supervisor, monitor-based) + - Key finding: swactor has all building blocks (monitor, spawn_restartable, lifecycle hooks, Down messages) + - Decision: Supervisor as a user-space actor built on existing primitives (like Ractor base crate) + - handle_down callback enables any actor to react to monitored deaths without making Down the Incoming type +- **Implementation**: Three features added to `src/actor.rs`: + 1. **`handle_down` callback on ActorInterface** — default no-op, called when monitored actor dies + and actor's Incoming type is NOT Down. Implemented via second downcast attempt in `handle_any`. + Fully backward-compatible: actors with `Incoming = Down` still receive via `handle()`. + 2. **`ctx.stop_actor(addr)`** — send graceful stop to another actor from handler context. + Uses StopSignal through normal message routing (PoisonPill semantics). + 3. **`Supervisor` actor** — manages child actors with configurable restart policies: + - `SupervisorStrategy::OneForOne` — only failed child is restarted + - `RestartPolicy::Permanent` — always restart + - `RestartPolicy::Transient` — restart only on Panicked, not Normal + - `RestartPolicy::Temporary` — never restart + - `ChildSpec` with id, restart policy, and factory closure `Fn(&Ctx) -> Result` + - Meltdown detection: stops itself when `total_restarts > max_restarts` + - Cascading shutdown: on_stop sends stop signals to all living children +- **Tests**: 10 new behavioral tests + - `handle_down_receives_death_notification` — handle_down callback fires on monitored death + - `handle_down_skipped_when_incoming_is_down` — backward compat: Incoming=Down uses handle() + - `ctx_stop_actor_stops_target` — one actor can stop another via ctx.stop_actor() + - `supervisor_restarts_permanent_child_on_panic` — panic → restart (OneForOne + Permanent) + - `supervisor_does_not_restart_transient_child_on_normal_stop` — Normal stop → no restart + - `supervisor_restarts_transient_child_on_panic` — Panicked → restart (Transient) + - `supervisor_never_restarts_temporary_child` — Temporary → never restart + - `supervisor_meltdown_after_max_restarts` — exceeding max_restarts stops supervisor + - `supervisor_one_for_one_only_restarts_failed_child` — multi-child, only crashed child affected + - `supervisor_on_stop_kills_children` — supervisor shutdown cascades to children +- **Result**: 138 tests pass (130 behavioral + 7 proptest + 1 doctest), zero warnings, full workspace compiles + ### Cycle 16: Benchmark New Features - **Scope**: Added benchmark group for features from Cycles 12-15 (named actors, groups, monitors, ask) - **New benchmarks** (5 total in `registry` group): diff --git a/docs/runtime.md b/docs/runtime.md index b554e09..d4f2a51 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -83,6 +83,7 @@ only way for actors to interact with the outside world. │ │ ctx.spawn_named(name, actor) -> Result │ │ │ │ ctx.spawn_restartable(a, f, max) -> Result │ │ │ │ ctx.stop_self() │ │ +│ │ ctx.stop_actor(addr) -> Result<(), Error> │ │ │ │ ctx.where_is(name) -> Option │ │ │ │ ctx.monitor(target) -> MonitorRef │ │ │ │ ctx.demonitor(mref) │ │ @@ -199,6 +200,57 @@ Factory-based restart after panic: // After max_restarts: permanently poisoned. ``` +## Supervision Trees + +The `Supervisor` actor manages child actors with configurable restart policies: + +``` + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, // only failed child restarted + 5, // max 5 restarts before meltdown + vec![ + ChildSpec::new("worker_a", RestartPolicy::Permanent, |ctx| { + ctx.spawn(MyWorker::new()) + }), + ChildSpec::new("worker_b", RestartPolicy::Transient, |ctx| { + ctx.spawn(MyOtherWorker::new()) + }), + ], + ); + let sup_addr = rt.spawn(sup)?; +``` + +Restart policies: +- `Permanent`: always restart +- `Transient`: restart only on panic, not normal stop +- `Temporary`: never restart + +Meltdown: supervisor stops itself when total restarts exceed `max_restarts`. +Cascading: supervisor stops all children in `on_stop`. + +### handle_down Callback + +Any actor can override `handle_down` to react to monitored actor deaths +without making `Down` its `Incoming` type: + +``` + impl ActorInterface for MyActor { + type Incoming = MyMsg; + // ... + fn handle_down(&mut self, ctx: &Ctx, down: Down) { + // React to monitored actor death + } + } +``` + +### ctx.stop_actor + +Actors can stop other actors from handlers: + +``` + ctx.stop_actor(other_addr)?; // PoisonPill semantics — queued after existing msgs +``` + ## Per-Worker Timers Deterministic tick-counting timers (not wall-clock): diff --git a/src/actor.rs b/src/actor.rs index e31ae26..b3f4857 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -24,6 +24,15 @@ pub trait ActorInterface: 'static + Send { /// NOT called when an actor is poisoned by panic — panicked actors may have /// corrupt state and calling methods on them is unsafe. fn on_stop(&mut self, _ctx: &Ctx) {} + + /// Called when a monitored actor dies (via [`Ctx::monitor`]). + /// + /// Override this to react to death notifications without making [`Down`] + /// your `Incoming` type. Default: no-op (the `Down` message is silently consumed). + /// + /// If your `Incoming` type IS `Down`, this method is never called — the + /// normal `handle()` receives the message instead. + fn handle_down(&mut self, _ctx: &Ctx, _down: Down) {} } /// A unique address for this actor. 32 bytes is overkill for a small application, @@ -106,11 +115,19 @@ where A: ActorInterface, { fn handle_any(&mut self, ctx: &Ctx, msg: Box) -> Option<&'static str> { - if let Ok(typed) = msg.downcast::() { - self.inner.handle(ctx, *typed); - Some(std::any::type_name::()) - } else { - None + let msg = match msg.downcast::() { + Ok(typed) => { + self.inner.handle(ctx, *typed); + return Some(std::any::type_name::()); + } + Err(msg) => msg, + }; + match msg.downcast::() { + Ok(down) => { + self.inner.handle_down(ctx, *down); + Some("swactor::actor::Down") + } + Err(_) => None, } } @@ -265,6 +282,15 @@ impl<'a> Ctx<'a> { self.inner.request_stop(self.self_addr); } + /// Send a graceful stop request to another actor. + /// + /// The target actor will process any messages already in its mailbox before + /// the stop signal, then its `on_stop()` hook is called and it is removed. + /// Uses PoisonPill semantics — queued after existing messages. + pub fn stop_actor(&self, addr: ActorAddress) -> Result<(), Error> { + self.inner.send_any(addr, Box::new(StopSignal)) + } + /// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks. /// /// The message is delivered as a normal mailbox message during the fire tick, @@ -381,3 +407,190 @@ impl<'a> Ctx<'a> { Ok(addr) } } + +// ─── Supervision ──────────────────────────────────────────────────────────── + +/// How a child should be restarted when it dies. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RestartPolicy { + /// Always restart, regardless of stop reason. + Permanent, + /// Restart only on abnormal exit (Panicked). Normal stops are final. + Transient, + /// Never restart. The child is removed on any exit. + Temporary, +} + +/// Strategy for handling child failures. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SupervisorStrategy { + /// Only restart the failed child. Other children are unaffected. + OneForOne, +} + +/// Specification for a supervised child actor. +/// +/// The `start` closure is called with `&Ctx` and should spawn the child actor +/// (typically via `ctx.spawn()`). The supervisor monitors the returned address +/// and applies the restart policy when the child dies. +/// +/// Children should be spawned with `ctx.spawn()`, not `ctx.spawn_restartable()`, +/// since the supervisor itself manages restarts. +pub struct ChildSpec { + /// Unique identifier for this child. + pub id: String, + /// How to restart this child. + pub restart: RestartPolicy, + /// Factory to spawn the child. Called with `&Ctx`, returns the child's address. + pub start: Arc Result + Send + Sync>, +} + +impl ChildSpec { + pub fn new( + id: impl Into, + restart: RestartPolicy, + start: impl Fn(&Ctx) -> Result + Send + Sync + 'static, + ) -> Self { + Self { + id: id.into(), + restart, + start: Arc::new(start), + } + } +} + +/// Tracked state for an active child within a supervisor. +struct ActiveChild { + addr: ActorAddress, + _monitor_ref: MonitorRef, +} + +/// A supervisor actor that manages child actors according to a restart strategy. +/// +/// Children are spawned during `on_start`. When a child dies, the supervisor +/// receives a [`Down`] notification via [`ActorInterface::handle_down`] and +/// applies the configured strategy and restart policy. +/// +/// # Restart Intensity +/// +/// The supervisor tracks total restarts. When `total_restarts > max_restarts`, +/// the supervisor stops itself (meltdown protection), escalating the failure +/// to its own supervisor if one exists. +/// +/// # Example +/// +/// ```ignore +/// let sup = Supervisor::new( +/// SupervisorStrategy::OneForOne, +/// 5, // max 5 restarts before meltdown +/// vec![ +/// ChildSpec::new("worker", RestartPolicy::Permanent, |ctx| { +/// ctx.spawn(MyWorker::new()) +/// }), +/// ], +/// ); +/// let sup_addr = rt.spawn(sup)?; +/// ``` +pub struct Supervisor { + strategy: SupervisorStrategy, + max_restarts: u32, + specs: Vec, + children: Vec>, + total_restarts: u32, +} + +impl Supervisor { + pub fn new( + strategy: SupervisorStrategy, + max_restarts: u32, + specs: Vec, + ) -> Self { + let children = (0..specs.len()).map(|_| None).collect(); + Self { + strategy, + max_restarts, + specs, + children, + total_restarts: 0, + } + } + + fn start_child(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> { + let addr = (self.specs[idx].start)(ctx)?; + let mref = ctx.monitor(addr); + self.children[idx] = Some(ActiveChild { + addr, + _monitor_ref: mref, + }); + Ok(()) + } + + fn find_child_idx(&self, addr: ActorAddress) -> Option { + self.children + .iter() + .position(|c| c.as_ref().map_or(false, |ac| ac.addr == addr)) + } +} + +impl ActorInterface for Supervisor { + type Incoming = (); + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: ()) {} + + fn on_start(&mut self, ctx: &Ctx) { + for idx in 0..self.specs.len() { + if let Err(e) = self.start_child(ctx, idx) { + eprintln!( + "swactor: supervisor failed to start child '{}': {}", + self.specs[idx].id, e + ); + } + } + } + + fn on_stop(&mut self, ctx: &Ctx) { + // Stop all living children on supervisor shutdown. + for child in self.children.iter().flatten() { + let _ = ctx.stop_actor(child.addr); + } + } + + fn handle_down(&mut self, ctx: &Ctx, down: Down) { + let Some(idx) = self.find_child_idx(down.addr) else { + return; + }; + self.children[idx] = None; + + let should_restart = match self.specs[idx].restart { + RestartPolicy::Permanent => true, + RestartPolicy::Transient => down.reason == StopReason::Panicked, + RestartPolicy::Temporary => false, + }; + + if !should_restart { + return; + } + + self.total_restarts += 1; + if self.total_restarts > self.max_restarts { + eprintln!( + "swactor: supervisor reached max restarts ({}), shutting down", + self.max_restarts + ); + ctx.stop_self(); + return; + } + + match self.strategy { + SupervisorStrategy::OneForOne => { + if let Err(e) = self.start_child(ctx, idx) { + eprintln!( + "swactor: supervisor failed to restart child '{}': {}", + self.specs[idx].id, e + ); + } + } + } + } +} diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index dc1fba5..07dd468 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -1,7 +1,10 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; -use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason}; +use swactor::actor::{ + ActorAddress, ActorInterface, ChildSpec, Down, MonitorRef, RestartPolicy, StopReason, + Supervisor, SupervisorStrategy, +}; use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig}; // ── Messages ──────────────────────────────────────────────────────────────── @@ -3637,3 +3640,448 @@ fn ask_reply_addr_is_accessible() { // The address should be valid (non-zero) assert_ne!(addr, ActorAddress::default()); } + +// ─── Supervisor Tests ────────────────────────────────────────────────────── + +/// Actor that panics after receiving a configurable number of messages. +struct PanicAfterN { + trigger: usize, + count: usize, + counter: Arc, +} + +impl ActorInterface for PanicAfterN { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: Ping) { + self.count += 1; + self.counter.fetch_add(1, Ordering::SeqCst); + let _ = ctx.send(msg.reply_to, Pong); + if self.count >= self.trigger { + panic!("intentional panic at message {}", self.count); + } + } +} + +// --- handle_down tests --- + +/// Given an actor with handle_down and a monitored target, +/// when the target dies, the watcher receives a Down via handle_down. +#[test] +fn handle_down_receives_death_notification() { + struct MonitoringTracker { + target: ActorAddress, + downs: Vec, + inbox: ActorAddress, + } + impl ActorInterface for MonitoringTracker { + type Incoming = Ping; + type Response = (); + fn on_start(&mut self, ctx: &Ctx) { + ctx.monitor(self.target); + } + fn handle(&mut self, ctx: &Ctx, _msg: Ping) { + let _ = ctx.send(self.inbox, Count(self.downs.len())); + } + fn handle_down(&mut self, _ctx: &Ctx, down: Down) { + self.downs.push(down); + } + } + + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + let target = rt.spawn(PanicActor).unwrap(); + let tracker = rt.spawn(MonitoringTracker { + target, + downs: vec![], + inbox: inbox_addr, + }).unwrap(); + rt.tick(); // on_start for both + + // Kill the target + rt.send_to(target, PanicMsg).unwrap(); + rt.tick(); // target panics + rt.tick(); // Down delivered to tracker via handle_down + + // Ask tracker how many downs it saw + rt.send_to(tracker, Ping { reply_to: inbox_addr }).unwrap(); + rt.tick(); + assert_eq!(inbox.try_recv(), Some(Count(1))); +} + +/// Given an actor whose Incoming type IS Down, handle_down is NOT called — +/// the Down goes through the normal handle() method (backward compatibility). +#[test] +fn handle_down_skipped_when_incoming_is_down() { + struct DownAsIncoming { + target: ActorAddress, + inbox: ActorAddress, + } + impl ActorInterface for DownAsIncoming { + type Incoming = Down; + type Response = (); + fn on_start(&mut self, ctx: &Ctx) { + ctx.monitor(self.target); + } + fn handle(&mut self, ctx: &Ctx, msg: Down) { + let _ = ctx.send(self.inbox, msg); + } + fn handle_down(&mut self, _ctx: &Ctx, _down: Down) { + panic!("handle_down must not be called when Incoming=Down"); + } + } + + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + let target = rt.spawn(PanicActor).unwrap(); + let _watcher = rt.spawn(DownAsIncoming { target, inbox: inbox_addr }).unwrap(); + rt.tick(); // on_start + + rt.send_to(target, PanicMsg).unwrap(); + rt.tick(); // panic + rt.tick(); // Down delivered through handle(), not handle_down + + let received = inbox.try_recv().expect("Down should be delivered via handle()"); + assert_eq!(received.reason, StopReason::Panicked); +} + +// --- ctx.stop_actor tests --- + +/// Given two actors, one can stop the other via ctx.stop_actor(). +#[test] +fn ctx_stop_actor_stops_target() { + #[derive(Clone)] + struct StopCmd { + target: ActorAddress, + } + struct Stopper; + impl ActorInterface for Stopper { + type Incoming = StopCmd; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: StopCmd) { + let _ = ctx.stop_actor(msg.target); + } + } + + let rt = Runtime::new(RuntimeConfig::default()); + let target = rt.spawn(PingPongActor).unwrap(); + let stopper = rt.spawn(Stopper).unwrap(); + rt.tick(); // on_start + + rt.send_to(stopper, StopCmd { target }).unwrap(); + rt.tick(); // stopper handles StopCmd → stop_actor(target) + rt.tick(); // StopSignal delivered to target, target stops + rt.tick(); // cleanup + + assert!(rt.send_to(target, Ping { reply_to: ActorAddress::default() }).is_err()); + // Stopper should still be alive + assert!(rt.send_to(stopper, StopCmd { target }).is_ok()); +} + +// --- Supervisor tests --- + +/// Given a supervisor with one permanent child, +/// when the child panics, the supervisor restarts it. +#[test] +fn supervisor_restarts_permanent_child_on_panic() { + let counter = Arc::new(AtomicUsize::new(0)); + let counter_c = counter.clone(); + let inbox_holder: Arc>> = + Arc::new(std::sync::Mutex::new(None)); + + let rt = Runtime::new(RuntimeConfig::default()); + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + *inbox_holder.lock().unwrap() = Some(inbox_addr); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ChildSpec::new("worker", RestartPolicy::Permanent, move |ctx| { + ctx.spawn(PanicAfterN { + trigger: 2, // panics on 2nd message + count: 0, + counter: counter_c.clone(), + }) + })], + ); + let _sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); // supervisor on_start → spawns child + rt.tick(); // child on_start + + // Find the child by checking stats + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 2); // supervisor + child + + // Send message to child — need to discover child address. + // We'll use the address map from stats. + let child_addr = stats.actors.iter() + .find(|(addr, _)| *addr != _sup_addr) + .map(|(addr, _)| *addr) + .unwrap(); + + // First message: child processes, increments counter + rt.send_to(child_addr, Ping { reply_to: inbox_addr }).unwrap(); + rt.tick(); + assert_eq!(counter.load(Ordering::SeqCst), 1); + + // Second message: child panics (trigger=2) + rt.send_to(child_addr, Ping { reply_to: inbox_addr }).unwrap(); + rt.tick(); // child panics and is poisoned + rt.tick(); // cleanup: Down delivered to supervisor via handle_down + rt.tick(); // supervisor restarts child (spawns new one) + rt.tick(); // new child on_start + + // Supervisor is still alive, and a new child exists + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 2); // supervisor + new child +} + +/// Given a supervisor with a transient child, +/// when the child stops normally, it is NOT restarted. +#[test] +fn supervisor_does_not_restart_transient_child_on_normal_stop() { + let rt = Runtime::new(RuntimeConfig::default()); + + struct StopsAfterFirst; + impl ActorInterface for StopsAfterFirst { + type Incoming = Ping; + type Response = (); + fn handle(&mut self, ctx: &Ctx, _msg: Ping) { + ctx.stop_self(); + } + } + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ChildSpec::new("worker", RestartPolicy::Transient, |ctx| { + ctx.spawn(StopsAfterFirst) + })], + ); + let sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); // supervisor on_start → child spawned + rt.tick(); // child on_start + + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 2); // sup + child + + // Find child address + let child_addr = stats.actors.iter() + .find(|(addr, _)| *addr != sup_addr) + .map(|(addr, _)| *addr) + .unwrap(); + + // Send message — child stops itself + rt.send_to(child_addr, Ping { reply_to: ActorAddress::default() }).unwrap(); + rt.tick(); // child handles, stops self + rt.tick(); // cleanup: Down(Normal) delivered to supervisor + rt.tick(); // supervisor sees Transient + Normal → no restart + + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 1); // only supervisor remains +} + +/// Given a supervisor with a transient child, +/// when the child panics, it IS restarted. +#[test] +fn supervisor_restarts_transient_child_on_panic() { + let rt = Runtime::new(RuntimeConfig::default()); + let counter = Arc::new(AtomicUsize::new(0)); + let counter_c = counter.clone(); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ChildSpec::new("worker", RestartPolicy::Transient, move |ctx| { + ctx.spawn(PanicAfterN { + trigger: 1, // panics on first message + count: 0, + counter: counter_c.clone(), + }) + })], + ); + let sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); // supervisor starts, spawns child + rt.tick(); // child on_start + + let child_addr = rt.stats().actors.iter() + .find(|(addr, _)| *addr != sup_addr) + .map(|(addr, _)| *addr) + .unwrap(); + + // Send message — child panics + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(child_addr, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); // child panics + rt.tick(); // Down(Panicked) → supervisor restarts + rt.tick(); // new child spawned + rt.tick(); // new child on_start + + // Supervisor + new child alive + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 2); +} + +/// Given a supervisor with a temporary child, +/// when the child dies (any reason), it is never restarted. +#[test] +fn supervisor_never_restarts_temporary_child() { + let rt = Runtime::new(RuntimeConfig::default()); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ChildSpec::new("worker", RestartPolicy::Temporary, |ctx| { + ctx.spawn(PanicActor) + })], + ); + let sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); // supervisor starts, spawns child + rt.tick(); // child on_start + + let child_addr = rt.stats().actors.iter() + .find(|(addr, _)| *addr != sup_addr) + .map(|(addr, _)| *addr) + .unwrap(); + + // Kill the child + rt.send_to(child_addr, PanicMsg).unwrap(); + rt.tick(); // panic + rt.tick(); // Down → supervisor sees Temporary → no restart + rt.tick(); // settle + + let stats = rt.stats(); + assert_eq!(stats.workers[0].num_actors, 1); // only supervisor +} + +/// Given a supervisor with max_restarts=2, +/// when more than 2 restarts occur, the supervisor stops itself (meltdown). +#[test] +fn supervisor_meltdown_after_max_restarts() { + let rt = Runtime::new(RuntimeConfig::default()); + let counter = Arc::new(AtomicUsize::new(0)); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 2, // only 2 restarts allowed + vec![ChildSpec::new("crasher", RestartPolicy::Permanent, { + let counter = counter.clone(); + move |ctx| { + ctx.spawn(PanicAfterN { + trigger: 1, + count: 0, + counter: counter.clone(), + }) + } + })], + ); + let sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); rt.tick(); // supervisor + child started + + // Crash the child 3 times (1 initial + 2 restarts = max, 3rd restart triggers meltdown) + for _ in 0..3 { + // Find current child + if let Some((child_addr, _)) = rt.stats().actors.iter() + .find(|(addr, _)| *addr != sup_addr) + { + let inbox = rt.new_inbox::().unwrap(); + let _ = rt.send_to(*child_addr, Ping { reply_to: *inbox.addr() }); + rt.tick(); // child panics + rt.tick(); // Down delivered → restart or meltdown + rt.tick(); // new child spawned (or supervisor stopped) + rt.tick(); // settle + } + } + + // After 3 crashes with max_restarts=2, supervisor should have stopped itself + let stats = rt.stats(); + let sup_alive = stats.actors.iter().any(|(addr, _)| *addr == sup_addr); + assert!(!sup_alive, "supervisor should have stopped after exceeding max_restarts"); +} + +/// Given a supervisor with multiple children, +/// when one child panics, only that child is restarted (OneForOne). +#[test] +fn supervisor_one_for_one_only_restarts_failed_child() { + let rt = Runtime::new(RuntimeConfig::default()); + let counter_a = Arc::new(AtomicUsize::new(0)); + let counter_b = Arc::new(AtomicUsize::new(0)); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ + ChildSpec::new("crasher", RestartPolicy::Permanent, { + let c = counter_a.clone(); + move |ctx| ctx.spawn_named("child_a", PanicAfterN { + trigger: 1, count: 0, counter: c.clone(), + }) + }), + ChildSpec::new("stable", RestartPolicy::Permanent, { + let c = counter_b.clone(); + move |ctx| ctx.spawn_named("child_b", CountingPingActor { counter: c.clone() }) + }), + ], + ); + let _sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); rt.tick(); // start up + + let child_a = rt.where_is("child_a").expect("child_a should be named"); + let child_b = rt.where_is("child_b").expect("child_b should be named"); + + // Send to child_b to prove it's alive + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(child_b, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); + let b_processed_before = counter_b.load(Ordering::SeqCst); + assert!(b_processed_before >= 1); + + // Crash child_a + rt.send_to(child_a, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); // child_a panics + rt.tick(); // Down → supervisor restarts child_a + rt.tick(); rt.tick(); // new child spawned + on_start + + // child_b should still be alive (same address, same name) + let child_b_after = rt.where_is("child_b").expect("child_b should still exist"); + assert_eq!(child_b, child_b_after, "child_b address should be unchanged"); + + rt.send_to(child_b, Ping { reply_to: *inbox.addr() }).unwrap(); + rt.tick(); + assert!(counter_b.load(Ordering::SeqCst) > b_processed_before, + "child_b should still be processing messages"); + + // Supervisor + 2 children should be alive + assert_eq!(rt.stats().workers[0].num_actors, 3); +} + +/// Given a supervisor that stops, its children also stop. +#[test] +fn supervisor_on_stop_kills_children() { + let rt = Runtime::new(RuntimeConfig::default()); + + let sup = Supervisor::new( + SupervisorStrategy::OneForOne, + 5, + vec![ + ChildSpec::new("a", RestartPolicy::Permanent, |ctx| ctx.spawn(PingPongActor)), + ChildSpec::new("b", RestartPolicy::Permanent, |ctx| ctx.spawn(PingPongActor)), + ], + ); + let sup_addr = rt.spawn(sup).unwrap(); + rt.tick(); rt.tick(); // start up + + assert_eq!(rt.stats().workers[0].num_actors, 3); // sup + 2 children + + // Stop the supervisor + rt.stop_actor(sup_addr).unwrap(); + rt.tick(); // StopSignal delivered to supervisor, on_stop sends stop to children + rt.tick(); // supervisor cleaned up, stop signals delivered to children + rt.tick(); // children stop + rt.tick(); // children cleaned up + + assert_eq!(rt.stats().workers[0].num_actors, 0); +}