From 8bb9810cc10123548b19d325fe0fee79a8eaac91 Mon Sep 17 00:00:00 2001 From: brunota20 Date: Thu, 18 Jun 2026 11:28:29 -0300 Subject: [PATCH] feat(supervisor): exponential-backoff restart with component reinstantiation (COW-1033) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a module traps in `on_event` (OutOfFuel, MemoryOutOfBounds, unhandled host error), the supervisor now: 1. Marks the module `alive = false` and increments `failure_count`. 2. Schedules a `next_attempt` instant via the new `runtime::restart_policy::backoff_for` (1s → 2s → 4s → ... cap 5 min). All dispatches before that instant skip the module. 3. On the first dispatch past the backoff window, the supervisor tears down the trapped wasmtime Store + component instance and re-instantiates from the cached `Component`. The instance state resets but host-side persistent state (local-store) survives so a module's progress counters live across restarts. 4. On a successful `on_event` after recovery, `failure_count` resets to 0 + `next_attempt = None`. ## Why the reinstantiation is required A wasmtime trap leaves the component instance poisoned: subsequent `call_on_event` returns "wasm trap: cannot enter component instance". Just refueling the Store does not recover. The supervisor caches the `Component`, `init_config`, and `http_allowlist` on `LoadedModule` at boot so a restart only needs a fresh Store + re-instantiation - the compiled component bytes are reused. ## New types / files - `crates/nexum-engine/src/runtime/restart_policy.rs`: `backoff_for(failure_count) -> Duration` with the 1s → 5min schedule. 4 unit tests covering the steady-state, first-failure, doubling, and cap arms. - `Supervisor` gains four cached backends (`engine`, `cow_pool`, `provider_pool`, `local_store`) so `reinstantiate_one(idx)` can rebuild the wasi Linker + HostState + Store + bindings on demand. - `LoadedModule` gains `component: Component`, `init_config: Config`, `http_allowlist: Vec` (all cloned at boot), plus `failure_count: u32` and `next_attempt: Option` for the schedule. ## Dispatch path changes `dispatch_block` and `dispatch_log` now restructure into two phases: 1. **Phase 1 (restart sweep)**: walk modules, collect indices of dead-but-due modules, call `reinstantiate_one` on each. Failed restarts bump the backoff again. Successful restarts flip `alive = true` so phase 2 dispatches the next event to them. 2. **Phase 2 (steady-state dispatch)**: unchanged from before - walk modules, dispatch where subscribed + alive. Trap path sets `next_attempt` + bumps `failure_count`; success path resets both. The structured logs from COW-1035 gain `failure_count` + `backoff_ms` on trap + `restart attempt` info lines on each restart. The `shepherd_module_restarts_total{module}` Prometheus counter from COW-1034 increments on every restart attempt. ## New fixture + integration test `modules/fixtures/flaky-bomb/` (test-only): traps via OutOfFuel on the first N events (N from `[config].fail_first_n`) and recovers afterwards. Uses local-store for the attempt counter because the wasm instance state resets on each reinstantiation; the counter persists in the host-side store so the module deterministically recovers after the configured N. `supervisor::tests::restart_flaky_module_recovers_after_backoff` (new): boots flaky-bomb with fail_first_n=1, dispatches, observes: - Dispatch 1: trap. alive=false, failure_count=1, next_attempt=+1s. - Immediate redispatch: skipped (still in backoff). - Sleep 1.1s. - Dispatch 3: restart fires, fresh instance attempts again. With attempt=2 > N=1, returns Ok. alive=true, failure_count=0, next_attempt=None. - Dispatch 4: steady-state, dispatches normally. Test wall-clock ~1.4s. ## Tests - `cargo test --workspace` -> 159 host tests + 6 doctests passing. +4 from `restart_policy` unit tests + 1 from the new integration test (was 154 + 6). - `cargo clippy --all-targets --workspace -- -D warnings` clean. - `cargo fmt --all --check` clean. - All existing resource-limit tests (COW-1036) still pass against the new dispatch shape: their assertions are against state *immediately* after the trap (before backoff elapses), so the restart machinery is transparent. - The `init_failure_marks_module_dead_and_excludes_from_dispatch` test (COW-1070) still passes: init-failed modules carry `next_attempt = None` so the restart sweep never picks them up. ## Out of scope - Persistence of `failure_count` / `next_attempt` across full engine restarts. The schedule resets on every boot; cross-engine persistence is a 0.3 follow-up. - WS reconnect-with-backoff for upstream RPC drops - that is COW-1071, a separate axis. - Operator-tunable backoff via `engine.toml::[engine.restart]`. The current constants are workspace literals in `runtime::restart_policy`; configurable in 0.3. - Module-side `on_restart` hook. Modules just see a fresh `init` call after a restart, same as boot. Linear: COW-1033. Fifth M4 issue landed; stacks on #38 (COW-1034). --- Cargo.toml | 1 + crates/nexum-engine/src/runtime/mod.rs | 1 + .../src/runtime/restart_policy.rs | 78 ++++++ crates/nexum-engine/src/supervisor.rs | 261 ++++++++++++++++-- crates/nexum-engine/src/supervisor/tests.rs | 111 +++++++- modules/fixtures/flaky-bomb/Cargo.toml | 13 + modules/fixtures/flaky-bomb/module.toml | 26 ++ modules/fixtures/flaky-bomb/src/lib.rs | 94 +++++++ 8 files changed, 562 insertions(+), 23 deletions(-) create mode 100644 crates/nexum-engine/src/runtime/restart_policy.rs create mode 100644 modules/fixtures/flaky-bomb/Cargo.toml create mode 100644 modules/fixtures/flaky-bomb/module.toml create mode 100644 modules/fixtures/flaky-bomb/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index 87034991..840e0a42 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "modules/examples/balance-tracker", "modules/examples/price-alert", "modules/examples/stop-loss", + "modules/fixtures/flaky-bomb", "modules/fixtures/fuel-bomb", "modules/fixtures/memory-bomb", "modules/twap-monitor", diff --git a/crates/nexum-engine/src/runtime/mod.rs b/crates/nexum-engine/src/runtime/mod.rs index 72ea95fd..fe041746 100644 --- a/crates/nexum-engine/src/runtime/mod.rs +++ b/crates/nexum-engine/src/runtime/mod.rs @@ -3,3 +3,4 @@ pub mod event_loop; pub mod limits; +pub mod restart_policy; diff --git a/crates/nexum-engine/src/runtime/restart_policy.rs b/crates/nexum-engine/src/runtime/restart_policy.rs new file mode 100644 index 00000000..d5938a0d --- /dev/null +++ b/crates/nexum-engine/src/runtime/restart_policy.rs @@ -0,0 +1,78 @@ +//! Supervisor module restart policy (COW-1033). +//! +//! When a module traps in `on_event`, the supervisor flips `alive = +//! false` and schedules a restart attempt with exponential backoff. +//! The next dispatch eligible for that module retries the call; on +//! success the failure counter resets so a module that recovers +//! lands back in the steady-state schedule with no further delay. +//! +//! Policy: +//! +//! | failure_count | next_attempt delay | +//! |---|---| +//! | 1 | 1s | +//! | 2 | 2s | +//! | 3 | 4s | +//! | ... | doubles | +//! | 9+ | capped at 5 minutes | +//! +//! State is in-memory per supervisor process. Persistence across +//! engine restarts is out of scope (a separate 0.3 / M5 follow-up +//! that lands alongside `submitted:{uid}` cross-restart dedup). + +use std::time::Duration; + +/// Hard cap on the restart backoff. After ~8 doublings we plateau +/// here. Tuneable in 0.3 via `engine.toml::[engine.restart]`. +pub const RESTART_MAX_BACKOFF: Duration = Duration::from_secs(300); + +/// Compute the wait window the supervisor honours before the next +/// restart attempt of a module that has trapped `failure_count` times +/// in a row. +/// +/// `failure_count = 0` is the steady-state value (no failures yet); +/// it returns `Duration::ZERO` so the supervisor can call this +/// unconditionally without a branch at the call site. +/// +/// `failure_count >= 1` is "the module just trapped"; the first +/// retry is 1 s, doubling on each subsequent trap, capped at 5 min. +pub fn backoff_for(failure_count: u32) -> Duration { + if failure_count == 0 { + return Duration::ZERO; + } + // 1 << (n - 1) doubles: 1, 2, 4, 8, 16, ..., 256 at n=9. + // saturating_sub keeps n=1 -> 1s; the .min(9) keeps the shift + // from overflowing on absurdly large failure counts. + let shift = failure_count.saturating_sub(1).min(9); + let secs = 1u64 << shift; + Duration::from_secs(secs).min(RESTART_MAX_BACKOFF) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn steady_state_is_zero() { + assert_eq!(backoff_for(0), Duration::ZERO); + } + + #[test] + fn first_failure_waits_one_second() { + assert_eq!(backoff_for(1), Duration::from_secs(1)); + } + + #[test] + fn doubling_progression() { + assert_eq!(backoff_for(2), Duration::from_secs(2)); + assert_eq!(backoff_for(3), Duration::from_secs(4)); + assert_eq!(backoff_for(4), Duration::from_secs(8)); + assert_eq!(backoff_for(5), Duration::from_secs(16)); + } + + #[test] + fn caps_at_five_minutes() { + assert_eq!(backoff_for(20), RESTART_MAX_BACKOFF); + assert_eq!(backoff_for(u32::MAX), RESTART_MAX_BACKOFF); + } +} diff --git a/crates/nexum-engine/src/supervisor.rs b/crates/nexum-engine/src/supervisor.rs index bb58fba9..6b3adac9 100644 --- a/crates/nexum-engine/src/supervisor.rs +++ b/crates/nexum-engine/src/supervisor.rs @@ -5,11 +5,18 @@ //! `Store`, and routes the event types declared in each manifest's //! `[[subscription]]` table. //! -//! Trap handling (BLEU-817): a wasmtime trap in `on_event` marks the -//! module as `alive = false` and removes it from all future dispatch. -//! The module's subscriptions remain registered (the event-loop -//! streams are not closed) but the dispatcher skips dead modules. -//! Full restart-with-backoff lands in 0.3. +//! Trap handling (BLEU-817 + COW-1033): a wasmtime trap in `on_event` +//! marks the module `alive = false`, increments `failure_count`, and +//! schedules a `next_attempt` instant via `runtime::restart_policy:: +//! backoff_for`. The next dispatch eligible after that instant +//! re-instantiates the component (fresh `Store` + bindings; the +//! wasm instance left by a trap is poisoned with "cannot enter +//! component instance") and re-calls `init`. On a successful +//! `on_event` the failure counter resets to 0. +//! +//! Modules whose `init` returned `Err(HostError)` are dead with +//! `next_attempt = None` and never get scheduled - the init failure +//! is treated as a manifest / config bug, not a transient (COW-1070). use std::collections::BTreeSet; use std::path::Path; @@ -33,6 +40,15 @@ use crate::runtime::limits::{DEFAULT_FUEL_PER_EVENT, DEFAULT_MEMORY_LIMIT}; /// event loop needs. pub struct Supervisor { modules: Vec, + /// Cached for COW-1033 module restart: re-instantiating a + /// trapped module requires a fresh wasmtime `Store` + `Linker`, + /// which in turn need the shared backends. All four types are + /// `Clone` (internally `Arc`-backed) so the supervisor takes + /// owned copies at boot. + engine: Engine, + cow_pool: OrderBookPool, + provider_pool: ProviderPool, + local_store: LocalStore, } struct LoadedModule { @@ -42,10 +58,31 @@ struct LoadedModule { /// Subscriptions copied from `module.toml`. The supervisor reads /// these on every event to decide whether to dispatch. subscriptions: Vec, - /// Set to `false` when `on_event` traps. Dead modules are silently - /// skipped on every subsequent dispatch. Full restart-with-backoff - /// lands in 0.3. + /// Cached for COW-1033 restart: re-instantiating from the original + /// wasm bytes avoids re-reading the file on every restart. The + /// `Component` itself is internally `Arc`-backed by wasmtime. + component: Component, + /// Cached for COW-1033 restart: the manifest's `[config]` we pass + /// to `Guest::init`. Cloning a `Vec<(String, String)>` is cheap. + init_config: Config, + /// Cached for COW-1033 restart: HTTP allowlist baked into the + /// `HostState` we rebuild on each re-instantiation. + http_allowlist: Vec, + /// Set to `false` when `on_event` traps. Dead modules are + /// excluded from dispatch until `next_attempt` is in the past + /// (COW-1033). Modules whose `init` failed have `alive = false` + /// + `next_attempt = None`, so they never come back. alive: bool, + /// Number of consecutive trap-style failures since the last + /// successful dispatch. Resets to 0 on success. Drives the + /// exponential backoff via `restart_policy::backoff_for`. + failure_count: u32, + /// Earliest instant at which the supervisor may retry this + /// module after a trap. `None` for healthy modules + for modules + /// whose `init` failed (the latter never get scheduled because + /// the dispatch fast-path checks `next_attempt` *and* requires + /// `alive = false` before flipping back). + next_attempt: Option, } impl Supervisor { @@ -72,7 +109,13 @@ impl Supervisor { } let alive = modules.iter().filter(|m| m.alive).count(); info!(loaded = modules.len(), alive, "supervisor up"); - Ok(Self { modules }) + Ok(Self { + modules, + engine: engine.clone(), + cow_pool: cow_pool.clone(), + provider_pool: provider_pool.clone(), + local_store: local_store.clone(), + }) } /// One-shot construction from a single ad-hoc `(component, manifest)` @@ -96,6 +139,10 @@ impl Supervisor { Self::load_one(engine, linker, &entry, cow_pool, provider_pool, local_store).await?; Ok(Self { modules: vec![loaded], + engine: engine.clone(), + cow_pool: cow_pool.clone(), + provider_pool: provider_pool.clone(), + local_store: local_store.clone(), }) } @@ -240,6 +287,11 @@ impl Supervisor { store, subscriptions: loaded_manifest.manifest.subscriptions.clone(), alive: init_succeeded, + failure_count: 0, + next_attempt: None, + component, + init_config: config, + http_allowlist: loaded_manifest.http_allowlist.clone(), }) } @@ -295,10 +347,116 @@ impl Supervisor { /// Dispatch a block event to every module subscribed to /// `block.chain_id`. Returns the number of modules invoked. /// Modules that trap are marked dead and excluded from future dispatch. + /// Rebuild a module from its cached `Component` + `init_config` + /// after a wasmtime trap (COW-1033). A trap leaves the original + /// `Store` + component instance in a poisoned state ("cannot + /// enter component instance" on the next call); the only way to + /// recover is to create a fresh `Store` + re-instantiate. The + /// `LoadedModule.subscriptions` and `LoadedModule.name` are + /// preserved so the dispatch routing keeps working. + /// + /// On success the module's `alive` flag is left for the caller + /// to flip; on failure (e.g. `init` returns Err again) the + /// module stays dead and the failure_count keeps climbing. + async fn reinstantiate_one(&mut self, idx: usize) -> Result<()> { + // Re-build the wasi linker. Cheap: just two `add_to_linker` + // calls against the cached `Engine`. + let mut linker = Linker::::new(&self.engine); + Shepherd::add_to_linker::>( + &mut linker, + |state| state, + )?; + wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; + + let wasi = WasiCtxBuilder::new().inherit_stdio().build(); + let limits = wasmtime::StoreLimitsBuilder::new() + .memory_size(DEFAULT_MEMORY_LIMIT) + .build(); + let module = &mut self.modules[idx]; + let mut store = Store::new( + &self.engine, + HostState { + wasi, + table: ResourceTable::new(), + limits, + monotonic_baseline: std::time::Instant::now(), + http_allowlist: module.http_allowlist.clone(), + module_namespace: module.name.clone(), + cow: self.cow_pool.clone(), + chain: self.provider_pool.clone(), + store: self.local_store.clone(), + }, + ); + store.limiter(|state| &mut state.limits); + store.set_fuel(DEFAULT_FUEL_PER_EVENT)?; + let bindings = Shepherd::instantiate_async(&mut store, &module.component, &linker) + .await + .map_err(Error::from) + .with_context(|| format!("reinstantiate {}", module.name))?; + match bindings.call_init(&mut store, &module.init_config).await? { + Ok(()) => {} + Err(e) => { + return Err(anyhow!( + "init returned host-error on restart: {} ({:?})", + e.message, + e.kind + )); + } + } + module.bindings = bindings; + module.store = store; + Ok(()) + } + pub async fn dispatch_block(&mut self, block: nexum::host::types::Block) -> usize { let chain_id = block.chain_id; let block_number = block.number; let event = nexum::host::types::Event::Block(block); + let now = std::time::Instant::now(); + + // 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. + 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) + }) + .collect(); + for idx in restart_candidates { + let name = self.modules[idx].name.clone(); + let failure_count = self.modules[idx].failure_count; + info!(module = %name, failure_count, "restart attempt"); + metrics::counter!( + "shepherd_module_restarts_total", + "module" => name.clone(), + ) + .increment(1); + match self.reinstantiate_one(idx).await { + Ok(()) => { + self.modules[idx].alive = true; + info!(module = %name, "restart succeeded"); + } + Err(e) => { + // Re-instantiation failed: bump the backoff + // again so the next attempt is further out. + let m = &mut self.modules[idx]; + m.failure_count = m.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(m.failure_count); + m.next_attempt = Some(std::time::Instant::now() + backoff); + error!( + module = %name, + failure_count = m.failure_count, + backoff_ms = backoff.as_millis() as u64, + error = %e, + "restart failed - will retry after backoff", + ); + } + } + } + let mut dispatched = 0; for module in &mut self.modules { if !module.alive { @@ -344,6 +502,12 @@ impl Supervisor { "event_kind" => "block", ) .record(elapsed.as_secs_f64()); + // COW-1033: successful dispatch clears the + // failure history. A module that recovered after + // N traps lands back in the steady-state + // schedule with no further delay. + module.failure_count = 0; + module.next_attempt = None; dispatched += 1; } Ok(Err(host_err)) => { @@ -370,14 +534,19 @@ impl Supervisor { Err(trap) => { let elapsed = start.elapsed(); let latency_ms = elapsed.as_millis() as u64; + module.failure_count = module.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(module.failure_count); + let next_attempt = std::time::Instant::now() + backoff; error!( module = %module.name, chain_id, event_kind = "block", block_number, latency_ms, + failure_count = module.failure_count, + backoff_ms = backoff.as_millis() as u64, error = %trap, - "on-event trapped - module marked dead, removed from dispatch", + "on-event trapped - module marked dead; will retry after backoff", ); metrics::counter!( "shepherd_module_errors_total", @@ -386,6 +555,7 @@ impl Supervisor { ) .increment(1); module.alive = false; + module.next_attempt = Some(next_attempt); } } } @@ -402,13 +572,47 @@ impl Supervisor { chain_id: u64, log: alloy_rpc_types_eth::Log, ) -> bool { - let target = match self.modules.iter_mut().find(|m| m.name == module_name) { - Some(m) => m, - None => { - warn!(module = %module_name, "no such module - dropping log"); - return false; - } + let now = std::time::Instant::now(); + 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-1033 restart-on-trap: re-instantiate before dispatch + // if the backoff window elapsed. See `dispatch_block` for + // the symmetric path. + let needs_restart = { + let m = &self.modules[idx]; + !m.alive && m.next_attempt.is_some_and(|t| t <= now) + }; + if needs_restart { + let name = self.modules[idx].name.clone(); + let failure_count = self.modules[idx].failure_count; + info!(module = %name, failure_count, "restart attempt"); + metrics::counter!( + "shepherd_module_restarts_total", + "module" => name.clone(), + ) + .increment(1); + match self.reinstantiate_one(idx).await { + Ok(()) => self.modules[idx].alive = true, + Err(e) => { + let m = &mut self.modules[idx]; + m.failure_count = m.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(m.failure_count); + m.next_attempt = Some(std::time::Instant::now() + backoff); + error!( + module = %name, + failure_count = m.failure_count, + error = %e, + "restart failed - will retry after backoff", + ); + return false; + } + } + } + + let target = &mut self.modules[idx]; if !target.alive { return false; } @@ -441,6 +645,8 @@ impl Supervisor { "event_kind" => "log", ) .record(elapsed.as_secs_f64()); + target.failure_count = 0; + target.next_attempt = None; true } Ok(Err(host_err)) => { @@ -468,14 +674,19 @@ impl Supervisor { Err(trap) => { let elapsed = start.elapsed(); let latency_ms = elapsed.as_millis() as u64; + target.failure_count = target.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(target.failure_count); + let next_attempt = std::time::Instant::now() + backoff; error!( module = %module_name, chain_id, event_kind = "log", block_number, latency_ms, + failure_count = target.failure_count, + backoff_ms = backoff.as_millis() as u64, error = %trap, - "on-event trapped - module marked dead, removed from dispatch", + "on-event trapped - module marked dead; will retry after backoff", ); metrics::counter!( "shepherd_module_errors_total", @@ -484,6 +695,7 @@ impl Supervisor { ) .increment(1); target.alive = false; + target.next_attempt = Some(next_attempt); false } } @@ -494,6 +706,21 @@ impl Supervisor { pub fn alive_count(&self) -> usize { self.modules.iter().filter(|m| m.alive).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 + /// `boot` pipeline. + #[cfg(test)] + pub(crate) fn empty_for_test(engine: &Engine, local_store: LocalStore) -> Self { + Self { + modules: Vec::new(), + engine: engine.clone(), + cow_pool: OrderBookPool::default(), + provider_pool: ProviderPool::empty(), + local_store, + } + } } /// Project an alloy `Log` onto the WIT `log` record. The chain id diff --git a/crates/nexum-engine/src/supervisor/tests.rs b/crates/nexum-engine/src/supervisor/tests.rs index d16b6356..6e2a0e68 100644 --- a/crates/nexum-engine/src/supervisor/tests.rs +++ b/crates/nexum-engine/src/supervisor/tests.rs @@ -4,9 +4,9 @@ use super::*; #[test] fn empty_supervisor_returns_no_subscriptions() { - let sup = Supervisor { - modules: Vec::new(), - }; + let engine = make_wasmtime_engine(); + let (_dir, store) = temp_local_store(); + let sup = Supervisor::empty_for_test(&engine, store); assert!(sup.block_chains().is_empty()); assert!(sup.log_subscriptions().is_empty()); assert_eq!(sup.module_count(), 0); @@ -28,9 +28,9 @@ fn empty_supervisor_returns_no_subscriptions() { async fn run_does_not_bail_when_both_stream_kinds_are_empty() { use std::time::{Duration, Instant}; - let mut supervisor = Supervisor { - modules: Vec::new(), - }; + let engine = make_wasmtime_engine(); + let (_dir, store) = temp_local_store(); + let mut supervisor = Supervisor::empty_for_test(&engine, store); let started = Instant::now(); let shutdown = tokio::time::sleep(Duration::from_millis(50)); @@ -660,6 +660,105 @@ async fn resource_limit_memory_bomb_traps_and_marks_module_dead() { assert_eq!(dispatched_again, 0); } +// ── COW-1033: supervisor auto-restart with exponential backoff ─────── +// +// flaky-bomb traps on the first N events (via wasm `unreachable!`) +// and recovers on event N+1. Exercises the full restart lifecycle: +// +// 1. Dispatch 1: trap -> alive=false, failure_count=1, next_attempt=+1s. +// 2. Immediate redispatch: skipped (next_attempt in the future). +// 3. After 1.1s: alive flipped back on, dispatch retried. +// 4. With fail_first_n=1, the second attempt succeeds -> failure_count +// resets to 0, next_attempt = None. +// +// Asserts the schedule shape end-to-end with real wall-clock. + +#[tokio::test] +async fn restart_flaky_module_recovers_after_backoff() { + let Some(wasm) = module_wasm_or_skip("flaky-bomb") else { + return; + }; + + let dir = tempfile::tempdir().unwrap(); + let manifest = dir.path().join("module.toml"); + // fail_first_n = 1 so the module traps once and recovers on the + // second dispatch attempt. Keeps the test wall-clock under 2 s. + std::fs::write( + &manifest, + r#" +[module] +name = "flaky-bomb" + +[capabilities] +required = ["logging", "local-store"] + +[[subscription]] +kind = "block" +chain_id = 1 + +[config] +fail_first_n = "1" +"#, + ) + .unwrap(); + + 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(); + let mut supervisor = Supervisor::boot_single( + &engine, + &linker, + &wasm, + Some(&manifest), + &cow_pool, + &provider_pool, + &store, + ) + .await + .expect("boot_single"); + assert_eq!(supervisor.alive_count(), 1); + + let block = nexum::host::types::Block { + chain_id: 1, + number: 1, + hash: vec![0; 32], + timestamp: 1_700_000_000_000, + }; + + // Dispatch 1: trap. Module marked dead with a +1s backoff. + let dispatched = supervisor.dispatch_block(block.clone()).await; + assert_eq!(dispatched, 0, "first dispatch trapped, no module accepted"); + assert_eq!(supervisor.alive_count(), 0, "module marked dead"); + + // Immediate redispatch (under the 1s backoff): still skipped. + let dispatched_immediate = supervisor.dispatch_block(block.clone()).await; + assert_eq!( + dispatched_immediate, 0, + "in-backoff module not eligible for redispatch yet", + ); + assert_eq!(supervisor.alive_count(), 0); + + // Wait for the 1s backoff window to elapse (+ a small fudge for + // scheduler jitter). + tokio::time::sleep(std::time::Duration::from_millis(1100)).await; + + // Dispatch 3: now eligible. fail_first_n=1 was satisfied on + // dispatch 1, so this attempt succeeds. The supervisor flips + // alive back on, dispatch lands, failure_count resets. + let dispatched_after_backoff = supervisor.dispatch_block(block.clone()).await; + assert_eq!( + dispatched_after_backoff, 1, + "module recovered after the backoff window", + ); + assert_eq!(supervisor.alive_count(), 1, "recovered + alive"); + + // Dispatch 4: steady-state, no backoff in play. Module is happy. + let dispatched_steady = supervisor.dispatch_block(block).await; + assert_eq!(dispatched_steady, 1); +} + // ── build_alloy_filter ──────────────────────────────────────────────── #[test] diff --git a/modules/fixtures/flaky-bomb/Cargo.toml b/modules/fixtures/flaky-bomb/Cargo.toml new file mode 100644 index 00000000..d6e3cd6e --- /dev/null +++ b/modules/fixtures/flaky-bomb/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "flaky-bomb" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true +description = "COW-1033 evil-by-design fixture: traps on the first N events (via unreachable!) and succeeds afterwards. The supervisor must exercise its exponential-backoff restart policy + reset the failure counter when the module recovers." + +[lib] +crate-type = ["cdylib"] + +[dependencies] +wit-bindgen = { version = "0.57", default-features = false, features = ["macros", "realloc"] } diff --git a/modules/fixtures/flaky-bomb/module.toml b/modules/fixtures/flaky-bomb/module.toml new file mode 100644 index 00000000..5237bdbd --- /dev/null +++ b/modules/fixtures/flaky-bomb/module.toml @@ -0,0 +1,26 @@ +# flaky-bomb test fixture (COW-1033). Subscribes to blocks; `on_event` +# traps via `unreachable!()` on the first N attempts, then recovers. +# Drives the supervisor's exponential-backoff restart policy through +# its full lifecycle. + +[module] +name = "flaky-bomb" +version = "0.1.0" +component = "sha256:0000000000000000000000000000000000000000000000000000000000000000" + +[capabilities] +required = ["logging", "local-store"] +optional = [] + +[capabilities.http] +allow = [] + +[[subscription]] +kind = "block" +chain_id = 1 + +[config] +# Number of consecutive events to trap on before recovering. Tests +# typically synthesise a manifest with `fail_first_n = "1"` to keep +# the test wall-clock short (only one 1 s backoff window to wait). +fail_first_n = "1" diff --git a/modules/fixtures/flaky-bomb/src/lib.rs b/modules/fixtures/flaky-bomb/src/lib.rs new file mode 100644 index 00000000..2bd9f1d3 --- /dev/null +++ b/modules/fixtures/flaky-bomb/src/lib.rs @@ -0,0 +1,94 @@ +//! # flaky-bomb (test fixture - COW-1033) +//! +//! Traps deterministically on the first N events and succeeds on +//! every subsequent event. Drives the supervisor's exponential- +//! backoff restart policy through its full lifecycle: +//! +//! 1. Dispatch 1: trap (failure_count = 1, next_attempt = +1s). +//! 2. (engine waits the backoff window) +//! 3. Dispatch 2 (eligible after 1s): trap again, failure_count = 2. +//! 4. ... +//! 5. Dispatch N+1: succeeds, failure_count resets to 0. +//! +//! N is config-supplied via `[config].fail_first_n`. The fixture +//! reads the value once during `init` into a `OnceLock` and keeps +//! a static `AtomicU32` counter across calls. +//! +//! Not a production module. Lives under `modules/fixtures/` so it is +//! obviously test-only. + +#![cfg_attr(not(test), warn(unused_crate_dependencies))] +#![allow(clippy::too_many_arguments)] + +wit_bindgen::generate!({ + path: "../../../wit/nexum-host", + world: "nexum:host/event-module", +}); + +use std::sync::OnceLock; + +use nexum::host::{local_store, logging, types}; + +/// Number of consecutive events to trap on. Set from `[config].fail_first_n` +/// at init; defaults to `1` (trap once, recover on second event). +static FAIL_FIRST_N: OnceLock = OnceLock::new(); + +const ATTEMPTS_KEY: &str = "attempts"; + +struct FlakyBomb; + +impl Guest for FlakyBomb { + fn init(config: Vec<(String, String)>) -> Result<(), HostError> { + let n: u32 = config + .iter() + .find(|(k, _)| k == "fail_first_n") + .and_then(|(_, v)| v.parse().ok()) + .unwrap_or(1); + FAIL_FIRST_N.set(n).ok(); + logging::log( + logging::Level::Info, + &format!("flaky-bomb init: will trap on the first {n} event(s)"), + ); + Ok(()) + } + + fn on_event(_event: types::Event) -> Result<(), HostError> { + // Read + increment the attempt counter from local-store. + // Survives wasm-side state resets (the supervisor's restart + // path tears down the Store; local-store is host-side and + // persistent within the supervisor's lifetime, exactly the + // store COW-1033 keeps across reinstantiations). + let prior = local_store::get(ATTEMPTS_KEY)? + .and_then(|b| <[u8; 4]>::try_from(b.as_slice()).ok()) + .map(u32::from_le_bytes) + .unwrap_or(0); + let attempt = prior + 1; + local_store::set(ATTEMPTS_KEY, &attempt.to_le_bytes())?; + + let n = FAIL_FIRST_N.get().copied().unwrap_or(1); + if attempt <= n { + logging::log( + logging::Level::Warn, + &format!("flaky-bomb attempt {attempt}/{n}: burning fuel to trigger OutOfFuel"), + ); + // Burn fuel until wasmtime traps with `OutOfFuel`. The + // supervisor catches the trap + schedules a backoff + // restart. After the backoff window the supervisor + // re-instantiates the component (fresh wasm Store), but + // local-store survives so the attempt counter keeps + // climbing across restarts. + let mut x: u64 = 0; + loop { + x = x.wrapping_add(1); + std::hint::black_box(x); + } + } + logging::log( + logging::Level::Info, + &format!("flaky-bomb attempt {attempt}: ok, recovered"), + ); + Ok(()) + } +} + +export!(FlakyBomb);