use std::time::Instant;
use super::{BackoffDelay, ChildIndex, SupervisionDecision, SupervisionStrategy};
use crate::actor::{RestartLimitExceeded, RestartLimiter};
use crate::message::ChildTerminated;
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RestartPlan {
pub stop: Vec<ChildIndex>,
pub restart: Vec<ChildIndex>,
}
impl RestartPlan {
pub const fn is_empty(&self) -> bool {
self.stop.is_empty() && self.restart.is_empty()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SlotView {
pub index: ChildIndex,
pub restartable: bool,
pub alive: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExpectedTermination {
Shutdown,
GroupStop {
then_restart: bool,
},
Stale,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SlotSnapshot {
pub index: ChildIndex,
pub restartable: bool,
pub expected: Option<ExpectedTermination>,
pub last_restart: Option<Instant>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SupervisionOutcome {
Ignore,
GroupStopLanded {
then_restart: bool,
},
Forget,
Restart {
plan: RestartPlan,
backoff: BackoffDelay,
},
Escalate(RestartLimitExceeded),
}
pub fn plan_restart(
decision: &SupervisionDecision,
failed: ChildIndex,
slots: &[SlotView],
) -> RestartPlan {
let from = match decision {
SupervisionDecision::RestartChild => {
let restart = slots
.iter()
.filter(|slot| slot.index == failed && slot.restartable)
.map(|slot| slot.index)
.collect();
return RestartPlan {
stop: Vec::new(),
restart,
};
}
SupervisionDecision::RestartAll => ChildIndex::new(0),
SupervisionDecision::RestartFrom(index) => ChildIndex::new(*index),
SupervisionDecision::NoRestart | SupervisionDecision::Escalate => {
return RestartPlan::default()
}
};
let mut affected: Vec<&SlotView> = slots.iter().filter(|slot| slot.index >= from).collect();
affected.sort_unstable_by_key(|slot| slot.index);
let restart = affected
.iter()
.filter(|slot| slot.restartable)
.map(|slot| slot.index)
.collect();
let stop = affected
.iter()
.rev()
.filter(|slot| slot.alive && slot.index != failed)
.map(|slot| slot.index)
.collect();
RestartPlan { stop, restart }
}
pub fn evaluate(
notification: &ChildTerminated,
slot: &SlotSnapshot,
strategy: SupervisionStrategy,
limiter: &mut RestartLimiter,
slots: &[SlotView],
now: Instant,
) -> SupervisionOutcome {
match slot.expected {
Some(ExpectedTermination::GroupStop { then_restart }) => {
return SupervisionOutcome::GroupStopLanded { then_restart }
}
Some(ExpectedTermination::Shutdown | ExpectedTermination::Stale) => {
return SupervisionOutcome::Ignore
}
None => {}
}
if !slot.restartable {
return SupervisionOutcome::Forget;
}
let decision = strategy.decide(notification, slot.index.get());
if matches!(decision, SupervisionDecision::NoRestart) {
return SupervisionOutcome::Forget;
}
let recovery_window = limiter.window();
if slot
.last_restart
.is_some_and(|last| now.saturating_duration_since(last) > recovery_window)
{
limiter.reset_consecutive();
}
if let Err(exceeded) = limiter.can_restart() {
return SupervisionOutcome::Escalate(exceeded);
}
let backoff = BackoffDelay::from(limiter.record_restart());
let plan = plan_restart(&decision, slot.index, slots);
if plan.is_empty() {
SupervisionOutcome::Forget
} else {
SupervisionOutcome::Restart { plan, backoff }
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use acton_ern::Ern;
use super::*;
use crate::actor::{RestartLimiterConfig, RestartPolicy, TerminationReason};
fn view(index: usize, restartable: bool, alive: bool) -> SlotView {
SlotView {
index: ChildIndex::new(index),
restartable,
alive,
}
}
fn four_healthy_slots() -> Vec<SlotView> {
(0..4).map(|index| view(index, true, true)).collect()
}
fn indices(raw: &[usize]) -> Vec<ChildIndex> {
raw.iter().copied().map(ChildIndex::new).collect()
}
fn notification(policy: RestartPolicy, reason: TerminationReason) -> ChildTerminated {
ChildTerminated::new(
Ern::with_root("child").expect("'child' is a valid Ern root"),
reason,
policy,
)
}
fn snapshot(index: usize) -> SlotSnapshot {
SlotSnapshot {
index: ChildIndex::new(index),
restartable: true,
expected: None,
last_restart: None,
}
}
const WINDOW_SECS: u64 = 60;
fn limiter(max_restarts: u32, initial_backoff_ms: u64, max_backoff_ms: u64) -> RestartLimiter {
RestartLimiter::new(RestartLimiterConfig {
enabled: true,
max_restarts,
window_secs: WINDOW_SECS,
initial_backoff_ms,
max_backoff_ms,
backoff_multiplier: 2.0,
})
}
#[test]
fn restart_child_restarts_only_the_failed_child_at_any_position() {
let slots = four_healthy_slots();
for position in 0..4 {
let plan = plan_restart(
&SupervisionDecision::RestartChild,
ChildIndex::new(position),
&slots,
);
assert_eq!(plan.restart, indices(&[position]));
assert!(
plan.stop.is_empty(),
"the failed child is already down, so nothing needs stopping"
);
}
}
#[test]
fn restart_child_plans_nothing_when_the_child_cannot_be_recreated() {
let slots = vec![view(0, true, true), view(1, false, true)];
let plan = plan_restart(
&SupervisionDecision::RestartChild,
ChildIndex::new(1),
&slots,
);
assert_eq!(plan, RestartPlan::default());
}
#[test]
fn restart_all_stops_in_reverse_order_and_restarts_in_start_order() {
let slots = four_healthy_slots();
let plan = plan_restart(&SupervisionDecision::RestartAll, ChildIndex::new(0), &slots);
assert_eq!(plan.restart, indices(&[0, 1, 2, 3]));
assert_eq!(
plan.stop,
indices(&[3, 2, 1]),
"slot 0 is the child that failed and is already down"
);
}
#[test]
fn the_failed_child_is_never_stopped_even_when_its_slot_still_reads_as_running() {
let slots = four_healthy_slots();
for failed in 0..4 {
let plan = plan_restart(
&SupervisionDecision::RestartAll,
ChildIndex::new(failed),
&slots,
);
assert!(
!plan.stop.contains(&ChildIndex::new(failed)),
"child {failed} terminated; stopping it would wait forever: {:?}",
plan.stop
);
assert!(
plan.restart.contains(&ChildIndex::new(failed)),
"and it is still the child that has to come back: {:?}",
plan.restart
);
}
}
#[test]
fn restart_all_skips_stopping_a_child_that_is_already_down() {
let mut slots = four_healthy_slots();
slots[2].alive = false;
let plan = plan_restart(&SupervisionDecision::RestartAll, ChildIndex::new(2), &slots);
assert_eq!(plan.stop, indices(&[3, 1, 0]), "slot 2 is already down");
assert_eq!(
plan.restart,
indices(&[0, 1, 2, 3]),
"slot 2 still needs starting"
);
}
#[test]
fn restart_all_stops_a_sibling_it_cannot_recreate() {
let mut slots = four_healthy_slots();
slots[1].restartable = false;
let plan = plan_restart(&SupervisionDecision::RestartAll, ChildIndex::new(0), &slots);
assert!(
plan.stop.contains(&ChildIndex::new(1)),
"an interdependent sibling must come down: {:?}",
plan.stop
);
assert!(
!plan.restart.contains(&ChildIndex::new(1)),
"it cannot be recreated: {:?}",
plan.restart
);
}
#[test]
fn rest_for_one_touches_only_the_failed_child_and_those_after_it() {
let mut slots = four_healthy_slots();
slots[2].alive = false;
let plan = plan_restart(
&SupervisionDecision::RestartFrom(2),
ChildIndex::new(2),
&slots,
);
assert_eq!(plan.stop, indices(&[3]), "slot 2 is already down");
assert_eq!(plan.restart, indices(&[2, 3]));
for untouched in [ChildIndex::new(0), ChildIndex::new(1)] {
assert!(!plan.stop.contains(&untouched));
assert!(!plan.restart.contains(&untouched));
}
}
#[test]
fn rest_for_one_from_the_first_child_matches_restart_all() {
let slots = four_healthy_slots();
let rest_for_one = plan_restart(
&SupervisionDecision::RestartFrom(0),
ChildIndex::new(0),
&slots,
);
let one_for_all =
plan_restart(&SupervisionDecision::RestartAll, ChildIndex::new(0), &slots);
assert_eq!(rest_for_one, one_for_all);
}
#[test]
fn rest_for_one_uses_the_strategys_index_over_the_failed_index() {
let slots = four_healthy_slots();
let plan = plan_restart(
&SupervisionDecision::RestartFrom(3),
ChildIndex::new(0),
&slots,
);
assert_eq!(plan.restart, indices(&[3]));
}
#[test]
fn no_restart_and_escalate_plan_nothing() {
let slots = four_healthy_slots();
for decision in [
SupervisionDecision::NoRestart,
SupervisionDecision::Escalate,
] {
let plan = plan_restart(&decision, ChildIndex::new(1), &slots);
assert_eq!(plan, RestartPlan::default(), "{decision}");
assert!(plan.is_empty());
}
}
#[test]
fn an_out_of_range_index_plans_nothing_instead_of_panicking() {
let slots = four_healthy_slots();
let by_failed = plan_restart(
&SupervisionDecision::RestartChild,
ChildIndex::new(9),
&slots,
);
assert_eq!(by_failed, RestartPlan::default());
let by_decision = plan_restart(
&SupervisionDecision::RestartFrom(9),
ChildIndex::new(9),
&slots,
);
assert_eq!(by_decision, RestartPlan::default());
}
#[test]
fn an_empty_child_list_plans_nothing_instead_of_panicking() {
for decision in [
SupervisionDecision::RestartChild,
SupervisionDecision::RestartAll,
SupervisionDecision::RestartFrom(0),
SupervisionDecision::NoRestart,
SupervisionDecision::Escalate,
] {
let plan = plan_restart(&decision, ChildIndex::new(0), &[]);
assert_eq!(plan, RestartPlan::default(), "{decision}");
}
}
#[test]
fn stop_always_descends_and_restart_always_ascends() {
let slots = vec![
view(3, true, true),
view(0, true, false),
view(2, false, true),
view(1, true, true),
];
let plan = plan_restart(&SupervisionDecision::RestartAll, ChildIndex::new(0), &slots);
assert!(
plan.stop.windows(2).all(|pair| pair[0] > pair[1]),
"stop must strictly descend: {:?}",
plan.stop
);
assert!(
plan.restart.windows(2).all(|pair| pair[0] < pair[1]),
"restart must strictly ascend: {:?}",
plan.restart
);
}
fn evaluate_expected(expected: Option<ExpectedTermination>) -> (SupervisionOutcome, usize) {
let mut limiter = limiter(5, 100, 10_000);
let slot = SlotSnapshot {
expected,
..snapshot(0)
};
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Normal),
&slot,
SupervisionStrategy::OneForAll,
&mut limiter,
&four_healthy_slots(),
Instant::now(),
);
(outcome, limiter.restarts_in_window())
}
#[test]
fn an_expected_stop_never_restarts_the_child_the_supervisor_just_stopped() {
for expected in [
ExpectedTermination::Shutdown,
ExpectedTermination::Stale,
ExpectedTermination::GroupStop { then_restart: true },
ExpectedTermination::GroupStop {
then_restart: false,
},
] {
let (outcome, charged) = evaluate_expected(Some(expected));
assert!(
!matches!(outcome, SupervisionOutcome::Restart { .. }),
"{expected:?} must not be read as a fresh failure, got {outcome:?}"
);
assert_eq!(
charged, 0,
"{expected:?} must not consume a restart allowance"
);
}
}
#[test]
fn a_shutdown_abandons_a_group_restart_rather_than_driving_it() {
let (outcome, _) = evaluate_expected(Some(ExpectedTermination::Shutdown));
assert_eq!(outcome, SupervisionOutcome::Ignore);
}
#[test]
fn a_stale_notice_is_ignored_rather_than_advancing_anything() {
let (outcome, _) = evaluate_expected(Some(ExpectedTermination::Stale));
assert_eq!(outcome, SupervisionOutcome::Ignore);
}
#[test]
fn a_group_stop_reports_that_it_landed_instead_of_being_ignored() {
for then_restart in [true, false] {
let (outcome, charged) =
evaluate_expected(Some(ExpectedTermination::GroupStop { then_restart }));
assert_eq!(
outcome,
SupervisionOutcome::GroupStopLanded { then_restart },
"a group stop must report which half of the group it is in"
);
assert_eq!(
charged, 0,
"the group was charged once when it was planned, not again per sibling"
);
}
}
#[test]
fn a_child_without_a_blueprint_is_forgotten_without_touching_the_limiter() {
let mut limiter = limiter(5, 100, 10_000);
let slot = SlotSnapshot {
restartable: false,
..snapshot(0)
};
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&slot,
SupervisionStrategy::OneForOne,
&mut limiter,
&four_healthy_slots(),
Instant::now(),
);
assert_eq!(outcome, SupervisionOutcome::Forget);
assert_eq!(limiter.restarts_in_window(), 0);
assert_eq!(limiter.consecutive_restarts(), 0);
}
#[test]
fn a_temporary_child_is_never_restarted() {
for reason in [
TerminationReason::Normal,
TerminationReason::Panic("x".into()),
TerminationReason::InboxClosed,
TerminationReason::ParentShutdown,
] {
let mut limiter = limiter(5, 100, 10_000);
let outcome = evaluate(
¬ification(RestartPolicy::Temporary, reason.clone()),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&four_healthy_slots(),
Instant::now(),
);
assert_eq!(outcome, SupervisionOutcome::Forget, "{reason}");
}
}
#[test]
fn a_transient_child_restarts_only_on_abnormal_termination() {
let mut normal = limiter(5, 100, 10_000);
let after_normal = evaluate(
¬ification(RestartPolicy::Transient, TerminationReason::Normal),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut normal,
&four_healthy_slots(),
Instant::now(),
);
assert_eq!(after_normal, SupervisionOutcome::Forget);
let mut panicked = limiter(5, 100, 10_000);
let after_panic = evaluate(
¬ification(RestartPolicy::Transient, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut panicked,
&four_healthy_slots(),
Instant::now(),
);
assert!(matches!(after_panic, SupervisionOutcome::Restart { .. }));
}
#[test]
fn a_parent_shutdown_never_restarts_even_a_permanent_child() {
let mut limiter = limiter(5, 100, 10_000);
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::ParentShutdown),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&four_healthy_slots(),
Instant::now(),
);
assert_eq!(outcome, SupervisionOutcome::Forget);
}
#[test]
fn backoff_compounds_across_consecutive_restarts_and_then_caps() {
let mut limiter = limiter(10, 100, 300);
let now = Instant::now();
let slots = four_healthy_slots();
let mut delays = Vec::new();
for _ in 0..4 {
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
now,
);
match outcome {
SupervisionOutcome::Restart { backoff, .. } => delays.push(backoff),
other => panic!("expected a restart, got {other:?}"),
}
}
assert_eq!(delays[0], BackoffDelay::from(Duration::from_millis(100)));
assert_eq!(delays[1], BackoffDelay::from(Duration::from_millis(200)));
assert_eq!(delays[2], BackoffDelay::from(Duration::from_millis(300)));
assert_eq!(
delays[3],
BackoffDelay::from(Duration::from_millis(300)),
"capped at max_backoff_ms"
);
}
#[test]
fn a_child_that_stayed_up_past_the_window_starts_its_backoff_over() {
let window = Duration::from_secs(WINDOW_SECS);
let mut limiter = limiter(10, 100, 10_000);
let start = Instant::now();
let slots = four_healthy_slots();
for _ in 0..2 {
let _ = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start,
);
}
assert_eq!(limiter.consecutive_restarts(), 2);
let recovered = SlotSnapshot {
last_restart: Some(start),
..snapshot(0)
};
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&recovered,
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start + window + Duration::from_secs(1),
);
match outcome {
SupervisionOutcome::Restart { backoff, .. } => assert_eq!(
backoff,
BackoffDelay::from(Duration::from_millis(100)),
"the backoff should start over, not compound to 400ms"
),
other => panic!("expected a restart, got {other:?}"),
}
}
#[test]
fn the_recovery_window_is_the_one_belonging_to_the_limiter_being_charged() {
let mut limiter = RestartLimiter::new(RestartLimiterConfig {
enabled: true,
max_restarts: 10,
window_secs: 600,
initial_backoff_ms: 100,
max_backoff_ms: 10_000,
backoff_multiplier: 2.0,
});
assert_eq!(limiter.window(), Duration::from_mins(10));
let start = Instant::now();
let slots = four_healthy_slots();
let _ = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start,
);
let up_for_a_minute = SlotSnapshot {
last_restart: Some(start),
..snapshot(0)
};
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&up_for_a_minute,
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start + Duration::from_secs(WINDOW_SECS + 1),
);
match outcome {
SupervisionOutcome::Restart { backoff, .. } => assert_eq!(
backoff,
BackoffDelay::from(Duration::from_millis(200)),
"a minute does not clear a ten-minute window, so the backoff compounds"
),
other => panic!("expected a restart, got {other:?}"),
}
}
#[test]
fn a_crash_inside_the_window_keeps_compounding_the_backoff() {
let mut limiter = limiter(10, 100, 10_000);
let start = Instant::now();
let slots = four_healthy_slots();
let _ = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start,
);
let recently_restarted = SlotSnapshot {
last_restart: Some(start),
..snapshot(0)
};
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&recently_restarted,
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
start + Duration::from_secs(5),
);
match outcome {
SupervisionOutcome::Restart { backoff, .. } => {
assert_eq!(backoff, BackoffDelay::from(Duration::from_millis(200)));
}
other => panic!("expected a restart, got {other:?}"),
}
}
#[test]
fn exhausting_the_allowance_escalates_instead_of_restarting() {
let mut limiter = limiter(2, 100, 10_000);
let now = Instant::now();
let slots = four_healthy_slots();
for attempt in 0..2 {
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
now,
);
assert!(
matches!(outcome, SupervisionOutcome::Restart { .. }),
"attempt {attempt} should still restart"
);
}
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
now,
);
match outcome {
SupervisionOutcome::Escalate(exceeded) => {
assert_eq!(exceeded.attempts, 2);
assert_eq!(exceeded.max_restarts, 2);
assert_eq!(exceeded.window_secs, WINDOW_SECS);
}
other => panic!("expected an escalation, got {other:?}"),
}
}
#[test]
fn a_disabled_limiter_never_escalates_and_never_delays() {
let mut limiter = RestartLimiter::new(RestartLimiterConfig::disabled());
let now = Instant::now();
let slots = four_healthy_slots();
for attempt in 0..20 {
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&slots,
now,
);
match outcome {
SupervisionOutcome::Restart { backoff, .. } => {
assert_eq!(backoff, BackoffDelay::NONE, "attempt {attempt}");
assert!(backoff.is_immediate());
}
other => panic!("expected a restart on attempt {attempt}, got {other:?}"),
}
}
}
#[test]
fn a_restart_that_would_plan_nothing_is_forgotten_rather_than_scheduled() {
let mut limiter = limiter(5, 100, 10_000);
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(0),
SupervisionStrategy::OneForOne,
&mut limiter,
&[],
Instant::now(),
);
assert_eq!(outcome, SupervisionOutcome::Forget);
}
#[test]
fn a_group_strategy_produces_a_group_plan() {
let mut limiter = limiter(5, 100, 10_000);
let outcome = evaluate(
¬ification(RestartPolicy::Permanent, TerminationReason::Panic("x".into())),
&snapshot(1),
SupervisionStrategy::RestForOne,
&mut limiter,
&four_healthy_slots(),
Instant::now(),
);
match outcome {
SupervisionOutcome::Restart { plan, .. } => {
assert_eq!(plan.restart, indices(&[1, 2, 3]));
assert_eq!(
plan.stop,
indices(&[3, 2]),
"child 1 is the one that terminated, so it is started but never stopped"
);
}
other => panic!("expected a restart, got {other:?}"),
}
}
}