diff --git a/crates/lance-graph-supervisor/Cargo.toml b/crates/lance-graph-supervisor/Cargo.toml index cbf177f27..c36fb9208 100644 --- a/crates/lance-graph-supervisor/Cargo.toml +++ b/crates/lance-graph-supervisor/Cargo.toml @@ -51,7 +51,7 @@ supervisor-lifecycle-audit = ["supervisor"] cycle-driver = ["dep:lance-graph-planner", "dep:tokio"] [dev-dependencies] -tokio = { version = "1", features = ["rt-multi-thread", "macros"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "test-util"] } static_assertions = "1" # D-V3-W2b integration probe ONLY: spawns KanbanActor over the REAL production # MailboxSoaOwner impl instead of the in-file TestBoard fake. dev-dependencies diff --git a/crates/lance-graph-supervisor/src/supervisor.rs b/crates/lance-graph-supervisor/src/supervisor.rs index e5a8d4c01..ccf788af2 100644 --- a/crates/lance-graph-supervisor/src/supervisor.rs +++ b/crates/lance-graph-supervisor/src/supervisor.rs @@ -28,6 +28,9 @@ const BACKOFF_INITIAL_MS: u64 = 100; const BACKOFF_CAP: Duration = Duration::from_secs(30); /// Crash count threshold for escalation (operator alert). const ESCALATION_CRASH_COUNT: u32 = 10; +/// Escalation counts crashes within this window (spec §5.3): a crash more +/// than this long after the previous one starts the count again at 1. +const CRASH_WINDOW: Duration = Duration::from_secs(5 * 60); // ─── Module info for a G slot ───────────────────────────────────────────────── @@ -46,6 +49,8 @@ pub struct ModuleEntry { pub struct ConsumerSlot { pub actor_ref: ActorRef, pub crash_count: u32, + /// When the crash that led to this respawn happened; `None` for a child + /// that has never crashed. pub last_crash_ts: Option, } @@ -53,13 +58,18 @@ impl ConsumerSlot { /// Exponential backoff delay for the next respawn. /// Formula: `min(100ms * 2^crash_count, 30s)`. pub fn backoff_delay(&self) -> Duration { - let shifts = self.crash_count.min(30); - let ms = BACKOFF_INITIAL_MS.saturating_mul(1u64.checked_shl(shifts).unwrap_or(u64::MAX)); - let delay = Duration::from_millis(ms.min(BACKOFF_CAP.as_millis() as u64)); - delay.min(BACKOFF_CAP) + backoff_delay(self.crash_count) } } +/// Backoff before respawning a child that has crashed `crash_count` times: +/// `min(100ms * 2^crash_count, 30s)`. +fn backoff_delay(crash_count: u32) -> Duration { + let shifts = crash_count.min(30); + let ms = BACKOFF_INITIAL_MS.saturating_mul(1u64.checked_shl(shifts).unwrap_or(u64::MAX)); + Duration::from_millis(ms.min(BACKOFF_CAP.as_millis() as u64)).min(BACKOFF_CAP) +} + // ─── Supervisor messages ────────────────────────────────────────────────────── /// Messages the supervisor actor accepts. @@ -77,11 +87,14 @@ pub enum SupervisorMsg { }, /// Graceful shutdown — stops all children then stops supervisor. Shutdown, - /// Internal: supervisor schedules a respawn of a dead child after backoff. + /// Internal: respawn a dead child. Sent only by the supervisor itself, + /// once the backoff for `crash_count` has elapsed; the respawned slot + /// carries `crash_count` forward. RespawnG { g: u32, version: u32, crash_count: u32, + crashed_at: Instant, }, } @@ -246,6 +259,7 @@ impl Actor for CallcenterSupervisor { g, version, crash_count, + crashed_at, } => { let entry = state .modules @@ -258,53 +272,21 @@ impl Actor for CallcenterSupervisor { return Ok(()); }; - if crash_count > ESCALATION_CRASH_COUNT { - tracing::error!( - g, - version, - crash_count, - "consumer actor crash threshold > 10 — NOT respawning; \ - operator action required (send ResetCrashCount)" - ); + // The backoff has already been waited out by the sender + // (`handle_supervisor_evt`). A slot that is live again needs + // nothing; this message never reschedules itself. Defensive + // only: ractor's registry would also refuse a second actor + // under the same name, so no test can tell this check apart. + if state.slots.contains_key(&(g, version)) { + tracing::debug!(g, version, "RespawnG: slot already live"); return Ok(()); } - // Backoff: 100ms * 2^crash_count, capped at 30s. - let delay = { - let shifts = crash_count.min(30); - let ms = BACKOFF_INITIAL_MS - .saturating_mul(1u64.checked_shl(shifts).unwrap_or(u64::MAX)); - Duration::from_millis(ms.min(BACKOFF_CAP.as_millis() as u64)).min(BACKOFF_CAP) - }; - - tracing::warn!( - g, - version, - crash_count, - delay_ms = delay.as_millis(), - "scheduling consumer actor respawn" - ); - - // Use tokio to sleep the backoff, then respawn. - let myself2 = myself.clone(); - tokio::spawn(async move { - tokio::time::sleep(delay).await; - if let Err(e) = myself2.cast(SupervisorMsg::RespawnG { - g, - version, - crash_count: 0, // reset count after backoff succeeds - }) { - tracing::error!(g, version, ?e, "failed to cast RespawnG after backoff"); - } - }); - - // Immediate spawn attempt (the tokio task above will re-cast with count=0 - // after the backoff if we need to retry). match spawn_consumer_actor(myself.clone(), state, &entry).await { Ok(()) => { if let Some(slot) = state.slots.get_mut(&(g, version)) { slot.crash_count = crash_count; - slot.last_crash_ts = Some(Instant::now()); + slot.last_crash_ts = Some(crashed_at); } tracing::info!(g, version, crash_count, "consumer actor respawned"); } @@ -334,20 +316,56 @@ impl Actor for CallcenterSupervisor { ?reason, "consumer actor terminated; scheduling respawn" ); - let crash_count = state - .slots - .get(&(g, version)) - .map(|s| s.crash_count + 1) - .unwrap_or(1); + // Count crashes within the window: a crash more than + // CRASH_WINDOW after the previous one starts again at 1. + // tokio's clock, so the window follows paused time in tests; + // outside tests it is the real clock. + let crashed_at = tokio::time::Instant::now().into_std(); + let crash_count = match state.slots.get(&(g, version)) { + Some(ConsumerSlot { + crash_count, + last_crash_ts: Some(prev), + .. + }) if crashed_at.duration_since(*prev) <= CRASH_WINDOW => crash_count + 1, + _ => 1, + }; state.slots.remove(&(g, version)); state.reverse_index.remove(&actor_id); - myself.cast(SupervisorMsg::RespawnG { + if crash_count > ESCALATION_CRASH_COUNT { + tracing::error!( + g, + version, + crash_count, + "consumer actor crashed more than 10 times within 5 minutes — NOT respawning; \ + operator action required" + ); + return Ok(()); + } + + // The delayed cast is the only respawn path: one per crash, + // after `min(100ms * 2^crash_count, 30s)`. + let delay = backoff_delay(crash_count); + tracing::warn!( g, version, crash_count, - })?; + delay_ms = delay.as_millis(), + "scheduling consumer actor respawn" + ); + let myself2 = myself.clone(); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + if let Err(e) = myself2.cast(SupervisorMsg::RespawnG { + g, + version, + crash_count, + crashed_at, + }) { + tracing::error!(g, version, ?e, "failed to cast RespawnG after backoff"); + } + }); } } Ok(()) diff --git a/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs new file mode 100644 index 000000000..d5234a3c0 --- /dev/null +++ b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs @@ -0,0 +1,141 @@ +//! Test: a crashed consumer actor is respawned once per crash, after the +//! exponential backoff, and not at all once the escalation threshold is passed. +//! +//! Issue: ISS-SUPERVISOR-RESPAWN-BACKOFF-NOT-APPLIED. Before the fix, the +//! `RespawnG` arm respawned at once (no delay) and also scheduled a delayed +//! `RespawnG { crash_count: 0 }` that re-ran the same arm every 100 ms forever, +//! so a child escalated past 10 crashes was brought back by that chain. +//! +//! Time is paused (`start_paused`), so the 30 s backoff cap costs nothing: +//! tokio advances the clock whenever every task is idle. + +#[cfg(feature = "supervisor")] +mod tests { + use lance_graph_supervisor::{CallcenterSupervisor, ModuleEntry, SupervisorMsg}; + use ractor::{Actor, ActorRef}; + use std::time::Duration; + + // The ractor registry is process-global and the tests run in parallel, so + // each test supervises its own G to keep the child names apart. + fn one_active_module(g: u32) -> Vec { + vec![ModuleEntry { + g, + version: 1, + mailbox_capacity: None, + is_active: true, + }] + } + + /// `(is_live, crash_count)` of slot `g` as the supervisor reports it. + async fn slot(sup: &ActorRef, g: u32) -> (bool, u32) { + let health = sup + .call(|tx| SupervisorMsg::Health { reply: tx }, None) + .await + .expect("health call") + .expect("health reply"); + let child = health + .children + .iter() + .find(|c| c.g == g) + .expect("G is in the module table"); + (child.is_live, child.crash_count) + } + + fn crash(g: u32) { + ractor::registry::where_is(format!("consumer_g_{g}")) + .expect("G is registered while live") + .stop(None); + } + + /// Backoff after crash `n`: `min(100 ms * 2^n, 30 s)`. + fn backoff(n: u32) -> Duration { + Duration::from_millis((100u64 << n.min(30)).min(30_000)) + } + + #[tokio::test(start_paused = true)] + async fn each_crash_respawns_once_after_its_backoff() { + const G: u32 = 2; + let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module(G)), ()) + .await + .expect("supervisor spawn"); + tokio::time::sleep(Duration::from_millis(10)).await; + assert_eq!(slot(&sup, G).await, (true, 0), "G live before any crash"); + + for n in 1..=2u32 { + crash(G); + // Half the backoff: the child must still be down. + tokio::time::sleep(backoff(n) / 2).await; + assert!( + !slot(&sup, G).await.0, + "crash {n}: respawned before its {:?} backoff", + backoff(n) + ); + // Past the backoff: up again, and the count is carried, not reset. + tokio::time::sleep(backoff(n)).await; + assert_eq!( + slot(&sup, G).await, + (true, n), + "crash {n}: one respawn after the backoff" + ); + } + + sup.stop(None); + let _ = handle.await; + } + + #[tokio::test(start_paused = true)] + async fn a_child_past_the_escalation_threshold_stays_down() { + const G: u32 = 3; + let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module(G)), ()) + .await + .expect("supervisor spawn"); + tokio::time::sleep(Duration::from_millis(10)).await; + + // Crashes 1..=10 are respawned (threshold is "more than 10"). + for n in 1..=10u32 { + crash(G); + tokio::time::sleep(backoff(n) + Duration::from_millis(50)).await; + assert_eq!(slot(&sup, G).await, (true, n), "crash {n} is respawned"); + } + + // Crash 11 escalates: no respawn, however long we wait. + crash(G); + tokio::time::sleep(Duration::from_secs(120)).await; + assert!(!slot(&sup, G).await.0, "crash 11 must not be respawned"); + + sup.stop(None); + let _ = handle.await; + } + + /// Spec §5.3: escalation is more than ten crashes within a five-minute + /// window. A child that ran quietly for longer than the window starts + /// counting again. + #[tokio::test(start_paused = true)] + async fn the_count_restarts_after_a_quiet_window() { + const G: u32 = 4; + let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module(G)), ()) + .await + .expect("supervisor spawn"); + tokio::time::sleep(Duration::from_millis(10)).await; + + for n in 1..=10u32 { + crash(G); + tokio::time::sleep(backoff(n) + Duration::from_millis(50)).await; + assert_eq!(slot(&sup, G).await, (true, n), "crash {n} is respawned"); + } + + // Six quiet minutes, then one more crash: the count restarts at 1 + // instead of escalating as crash 11. + tokio::time::sleep(Duration::from_secs(6 * 60)).await; + crash(G); + tokio::time::sleep(backoff(1) + Duration::from_millis(50)).await; + assert_eq!( + slot(&sup, G).await, + (true, 1), + "a crash after a quiet window counts from 1" + ); + + sup.stop(None); + let _ = handle.await; + } +}