feat(supervisor): poison-pill detection + quarantine (COW-1032)

crates/nexum-engine/src/runtime/mod.rs

@@ -3,4 +3,5 @@
 
 pub mod event_loop;
 pub mod limits;
+pub mod poison_policy;
 pub mod restart_policy;

crates/nexum-engine/src/runtime/poison_policy.rs

@@ -0,0 +1,91 @@
+//! Supervisor poison-pill policy (COW-1032).
+//!
+//! Modules that trap more than `max_failures` times within a sliding
+//! `window` are marked **poisoned**: the supervisor stops dispatching
+//! events to them entirely (no further restart attempts), bumps a
+//! `shepherd_module_poisoned{module}` gauge to 1, and logs the
+//! quarantine event so an operator can investigate. Recovery
+//! requires an operator-driven full engine restart (today): remove
+//! the entry from `engine.toml::[[modules]]`, kill the process, fix
+//! the module, restart.
+//!
+//! ## Difference from the restart policy (COW-1033)
+//!
+//! `restart_policy::backoff_for` schedules retries for transient
+//! traps; the failure counter resets on a successful dispatch. The
+//! poison policy is the *sustained-failure* escalation: if a module
+//! is still trapping after `max_failures` retries inside `window`,
+//! it stops being a transient and becomes a permanent failure that
+//! exhausts an operator's restart budget without ever recovering.
+//! Stop retrying.
+//!
+//! The two policies share `LoadedModule.failure_count` for the
+//! consecutive-failure semantic; poison adds a `failure_timestamps`
+//! ring so the window check is independent of how the failures are
+//! spaced (one second apart vs nine minutes apart both count toward
+//! the same window).
+
+use std::time::Duration;
+
+/// Production defaults: 5 traps within 10 minutes -> quarantine.
+/// Aggressive enough to catch a deterministically broken module
+/// without waiting out the full exponential backoff (the 5th trap
+/// happens at ~31 s into the schedule: 1+2+4+8+16 s); lenient
+/// enough that a one-off RPC blip during a real cow-api submit does
+/// not get a module quarantined.
+pub const POISON_MAX_FAILURES: u32 = 5;
+pub const POISON_WINDOW: Duration = Duration::from_secs(600);
+
+/// Configurable poison-pill thresholds. Constructed via
+/// [`PoisonPolicy::default`] for production; tests can shorten both
+/// values via [`PoisonPolicy::new`] so the integration test does
+/// not have to wait out the full real-world schedule.
+#[derive(Debug, Clone, Copy)]
+pub struct PoisonPolicy {
+    /// Maximum traps within `window` before the module is poisoned.
+    pub max_failures: u32,
+    /// Sliding window the failures are counted across.
+    pub window: Duration,
+}
+
+impl PoisonPolicy {
+    pub const fn new(max_failures: u32, window: Duration) -> Self {
+        Self {
+            max_failures,
+            window,
+        }
+    }
+}
+
+impl Default for PoisonPolicy {
+    fn default() -> Self {
+        Self::new(POISON_MAX_FAILURES, POISON_WINDOW)
+    }
+}
+
+/// Return `true` when `failure_count` failures inside `window`
+/// crosses the configured threshold.
+pub fn should_poison(policy: PoisonPolicy, recent_failures: u32) -> bool {
+    recent_failures >= policy.max_failures
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn default_is_production_constants() {
+        let p = PoisonPolicy::default();
+        assert_eq!(p.max_failures, POISON_MAX_FAILURES);
+        assert_eq!(p.window, POISON_WINDOW);
+    }
+
+    #[test]
+    fn poisons_at_threshold() {
+        let p = PoisonPolicy::new(3, Duration::from_secs(60));
+        assert!(!should_poison(p, 0));
+        assert!(!should_poison(p, 2));
+        assert!(should_poison(p, 3));
+        assert!(should_poison(p, 100));
+    }
+}

crates/nexum-engine/src/supervisor.rs

@@ -49,6 +49,10 @@ pub struct Supervisor {
     cow_pool: OrderBookPool,
     provider_pool: ProviderPool,
     local_store: LocalStore,
+    /// COW-1032 poison-pill thresholds. Defaults to the production
+    /// constants (5 failures / 10 min); tests inject tighter values
+    /// via `boot_with_poison_policy` / `empty_for_test`.
+    poison_policy: crate::runtime::poison_policy::PoisonPolicy,
 }
 
 struct LoadedModule {
@@ -83,6 +87,15 @@ struct LoadedModule {
     /// the dispatch fast-path checks `next_attempt` *and* requires
     /// `alive = false` before flipping back).
     next_attempt: Option<std::time::Instant>,
+    /// Sliding-window record of recent trap timestamps for the
+    /// poison-pill check (COW-1032). Entries older than the
+    /// `PoisonPolicy.window` are dropped on each push.
+    failure_timestamps: std::collections::VecDeque<std::time::Instant>,
+    /// Once `true` the module is permanently quarantined: no restart
+    /// attempts, no dispatches, no metric churn. Recovery requires
+    /// an operator-driven full engine restart with the module
+    /// removed from `engine.toml::[[modules]]`.
+    poisoned: bool,
 }
 
 impl Supervisor {
@@ -115,6 +128,7 @@ impl Supervisor {
             cow_pool: cow_pool.clone(),
             provider_pool: provider_pool.clone(),
             local_store: local_store.clone(),
+            poison_policy: crate::runtime::poison_policy::PoisonPolicy::default(),
         })
     }
 
@@ -143,9 +157,23 @@ impl Supervisor {
             cow_pool: cow_pool.clone(),
             provider_pool: provider_pool.clone(),
             local_store: local_store.clone(),
+            poison_policy: crate::runtime::poison_policy::PoisonPolicy::default(),
         })
     }
 
+    /// Override the poison-pill policy. Tests use this to inject
+    /// tighter thresholds (e.g. 3 failures in 60 s) so the
+    /// integration suite does not wait out the production 5/10min
+    /// schedule. Returns `self` so it can be chained off `boot_single`.
+    #[cfg(test)]
+    pub(crate) fn with_poison_policy(
+        mut self,
+        policy: crate::runtime::poison_policy::PoisonPolicy,
+    ) -> Self {
+        self.poison_policy = policy;
+        self
+    }
+
     async fn load_one(
         engine: &Engine,
         linker: &Linker<HostState>,
@@ -292,6 +320,8 @@ impl Supervisor {
             component,
             init_config: config,
             http_allowlist: loaded_manifest.http_allowlist.clone(),
+            failure_timestamps: std::collections::VecDeque::new(),
+            poisoned: false,
         })
     }
 
@@ -413,16 +443,22 @@ impl Supervisor {
         let block_number = block.number;
         let event = nexum::host::types::Event::Block(block);
         let now = std::time::Instant::now();
+        let poison_policy = self.poison_policy;
 
         // COW-1033 phase 1: find dead modules whose backoff window
         // has elapsed and re-instantiate them in place. The wasmtime
         // store + component instance left by a trap is poisoned
         // ("cannot enter component instance" on the next call), so
         // recovery requires a fresh Store + re-instantiated bindings.
+        //
+        // COW-1032: poisoned modules are excluded from the restart
+        // sweep entirely. Once quarantined they stay dead until
+        // an operator removes them from `engine.toml::[[modules]]`
+        // and restarts the engine.
         let restart_candidates: Vec<usize> = (0..self.modules.len())
             .filter(|&i| {
                 let m = &self.modules[i];
-                !m.alive && m.next_attempt.is_some_and(|t| t <= now)
+                !m.poisoned && !m.alive && m.next_attempt.is_some_and(|t| t <= now)
             })
             .collect();
         for idx in restart_candidates {
@@ -459,7 +495,7 @@ impl Supervisor {
 
         let mut dispatched = 0;
         for module in &mut self.modules {
-            if !module.alive {
+            if module.poisoned || !module.alive {
                 continue;
             }
             let subscribed = module
@@ -556,6 +592,7 @@ impl Supervisor {
                     .increment(1);
                     module.alive = false;
                     module.next_attempt = Some(next_attempt);
+                    record_failure_and_maybe_poison(module, poison_policy, &trap.to_string());
                 }
             }
         }
@@ -573,11 +610,20 @@ impl Supervisor {
         log: alloy_rpc_types_eth::Log,
     ) -> bool {
         let now = std::time::Instant::now();
+        let poison_policy = self.poison_policy;
         let Some(idx) = self.modules.iter().position(|m| m.name == module_name) else {
             warn!(module = %module_name, "no such module - dropping log");
             return false;
         };
 
+        // COW-1032 poison-pill: quarantined modules get no log
+        // dispatches at all - same as block. The check happens
+        // before the restart sweep so a poisoned module never
+        // triggers a restart attempt.
+        if self.modules[idx].poisoned {
+            return false;
+        }
+
         // COW-1033 restart-on-trap: re-instantiate before dispatch
         // if the backoff window elapsed. See `dispatch_block` for
         // the symmetric path.
@@ -696,6 +742,7 @@ impl Supervisor {
                 .increment(1);
                 target.alive = false;
                 target.next_attempt = Some(next_attempt);
+                record_failure_and_maybe_poison(target, poison_policy, &trap.to_string());
                 false
             }
         }
@@ -707,6 +754,13 @@ impl Supervisor {
         self.modules.iter().filter(|m| m.alive).count()
     }
 
+    /// COW-1032: also expose a per-module poisoned state for
+    /// metrics + integration tests.
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub fn poisoned_count(&self) -> usize {
+        self.modules.iter().filter(|m| m.poisoned).count()
+    }
+
     /// Build a zero-module supervisor with synthetic shared
     /// backends. Used by the unit tests that need a `Supervisor` to
     /// poke its public surface without going through the full
@@ -719,10 +773,49 @@ impl Supervisor {
             cow_pool: OrderBookPool::default(),
             provider_pool: ProviderPool::empty(),
             local_store,
+            poison_policy: crate::runtime::poison_policy::PoisonPolicy::default(),
         }
     }
 }
 
+/// COW-1032: push the current trap timestamp into the module's
+/// failure-window ring, drop entries older than the policy window,
+/// and flip `poisoned = true` once the window holds more than
+/// `policy.max_failures` traps. The first transition emits the
+/// `shepherd_module_poisoned` gauge + a structured WARN.
+fn record_failure_and_maybe_poison(
+    module: &mut LoadedModule,
+    policy: crate::runtime::poison_policy::PoisonPolicy,
+    last_error: &str,
+) {
+    let now = std::time::Instant::now();
+    // Prune entries outside the window.
+    while let Some(&front) = module.failure_timestamps.front() {
+        if now.duration_since(front) > policy.window {
+            module.failure_timestamps.pop_front();
+        } else {
+            break;
+        }
+    }
+    module.failure_timestamps.push_back(now);
+    let recent = module.failure_timestamps.len() as u32;
+    if crate::runtime::poison_policy::should_poison(policy, recent) && !module.poisoned {
+        module.poisoned = true;
+        warn!(
+            module = %module.name,
+            recent_failures = recent,
+            window_secs = policy.window.as_secs(),
+            last_error,
+            "module poisoned - quarantined; remove from engine.toml + restart to clear",
+        );
+        metrics::gauge!(
+            "shepherd_module_poisoned",
+            "module" => module.name.clone(),
+        )
+        .set(1.0);
+    }
+}
+
 /// Project an alloy `Log` onto the WIT `log` record. The chain id
 /// is not on the alloy log (the subscription context carries it),
 /// so we receive it alongside.

crates/nexum-engine/src/supervisor/tests.rs

@@ -759,6 +759,99 @@ fail_first_n = "1"
     assert_eq!(dispatched_steady, 1);
 }
 
+// ── COW-1032: poison-pill quarantine ──────────────────────────────────
+//
+// fuel-bomb (the COW-1036 fixture) traps on every dispatch. With a
+// tight poison policy (3 failures / 60 s) we can observe the
+// supervisor escalate from "retry" to "permanent quarantine" inside
+// ~4 s of wall clock:
+//
+//   trap 1: failure_count=1, next_attempt=+1s
+//   sleep 1.1s
+//   trap 2: failure_count=2, next_attempt=+2s
+//   sleep 2.1s
+//   trap 3: failure_count=3 -> POISONED. Recent failures hit the
+//           window threshold; the supervisor stops attempting
+//           restarts entirely. Subsequent dispatches skip the
+//           module silently.
+//
+// Tests assert each transition + the post-quarantine no-op semantic.
+
+#[tokio::test]
+async fn poison_pill_quarantines_module_after_threshold() {
+    let Some(wasm) = module_wasm_or_skip("fuel-bomb") else {
+        return;
+    };
+    let manifest = production_module_toml("modules/fixtures/fuel-bomb/module.toml");
+    let engine = make_wasmtime_engine();
+    let linker = make_linker(&engine);
+    let cow_pool = crate::host::cow_orderbook::OrderBookPool::default();
+    let provider_pool = crate::host::provider_pool::ProviderPool::empty();
+    let (_dir, store) = temp_local_store();
+
+    // Tight policy: 3 failures in 60 s -> quarantine. Keeps the
+    // test wall-clock under 4 s.
+    let policy =
+        crate::runtime::poison_policy::PoisonPolicy::new(3, std::time::Duration::from_secs(60));
+    let mut supervisor = Supervisor::boot_single(
+        &engine,
+        &linker,
+        &wasm,
+        Some(&manifest),
+        &cow_pool,
+        &provider_pool,
+        &store,
+    )
+    .await
+    .expect("boot_single")
+    .with_poison_policy(policy);
+
+    assert_eq!(supervisor.module_count(), 1);
+    assert_eq!(supervisor.alive_count(), 1);
+    assert_eq!(supervisor.poisoned_count(), 0);
+
+    let block = nexum::host::types::Block {
+        chain_id: 1,
+        number: 1,
+        hash: vec![0; 32],
+        timestamp: 1_700_000_000_000,
+    };
+
+    // Trap 1.
+    let dispatched = supervisor.dispatch_block(block.clone()).await;
+    assert_eq!(dispatched, 0);
+    assert_eq!(supervisor.alive_count(), 0);
+    assert_eq!(supervisor.poisoned_count(), 0, "1 trap < threshold");
+    tokio::time::sleep(std::time::Duration::from_millis(1_100)).await;
+
+    // Trap 2.
+    let dispatched = supervisor.dispatch_block(block.clone()).await;
+    assert_eq!(dispatched, 0);
+    assert_eq!(supervisor.poisoned_count(), 0, "2 traps < threshold");
+    tokio::time::sleep(std::time::Duration::from_millis(2_100)).await;
+
+    // Trap 3 -> POISONED.
+    let dispatched = supervisor.dispatch_block(block.clone()).await;
+    assert_eq!(dispatched, 0);
+    assert_eq!(
+        supervisor.poisoned_count(),
+        1,
+        "3 traps inside window -> module quarantined",
+    );
+
+    // Post-quarantine: immediately re-dispatch. A poisoned module
+    // is excluded regardless of how much time has passed; the
+    // backoff timer is no longer load-bearing. We do NOT wait for
+    // the would-be next_attempt because the test just needs to
+    // observe the "skipped silently" semantic, not the timing.
+    let dispatched = supervisor.dispatch_block(block).await;
+    assert_eq!(
+        dispatched, 0,
+        "poisoned module excluded from dispatch forever",
+    );
+    assert_eq!(supervisor.poisoned_count(), 1);
+}
+
 // ── build_alloy_filter ────────────────────────────────────────────────
 
 #[test]