diff --git a/crates/nexum-engine/src/runtime/mod.rs b/crates/nexum-engine/src/runtime/mod.rs index fe041746..b0639297 100644 --- a/crates/nexum-engine/src/runtime/mod.rs +++ b/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; diff --git a/crates/nexum-engine/src/runtime/poison_policy.rs b/crates/nexum-engine/src/runtime/poison_policy.rs new file mode 100644 index 00000000..cf6a02c8 --- /dev/null +++ b/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)); + } +} diff --git a/crates/nexum-engine/src/supervisor.rs b/crates/nexum-engine/src/supervisor.rs index 6b3adac9..78a5c200 100644 --- a/crates/nexum-engine/src/supervisor.rs +++ b/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, + /// 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, + /// 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, @@ -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 = (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. diff --git a/crates/nexum-engine/src/supervisor/tests.rs b/crates/nexum-engine/src/supervisor/tests.rs index 6e2a0e68..bb15ba16 100644 --- a/crates/nexum-engine/src/supervisor/tests.rs +++ b/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]