feat: supervision trees with handle_down callback and Supervisor actor (Cycle 17)
Add three features to enable Erlang-style supervision: 1. ActorInterface::handle_down() — callback for monitored actor deaths, allowing actors to react to Down messages without making Down their Incoming type. Backward-compatible: actors with Incoming=Down still receive through handle(). 2. ctx.stop_actor(addr) — send graceful stop to another actor from handler context using PoisonPill semantics. 3. Supervisor actor — manages child actors with configurable restart policies (Permanent/Transient/Temporary), OneForOne strategy, and meltdown detection (max_restarts). Built entirely on existing primitives (monitor, spawn, Down, lifecycle hooks). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
8c68a59b45
commit
a70bd86db2
4 changed files with 754 additions and 7 deletions
|
|
@ -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<ActorAddress>`
|
||||
- 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):
|
||||
|
|
|
|||
|
|
@ -83,6 +83,7 @@ only way for actors to interact with the outside world.
|
|||
│ │ ctx.spawn_named(name, actor) -> Result<ActorAddress, Error> │ │
|
||||
│ │ ctx.spawn_restartable(a, f, max) -> Result<ActorAddress, Error> │ │
|
||||
│ │ ctx.stop_self() │ │
|
||||
│ │ ctx.stop_actor(addr) -> Result<(), Error> │ │
|
||||
│ │ ctx.where_is(name) -> Option<ActorAddress> │ │
|
||||
│ │ 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):
|
||||
|
|
|
|||
221
src/actor.rs
221
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<dyn Any + Send>) -> Option<&'static str> {
|
||||
if let Ok(typed) = msg.downcast::<A::Incoming>() {
|
||||
let msg = match msg.downcast::<A::Incoming>() {
|
||||
Ok(typed) => {
|
||||
self.inner.handle(ctx, *typed);
|
||||
Some(std::any::type_name::<A::Incoming>())
|
||||
} else {
|
||||
None
|
||||
return Some(std::any::type_name::<A::Incoming>());
|
||||
}
|
||||
Err(msg) => msg,
|
||||
};
|
||||
match msg.downcast::<Down>() {
|
||||
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<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
|
||||
}
|
||||
|
||||
impl ChildSpec {
|
||||
pub fn new(
|
||||
id: impl Into<String>,
|
||||
restart: RestartPolicy,
|
||||
start: impl Fn(&Ctx) -> Result<ActorAddress, Error> + 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<ChildSpec>,
|
||||
children: Vec<Option<ActiveChild>>,
|
||||
total_restarts: u32,
|
||||
}
|
||||
|
||||
impl Supervisor {
|
||||
pub fn new(
|
||||
strategy: SupervisorStrategy,
|
||||
max_restarts: u32,
|
||||
specs: Vec<ChildSpec>,
|
||||
) -> 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<usize> {
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<AtomicUsize>,
|
||||
}
|
||||
|
||||
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<Down>,
|
||||
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::<Count>().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::<Down>().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<std::sync::Mutex<Option<ActorAddress>>> =
|
||||
Arc::new(std::sync::Mutex::new(None));
|
||||
|
||||
let rt = Runtime::new(RuntimeConfig::default());
|
||||
let inbox = rt.new_inbox::<Pong>().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::<Pong>().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::<Pong>().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::<Pong>().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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue