From 44b6d6a7bb0af342e3e9b8b4a7d66bcb65cae312 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Fri, 17 Jul 2026 00:43:22 +0000 Subject: [PATCH] runtime: fold venue adapters into the restart and poison sweeps A trap in a routed adapter call marks a shared Liveness dead; the dispatch-time sweeps restart the provider after the module backoff policy and quarantine a crash-looper per the poison policy. A dead venue resolves to unavailable, distinct from unknown-venue, and adapter_alive_count reports live routability. The flaky-venue fixture drives the trap-to-recovery and poison paths end to end. --- .github/workflows/ci.yml | 6 +- Cargo.lock | 8 + Cargo.toml | 1 + crates/nexum-runtime/src/host/actor.rs | 61 +++- crates/nexum-runtime/src/host/extension.rs | 4 + .../nexum-runtime/src/host/venue_registry.rs | 129 +++++-- crates/nexum-runtime/src/supervisor.rs | 325 +++++++++++++++--- crates/nexum-runtime/src/supervisor/tests.rs | 167 ++++++++- justfile | 4 +- modules/fixtures/flaky-venue/Cargo.toml | 17 + modules/fixtures/flaky-venue/module.toml | 19 + modules/fixtures/flaky-venue/src/lib.rs | 76 ++++ 12 files changed, 731 insertions(+), 86 deletions(-) create mode 100644 modules/fixtures/flaky-venue/Cargo.toml create mode 100644 modules/fixtures/flaky-venue/module.toml create mode 100644 modules/fixtures/flaky-venue/src/lib.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 186bce4d..42cc34c0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,7 +78,7 @@ jobs: - uses: taiki-e/install-action@43aecc8d72668fbcfe75c31400bc4f890f1c5853 # v2.83.2 with: tool: nextest - # Build all 15 guest module wasms ONCE (release/wasm32-wasip2): the single + # Build all 16 guest module wasms ONCE (release/wasm32-wasip2): the single # source of truth for guest buildability and the artifacts the integration # tests load. Replaces the deleted 9-way build-module matrix, which recompiled # the shared wasm dependency graph ~9x cold. Per-module size report folded in; @@ -88,8 +88,8 @@ jobs: cargo build --release --target wasm32-wasip2 --locked \ -p example -p twap-monitor -p ethflow-watcher -p price-alert \ -p balance-tracker -p stop-loss -p http-probe -p echo-venue \ - -p echo-client -p clock-reader -p flaky-bomb -p fuel-bomb \ - -p memory-bomb -p panic-bomb -p slow-host + -p echo-client -p clock-reader -p flaky-bomb -p flaky-venue \ + -p fuel-bomb -p memory-bomb -p panic-bomb -p slow-host { echo "### module .wasm sizes" echo "| module | bytes |" diff --git a/Cargo.lock b/Cargo.lock index 0b867ea8..217777ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2377,6 +2377,14 @@ dependencies = [ "wit-bindgen 0.59.0", ] +[[package]] +name = "flaky-venue" +version = "0.1.0" +dependencies = [ + "nexum-venue-sdk", + "wit-bindgen 0.58.0", +] + [[package]] name = "fnv" version = "1.0.7" diff --git a/Cargo.toml b/Cargo.toml index d4ee5eb6..418564d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ members = [ "modules/examples/stop-loss", "modules/fixtures/clock-reader", "modules/fixtures/flaky-bomb", + "modules/fixtures/flaky-venue", "modules/fixtures/fuel-bomb", "modules/fixtures/memory-bomb", "modules/fixtures/panic-bomb", diff --git a/crates/nexum-runtime/src/host/actor.rs b/crates/nexum-runtime/src/host/actor.rs index 10d5c461..f55bf2c7 100644 --- a/crates/nexum-runtime/src/host/actor.rs +++ b/crates/nexum-runtime/src/host/actor.rs @@ -4,7 +4,8 @@ //! caller, and each instance sits behind an [`ActorSlot`] async mutex held //! across the guest await, so one store never runs two guest calls at once. -use std::sync::Arc; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Instant; use tokio::sync::Mutex as AsyncMutex; use wasmtime::Store; @@ -16,6 +17,47 @@ use super::state::HostState; /// is not `Sync`; concurrent callers queue here. pub type ActorSlot = Arc>; +/// Shared liveness of one supervised component. The store marks it dead on +/// a trap, recording when, so the supervisor's restart sweep can count the +/// backoff from the death rather than from the sweep that observed it. +/// Cloning shares the flag. Starts alive. +#[derive(Clone, Debug, Default)] +pub struct Liveness(Arc>>); + +impl Liveness { + /// Whether the component is currently callable. + pub fn is_alive(&self) -> bool { + self.lock().is_none() + } + + /// When the component died, while it is dead. + pub fn dead_since(&self) -> Option { + *self.lock() + } + + /// Mark the component dead: its store trapped and is unusable. Keeps + /// the first death instant when already dead. + pub fn mark_dead(&self) { + let mut died_at = self.lock(); + if died_at.is_none() { + *died_at = Some(Instant::now()); + } + } + + /// Mark the component alive again after a restart. + pub fn mark_alive(&self) { + *self.lock() = None; + } + + /// The flag, recovered from a poisoned lock: the state is a bare + /// `Option`, valid under any interleaving. + fn lock(&self) -> MutexGuard<'_, Option> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + /// A guest call failed outside the component's typed error space. #[derive(Debug, thiserror::Error)] pub enum ActorFault { @@ -30,22 +72,26 @@ pub enum ActorFault { /// A supervised component store: refuelled before each guest call so every /// invocation starts from a full budget, with traps projected onto -/// [`ActorFault`]. +/// [`ActorFault`] and recorded on the shared [`Liveness`]. pub struct SupervisedStore { store: Store>, fuel_per_call: u64, + liveness: Liveness, } impl SupervisedStore { - /// Supervise an instantiated store with a per-call fuel budget. - pub fn new(store: Store>, fuel_per_call: u64) -> Self { + /// Supervise an instantiated store with a per-call fuel budget, + /// reporting traps on `liveness`. + pub fn new(store: Store>, fuel_per_call: u64, liveness: Liveness) -> Self { Self { store, fuel_per_call, + liveness, } } - /// Refuel, then run one guest call against the store. + /// Refuel, then run one guest call against the store. A trap marks the + /// shared liveness dead: the store is poisoned until reinstantiated. pub async fn call( &mut self, call: impl AsyncFnOnce(&mut Store>) -> wasmtime::Result, @@ -53,6 +99,9 @@ impl SupervisedStore { self.store .set_fuel(self.fuel_per_call) .map_err(ActorFault::Refuel)?; - call(&mut self.store).await.map_err(ActorFault::Trap) + call(&mut self.store).await.map_err(|trap| { + self.liveness.mark_dead(); + ActorFault::Trap(trap) + }) } } diff --git a/crates/nexum-runtime/src/host/extension.rs b/crates/nexum-runtime/src/host/extension.rs index a282d4ba..00d82442 100644 --- a/crates/nexum-runtime/src/host/extension.rs +++ b/crates/nexum-runtime/src/host/extension.rs @@ -11,6 +11,7 @@ use async_trait::async_trait; use wasmtime::Store; use wasmtime::component::{Component, Linker}; +use crate::host::actor::Liveness; use crate::host::component::RuntimeTypes; use crate::host::state::HostState; use crate::manifest::NamespaceCaps; @@ -84,6 +85,9 @@ pub struct ProviderInstance<'a, T: RuntimeTypes> { pub config: Vec<(String, String)>, /// Fuel budget applied before each routed guest call. pub fuel_per_call: u64, + /// Shared liveness the installed instance reports traps on and the + /// supervisor's restart sweep reads. + pub liveness: Liveness, } /// Outcome of one provider install. diff --git a/crates/nexum-runtime/src/host/venue_registry.rs b/crates/nexum-runtime/src/host/venue_registry.rs index 46c5485c..301e371e 100644 --- a/crates/nexum-runtime/src/host/venue_registry.rs +++ b/crates/nexum-runtime/src/host/venue_registry.rs @@ -41,7 +41,7 @@ use crate::bindings::{ IntentHeader, IntentStatus, IntentStatusUpdate, Quotation, RateLimit, SubmitOutcome, VenueAdapter, VenueError, nexum, }; -use crate::host::actor::{ActorFault, ActorSlot, SupervisedStore}; +use crate::host::actor::{ActorFault, ActorSlot, Liveness, SupervisedStore}; use crate::host::component::RuntimeTypes; use crate::host::extension::{ HostService, Installed, ProviderInstance, ProviderKind, downcast_service, @@ -225,10 +225,16 @@ pub struct VenueActor { } impl VenueActor { - /// Wrap an instantiated adapter store for routing. - pub fn new(store: Store>, bindings: VenueAdapter, fuel_per_call: u64) -> Self { + /// Wrap an instantiated adapter store for routing, reporting traps on + /// the shared `liveness`. + pub fn new( + store: Store>, + bindings: VenueAdapter, + fuel_per_call: u64, + liveness: Liveness, + ) -> Self { Self { - actor: SupervisedStore::new(store, fuel_per_call), + actor: SupervisedStore::new(store, fuel_per_call, liveness), bindings, } } @@ -302,6 +308,15 @@ impl VenueInvoker for VenueActor { /// One installed adapter behind its serialising slot. type AdapterSlot = ActorSlot; +/// One installed venue: the adapter slot plus the liveness the supervisor's +/// sweep shares with the actor. A dead entry stays installed, so the venue +/// resolves to `unavailable` (temporarily dead) rather than `unknown-venue` +/// (never installed) until the sweep restarts it. +struct InstalledVenue { + slot: AdapterSlot, + liveness: Liveness, +} + /// Per-caller charge history, pruned to the quota window on each touch. #[derive(Default)] struct QuotaLedger { @@ -356,7 +371,7 @@ fn status_body(status: IntentStatus) -> StatusBody { /// install through the shared handle at provider boot, before any client /// call routes. struct VenueRegistryInner { - adapters: Mutex>, + adapters: Mutex>, guard: Arc, quota: SubmitQuota, ledger: Mutex, @@ -382,31 +397,44 @@ impl VenueRegistry { /// extension's. pub const NAMESPACE: &'static str = "videre"; - /// Install an adapter under its venue id. Rejects a duplicate id: two + /// Install an adapter under its venue id, sharing `liveness` with its + /// invoker. Rejects a duplicate id while the incumbent is alive: two /// adapters answering the same venue would silently shadow one another, - /// which is a config error worth failing boot over. + /// which is a config error worth failing boot over. A dead incumbent is + /// replaced: that is the sweep restarting a trapped adapter. pub fn install( &self, venue: VenueId, + liveness: Liveness, invoker: impl VenueInvoker + 'static, ) -> Result<(), DuplicateVenue> { let mut adapters = self.inner.adapters.lock().expect("adapter map poisoned"); - if adapters.contains_key(&venue) { + if adapters.get(&venue).is_some_and(|v| v.liveness.is_alive()) { return Err(DuplicateVenue { venue }); } - adapters.insert(venue, Arc::new(AsyncMutex::new(invoker))); + adapters.insert( + venue, + InstalledVenue { + slot: Arc::new(AsyncMutex::new(invoker)), + liveness, + }, + ); Ok(()) } - /// Resolve a venue id to its installed adapter slot. + /// Resolve a venue id to its installed adapter slot. An uninstalled + /// venue is `unknown-venue`; an installed but dead one is `unavailable` + /// pending the supervisor's restart sweep, without touching its + /// poisoned store. fn resolve(&self, venue: &VenueId) -> Result { - self.inner - .adapters - .lock() - .expect("adapter map poisoned") - .get(venue) - .cloned() - .ok_or(VenueError::UnknownVenue) + let adapters = self.inner.adapters.lock().expect("adapter map poisoned"); + let installed = adapters.get(venue).ok_or(VenueError::UnknownVenue)?; + if !installed.liveness.is_alive() { + return Err(VenueError::Unavailable(format!( + "venue {venue} is dead pending restart" + ))); + } + Ok(Arc::clone(&installed.slot)) } /// Whether `caller` has budget left in the current window. Read-only: it @@ -580,8 +608,8 @@ impl VenueRegistry { } let mut updates = Vec::new(); for (venue, receipt) in snapshot { - // Installed adapters never leave the registry, so a resolve - // failure here is unreachable; skip defensively regardless. + // A dead venue fails to resolve; its watch stays for the + // cadence after the sweep restarts the adapter. let Ok(slot) = self.resolve(&venue) else { continue; }; @@ -724,6 +752,7 @@ impl ProviderKind for VenueAdapterKind { mut store, config, fuel_per_call, + liveness, } = instance; let bindings = VenueAdapter::instantiate_async(&mut store, component, linker) .await @@ -750,7 +779,8 @@ impl ProviderKind for VenueAdapterKind { registry .install( venue_id.clone(), - VenueActor::new(store, bindings, fuel_per_call), + liveness.clone(), + VenueActor::new(store, bindings, fuel_per_call, liveness), ) .with_context(|| format!("install adapter {venue_id}"))?; Ok(Installed::Live) @@ -1062,7 +1092,9 @@ mod tests { builder = builder.with_guard(guard); } let registry = builder.build(); - registry.install(cow(), adapter).expect("install adapter"); + registry + .install(cow(), Liveness::default(), adapter) + .expect("install adapter"); registry } @@ -1367,14 +1399,61 @@ mod tests { let a = Arc::new(StubCalls::default()); let b = Arc::new(StubCalls::default()); registry - .install(cow(), StubAdapter::new(a)) + .install(cow(), Liveness::default(), StubAdapter::new(a)) .expect("first install"); let err = registry - .install(cow(), StubAdapter::new(b)) + .install(cow(), Liveness::default(), StubAdapter::new(b)) .expect_err("second install collides"); assert_eq!(err.venue, cow()); } + #[tokio::test] + async fn dead_venue_is_unavailable_not_unknown() { + let calls = Arc::new(StubCalls::default()); + let liveness = Liveness::default(); + let registry = VenueRegistryBuilder::new(SubmitQuota::default()).build(); + registry + .install(cow(), liveness.clone(), StubAdapter::new(calls.clone())) + .expect("install adapter"); + liveness.mark_dead(); + + // Temporarily dead resolves distinctly from never installed, and + // the dead adapter's slot is never entered. + assert!(matches!( + registry.submit("mod-a", &cow(), b"b".to_vec()).await, + Err(VenueError::Unavailable(_)) + )); + assert!(matches!( + registry + .submit("mod-a", &VenueId::from("unlisted"), b"b".to_vec()) + .await, + Err(VenueError::UnknownVenue) + )); + assert_eq!(calls.derive.load(Ordering::SeqCst), 0); + } + + #[test] + fn a_dead_incumbent_is_replaced_on_reinstall() { + let registry = VenueRegistryBuilder::new(SubmitQuota::default()).build(); + let liveness = Liveness::default(); + registry + .install( + cow(), + liveness.clone(), + StubAdapter::new(Arc::new(StubCalls::default())), + ) + .expect("first install"); + liveness.mark_dead(); + registry + .install( + cow(), + Liveness::default(), + StubAdapter::new(Arc::new(StubCalls::default())), + ) + .expect("a restart replaces the dead incumbent"); + assert_eq!(registry.venue_count(), 1); + } + #[test] fn zero_quota_saturates_to_one() { let registry = @@ -1523,7 +1602,9 @@ mod tests { let registry = VenueRegistryBuilder::new(SubmitQuota::default()) .with_watch_limit(watch_limit) .build(); - registry.install(cow(), adapter).expect("install adapter"); + registry + .install(cow(), Liveness::default(), adapter) + .expect("install adapter"); registry } diff --git a/crates/nexum-runtime/src/supervisor.rs b/crates/nexum-runtime/src/supervisor.rs index 92894f6b..029fa1e3 100644 --- a/crates/nexum-runtime/src/supervisor.rs +++ b/crates/nexum-runtime/src/supervisor.rs @@ -18,6 +18,12 @@ //! `next_attempt = None` and never get scheduled - the init failure //! is treated as a manifest / config bug, not a transient. //! +//! Providers (venue adapters) ride the same sweeps: a trap inside a +//! routed call flips the [`Liveness`] their actor shares with the +//! supervisor, the venue resolves to `unavailable` (not +//! `unknown-venue`) while dead, and the sweep reinstalls the provider +//! after the same backoff and poison policies. +//! //! Multi-chain isolation: `dispatch_block(block)` walks //! every module but only enters those whose subscriptions match //! `block.chain_id`. Per-module restart / poison / fuel limits are @@ -43,6 +49,7 @@ use crate::bindings::{Config, EventModule, nexum}; use crate::engine_config::{ AdapterEntry, EngineConfig, ModuleEntry, ModuleLimits, OutboundHttpLimits, }; +use crate::host::actor::Liveness; use crate::host::component::{Components, RuntimeTypes, StateHandle, StateStore}; use crate::host::extension::{ Extension, HostService, HostServices, Installed, ProviderInstance, ProviderKind, @@ -64,10 +71,12 @@ use crate::manifest::{ /// the component seam backends. pub struct Supervisor { modules: Vec>, - /// Venue adapters loaded at boot, whether or not `init` succeeded. - adapters_total: usize, - /// Adapters whose `init` succeeded and that are installed for routing. - adapters_alive: usize, + /// Providers (venue adapters) loaded at boot, whether or not `init` + /// succeeded. Swept for restart and poison alongside the modules. + providers: Vec, + /// Registered provider kinds paired with their services, kept for the + /// provider restart sweep to reinstall through. + kinds: ProviderKinds, /// Cached for module restart: re-instantiating a trapped module /// requires a fresh wasmtime `Store` + `Linker`, which in turn need /// the shared backends. The `Components` bundle is cheaply cloned @@ -252,6 +261,41 @@ struct LoadedModule { dispatch_bucket: crate::runtime::dispatch_rate::TokenBucket, } +/// One loaded provider (venue adapter). Mirrors [`LoadedModule`]'s restart +/// and poison bookkeeping; liveness is shared with the installed actor, +/// which marks it dead on a trap, and read back by the sweep. +struct LoadedProvider { + /// The provider's namespace: its manifest name, and its venue id. + name: String, + /// Registered kind spelling the restart sweep reinstalls through. + kind: &'static str, + /// Cached for restart, like a module's. + component: Component, + /// Cached for restart: the manifest `[config]` handed to `init`. + init_config: Config, + /// Cached for restart: the operator's transport grants. + http_allow: Vec, + messaging_topics: Vec, + /// Cached for restart: the engine `[limits]` applied at boot. + http_limits: OutboundHttpLimits, + fuel_per_call: u64, + memory_limit: usize, + chain_response_max_bytes: usize, + local_store_bytes: u64, + /// Trap flag shared with the installed actor. + liveness: Liveness, + /// Sequence of the run currently installed; restarts increment it. + run_seq: u64, + /// The sweep's view of `liveness`: a `true` here against a dead + /// liveness is an unrecorded trap. Boot init failure leaves it `false` + /// with `next_attempt = None`, permanent like a module's. + alive: bool, + failure_count: u32, + next_attempt: Option, + failure_timestamps: std::collections::VecDeque, + poisoned: bool, +} + /// One registered provider kind paired with the service its installs bind to. type ProviderRow = (Box>, Arc); @@ -330,10 +374,9 @@ impl Supervisor { // module store built below already routes to the installed venues. // Providers link only their kind's scoped imports. let provider_registry = CapabilityRegistry::provider(); - let adapters_total = engine_cfg.adapters.len(); - let mut adapters_alive = 0; + let mut providers = Vec::with_capacity(engine_cfg.adapters.len()); for entry in &engine_cfg.adapters { - let installed = Self::load_provider( + let loaded = Self::load_provider( engine, entry, components, @@ -344,9 +387,7 @@ impl Supervisor { ) .await .with_context(|| format!("load provider {}", entry.path.display()))?; - if installed == Installed::Live { - adapters_alive += 1; - } + providers.push(loaded); } let mut modules = Vec::with_capacity(engine_cfg.modules.len()); @@ -366,17 +407,18 @@ impl Supervisor { modules.push(loaded); } let alive = modules.iter().filter(|m| m.alive).count(); + let adapters_alive = providers.iter().filter(|p| p.alive).count(); info!( loaded = modules.len(), alive, - adapters = adapters_total, + adapters = providers.len(), adapters_alive, "supervisor up" ); Ok(Self { modules, - adapters_total, - adapters_alive, + providers, + kinds, engine: engine.clone(), components: components.clone(), extensions: extensions.to_vec(), @@ -425,8 +467,8 @@ impl Supervisor { .await?; Ok(Self { modules: vec![loaded], - adapters_total: 0, - adapters_alive: 0, + providers: Vec::new(), + kinds: ProviderKinds::new(), engine: engine.clone(), components: components.clone(), extensions: extensions.to_vec(), @@ -691,8 +733,8 @@ impl Supervisor { /// declared kind against the registered provider kinds, enforce the /// scoped-transport capability set, build a supervised store carrying /// the operator's HTTP and messaging grants, and hand the instance to - /// its kind to instantiate and install. [`Installed::Dead`] marks a - /// failed guest `init`: loaded and counted, but not routable. + /// its kind to instantiate and install. A failed guest `init` loads the + /// provider dead and unroutable, permanently like a module's. // One flat argument per shared input threaded onto the store, matching // the module load path. #[allow(clippy::too_many_arguments)] @@ -704,7 +746,7 @@ impl Supervisor { registry: &CapabilityRegistry, clocks: Option<&WasiClockOverride>, kinds: &ProviderKinds, - ) -> Result { + ) -> Result { let manifest_path = resolve_manifest_path(&entry.path, entry.manifest.as_deref()); let loaded_manifest: LoadedManifest = match manifest_path.as_deref() { Some(p) if p.exists() => { @@ -801,22 +843,48 @@ impl Supervisor { )?; let config: Config = if loaded_manifest.config.is_empty() { - vec![("name".into(), namespace)] + vec![("name".into(), namespace.clone())] } else { loaded_manifest.config.clone() }; - kind.install( - ProviderInstance { - component: &component, - linker: &linker, - store, - config, - fuel_per_call: limits_cfg.fuel(), - }, - service, - ) - .await - .with_context(|| format!("install {}", entry.path.display())) + let liveness = Liveness::default(); + let installed = kind + .install( + ProviderInstance { + component: &component, + linker: &linker, + store, + config: config.clone(), + fuel_per_call: limits_cfg.fuel(), + liveness: liveness.clone(), + }, + service, + ) + .await + .with_context(|| format!("install {}", entry.path.display()))?; + if installed == Installed::Dead { + liveness.mark_dead(); + } + Ok(LoadedProvider { + name: namespace, + kind: kind.kind(), + component, + init_config: config, + http_allow: entry.http_allow.clone(), + messaging_topics: entry.messaging_topics.clone(), + http_limits: limits_cfg.http(), + fuel_per_call: limits_cfg.fuel(), + memory_limit: limits_cfg.memory(), + chain_response_max_bytes: limits_cfg.chain_response_max_bytes(), + local_store_bytes: limits_cfg.state_bytes(), + liveness, + run_seq: 0, + alive: installed == Installed::Live, + failure_count: 0, + next_attempt: None, + failure_timestamps: std::collections::VecDeque::new(), + poisoned: false, + }) } /// Number of modules currently loaded. @@ -826,14 +894,17 @@ impl Supervisor { /// Number of venue adapters loaded at boot, alive or not. pub fn adapter_count(&self) -> usize { - self.adapters_total + self.providers.len() } - /// Number of adapters whose `init` succeeded and that are installed in the - /// venue registry for routing. + /// Number of adapters currently alive and routable. Live: a trap drops + /// it, the restart sweep raises it again. #[cfg_attr(not(test), allow(dead_code))] pub fn adapter_alive_count(&self) -> usize { - self.adapters_alive + self.providers + .iter() + .filter(|p| p.liveness.is_alive()) + .count() } /// Chains any **alive** module asked for block events on. Dead modules @@ -1024,6 +1095,7 @@ impl Supervisor { for idx in restart_candidates { self.try_restart(idx).await; } + self.sweep_providers().await; let mut dispatched = 0; let candidate_indices: Vec = (0..self.modules.len()) @@ -1087,6 +1159,7 @@ impl Supervisor { cursor_key: Option<&str>, ) -> bool { let now = std::time::Instant::now(); + self.sweep_providers().await; let Some(idx) = self.modules.iter().position(|m| m.name == module_name) else { warn!(module = %module_name, "no such module - dropping chain-log"); return false; @@ -1176,6 +1249,7 @@ impl Supervisor { for idx in restart_candidates { self.try_restart(idx).await; } + self.sweep_providers().await; let candidate_indices: Vec = (0..self.modules.len()) .filter(|&i| { @@ -1418,6 +1492,133 @@ impl Supervisor { } } + /// Fold providers into the recovery path: record any trap the shared + /// liveness reports (backoff plus poison bookkeeping), then reinstall + /// dead, unpoisoned providers whose backoff has elapsed. Runs at the + /// head of every dispatch, beside the module restart sweep. + async fn sweep_providers(&mut self) { + let now = std::time::Instant::now(); + let policy = self.poison_policy; + for idx in 0..self.providers.len() { + let provider = &mut self.providers[idx]; + if provider.alive + && let Some(died_at) = provider.liveness.dead_since() + { + provider.alive = false; + provider.failure_count = provider.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(provider.failure_count); + // Backoff counts from the death, not from this sweep, so a + // trap whose backoff already elapsed restarts right below. + provider.next_attempt = Some(died_at.checked_add(backoff).unwrap_or(now)); + warn!( + adapter = %provider.name, + failure_count = provider.failure_count, + backoff_ms = backoff.as_millis() as u64, + "adapter trapped - marked dead; will restart after backoff", + ); + metrics::counter!( + "shepherd_adapter_errors_total", + "adapter" => provider.name.clone(), + "error_kind" => "trap", + ) + .increment(1); + if let Some(recent) = poison_crossed(&mut provider.failure_timestamps, policy) + && !provider.poisoned + { + provider.poisoned = true; + warn!( + adapter = %provider.name, + recent_failures = recent, + window_secs = policy.window.as_secs(), + "adapter poisoned - quarantined; remove from engine.toml + restart to clear", + ); + metrics::gauge!( + "shepherd_adapter_poisoned", + "adapter" => provider.name.clone(), + ) + .set(1.0); + } + } + let provider = &self.providers[idx]; + if !provider.poisoned + && !provider.alive + && provider.next_attempt.is_some_and(|t| t <= now) + { + self.try_restart_provider(idx).await; + } + } + } + + /// Attempt to reinstall a dead provider in place: fresh store, fresh + /// instance, `init`, and a re-install replacing the dead slot. On + /// success the shared liveness is revived; on failure the backoff + /// slides further out, like a module restart. + async fn try_restart_provider(&mut self, idx: usize) { + let name = self.providers[idx].name.clone(); + let failure_count = self.providers[idx].failure_count; + info!(adapter = %name, failure_count, "adapter restart attempt"); + metrics::counter!( + "shepherd_adapter_restarts_total", + "adapter" => name.clone(), + ) + .increment(1); + let outcome = self.reinstall_provider(idx).await; + let provider = &mut self.providers[idx]; + match outcome { + Ok(Installed::Live) => { + provider.run_seq += 1; + provider.liveness.mark_alive(); + provider.alive = true; + provider.failure_count = 0; + provider.next_attempt = None; + info!(adapter = %name, "adapter restart succeeded"); + } + Ok(Installed::Dead) => { + defer_provider_restart(provider, "init returned fault on restart"); + } + Err(e) => defer_provider_restart(provider, &format!("{e:#}")), + } + } + + /// Rebuild a provider from its cached component and grants, then hand + /// it back to its kind to instantiate and install over the dead slot. + async fn reinstall_provider(&mut self, idx: usize) -> Result { + let provider = &self.providers[idx]; + let (kind, service) = self + .kinds + .get(provider.kind) + .ok_or_else(|| anyhow!("provider kind {} is not registered", provider.kind))?; + let linker = build_provider_linker::(&self.engine, kind.as_ref())?; + // A restart is a new run, like a module's. + let run = RunId::new(provider.name.clone(), provider.run_seq + 1); + let store = Self::build_store( + &self.engine, + &self.components, + run, + provider.http_allow.clone(), + provider.http_limits, + provider.messaging_topics.clone(), + provider.memory_limit, + provider.fuel_per_call, + provider.chain_response_max_bytes, + provider.local_store_bytes, + self.clocks.as_ref(), + HostServices::default(), + )?; + kind.install( + ProviderInstance { + component: &provider.component, + linker: &linker, + store, + config: provider.init_config.clone(), + fuel_per_call: provider.fuel_per_call, + liveness: provider.liveness.clone(), + }, + service, + ) + .await + } + /// Count of modules currently alive. A module is not alive when its /// `init` returned `Err` (permanent, never retried) or when `on_event` /// trapped and its restart backoff has not yet elapsed. @@ -1591,28 +1792,38 @@ enum DispatchOutcome { RateLimited, } -/// 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, +/// Push the current trap timestamp into a component's failure-window +/// ring, drop entries older than the policy window, and report the +/// recent-failure count once it crosses `policy.max_failures`. Shared by +/// the module and provider poison sweeps. +fn poison_crossed( + failure_timestamps: &mut std::collections::VecDeque, policy: crate::runtime::poison_policy::PoisonPolicy, - last_error: &str, -) { +) -> Option { let now = std::time::Instant::now(); - // Prune entries outside the window. - while let Some(&front) = module.failure_timestamps.front() { + while let Some(&front) = failure_timestamps.front() { if now.duration_since(front) > policy.window { - module.failure_timestamps.pop_front(); + 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 { + failure_timestamps.push_back(now); + let recent = failure_timestamps.len() as u32; + crate::runtime::poison_policy::should_poison(policy, recent).then_some(recent) +} + +/// Flip `poisoned = true` once the module's failure window crosses the +/// policy threshold. 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, +) { + if let Some(recent) = poison_crossed(&mut module.failure_timestamps, policy) + && !module.poisoned + { module.poisoned = true; warn!( module = %module.name, @@ -1629,6 +1840,20 @@ fn record_failure_and_maybe_poison( } } +/// Slide a failed provider restart's next attempt further out. +fn defer_provider_restart(provider: &mut LoadedProvider, error: &str) { + provider.failure_count = provider.failure_count.saturating_add(1); + let backoff = crate::runtime::restart_policy::backoff_for(provider.failure_count); + provider.next_attempt = Some(std::time::Instant::now() + backoff); + error!( + adapter = %provider.name, + failure_count = provider.failure_count, + backoff_ms = backoff.as_millis() as u64, + error, + "adapter restart failed - will retry after backoff", + ); +} + /// Persisted per-chain progress key; must stay numeric for data compat. fn progress_key(chain: Chain) -> String { format!("last_dispatched_block:{}", chain.id()) diff --git a/crates/nexum-runtime/src/supervisor/tests.rs b/crates/nexum-runtime/src/supervisor/tests.rs index 3830e6f3..0bf06086 100644 --- a/crates/nexum-runtime/src/supervisor/tests.rs +++ b/crates/nexum-runtime/src/supervisor/tests.rs @@ -651,7 +651,11 @@ fn scripted_registry(adapter: ScriptedAdapter) -> crate::host::venue_registry::V ) .build(); registry - .install(crate::host::venue_registry::VenueId::from("cow"), adapter) + .install( + crate::host::venue_registry::VenueId::from("cow"), + crate::host::actor::Liveness::default(), + adapter, + ) .expect("install"); registry } @@ -2959,3 +2963,164 @@ async fn boot_admits_a_venue_adapter_manifest_past_the_kind_gate() { "the kind gate passed rather than rejecting: {msg}", ); } + +// ── venue-adapter trap recovery ─────────────────────────────────────── + +/// Boot one flaky-venue adapter over the mock chain, whose head starts at +/// the fixture's poison sentinel. Returns the chain handle so the test can +/// let the venue recover. +async fn boot_flaky_venue( + adapter_wasm: PathBuf, + limits: crate::engine_config::ModuleLimits, +) -> ( + Supervisor, + crate::test_utils::MockChainProvider, +) { + use crate::engine_config::AdapterEntry; + use crate::host::component::ChainMethod; + + let chain = crate::test_utils::MockChainProvider::new(); + chain.on_method(ChainMethod::EthBlockNumber, "\"0xdead\""); + let components = crate::test_utils::mock_components_from( + chain.clone(), + crate::test_utils::MockStateStore::new(), + ); + let engine = make_wasmtime_engine(); + let linker = crate::supervisor::build_linker::(&engine, &[]) + .expect("build_linker"); + let config = EngineConfig { + adapters: vec![AdapterEntry { + path: adapter_wasm, + manifest: Some(fixture_module_toml( + "modules/fixtures/flaky-venue/module.toml", + )), + http_allow: Vec::new(), + messaging_topics: Vec::new(), + }], + limits, + ..Default::default() + }; + let supervisor = Supervisor::boot(&engine, &linker, &config, &components, &[], None) + .await + .expect("boot"); + (supervisor, chain) +} + +/// A test block that drives the dispatch-time sweeps. +fn sweep_block() -> nexum::host::types::Block { + nexum::host::types::Block { + chain_id: 1, + number: 1, + hash: vec![0; 32], + timestamp: 1_700_000_000_000, + } +} + +/// The full trap-to-recovery lifecycle over a real wasm adapter: a trapped +/// venue is temporarily dead (`unavailable`, not `unknown-venue`) and the +/// provider restart sweep reinstantiates it after backoff, after which a +/// submit succeeds again. +#[tokio::test] +async fn e2e_trapped_adapter_is_swept_and_restarts() { + use crate::bindings::{SubmitOutcome, VenueError}; + use crate::host::component::ChainMethod; + use crate::host::venue_registry::VenueId; + + let Some(wasm) = module_wasm_or_skip("flaky-venue") else { + return; + }; + let (mut supervisor, chain) = + boot_flaky_venue(wasm, crate::engine_config::ModuleLimits::default()).await; + assert_eq!(supervisor.adapter_count(), 1); + assert_eq!(supervisor.adapter_alive_count(), 1, "boots alive"); + let registry = supervisor.venue_registry().expect("registry service"); + let venue = VenueId::from("flaky-venue"); + + // The poison head detonates submit: the guest panic traps the store + // and the shared liveness drops. + let err = registry + .submit("mod-a", &venue, b"body".to_vec()) + .await + .expect_err("the poison head traps the adapter"); + assert!(matches!(err, VenueError::Unavailable(_)), "{err:?}"); + assert_eq!( + supervisor.adapter_alive_count(), + 0, + "the trap drops liveness" + ); + + // Temporarily dead resolves distinctly from never installed. + assert!(matches!( + registry.submit("mod-a", &venue, b"body".to_vec()).await, + Err(VenueError::Unavailable(_)) + )); + assert!(matches!( + registry + .submit("mod-a", &VenueId::from("unlisted"), b"body".to_vec()) + .await, + Err(VenueError::UnknownVenue) + )); + + // The venue recovers; past the 1s backoff the dispatch-time sweep + // reinstalls the adapter on a fresh store. + chain.on_method(ChainMethod::EthBlockNumber, "\"0x1\""); + tokio::time::sleep(std::time::Duration::from_millis(1_200)).await; + supervisor.dispatch_block(sweep_block()).await; + assert_eq!(supervisor.adapter_alive_count(), 1, "the sweep revived it"); + let outcome = registry + .submit("mod-a", &venue, b"body".to_vec()) + .await + .expect("the recovered adapter accepts"); + assert!(matches!(outcome, SubmitOutcome::Accepted(r) if r == b"body")); +} + +/// A crash-looping adapter is quarantined by the provider poison sweep: +/// at the threshold the restarts stop, and the venue stays dead past every +/// backoff until an operator intervenes. +#[tokio::test] +async fn e2e_crash_looping_adapter_is_poisoned() { + use crate::bindings::VenueError; + use crate::engine_config::PoisonLimitsSection; + use crate::host::venue_registry::VenueId; + + let Some(wasm) = module_wasm_or_skip("flaky-venue") else { + return; + }; + let limits = ModuleLimits { + poison: PoisonLimitsSection { + max_failures: Some(2), + window_secs: Some(600), + }, + ..ModuleLimits::default() + }; + // The chain head stays at the poison sentinel for the whole test: every + // submit after a restart traps again. + let (mut supervisor, _chain) = boot_flaky_venue(wasm, limits).await; + let registry = supervisor.venue_registry().expect("registry service"); + let venue = VenueId::from("flaky-venue"); + + // Trap 1, then a successful restart past the 1s backoff. + let _ = registry.submit("mod-a", &venue, b"body".to_vec()).await; + tokio::time::sleep(std::time::Duration::from_millis(1_200)).await; + supervisor.dispatch_block(sweep_block()).await; + assert_eq!(supervisor.adapter_alive_count(), 1, "first restart lands"); + + // Trap 2 crosses the 2-failure threshold: the sweep quarantines the + // adapter instead of scheduling another restart. + let _ = registry.submit("mod-a", &venue, b"body".to_vec()).await; + supervisor.dispatch_block(sweep_block()).await; + assert_eq!(supervisor.adapter_alive_count(), 0, "quarantined"); + + // Past every backoff the poisoned adapter stays dead and unavailable. + tokio::time::sleep(std::time::Duration::from_millis(1_500)).await; + supervisor.dispatch_block(sweep_block()).await; + assert_eq!( + supervisor.adapter_alive_count(), + 0, + "no restart while poisoned" + ); + assert!(matches!( + registry.submit("mod-a", &venue, b"body".to_vec()).await, + Err(VenueError::Unavailable(_)) + )); +} diff --git a/justfile b/justfile index 964bba79..b1311631 100644 --- a/justfile +++ b/justfile @@ -97,6 +97,6 @@ ci: cargo build --release --target wasm32-wasip2 \ -p example -p twap-monitor -p ethflow-watcher -p price-alert \ -p balance-tracker -p stop-loss -p http-probe -p echo-venue \ - -p echo-client -p clock-reader -p flaky-bomb -p fuel-bomb -p memory-bomb \ - -p panic-bomb -p slow-host + -p echo-client -p clock-reader -p flaky-bomb -p flaky-venue -p fuel-bomb \ + -p memory-bomb -p panic-bomb -p slow-host cargo test --workspace --all-features --no-fail-fast diff --git a/modules/fixtures/flaky-venue/Cargo.toml b/modules/fixtures/flaky-venue/Cargo.toml new file mode 100644 index 00000000..5237feb0 --- /dev/null +++ b/modules/fixtures/flaky-venue/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "flaky-venue" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true +description = "Evil-by-design venue adapter fixture: submit panics while the chain head reads as the poison sentinel and accepts once it stops. Drives the supervisor's provider trap-to-recovery and poison sweeps." + +[lints] +workspace = true + +[lib] +crate-type = ["cdylib"] + +[dependencies] +nexum-venue-sdk = { path = "../../../crates/nexum-venue-sdk" } +wit-bindgen = { version = "0.58", default-features = false, features = ["macros", "realloc"] } diff --git a/modules/fixtures/flaky-venue/module.toml b/modules/fixtures/flaky-venue/module.toml new file mode 100644 index 00000000..c7667b9d --- /dev/null +++ b/modules/fixtures/flaky-venue/module.toml @@ -0,0 +1,19 @@ +# flaky-venue adapter manifest - the trap-to-recovery test fixture. +# Declares the chain capability its poison-sentinel read requires. + +[module] +name = "flaky-venue" +version = "0.1.0" +kind = "venue-adapter" +# Placeholder content hash; parsed but not verified in 0.2. +component = "sha256:0000000000000000000000000000000000000000000000000000000000000000" + +[capabilities] +required = ["chain"] +optional = [] + +[capabilities.http] +allow = [] + +[config] +name = "flaky-venue" diff --git a/modules/fixtures/flaky-venue/src/lib.rs b/modules/fixtures/flaky-venue/src/lib.rs new file mode 100644 index 00000000..af1c9d9b --- /dev/null +++ b/modules/fixtures/flaky-venue/src/lib.rs @@ -0,0 +1,76 @@ +//! # flaky-venue (test fixture) +//! +//! A venue adapter whose `submit` panics (traps the store) while the chain +//! head reads as the poison sentinel `0xdead`, and accepts once the head +//! moves on. The controlling test programs the mock chain backend, so the +//! trap is deterministic and the recovery is host-driven: the supervisor's +//! sweep must reinstantiate the adapter before a submit succeeds again. +//! +//! Not a production adapter. Lives under `modules/fixtures/` so it is +//! obviously test-only. + +// wit_bindgen::generate! expands to host-import shims whose arity matches +// the WIT signatures, which can exceed clippy's too-many-arguments threshold. +#![cfg_attr(not(test), warn(unused_crate_dependencies))] +#![allow(clippy::too_many_arguments)] + +use nexum::host::chain; +use videre::types::types::{ + AuthScheme, IntentHeader, IntentStatus, Quotation, Settlement, SubmitOutcome, VenueError, +}; +use videre::value_flow::types::{Asset, AssetAmount}; + +/// The chain-head response that detonates `submit`. +const POISON_HEAD: &str = "0xdead"; + +struct FlakyVenue; + +#[nexum_venue_sdk::venue] +impl FlakyVenue { + fn init(_config: Config) -> Result<(), Fault> { + Ok(()) + } + + fn derive_header(_body: Vec) -> Result { + Ok(IntentHeader { + gives: zero_native(), + wants: zero_native(), + settlement: Settlement { chain: 1 }, + authorisation: AuthScheme::Eip1271, + }) + } + + fn quote(_body: Vec) -> Result { + Ok(Quotation { + gives: zero_native(), + wants: zero_native(), + fee: zero_native(), + valid_until_ms: u64::MAX, + }) + } + + fn submit(body: Vec) -> Result { + let head = chain::request(1, "eth_blockNumber", "[]") + .map_err(|_| VenueError::Unavailable("chain read failed".into()))?; + // The sentinel detonates the fixture: a guest panic traps the + // store, which is what the sweep under test must recover from. + assert!(!head.contains(POISON_HEAD), "flaky-venue poison head"); + Ok(SubmitOutcome::Accepted(body)) + } + + fn status(_receipt: Vec) -> Result { + Ok(IntentStatus::Open) + } + + fn cancel(_receipt: Vec) -> Result<(), VenueError> { + Ok(()) + } +} + +/// A zero native amount. +fn zero_native() -> AssetAmount { + AssetAmount { + asset: Asset::Native, + amount: Vec::new(), + } +}