fix(supervisor): respawn a crashed consumer once, after its backoff

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

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,20 +49,27 @@ pub struct ModuleEntry {
 pub struct ConsumerSlot {
     pub actor_ref: ActorRef<ConsumerEnvelope>,
     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<Instant>,
 }
 
 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(())

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<ModuleEntry> {
+        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<SupervisorMsg>, 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;
+    }
+}