zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: OneForAll and RestForOne supervisor strategies (Cycle 18)

Browse source 
Add coordinated restart strategies to the Supervisor actor. OneForAll
restarts all children when one fails; RestForOne restarts the failed
child and all children after it in spec order. Uses a SupervisorPhase
state machine to coordinate stop signals and Down confirmations.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Developer 2026-02-12 14:12:28 +00:00
parent a70bd86db2
commit 771c38c863
4 changed files with 294 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
CLAUDE/notes/progress.md
View file

@ -2,7 +2,7 @@
## Current Stage: Phase 1 — Research + First Improvement Cycle
### Status: Cycle 17 COMPLETE
### Status: Cycle 18 COMPLETE
## Plan Overview
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
@ -158,6 +158,26 @@
- `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 18: OneForAll + RestForOne Supervisor Strategies
- **Research**: Investigated SmallBox/InlineAny optimization (44% queue throughput improvement)
but deferred due to unsafe code risk and 32+ call-site changes violating structural constraints.
Chose to extend Supervisor with remaining Erlang-style strategies instead.
- **Implementation**: Extended `Supervisor` in `src/actor.rs` with coordinated restart strategies:
- `SupervisorStrategy::OneForAll` — all children restarted when one fails
- `SupervisorStrategy::RestForOne` — failed child + all children after it (in spec order) restarted
- `SupervisorPhase` state machine: `Normal` (steady state) | `Stopping { awaiting, restart_set }` (coordinated)
- During coordinated restart: supervisor stops living siblings, waits for all Down confirmations,
then restarts the full restart set in spec order
- `begin_coordinated_restart(ctx, indices)` — sends stop signals, transitions to Stopping phase
- `finish_restart(ctx)` — called when all awaiting Downs received, restarts from spec order
- `check_intensity()` factored out for restart budget checking
- Already-dead children are handled: if all targets are already dead, immediate restart (no Stopping phase)
- **Tests**: 3 new behavioral tests
- `supervisor_one_for_all_restarts_all_on_single_failure` — one child panics, all 3 get new addresses
- `supervisor_rest_for_one_restarts_rest_after_failed` — child_b panics, child_a unchanged, child_c restarted
- `supervisor_one_for_all_waits_for_all_downs_before_restart` — verifies coordinated shutdown completes before restart
- **Result**: 141 tests pass (133 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):

13
docs/runtime.md
View file

@ -206,7 +206,9 @@ The `Supervisor` actor manages child actors with configurable restart policies:
```
let sup = Supervisor::new(
SupervisorStrategy::OneForOne, // only failed child restarted
SupervisorStrategy::OneForOne, // only the failed child is restarted
// Also: OneForAll — all children restarted when one fails
// RestForOne — failed child + all children after it restarted
5, // max 5 restarts before meltdown
vec![
ChildSpec::new("worker_a", RestartPolicy::Permanent, |ctx| {
@ -225,6 +227,15 @@ Restart policies:
- `Transient`: restart only on panic, not normal stop
- `Temporary`: never restart
Strategies:
- `OneForOne`: only the failed child is restarted (default)
- `OneForAll`: all children are stopped and restarted when one fails
- `RestForOne`: the failed child and all children after it (in spec order) are restarted
Coordinated restart (OneForAll/RestForOne): the supervisor enters a `Stopping` phase,
sends stop signals to affected siblings, waits for all `Down` confirmations, then
restarts the full set in spec order. Already-dead children are handled immediately.
Meltdown: supervisor stops itself when total restarts exceed `max_restarts`.
Cascading: supervisor stops all children in `on_stop`.

119
src/actor.rs
View file

@ -426,6 +426,11 @@ pub enum RestartPolicy {
pub enum SupervisorStrategy {
/// Only restart the failed child. Other children are unaffected.
OneForOne,
/// Terminate all children and restart them all in spec order.
OneForAll,
/// Terminate children started after the failed child, then restart
/// the failed child and all terminated children in spec order.
RestForOne,
}
/// Specification for a supervised child actor.
@ -465,12 +470,36 @@ struct ActiveChild {
_monitor_ref: MonitorRef,
}
/// Internal phase for coordinating multi-child restart (OneForAll, RestForOne).
///
/// In `Normal` phase, the supervisor processes Down messages and applies the strategy.
/// When a coordinated restart is needed, it transitions to `Stopping` (sends stop
/// signals, waits for Down confirmations) then restarts all affected children.
enum SupervisorPhase {
/// Normal operation — process Down messages and apply strategy.
Normal,
/// Waiting for children to confirm death before restarting.
Stopping {
/// Children we're still waiting for Down confirmation.
awaiting: Vec<ActorAddress>,
/// Spec indices to restart once all confirmations received.
restart_set: Vec<usize>,
},
}
/// 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.
///
/// # Strategies
///
/// - **OneForOne**: Only the failed child is restarted.
/// - **OneForAll**: All children are stopped, then all restarted in spec order.
/// - **RestForOne**: The failed child and all children started after it are
/// stopped, then restarted in spec order.
///
/// # Restart Intensity
///
/// The supervisor tracks total restarts. When `total_restarts > max_restarts`,
@ -497,6 +526,7 @@ pub struct Supervisor {
specs: Vec<ChildSpec>,
children: Vec<Option<ActiveChild>>,
total_restarts: u32,
phase: SupervisorPhase,
}
impl Supervisor {
@ -512,6 +542,7 @@ impl Supervisor {
specs,
children,
total_restarts: 0,
phase: SupervisorPhase::Normal,
}
}
@ -530,6 +561,61 @@ impl Supervisor {
.iter()
.position(|c| c.as_ref().map_or(false, |ac| ac.addr == addr))
}
/// Check meltdown intensity — returns true if we should stop.
fn check_intensity(&mut self) -> bool {
self.total_restarts += 1;
self.total_restarts > self.max_restarts
}
/// Try to finish the coordinated restart: restart all children in `restart_set`.
fn finish_restart(&mut self, ctx: &Ctx) {
let restart_set = match &mut self.phase {
SupervisorPhase::Stopping { restart_set, .. } => {
std::mem::take(restart_set)
}
_ => return,
};
self.phase = SupervisorPhase::Normal;
for idx in restart_set {
if let Err(e) = self.start_child(ctx, idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[idx].id, e
);
}
}
}
/// Begin a coordinated restart for the given spec indices.
/// Stops any living children in the set, then waits for their Down messages.
fn begin_coordinated_restart(&mut self, ctx: &Ctx, restart_indices: Vec<usize>) {
let mut awaiting = Vec::new();
for &idx in &restart_indices {
if let Some(child) = self.children[idx].take() {
let _ = ctx.stop_actor(child.addr);
awaiting.push(child.addr);
}
}
if awaiting.is_empty() {
// All children already dead — restart immediately.
for idx in &restart_indices {
if let Err(e) = self.start_child(ctx, *idx) {
eprintln!(
"swactor: supervisor failed to restart child '{}': {}",
self.specs[*idx].id, e
);
}
}
} else {
self.phase = SupervisorPhase::Stopping {
awaiting,
restart_set: restart_indices,
};
}
}
}
impl ActorInterface for Supervisor {
@ -550,13 +636,31 @@ impl ActorInterface for Supervisor {
}
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) {
// During coordinated restart: track Down confirmations.
if matches!(self.phase, SupervisorPhase::Stopping { .. }) {
// Clear from children tracking
if let Some(idx) = self.find_child_idx(down.addr) {
self.children[idx] = None;
}
// Remove from awaiting list
if let SupervisorPhase::Stopping { awaiting, .. } = &mut self.phase {
awaiting.retain(|a| *a != down.addr);
}
let done = matches!(&self.phase,
SupervisorPhase::Stopping { awaiting, .. } if awaiting.is_empty());
if done {
self.finish_restart(ctx);
}
return;
}
// Normal phase: handle child death.
let Some(idx) = self.find_child_idx(down.addr) else {
return;
};
@ -572,8 +676,7 @@ impl ActorInterface for Supervisor {
return;
}
self.total_restarts += 1;
if self.total_restarts > self.max_restarts {
if self.check_intensity() {
eprintln!(
"swactor: supervisor reached max restarts ({}), shutting down",
self.max_restarts
@ -591,6 +694,16 @@ impl ActorInterface for Supervisor {
);
}
}
SupervisorStrategy::OneForAll => {
// Stop all other living children, then restart all in order.
let restart_indices: Vec<usize> = (0..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
SupervisorStrategy::RestForOne => {
// Stop children after the failed one, then restart failed + rest.
let restart_indices: Vec<usize> = (idx..self.specs.len()).collect();
self.begin_coordinated_restart(ctx, restart_indices);
}
}
}
}

145
tests/runtime_api.rs
View file

@ -4058,6 +4058,151 @@ fn supervisor_one_for_one_only_restarts_failed_child() {
assert_eq!(rt.stats().workers[0].num_actors, 3);
}
/// Given a OneForAll supervisor with 3 children,
/// when one child panics, ALL children are stopped and restarted in spec order.
#[test]
fn supervisor_one_for_all_restarts_all_on_single_failure() {
let rt = Runtime::new(RuntimeConfig::default());
let counter_a = Arc::new(AtomicUsize::new(0));
let counter_b = Arc::new(AtomicUsize::new(0));
let counter_c = Arc::new(AtomicUsize::new(0));
let sup = Supervisor::new(
SupervisorStrategy::OneForAll,
5,
vec![
ChildSpec::new("a", RestartPolicy::Permanent, {
let c = counter_a.clone();
move |ctx| ctx.spawn_named("ofa_a", PanicAfterN {
trigger: 1, count: 0, counter: c.clone(),
})
}),
ChildSpec::new("b", RestartPolicy::Permanent, {
let c = counter_b.clone();
move |ctx| ctx.spawn_named("ofa_b", CountingPingActor { counter: c.clone() })
}),
ChildSpec::new("c", RestartPolicy::Permanent, {
let c = counter_c.clone();
move |ctx| ctx.spawn_named("ofa_c", CountingPingActor { counter: c.clone() })
}),
],
);
let _sup_addr = rt.spawn(sup).unwrap();
rt.tick(); rt.tick(); // startup
let old_b = rt.where_is("ofa_b").expect("ofa_b exists");
let old_c = rt.where_is("ofa_c").expect("ofa_c exists");
let child_a = rt.where_is("ofa_a").expect("ofa_a exists");
// Crash child_a
let inbox = rt.new_inbox::<Pong>().unwrap();
rt.send_to(child_a, Ping { reply_to: *inbox.addr() }).unwrap();
rt.tick(); // child_a panics
// supervisor receives Down(a) → OneForAll → stops b and c
for _ in 0..8 { rt.tick(); } // wait for stops, Downs, restarts, on_starts
// All 3 children should be alive with NEW addresses (old ones are dead)
let stats = rt.stats();
assert_eq!(stats.workers[0].num_actors, 4); // sup + 3 new children
// The old addresses for b and c should be gone (they were stopped and re-created)
// New names should be re-registered
let new_b = rt.where_is("ofa_b").expect("ofa_b re-registered after restart");
let new_c = rt.where_is("ofa_c").expect("ofa_c re-registered after restart");
assert_ne!(old_b, new_b, "child_b should have a new address after restart");
assert_ne!(old_c, new_c, "child_c should have a new address after restart");
}
/// Given a RestForOne supervisor with children [a, b, c],
/// when child b panics, children b and c are restarted (children after b in spec order).
/// Child a is unaffected.
#[test]
fn supervisor_rest_for_one_restarts_rest_after_failed() {
let rt = Runtime::new(RuntimeConfig::default());
let counter_a = Arc::new(AtomicUsize::new(0));
let counter_b = Arc::new(AtomicUsize::new(0));
let counter_c = Arc::new(AtomicUsize::new(0));
let sup = Supervisor::new(
SupervisorStrategy::RestForOne,
5,
vec![
ChildSpec::new("a", RestartPolicy::Permanent, {
let c = counter_a.clone();
move |ctx| ctx.spawn_named("rfo_a", CountingPingActor { counter: c.clone() })
}),
ChildSpec::new("b", RestartPolicy::Permanent, {
let c = counter_b.clone();
move |ctx| ctx.spawn_named("rfo_b", PanicAfterN {
trigger: 1, count: 0, counter: c.clone(),
})
}),
ChildSpec::new("c", RestartPolicy::Permanent, {
let c = counter_c.clone();
move |ctx| ctx.spawn_named("rfo_c", CountingPingActor { counter: c.clone() })
}),
],
);
let _sup_addr = rt.spawn(sup).unwrap();
rt.tick(); rt.tick(); // startup
let old_a = rt.where_is("rfo_a").expect("rfo_a exists");
let old_c = rt.where_is("rfo_c").expect("rfo_c exists");
let child_b = rt.where_is("rfo_b").expect("rfo_b exists");
// Crash child_b
let inbox = rt.new_inbox::<Pong>().unwrap();
rt.send_to(child_b, Ping { reply_to: *inbox.addr() }).unwrap();
rt.tick(); // child_b panics
// supervisor: Down(b) → RestForOne → stops c (rest after b), then restarts b+c
for _ in 0..8 { rt.tick(); }
// All 3 children should be alive
let stats = rt.stats();
assert_eq!(stats.workers[0].num_actors, 4); // sup + 3 children
// child_a should be UNCHANGED (not affected by RestForOne)
let new_a = rt.where_is("rfo_a").expect("rfo_a still exists");
assert_eq!(old_a, new_a, "child_a should not be restarted in RestForOne when b fails");
// child_c should have a NEW address (it was stopped and re-created)
let new_c = rt.where_is("rfo_c").expect("rfo_c re-registered");
assert_ne!(old_c, new_c, "child_c should have a new address after RestForOne restart");
}
/// Given a OneForAll supervisor, when the last child of the failed set confirms death,
/// all children are restarted in spec order (not reverse).
#[test]
fn supervisor_one_for_all_waits_for_all_downs_before_restart() {
let rt = Runtime::new(RuntimeConfig::default());
let sup = Supervisor::new(
SupervisorStrategy::OneForAll,
5,
vec![
ChildSpec::new("x", RestartPolicy::Permanent, |ctx| ctx.spawn(PingPongActor)),
ChildSpec::new("y", RestartPolicy::Permanent, |ctx| ctx.spawn(PingPongActor)),
],
);
let sup_addr = rt.spawn(sup).unwrap();
rt.tick(); rt.tick(); // startup
assert_eq!(rt.stats().workers[0].num_actors, 3); // sup + 2 children
// Stop one child (graceful stop triggers OneForAll)
let actors: Vec<_> = rt.stats().actors.iter()
.filter(|(addr, _)| *addr != sup_addr)
.map(|(addr, _)| *addr)
.collect();
rt.stop_actor(actors[0]).unwrap();
// Tick enough times for full cycle: stop → Down → supervisor stops other → Down → restart all
for _ in 0..10 { rt.tick(); }
// Should have supervisor + 2 new children
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() {