diff --git a/crates/nexum-runtime/src/supervisor/lifecycle.rs b/crates/nexum-runtime/src/supervisor/lifecycle.rs index 2c903ba..8fbcfd9 100644 --- a/crates/nexum-runtime/src/supervisor/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/lifecycle.rs @@ -3,18 +3,16 @@ use std::collections::VecDeque; use std::time::{Duration, Instant}; -use anyhow::{Context, Error, Result, anyhow}; +use anyhow::{Result, anyhow}; use super::Shared; -use super::load::{LoadedModule, LoadedProvider, run_init}; +use super::load::{LoadedModule, LoadedProvider, install_provider, instantiate_module}; use super::role::{Role, report_restart_attempt, report_restart_outcome, report_trap}; -use super::store::{self, build_linker, build_provider_linker}; -use crate::bindings::EventModule; +use super::store::{build_linker, fresh_run_store}; use crate::digest::ContentDigest; use crate::host::actor::Liveness; use crate::host::component::RuntimeTypes; -use crate::host::extension::{HostServices, Installed, ProviderInstance}; -use crate::host::logs::RunId; +use crate::host::extension::Installed; use crate::module_id::ModuleId; use crate::runtime::poison_policy::{PoisonPolicy, should_poison}; use crate::runtime::restart_policy::backoff_for; @@ -69,6 +67,11 @@ impl Health { } } + /// The boot verdict: a failed `init` loads the item dead, permanently. + pub(super) fn from_init(ok: bool) -> Self { + if ok { Self::alive() } else { Self::dead() } + } + pub(super) fn dispatchable(&self) -> bool { matches!(self.state, LifecycleState::Alive) } @@ -193,35 +196,22 @@ impl Sweepable for LoadedModule { // Must match the boot-time linker: core interfaces plus every extension hook. let linker = build_linker::(&shared.engine, &shared.extensions)?; // A restart is a new run; the dead run's logs stay readable until evicted. - let run = RunId::new(self.name.clone(), self.live.run.seq + 1); - let mut store = store::build( + let (run, mut store) = fresh_run_store( shared, + &self.name, + self.live.run.seq + 1, &self.seed.spec, - run.clone(), - shared.services.clone(), + Role::Module, )?; - let bindings = - EventModule::instantiate_async(&mut store, &self.seed.artifact.component, &linker) - .await - .map_err(Error::from) - .with_context(|| format!("reinstantiate {}", self.name))?; + let (bindings, init) = + instantiate_module(&linker, &self.seed, &self.name, &mut store).await?; // An init fault defers the restart; only at boot is it permanent. - match run_init( - &bindings, - &mut store, - &self.seed.artifact.init_config, - self.seed.event_deadline, - ) - .await? - { - Ok(()) => {} - Err(e) => { - return Err(anyhow!( - "init returned fault on restart: {} ({})", - crate::host::error::fault_message(&e), - crate::host::error::fault_label(&e), - )); - } + if let Err(e) = init { + return Err(anyhow!( + "init returned fault on restart: {} ({})", + crate::host::error::fault_message(&e), + crate::host::error::fault_label(&e), + )); } self.live.bindings = bindings; self.live.store = store; @@ -257,32 +247,26 @@ impl Sweepable for LoadedProvider { /// Run and liveness commit only on a live install. async fn revive(&mut self, shared: &Shared) -> Result<()> { - let (kind, service) = shared + let row = shared .kinds .get(self.kind) .ok_or_else(|| anyhow!("provider kind {} is not registered", self.kind))?; - let linker = build_provider_linker::(&shared.engine, kind.as_ref())?; - let run = RunId::new(self.name.clone(), self.run.seq + 1); - let store = store::build( + let (run, store) = fresh_run_store( shared, + &self.name, + self.run.seq + 1, &self.seed.spec, - run.clone(), - HostServices::default(), + Role::Adapter, )?; - match kind - .install( - ProviderInstance { - component: &self.seed.artifact.component, - linker: &linker, - store, - config: self.seed.artifact.init_config.clone(), - sections: &self.sections, - fuel_per_call: self.seed.spec.fuel, - liveness: self.liveness.clone(), - }, - service, - ) - .await? + match install_provider( + shared, + row, + &self.seed, + &self.sections, + store, + self.liveness.clone(), + ) + .await? { Installed::Live => { self.run = run; diff --git a/crates/nexum-runtime/src/supervisor/load.rs b/crates/nexum-runtime/src/supervisor/load.rs index 6cdfa1d..a1dccdf 100644 --- a/crates/nexum-runtime/src/supervisor/load.rs +++ b/crates/nexum-runtime/src/supervisor/load.rs @@ -15,8 +15,10 @@ use super::artifact::read_verified_component; use super::dispatch::with_dispatch_deadline; use super::lifecycle::Health; use super::prepass::manifest_namespace; +use super::role::Role; use super::store::{ - self, HostStore, ResolvedLimits, StoreSpec, build_provider_linker, resolve_module_limits, + HostStore, ResolvedLimits, StoreSpec, build_provider_linker, fresh_run_store, + resolve_module_limits, }; use crate::bindings::nexum::host::types::Fault; use crate::bindings::{Config, EventModule}; @@ -24,7 +26,7 @@ use crate::digest::ContentDigest; use crate::engine_config::{AdapterEntry, ModuleEntry, ModuleLimits}; use crate::host::actor::Liveness; use crate::host::component::RuntimeTypes; -use crate::host::extension::{HostServices, Installed, ProviderInstance, ProviderManifest}; +use crate::host::extension::{Installed, ProviderInstance, ProviderManifest}; use crate::host::logs::RunId; use crate::host::state::HostState; use crate::manifest::{self, CapabilityRegistry, ComponentKind, LoadedManifest, Subscription}; @@ -40,18 +42,33 @@ pub(super) struct CachedArtifact { pub(super) init_config: Config, } -/// Everything needed to rebuild a module's store and re-run `init`. -pub(super) struct ModuleSeed { +/// Everything needed to rebuild a store and re-run `init` or reinstall. +pub(super) struct Seed { pub(super) artifact: CachedArtifact, pub(super) spec: StoreSpec, /// Wall-clock bound on a whole dispatch, host calls included. pub(super) event_deadline: Duration, } -/// Everything needed to rebuild a provider's store and reinstall it. -pub(super) struct ProviderSeed { - pub(super) artifact: CachedArtifact, - pub(super) spec: StoreSpec, +impl Seed { + /// The borrow of the cached component ends when `install` returns. + pub(super) fn instance<'a, T: RuntimeTypes>( + &'a self, + linker: &'a Linker>, + sections: &'a manifest::ExtensionSections, + store: HostStore, + liveness: Liveness, + ) -> ProviderInstance<'a, T> { + ProviderInstance { + component: &self.artifact.component, + linker, + store, + config: self.artifact.init_config.clone(), + sections, + fuel_per_call: self.spec.fuel, + liveness, + } + } } /// Restarts replace bindings, store, and run; the rate bucket carries across. @@ -65,7 +82,7 @@ pub(super) struct LiveInstance { pub(super) struct LoadedModule { pub(super) name: ModuleId, pub(super) live: LiveInstance, - pub(super) seed: ModuleSeed, + pub(super) seed: Seed, pub(super) subscriptions: Vec, pub(super) health: Health, } @@ -75,7 +92,7 @@ pub(super) struct LoadedProvider { pub(super) name: ModuleId, pub(super) kind: &'static str, pub(super) sections: manifest::ExtensionSections, - pub(super) seed: ProviderSeed, + pub(super) seed: Seed, /// Trap signal shared with the installed actor; feeds `health` at /// sweep time and carries no lifecycle authority of its own. pub(super) liveness: Liveness, @@ -129,7 +146,7 @@ fn default_init_config(config: &Config, namespace: &str) -> Config { /// Runs under the dispatch deadline so a hung host call cannot park boot or a /// restart; a deadline hit or trap is `Err`, a guest fault `Ok(Err(fault))`. -pub(super) async fn run_init( +async fn run_init( bindings: &EventModule, store: &mut HostStore, config: &Config, @@ -141,6 +158,45 @@ pub(super) async fn run_init( .map_err(Error::from) } +/// Instantiates the cached component on a fresh store and runs `init`; what +/// a guest init fault means (dead at boot, deferred on restart) stays with +/// the caller. +pub(super) async fn instantiate_module( + linker: &Linker>, + seed: &Seed, + name: &ModuleId, + store: &mut HostStore, +) -> Result<(EventModule, Result<(), Fault>)> { + let bindings = EventModule::instantiate_async(&mut *store, &seed.artifact.component, linker) + .await + .map_err(Error::from) + .with_context(|| format!("instantiate {name}"))?; + let init = run_init( + &bindings, + store, + &seed.artifact.init_config, + seed.event_deadline, + ) + .await?; + Ok((bindings, init)) +} + +/// Builds the kind's linker and installs on the given store; a `Dead` +/// verdict carries no error, its meaning stays with the caller. +pub(super) async fn install_provider( + shared: &Shared, + row: &ProviderRow, + seed: &Seed, + sections: &manifest::ExtensionSections, + store: HostStore, + liveness: Liveness, +) -> Result { + let (kind, service) = row; + let linker = build_provider_linker::(&shared.engine, kind.as_ref())?; + kind.install(seed.instance(&linker, sections, store, liveness), service) + .await +} + /// A failed `init` loads the module dead; the dispatcher skips it. pub(super) async fn module( shared: &Shared, @@ -193,34 +249,34 @@ pub(super) async fn module( chain_response_max_bytes: limits_cfg.chain_response_max_bytes(), state_quota: state_bytes, }; - let run = RunId::new(module_namespace.clone(), 0); - let mut store = store::build(shared, &spec, run.clone(), shared.services.clone())?; - let bindings = EventModule::instantiate_async(&mut store, &component, linker) - .await - .map_err(Error::from) - .with_context(|| format!("instantiate {}", entry.path.display()))?; - let config = default_init_config(&loaded_manifest.config, module_namespace.as_str()); + let seed = Seed { + artifact: CachedArtifact { + component, + digest, + init_config: config, + }, + spec, + event_deadline: limits_cfg.event_deadline(), + }; + let (run, mut store) = fresh_run_store(shared, &module_namespace, 0, &seed.spec, Role::Module)?; + let (bindings, init) = instantiate_module(linker, &seed, &module_namespace, &mut store).await?; // A failed `init` leaves guest state uninitialised, so the module loads dead. - let init_succeeded = - match run_init(&bindings, &mut store, &config, limits_cfg.event_deadline()).await? { - Ok(()) => { - info!(module = %module_namespace, "init succeeded"); - true - } - Err(e) => { - warn!( - module = %module_namespace, - kind = crate::host::error::fault_label(&e), - message = %crate::host::error::fault_message(&e), - "init failed - module loaded but marked dead; dispatcher will skip it", - ); - false - } - }; - // Refuel after init so the first on_event starts with a full budget. - store.set_fuel(fuel)?; - + let init_succeeded = match init { + Ok(()) => { + info!(module = %module_namespace, "init succeeded"); + true + } + Err(e) => { + warn!( + module = %module_namespace, + kind = crate::host::error::fault_label(&e), + message = %crate::host::error::fault_message(&e), + "init failed - module loaded but marked dead; dispatcher will skip it", + ); + false + } + }; // Unserviceable subscriptions warn; an undeclared extension kind refuses. let extension_kinds = extension_subscription_vocabulary(&shared.extensions); for sub in &loaded_manifest.manifest.subscriptions { @@ -247,21 +303,9 @@ pub(super) async fn module( run, dispatch_bucket: TokenBucket::new(limits_cfg.dispatch_rate(), Instant::now()), }, - seed: ModuleSeed { - artifact: CachedArtifact { - component, - digest, - init_config: config, - }, - spec, - event_deadline: limits_cfg.event_deadline(), - }, + seed, subscriptions: loaded_manifest.manifest.subscriptions.clone(), - health: if init_succeeded { - Health::alive() - } else { - Health::dead() - }, + health: Health::from_init(init_succeeded), }) } @@ -278,7 +322,7 @@ pub(super) async fn provider( // import fails after compile; the linker withholds the core interfaces. let registry = CapabilityRegistry::provider(); let sections = loaded_manifest.manifest.extensions.clone(); - let ((kind, service), component, digest) = admit_and_verify( + let (row, component, digest) = admit_and_verify( shared, namespace.as_str(), &entry.path, @@ -291,7 +335,7 @@ pub(super) async fn provider( .with_context(|| format!("install refused for {}", entry.path.display()))?; } // An unregistered kind refuses before compile. - let (kind, service): &ProviderRow = match &loaded_manifest.manifest.module.kind { + let row: &ProviderRow = match &loaded_manifest.manifest.module.kind { ComponentKind::Worker => { return Err(anyhow!( "{} declares the worker kind; an [[adapters]] entry requires a \ @@ -313,12 +357,13 @@ pub(super) async fn provider( }; info!( component = %entry.path.display(), - kind = kind.kind(), + kind = row.0.kind(), "compiling provider component", ); - Ok((kind, service)) + Ok(row) }, )?; + let kind = row.0.as_ref(); info!( provider = %namespace, @@ -330,7 +375,6 @@ pub(super) async fn provider( "applied provider resource limits and transport scope", ); - let linker = build_provider_linker::(&shared.engine, kind.as_ref())?; let spec = StoreSpec { http_allowlist: entry.http_allow.clone(), http_limits: limits_cfg.http(), @@ -340,28 +384,22 @@ pub(super) async fn provider( chain_response_max_bytes: limits_cfg.chain_response_max_bytes(), state_quota: limits_cfg.state_bytes(), }; - let run = RunId::new(namespace.clone(), 0); - // The store carries an empty service map: the shared map holds the - // registry that owns this store, and carrying it here would cycle. - let store = store::build(shared, &spec, run.clone(), HostServices::default())?; - let config = default_init_config(&loaded_manifest.config, namespace.as_str()); + let seed = Seed { + artifact: CachedArtifact { + component, + digest, + init_config: config, + }, + spec, + event_deadline: limits_cfg.event_deadline(), + }; let liveness = Liveness::default(); - let installed = kind - .install( - ProviderInstance { - component: &component, - linker: &linker, - store, - config: config.clone(), - sections: §ions, - fuel_per_call: limits_cfg.fuel(), - liveness: liveness.clone(), - }, - service, - ) + let (run, store) = fresh_run_store(shared, &namespace, 0, &seed.spec, Role::Adapter)?; + let installed = install_provider(shared, row, &seed, §ions, store, liveness.clone()) .await .with_context(|| format!("install {}", entry.path.display()))?; + // A dead install at boot is permanent; the liveness records it for the sweep. if installed == Installed::Dead { liveness.mark_dead(); } @@ -369,20 +407,9 @@ pub(super) async fn provider( name: namespace, kind: kind.kind(), sections, - seed: ProviderSeed { - artifact: CachedArtifact { - component, - digest, - init_config: config, - }, - spec, - }, + seed, liveness, run, - health: if installed == Installed::Live { - Health::alive() - } else { - Health::dead() - }, + health: Health::from_init(installed == Installed::Live), }) } diff --git a/crates/nexum-runtime/src/supervisor/store.rs b/crates/nexum-runtime/src/supervisor/store.rs index c30b040..548b7f1 100644 --- a/crates/nexum-runtime/src/supervisor/store.rs +++ b/crates/nexum-runtime/src/supervisor/store.rs @@ -8,6 +8,7 @@ use wasmtime::{Engine, Store}; use wasmtime_wasi::{HostMonotonicClock, HostWallClock, WasiCtxBuilder}; use super::Shared; +use super::role::Role; use crate::bindings::EventModule; use crate::engine_config::{ModuleLimits, OutboundHttpLimits}; use crate::host::component::{RuntimeTypes, StateHandle, StateStore}; @@ -16,6 +17,7 @@ use crate::host::http::HttpGate; use crate::host::logs::{LogSource, RunId, StdioStream}; use crate::host::state::HostState; use crate::manifest::ResourceSection; +use crate::module_id::ModuleId; pub(super) type HostStore = Store>; @@ -98,14 +100,32 @@ pub(super) struct StoreSpec { pub(super) state_quota: u64, } -/// Takes a freshly minted [`RunId`]; `services` is the module map, empty -/// for a provider store. -pub(super) fn build( +/// Mints the run identity for `name` at `seq` and builds its store. +pub(super) fn fresh_run_store( + shared: &Shared, + name: &ModuleId, + seq: u64, + spec: &StoreSpec, + role: Role, +) -> Result<(RunId, HostStore)> { + let run = RunId::new(name.clone(), seq); + let store = build(shared, spec, run.clone(), role)?; + Ok((run, store)) +} + +/// Takes a freshly minted [`RunId`]; `role` picks the service map. +fn build( shared: &Shared, spec: &StoreSpec, run: RunId, - services: HostServices, + role: Role, ) -> Result> { + // A provider store carries an empty service map: the shared map holds + // the registry that owns this store, and carrying it here would cycle. + let services = match role { + Role::Module => shared.services.clone(), + Role::Adapter => HostServices::default(), + }; let namespace: &str = run.module.as_str(); // Stdio is captured as tagged log records, stdin stays closed; the ctx // grants no network, so the allowlisted wasi:http gate is the only live path. diff --git a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs index 8d46617..2c9e28b 100644 --- a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs @@ -1,25 +1,35 @@ //! Lifecycle: init failure, traps, restart backoff, and poison quarantine. use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; use super::*; use crate::supervisor::lifecycle::sweep; -/// price-alert loads cleanly, then `init` rejects the unparseable +/// price-alert's `[config]`; `not-a-number` makes `init` reject the /// `threshold` with `fault.invalid-input`. -fn bad_threshold_price_alert() -> TestManifest { - TestManifest::new("price-alert") +fn price_alert_config(threshold: &str) -> crate::bindings::Config { + vec![ + ( + "oracle_address".into(), + "0x694AA1769357215DE4FAC081bf1f309aDC325306".into(), + ), + ("decimals".into(), "8".into()), + ("threshold".into(), threshold.to_owned()), + ("direction".into(), "below".into()), + ("every_n_blocks".into(), "1".into()), + ] +} + +fn price_alert(threshold: &str) -> TestManifest { + let mut manifest = TestManifest::new("price-alert") .cap("logging") .cap("chain") - .block_sub(SEPOLIA) - .config( - "oracle_address", - "0x694AA1769357215DE4FAC081bf1f309aDC325306", - ) - .config("decimals", "8") - .config("threshold", "not-a-number") - .config("direction", "below") - .config("every_n_blocks", "1") + .block_sub(SEPOLIA); + for (key, value) in price_alert_config(threshold) { + manifest = manifest.config(key, value); + } + manifest } /// Loaded but dead: no dispatch, no chain-facing subscription, and the @@ -33,7 +43,7 @@ async fn init_failure_marks_module_dead_excluding_dispatch_and_subscriptions() { // paths are exercised. let mut booted = BootScenario::new() .wasm(wasm) - .module(bad_threshold_price_alert().chain_log_sub_filtered( + .module(price_alert("not-a-number").chain_log_sub_filtered( SEPOLIA, Some("0xbA3cB449bD2B4ADddBc894D8697F5170800EAdeC"), Some("0xcf5f9de2984132265203b5c335b25727702ca77262ff622e136baa7362bf1da9"), @@ -77,7 +87,7 @@ async fn alive_module_subscriptions_survive_alongside_dead_module() { return; }; let booted = BootScenario::new() - .module(Entry::new(bad_threshold_price_alert()).wasm(price_alert_wasm)) + .module(Entry::new(price_alert("not-a-number")).wasm(price_alert_wasm)) .module( Entry::new(TestManifest::new("example").cap("logging").block_sub(1)).wasm(example_wasm), ) @@ -103,6 +113,69 @@ async fn alive_module_subscriptions_survive_alongside_dead_module() { ); } +/// Boot and restart share one instantiate-and-init helper, so the verdict on +/// the identical fault is the call sites' alone: dead forever, or deferred. +#[tokio::test] +async fn the_same_init_fault_kills_at_boot_and_only_defers_on_restart() { + let Some(wasm) = module_wasm_or_skip("price-alert") else { + return; + }; + + let booted = BootScenario::new() + .wasm(wasm.clone()) + .module(price_alert("not-a-number")) + .boot() + .await + .expect("the module loads; only init fails"); + let dead = &booted.supervisor.modules[0]; + assert!(!dead.health.dispatchable(), "a boot init fault loads dead"); + assert!( + !dead + .health + .due_restart(Instant::now() + Duration::from_secs(3600)), + "a boot init fault schedules no restart, ever", + ); + + let mut booted = BootScenario::new() + .wasm(wasm) + .module(price_alert("2500.50")) + .boot() + .await + .expect("boot"); + assert!( + booted.supervisor.modules[0].health.dispatchable(), + "a parseable threshold loads alive", + ); + // The revive re-runs `init` off the seed, so swapping the seed's config + // reaches the restart path and nothing else. + booted.supervisor.modules[0].seed.artifact.init_config = price_alert_config("not-a-number"); + let policy = booted.supervisor.policy; + let died_at = Instant::now(); + booted.supervisor.modules[0] + .health + .record_trap(died_at, died_at, policy); + let due = died_at + Duration::from_secs(5); + sweep( + &booted.supervisor.shared, + std::slice::from_mut(&mut booted.supervisor.modules[0]), + policy, + due, + ) + .await; + + let module = &booted.supervisor.modules[0]; + assert_eq!(module.live.run.seq, 0, "a failed revive commits no run"); + assert_eq!( + module.health.failure_count(), + 2, + "one recorded trap plus one deferred restart", + ); + assert!( + module.health.due_restart(due + Duration::from_secs(3600)), + "a restart init fault defers instead of killing", + ); +} + /// The host catches the bomb's trap without panicking, marks the module /// dead, and never re-enters it. async fn bomb_traps_and_marks_module_dead(module: &str) { @@ -391,7 +464,7 @@ fn scripted_provider(engine: &wasmtime::Engine) -> crate::supervisor::load::Load name: "scripted".into(), kind: "scripted-adapter", sections: manifest::ExtensionSections::default(), - seed: crate::supervisor::load::ProviderSeed { + seed: crate::supervisor::load::Seed { artifact: crate::supervisor::load::CachedArtifact { component: wasmtime::component::Component::new(engine, EMPTY_COMPONENT) .expect("empty component"), @@ -407,6 +480,7 @@ fn scripted_provider(engine: &wasmtime::Engine) -> crate::supervisor::load::Load chain_response_max_bytes: limits.chain_response_max_bytes(), state_quota: limits.state_bytes(), }, + event_deadline: limits.event_deadline(), }, liveness: crate::host::actor::Liveness::default(), run: crate::host::logs::RunId::new("scripted", 0),