From 53cfd4ba083736e01f6e750fb57dc40281dea35b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 17:53:12 +0000 Subject: [PATCH 1/5] test(supervisor): respawn waits for its backoff and stops after escalation Two paused-time tests for ISS-SUPERVISOR-RESPAWN-BACKOFF-NOT-APPLIED. Both fail on the current RespawnG arm: the first respawn happens before its 200 ms backoff, and a child escalated past 10 crashes is brought back by the self-rescheduling RespawnG { crash_count: 0 } chain. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- crates/lance-graph-supervisor/Cargo.toml | 2 +- .../tests/supervisor_respawn_backoff.rs | 102 ++++++++++++++++++ 2 files changed, 103 insertions(+), 1 deletion(-) create mode 100644 crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs 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/tests/supervisor_respawn_backoff.rs b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs new file mode 100644 index 000000000..217c5264f --- /dev/null +++ b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs @@ -0,0 +1,102 @@ +//! 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; + + fn one_active_module() -> Vec { + vec![ModuleEntry { + g: 2, + version: 1, + mailbox_capacity: None, + is_active: true, + }] + } + + /// `(is_live, crash_count)` of G=2 as the supervisor reports it. + async fn g2(sup: &ActorRef) -> (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 == 2) + .expect("G=2 is in the module table"); + (child.is_live, child.crash_count) + } + + fn crash_g2() { + ractor::registry::where_is("consumer_g_2".to_string()) + .expect("G=2 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() { + let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module()), ()) + .await + .expect("supervisor spawn"); + tokio::time::sleep(Duration::from_millis(10)).await; + assert_eq!(g2(&sup).await, (true, 0), "G=2 live before any crash"); + + for n in 1..=2u32 { + crash_g2(); + // Half the backoff: the child must still be down. + tokio::time::sleep(backoff(n) / 2).await; + assert_eq!( + g2(&sup).await.0, + false, + "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!(g2(&sup).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() { + let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module()), ()) + .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_g2(); + tokio::time::sleep(backoff(n) + Duration::from_millis(50)).await; + assert_eq!(g2(&sup).await, (true, n), "crash {n} is respawned"); + } + + // Crash 11 escalates: no respawn, however long we wait. + crash_g2(); + tokio::time::sleep(Duration::from_secs(120)).await; + assert!(!g2(&sup).await.0, "crash 11 must not be respawned"); + + sup.stop(None); + let _ = handle.await; + } +} From 9cc38d9e5792dc649fcfdf72e4b87eb10f346e26 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 17:54:26 +0000 Subject: [PATCH 2/5] fix(supervisor): respawn once per crash, after the backoff The respawn path is now the delayed cast alone. On ActorTerminated the supervisor checks the escalation threshold, then casts RespawnG with the crash count after min(100ms * 2^n, 30s). RespawnG spawns the child, carries the count onto the new slot, does nothing if the slot is already live, and never reschedules itself. Before, it spawned at once and started a RespawnG { crash_count: 0 } chain that repeated every 100 ms forever and respawned children escalated past 10 crashes. The escalation message no longer names a ResetCrashCount message that does not exist. The two backoff tests now supervise different G slots: ractor's registry is process-global and the tests run in parallel. Closes ISS-SUPERVISOR-RESPAWN-BACKOFF-NOT-APPLIED. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- .../lance-graph-supervisor/src/supervisor.rs | 92 +++++++++---------- .../tests/supervisor_respawn_backoff.rs | 49 +++++----- 2 files changed, 74 insertions(+), 67 deletions(-) diff --git a/crates/lance-graph-supervisor/src/supervisor.rs b/crates/lance-graph-supervisor/src/supervisor.rs index e5a8d4c01..08198e497 100644 --- a/crates/lance-graph-supervisor/src/supervisor.rs +++ b/crates/lance-graph-supervisor/src/supervisor.rs @@ -53,13 +53,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,7 +82,9 @@ 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, @@ -258,48 +265,14 @@ 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. + 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)) { @@ -343,11 +316,38 @@ impl Actor for CallcenterSupervisor { 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 crash threshold > 10 — 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, + }) { + 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 index 217c5264f..039f0becf 100644 --- a/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs +++ b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs @@ -15,17 +15,19 @@ mod tests { use ractor::{Actor, ActorRef}; use std::time::Duration; - fn one_active_module() -> Vec { + // 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: 2, + g, version: 1, mailbox_capacity: None, is_active: true, }] } - /// `(is_live, crash_count)` of G=2 as the supervisor reports it. - async fn g2(sup: &ActorRef) -> (bool, u32) { + /// `(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 @@ -34,14 +36,14 @@ mod tests { let child = health .children .iter() - .find(|c| c.g == 2) - .expect("G=2 is in the module table"); + .find(|c| c.g == g) + .expect("G is in the module table"); (child.is_live, child.crash_count) } - fn crash_g2() { - ractor::registry::where_is("consumer_g_2".to_string()) - .expect("G=2 is registered while live") + fn crash(g: u32) { + ractor::registry::where_is(format!("consumer_g_{g}")) + .expect("G is registered while live") .stop(None); } @@ -52,25 +54,29 @@ mod tests { #[tokio::test(start_paused = true)] async fn each_crash_respawns_once_after_its_backoff() { - let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module()), ()) + 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!(g2(&sup).await, (true, 0), "G=2 live before any crash"); + assert_eq!(slot(&sup, G).await, (true, 0), "G live before any crash"); for n in 1..=2u32 { - crash_g2(); + crash(G); // Half the backoff: the child must still be down. tokio::time::sleep(backoff(n) / 2).await; - assert_eq!( - g2(&sup).await.0, - false, + 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!(g2(&sup).await, (true, n), "crash {n}: one respawn after the backoff"); + assert_eq!( + slot(&sup, G).await, + (true, n), + "crash {n}: one respawn after the backoff" + ); } sup.stop(None); @@ -79,22 +85,23 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_child_past_the_escalation_threshold_stays_down() { - let (sup, handle) = Actor::spawn(None, CallcenterSupervisor::new(one_active_module()), ()) + 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_g2(); + crash(G); tokio::time::sleep(backoff(n) + Duration::from_millis(50)).await; - assert_eq!(g2(&sup).await, (true, n), "crash {n} is respawned"); + assert_eq!(slot(&sup, G).await, (true, n), "crash {n} is respawned"); } // Crash 11 escalates: no respawn, however long we wait. - crash_g2(); + crash(G); tokio::time::sleep(Duration::from_secs(120)).await; - assert!(!g2(&sup).await.0, "crash 11 must not be respawned"); + assert!(!slot(&sup, G).await.0, "crash 11 must not be respawned"); sup.stop(None); let _ = handle.await; From 0c8e1d8ef130997c7627a3e68894c3deafbbe14c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 17:54:58 +0000 Subject: [PATCH 3/5] docs(supervisor): mark the live-slot check in RespawnG as defensive A disable run shows no test can distinguish it: ractor's registry refuses a duplicate actor name anyway. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- crates/lance-graph-supervisor/src/supervisor.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/lance-graph-supervisor/src/supervisor.rs b/crates/lance-graph-supervisor/src/supervisor.rs index 08198e497..9805025e6 100644 --- a/crates/lance-graph-supervisor/src/supervisor.rs +++ b/crates/lance-graph-supervisor/src/supervisor.rs @@ -267,7 +267,9 @@ impl Actor for CallcenterSupervisor { // 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. + // 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(()); From 576bbb6367555f2f1e218534c4b995b55b463e0a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 18:05:59 +0000 Subject: [PATCH 4/5] test(supervisor): the crash count restarts after a quiet five-minute window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec §5.3 escalates on more than ten crashes within five minutes; the current count is lifetime-wide (codex P1 on #1540). Fails on the current code. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- .../tests/supervisor_respawn_backoff.rs | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs index 039f0becf..d5234a3c0 100644 --- a/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs +++ b/crates/lance-graph-supervisor/tests/supervisor_respawn_backoff.rs @@ -106,4 +106,36 @@ mod tests { 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; + } } From dd3ab5d47eb96df1978bb1c59390f86d87991658 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 18:06:42 +0000 Subject: [PATCH 5/5] fix(supervisor): escalate on more than 10 crashes within five minutes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec §5.3 escalates on crash_count > 10 within a five-minute window; the count was lifetime-wide, so any eleventh crash removed a child however far apart the crashes were (codex P1). A crash more than CRASH_WINDOW after the previous one now starts the count at 1. last_crash_ts records the crash time, carried through RespawnG (new field crashed_at), taken from tokio's clock so the window follows paused time in tests and the real clock otherwise. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011BMJnDBAfzcuQh5HeT39Rh --- .../lance-graph-supervisor/src/supervisor.rs | 30 ++++++++++++++----- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/crates/lance-graph-supervisor/src/supervisor.rs b/crates/lance-graph-supervisor/src/supervisor.rs index 9805025e6..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, } @@ -89,6 +94,7 @@ pub enum SupervisorMsg { g: u32, version: u32, crash_count: u32, + crashed_at: Instant, }, } @@ -253,6 +259,7 @@ impl Actor for CallcenterSupervisor { g, version, crash_count, + crashed_at, } => { let entry = state .modules @@ -279,7 +286,7 @@ impl Actor for CallcenterSupervisor { 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"); } @@ -309,11 +316,19 @@ 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); @@ -323,7 +338,7 @@ impl Actor for CallcenterSupervisor { g, version, crash_count, - "consumer actor crash threshold > 10 — NOT respawning; \ + "consumer actor crashed more than 10 times within 5 minutes — NOT respawning; \ operator action required" ); return Ok(()); @@ -346,6 +361,7 @@ impl Actor for CallcenterSupervisor { g, version, crash_count, + crashed_at, }) { tracing::error!(g, version, ?e, "failed to cast RespawnG after backoff"); }