From bfc9a784e859ee26274e1c699902448190c0d214 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:05:37 +0000 Subject: [PATCH 1/6] refactor(supervisor): promote Role to its own module and drop the dead name fallbacks AI Assistance: Claude Fable 5 used for implementation and tests --- .../nexum-runtime/src/supervisor/lifecycle.rs | 44 +----------- crates/nexum-runtime/src/supervisor/load.rs | 7 +- crates/nexum-runtime/src/supervisor/mod.rs | 7 +- .../nexum-runtime/src/supervisor/prepass.rs | 40 ++++------- crates/nexum-runtime/src/supervisor/role.rs | 68 +++++++++++++++++++ 5 files changed, 87 insertions(+), 79 deletions(-) create mode 100644 crates/nexum-runtime/src/supervisor/role.rs diff --git a/crates/nexum-runtime/src/supervisor/lifecycle.rs b/crates/nexum-runtime/src/supervisor/lifecycle.rs index ceb22be..273b278 100644 --- a/crates/nexum-runtime/src/supervisor/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/lifecycle.rs @@ -8,6 +8,7 @@ use tracing::{error, info, warn}; use super::Shared; use super::load::{LoadedModule, LoadedProvider, run_init}; +use super::role::Role; use super::store::{self, build_linker, build_provider_linker}; use crate::bindings::EventModule; use crate::digest::ContentDigest; @@ -146,49 +147,6 @@ impl Health { } } -/// Keys the per-role metric names and the compile-time tracing field keys. -#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr)] -pub(super) enum Role { - Module, - Adapter, -} - -impl Role { - const fn label(self) -> &'static str { - match self { - Self::Module => "module", - Self::Adapter => "adapter", - } - } - - const fn errors_total(self) -> &'static str { - match self { - Self::Module => "nexum_runtime_module_errors_total", - Self::Adapter => "nexum_runtime_adapter_errors_total", - } - } - - const fn restarts_total(self) -> &'static str { - match self { - Self::Module => "nexum_runtime_module_restarts_total", - Self::Adapter => "nexum_runtime_adapter_restarts_total", - } - } - - const fn poisoned_gauge(self) -> &'static str { - match self { - Self::Module => "nexum_runtime_module_poisoned", - Self::Adapter => "nexum_runtime_adapter_poisoned", - } - } - - /// A provider reinstall is a fresh instance, so its curve resets; a - /// module recovers in place and keeps climbing. - const fn resets_failure_count(self) -> bool { - matches!(self, Self::Adapter) - } -} - /// Run identity mints and commits only inside a successful [`Sweepable::revive`]; /// a failed attempt never advances the run sequence. pub(super) trait Sweepable { diff --git a/crates/nexum-runtime/src/supervisor/load.rs b/crates/nexum-runtime/src/supervisor/load.rs index 6f167d3..6cdfa1d 100644 --- a/crates/nexum-runtime/src/supervisor/load.rs +++ b/crates/nexum-runtime/src/supervisor/load.rs @@ -14,7 +14,7 @@ use super::admission::{ use super::artifact::read_verified_component; use super::dispatch::with_dispatch_deadline; use super::lifecycle::Health; -use super::prepass::{MODULE_FALLBACK_NAME, PROVIDER_FALLBACK_NAME, manifest_namespace}; +use super::prepass::manifest_namespace; use super::store::{ self, HostStore, ResolvedLimits, StoreSpec, build_provider_linker, resolve_module_limits, }; @@ -151,8 +151,7 @@ pub(super) async fn module( require_component_digest: bool, provider_manifests: &[ProviderManifest], ) -> Result> { - let module_namespace: ModuleId = - manifest_namespace(&loaded_manifest, MODULE_FALLBACK_NAME).into(); + let module_namespace: ModuleId = manifest_namespace(&loaded_manifest).into(); let registry = capability_registry(&shared.extensions); let sections = &loaded_manifest.manifest.extensions; let ((), component, digest) = admit_and_verify( @@ -274,7 +273,7 @@ pub(super) async fn provider( limits_cfg: &ModuleLimits, require_component_digest: bool, ) -> Result { - let namespace: ModuleId = manifest_namespace(&loaded_manifest, PROVIDER_FALLBACK_NAME).into(); + let namespace: ModuleId = manifest_namespace(&loaded_manifest).into(); // A core-only declaration fails at manifest load; an undeclared gated // import fails after compile; the linker withholds the core interfaces. let registry = CapabilityRegistry::provider(); diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index 032d5d3..007a41a 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -8,6 +8,7 @@ mod dispatch; mod lifecycle; mod load; mod prepass; +mod role; mod store; mod subscriptions; @@ -30,9 +31,7 @@ use crate::runtime::poison_policy::PoisonPolicy; use admission::{ProviderKinds, capability_registry, enforce_extension_uniqueness, provider_kinds}; use cursors::ChainLogCursors; use load::{LoadedModule, LoadedProvider}; -use prepass::{ - MODULE_FALLBACK_NAME, enforce_configured_chains, load_required_manifest, manifest_namespace, -}; +use prepass::{enforce_configured_chains, load_required_manifest, manifest_namespace}; /// Owns every loaded module and provider and exposes the dispatch surface. pub struct Supervisor { @@ -123,7 +122,7 @@ impl Supervisor { let loaded_manifest = load_required_manifest(&entry.path, entry.manifest.as_deref(), ®istry, "module")?; enforce_configured_chains( - &manifest_namespace(&loaded_manifest, MODULE_FALLBACK_NAME), + &manifest_namespace(&loaded_manifest), &loaded_manifest, &env.configured_chains, )?; diff --git a/crates/nexum-runtime/src/supervisor/prepass.rs b/crates/nexum-runtime/src/supervisor/prepass.rs index fe95606..b4ee82f 100644 --- a/crates/nexum-runtime/src/supervisor/prepass.rs +++ b/crates/nexum-runtime/src/supervisor/prepass.rs @@ -8,6 +8,7 @@ use alloy_chains::Chain; use anyhow::{Context, Error, Result, anyhow}; use tracing::{info, warn}; +use super::role::Role; use crate::engine_config::EngineConfig; use crate::manifest::{self, CapabilityRegistry, LoadedManifest, Subscription}; @@ -33,17 +34,9 @@ pub(super) fn claim_namespace( Ok(()) } -pub(super) const MODULE_FALLBACK_NAME: &str = "module"; - -pub(super) const PROVIDER_FALLBACK_NAME: &str = "provider"; - -/// `[module].name`, or `fallback` when it is empty. -pub(super) fn manifest_namespace(loaded: &LoadedManifest, fallback: &str) -> String { - if loaded.manifest.module.name.is_empty() { - fallback.to_owned() - } else { - loaded.manifest.module.name.clone() - } +/// `[module].name`; manifest parse already refused a blank one. +pub(super) fn manifest_namespace(loaded: &LoadedManifest) -> String { + loaded.manifest.module.name.clone() } /// Missing or unresolved refuses the boot. @@ -180,10 +173,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R .map(|e| (&e.path, e.manifest.as_deref())), &provider_registry, RolePass { - manifest_role: "provider", - claim_role: "adapter", - context: "load provider", - fallback: PROVIDER_FALLBACK_NAME, + role: Role::Adapter, chains: None, }, &mut ledger, @@ -195,10 +185,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R .map(|e| (&e.path, e.manifest.as_deref())), registry, RolePass { - manifest_role: "module", - claim_role: "module", - context: "load module", - fallback: MODULE_FALLBACK_NAME, + role: Role::Module, chains: Some(&configured_chains), }, &mut ledger, @@ -210,10 +197,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R } struct RolePass<'a> { - manifest_role: &'static str, - claim_role: &'static str, - context: &'static str, - fallback: &'static str, + role: Role, chains: Option<&'a ConfiguredChains>, } @@ -226,13 +210,13 @@ fn load_role_manifests<'a>( ) -> Result> { let mut manifests = Vec::new(); for (path, explicit) in entries { - let loaded = load_required_manifest(path, explicit, registry, pass.manifest_role) - .with_context(|| format!("{} {}", pass.context, path.display()))?; - let namespace = manifest_namespace(&loaded, pass.fallback); - claim_namespace(ledger, &namespace, pass.claim_role, path)?; + let loaded = load_required_manifest(path, explicit, registry, pass.role.manifest_role()) + .with_context(|| format!("{} {}", pass.role.load_context(), path.display()))?; + let namespace = manifest_namespace(&loaded); + claim_namespace(ledger, &namespace, pass.role.claim_role(), path)?; if let Some(chains) = pass.chains { enforce_configured_chains(&namespace, &loaded, chains) - .with_context(|| format!("{} {}", pass.context, path.display()))?; + .with_context(|| format!("{} {}", pass.role.load_context(), path.display()))?; } manifests.push(loaded); } diff --git a/crates/nexum-runtime/src/supervisor/role.rs b/crates/nexum-runtime/src/supervisor/role.rs new file mode 100644 index 0000000..b719eb1 --- /dev/null +++ b/crates/nexum-runtime/src/supervisor/role.rs @@ -0,0 +1,68 @@ +//! One role vocabulary for the load pass, the namespace ledger, the +//! metric names, and the compile-time tracing field keys. + +/// Keys the per-role metric names and the compile-time tracing field keys. +#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr)] +pub(super) enum Role { + Module, + Adapter, +} + +impl Role { + pub(super) const fn label(self) -> &'static str { + match self { + Self::Module => "module", + Self::Adapter => "adapter", + } + } + + /// The manifest-facing spelling: an adapter entry loads a provider manifest. + pub(super) const fn manifest_role(self) -> &'static str { + match self { + Self::Module => "module", + Self::Adapter => "provider", + } + } + + /// The ledger spelling: `engine.toml` names the section `[[adapters]]`. + pub(super) const fn claim_role(self) -> &'static str { + match self { + Self::Module => "module", + Self::Adapter => "adapter", + } + } + + pub(super) const fn load_context(self) -> &'static str { + match self { + Self::Module => "load module", + Self::Adapter => "load provider", + } + } + + pub(super) const fn errors_total(self) -> &'static str { + match self { + Self::Module => "nexum_runtime_module_errors_total", + Self::Adapter => "nexum_runtime_adapter_errors_total", + } + } + + pub(super) const fn restarts_total(self) -> &'static str { + match self { + Self::Module => "nexum_runtime_module_restarts_total", + Self::Adapter => "nexum_runtime_adapter_restarts_total", + } + } + + pub(super) const fn poisoned_gauge(self) -> &'static str { + match self { + Self::Module => "nexum_runtime_module_poisoned", + Self::Adapter => "nexum_runtime_adapter_poisoned", + } + } + + /// A provider reinstall is a fresh instance, so its curve resets; a + /// module recovers in place and keeps climbing. + pub(super) const fn resets_failure_count(self) -> bool { + matches!(self, Self::Adapter) + } +} From 61a90096de12a34cbe236cddf089c44651f35db7 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:06:20 +0000 Subject: [PATCH 2/6] refactor(supervisor): fold the provider and module load loops into one AI Assistance: Claude Fable 5 used for implementation and tests --- crates/nexum-runtime/src/supervisor/mod.rs | 99 +++++++++++----------- 1 file changed, 49 insertions(+), 50 deletions(-) diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index 007a41a..6b00152 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -32,6 +32,7 @@ use admission::{ProviderKinds, capability_registry, enforce_extension_uniqueness use cursors::ChainLogCursors; use load::{LoadedModule, LoadedProvider}; use prepass::{enforce_configured_chains, load_required_manifest, manifest_namespace}; +use role::Role; /// Owns every loaded module and provider and exposes the dispatch surface. pub struct Supervisor { @@ -88,14 +89,41 @@ impl Supervisor { let prepass = prepass::run(engine_cfg, ®istry)?; // Providers boot first, so every module store built after already // routes to the installed instances. - let providers = load_providers(&shared, engine_cfg, prepass.adapter_manifests).await?; + let providers = load_role( + &engine_cfg.adapters, + prepass.adapter_manifests, + Role::Adapter, + |e| &e.path, + async |entry, manifest| { + load::provider( + &shared, + entry, + manifest, + &engine_cfg.limits, + engine_cfg.engine.require_component_digest, + ) + .await + }, + ) + .await?; let provider_manifests = project_manifests(&providers); - let modules = load_modules( - &shared, - linker, - engine_cfg, + let modules = load_role( + &engine_cfg.modules, prepass.module_manifests, - &provider_manifests, + Role::Module, + |e| &e.path, + async |entry, manifest| { + load::module( + &shared, + linker, + entry, + manifest, + &engine_cfg.limits, + engine_cfg.engine.require_component_digest, + &provider_manifests, + ) + .await + }, ) .await?; Ok(assemble( @@ -209,26 +237,23 @@ fn wire_extensions( }) } -/// Load every `[[adapters]]` entry, in declaration order. -async fn load_providers( - shared: &Shared, - engine_cfg: &EngineConfig, +/// One entry per manifest, in declaration order; every refusal names the +/// role and the entry path. +async fn load_role( + entries: &[E], manifests: Vec, -) -> Result> { - let mut providers = Vec::with_capacity(engine_cfg.adapters.len()); - for (entry, loaded_manifest) in engine_cfg.adapters.iter().zip(manifests) { - let loaded = load::provider( - shared, - entry, - loaded_manifest, - &engine_cfg.limits, - engine_cfg.engine.require_component_digest, - ) - .await - .with_context(|| format!("load provider {}", entry.path.display()))?; - providers.push(loaded); + role: Role, + path: impl Fn(&E) -> &std::path::Path, + load: impl AsyncFn(&E, crate::manifest::LoadedManifest) -> Result, +) -> Result> { + let mut out = Vec::with_capacity(entries.len()); + for (entry, manifest) in entries.iter().zip(manifests) { + let loaded = load(entry, manifest) + .await + .with_context(|| format!("{} {}", role.load_context(), path(entry).display()))?; + out.push(loaded); } - Ok(providers) + Ok(out) } /// The providers' manifests as the worker install predicates see them. @@ -244,32 +269,6 @@ fn project_manifests(providers: &[LoadedProvider]) -> Vec { .collect() } -/// In declaration order, against the installed providers. -async fn load_modules( - shared: &Shared, - linker: &Linker>, - engine_cfg: &EngineConfig, - manifests: Vec, - provider_manifests: &[ProviderManifest], -) -> Result>> { - let mut modules = Vec::with_capacity(engine_cfg.modules.len()); - for (entry, loaded_manifest) in engine_cfg.modules.iter().zip(manifests) { - let loaded = load::module( - shared, - linker, - entry, - loaded_manifest, - &engine_cfg.limits, - engine_cfg.engine.require_component_digest, - provider_manifests, - ) - .await - .with_context(|| format!("load module {}", entry.path.display()))?; - modules.push(loaded); - } - Ok(modules) -} - fn assemble( shared: Shared, modules: Vec>, From 39645445eb68730d823c3d82c111c25f7a682313 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:08:19 +0000 Subject: [PATCH 3/6] refactor(supervisor): extract role-keyed telemetry helpers AI Assistance: Claude Fable 5 used for implementation and tests --- .../nexum-runtime/src/supervisor/dispatch.rs | 21 +-- .../nexum-runtime/src/supervisor/lifecycle.rs | 98 ++---------- crates/nexum-runtime/src/supervisor/role.rs | 148 ++++++++++++++++++ 3 files changed, 165 insertions(+), 102 deletions(-) diff --git a/crates/nexum-runtime/src/supervisor/dispatch.rs b/crates/nexum-runtime/src/supervisor/dispatch.rs index c2843b6..3419e06 100644 --- a/crates/nexum-runtime/src/supervisor/dispatch.rs +++ b/crates/nexum-runtime/src/supervisor/dispatch.rs @@ -10,6 +10,7 @@ use tracing_core::Level; use super::Supervisor; use super::cursors::{commit_chain_log_cursor, persist_progress_marker}; use super::lifecycle::{revive_one, sweep}; +use super::role::{Role, report_poison}; use crate::bindings::nexum; use crate::host::component::RuntimeTypes; use crate::host::extension::ExtensionEvent; @@ -288,21 +289,13 @@ impl Supervisor { format!("run terminated abnormally: {}", trap.root_cause()), )); if let Some(recent) = verdict.poisoned { - // A string field, not a `Display` one: the two record - // through different visitor methods and print differently. - let last_error = trap.to_string(); - warn!( - module = %module.name, - recent_failures = recent, - window_secs = poison_policy.window.as_secs(), - last_error, - "module poisoned - quarantined; remove from engine.toml + restart to clear", + report_poison( + Role::Module, + &module.name, + recent, + poison_policy.window, + Some(trap.to_string()), ); - metrics::gauge!( - "nexum_runtime_module_poisoned", - "module" => module.name.clone(), - ) - .set(1.0); } DispatchOutcome::Trapped } diff --git a/crates/nexum-runtime/src/supervisor/lifecycle.rs b/crates/nexum-runtime/src/supervisor/lifecycle.rs index 273b278..2c903ba 100644 --- a/crates/nexum-runtime/src/supervisor/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/lifecycle.rs @@ -4,11 +4,10 @@ use std::collections::VecDeque; use std::time::{Duration, Instant}; use anyhow::{Context, Error, Result, anyhow}; -use tracing::{error, info, warn}; use super::Shared; use super::load::{LoadedModule, LoadedProvider, run_init}; -use super::role::Role; +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 crate::digest::ContentDigest; @@ -316,47 +315,7 @@ pub(super) async fn sweep>( let policy = item.poison_policy(engine_default); if let Some(died_at) = item.detect_death() { let verdict = item.health_mut().record_trap(died_at, now, policy); - match S::ROLE { - Role::Module => warn!( - module = %item.name(), - failure_count = verdict.failure_count, - backoff_ms = verdict.backoff.as_millis() as u64, - "module trapped - marked dead; will restart after backoff", - ), - Role::Adapter => warn!( - adapter = %item.name(), - failure_count = verdict.failure_count, - backoff_ms = verdict.backoff.as_millis() as u64, - "adapter trapped - marked dead; will restart after backoff", - ), - } - metrics::counter!( - S::ROLE.errors_total(), - S::ROLE.label() => item.name().clone(), - "error_kind" => "trap", - ) - .increment(1); - if let Some(recent) = verdict.poisoned { - match S::ROLE { - Role::Module => warn!( - module = %item.name(), - recent_failures = recent, - window_secs = policy.window.as_secs(), - "module poisoned - quarantined; remove from engine.toml + restart to clear", - ), - Role::Adapter => warn!( - adapter = %item.name(), - recent_failures = recent, - window_secs = policy.window.as_secs(), - "adapter poisoned - quarantined; remove from engine.toml + restart to clear", - ), - } - metrics::gauge!( - S::ROLE.poisoned_gauge(), - S::ROLE.label() => item.name().clone(), - ) - .set(1.0); - } + report_trap(S::ROLE, item.name(), &verdict, policy); } if item.health().due_restart(now) { revive_one(shared, item, now).await; @@ -369,59 +328,22 @@ pub(super) async fn revive_one>( item: &mut S, now: Instant, ) { - let failure_count = item.health().failure_count(); // Revives reuse the cached component, so the boot-time digest holds. - match S::ROLE { - Role::Module => info!( - module = %item.name(), - failure_count, - digest = %item.digest(), - "restart attempt", - ), - Role::Adapter => info!( - adapter = %item.name(), - failure_count, - digest = %item.digest(), - "adapter restart attempt", - ), - } - metrics::counter!( - S::ROLE.restarts_total(), - S::ROLE.label() => item.name().clone(), - ) - .increment(1); + report_restart_attempt( + S::ROLE, + item.name(), + item.health().failure_count(), + item.digest(), + ); match item.revive(shared).await { Ok(()) => { item.health_mut() .restart_succeeded(S::ROLE.resets_failure_count()); - match S::ROLE { - Role::Module => info!(module = %item.name(), "restart succeeded"), - Role::Adapter => info!(adapter = %item.name(), "adapter restart succeeded"), - } + report_restart_outcome(S::ROLE, item.name(), Ok(())); } Err(e) => { let deferral = item.health_mut().defer_restart(now); - match S::ROLE { - Role::Module => error!( - module = %item.name(), - failure_count = deferral.failure_count, - backoff_ms = deferral.backoff.as_millis() as u64, - error = %e, - "restart failed - will retry after backoff", - ), - Role::Adapter => { - // A string field, not a `Display` one: the two record - // through different visitor methods and print differently. - let error = format!("{e:#}"); - error!( - adapter = %item.name(), - failure_count = deferral.failure_count, - backoff_ms = deferral.backoff.as_millis() as u64, - error, - "adapter restart failed - will retry after backoff", - ); - } - } + report_restart_outcome(S::ROLE, item.name(), Err((deferral, e))); } } } diff --git a/crates/nexum-runtime/src/supervisor/role.rs b/crates/nexum-runtime/src/supervisor/role.rs index b719eb1..60e7f22 100644 --- a/crates/nexum-runtime/src/supervisor/role.rs +++ b/crates/nexum-runtime/src/supervisor/role.rs @@ -1,6 +1,16 @@ //! One role vocabulary for the load pass, the namespace ledger, the //! metric names, and the compile-time tracing field keys. +use std::time::Duration; + +use anyhow::Error; +use tracing::{error, info, warn}; + +use super::lifecycle::{Deferral, TrapVerdict}; +use crate::digest::ContentDigest; +use crate::module_id::ModuleId; +use crate::runtime::poison_policy::PoisonPolicy; + /// Keys the per-role metric names and the compile-time tracing field keys. #[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr)] pub(super) enum Role { @@ -66,3 +76,141 @@ impl Role { matches!(self, Self::Adapter) } } + +/// A recorded trap: the death log, the error counter, and the poison +/// transition when the verdict crossed. +pub(super) fn report_trap( + role: Role, + name: &ModuleId, + verdict: &TrapVerdict, + policy: PoisonPolicy, +) { + match role { + Role::Module => warn!( + module = %name, + failure_count = verdict.failure_count, + backoff_ms = verdict.backoff.as_millis() as u64, + "module trapped - marked dead; will restart after backoff", + ), + Role::Adapter => warn!( + adapter = %name, + failure_count = verdict.failure_count, + backoff_ms = verdict.backoff.as_millis() as u64, + "adapter trapped - marked dead; will restart after backoff", + ), + } + metrics::counter!( + role.errors_total(), + role.label() => name.clone(), + "error_kind" => "trap", + ) + .increment(1); + if let Some(recent) = verdict.poisoned { + report_poison(role, name, recent, policy.window, None); + } +} + +/// The poison-transition trio: quarantine warn plus gauge; `last_error` +/// rides only when the caller held the trap itself. +pub(super) fn report_poison( + role: Role, + name: &ModuleId, + recent_failures: u32, + window: Duration, + last_error: Option, +) { + let window_secs = window.as_secs(); + match (role, last_error) { + (Role::Module, Some(last_error)) => warn!( + module = %name, + recent_failures, + window_secs, + last_error, + "module poisoned - quarantined; remove from engine.toml + restart to clear", + ), + (Role::Module, None) => warn!( + module = %name, + recent_failures, + window_secs, + "module poisoned - quarantined; remove from engine.toml + restart to clear", + ), + (Role::Adapter, Some(last_error)) => warn!( + adapter = %name, + recent_failures, + window_secs, + last_error, + "adapter poisoned - quarantined; remove from engine.toml + restart to clear", + ), + (Role::Adapter, None) => warn!( + adapter = %name, + recent_failures, + window_secs, + "adapter poisoned - quarantined; remove from engine.toml + restart to clear", + ), + } + metrics::gauge!( + role.poisoned_gauge(), + role.label() => name.clone(), + ) + .set(1.0); +} + +pub(super) fn report_restart_attempt( + role: Role, + name: &ModuleId, + failure_count: u32, + digest: ContentDigest, +) { + match role { + Role::Module => info!( + module = %name, + failure_count, + digest = %digest, + "restart attempt", + ), + Role::Adapter => info!( + adapter = %name, + failure_count, + digest = %digest, + "adapter restart attempt", + ), + } + metrics::counter!( + role.restarts_total(), + role.label() => name.clone(), + ) + .increment(1); +} + +pub(super) fn report_restart_outcome( + role: Role, + name: &ModuleId, + outcome: Result<(), (Deferral, Error)>, +) { + match (role, outcome) { + (Role::Module, Ok(())) => info!(module = %name, "restart succeeded"), + (Role::Adapter, Ok(())) => info!(adapter = %name, "adapter restart succeeded"), + (Role::Module, Err((deferral, e))) => { + // A string field, not a `Display` one: it carries the full + // context chain through the string visitor. + let error = format!("{e:#}"); + error!( + module = %name, + failure_count = deferral.failure_count, + backoff_ms = deferral.backoff.as_millis() as u64, + error, + "restart failed - will retry after backoff", + ); + } + (Role::Adapter, Err((deferral, e))) => { + let error = format!("{e:#}"); + error!( + adapter = %name, + failure_count = deferral.failure_count, + backoff_ms = deferral.backoff.as_millis() as u64, + error, + "adapter restart failed - will retry after backoff", + ); + } + } +} From d1d7eefc2d80050892dd01d51352edcfcd25bf32 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:11:49 +0000 Subject: [PATCH 4/6] fix(supervisor): fail closed on adapter chain subscriptions and chain-log filters AI Assistance: Claude Fable 5 used for implementation and tests --- crates/nexum-runtime/src/supervisor/mod.rs | 5 +- .../nexum-runtime/src/supervisor/prepass.rs | 50 ++++++++++---- .../src/supervisor/subscriptions.rs | 65 ++++++++----------- .../src/supervisor/tests/chain_gate.rs | 51 ++++++++++++++- 4 files changed, 117 insertions(+), 54 deletions(-) diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index 6b00152..f9a5ab5 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -31,7 +31,7 @@ use crate::runtime::poison_policy::PoisonPolicy; use admission::{ProviderKinds, capability_registry, enforce_extension_uniqueness, provider_kinds}; use cursors::ChainLogCursors; use load::{LoadedModule, LoadedProvider}; -use prepass::{enforce_configured_chains, load_required_manifest, manifest_namespace}; +use prepass::{enforce_subscriptions, load_required_manifest, manifest_namespace}; use role::Role; /// Owns every loaded module and provider and exposes the dispatch surface. @@ -149,7 +149,8 @@ impl Supervisor { let registry = capability_registry(&shared.extensions); let loaded_manifest = load_required_manifest(&entry.path, entry.manifest.as_deref(), ®istry, "module")?; - enforce_configured_chains( + enforce_subscriptions( + Role::Module, &manifest_namespace(&loaded_manifest), &loaded_manifest, &env.configured_chains, diff --git a/crates/nexum-runtime/src/supervisor/prepass.rs b/crates/nexum-runtime/src/supervisor/prepass.rs index b4ee82f..8d93308 100644 --- a/crates/nexum-runtime/src/supervisor/prepass.rs +++ b/crates/nexum-runtime/src/supervisor/prepass.rs @@ -9,6 +9,7 @@ use anyhow::{Context, Error, Result, anyhow}; use tracing::{info, warn}; use super::role::Role; +use super::subscriptions::build_alloy_filter; use crate::engine_config::EngineConfig; use crate::manifest::{self, CapabilityRegistry, LoadedManifest, Subscription}; @@ -112,9 +113,11 @@ impl ConfiguredChains { } } -/// Refuse any subscription naming a chain absent from `[chains]`, before any guest code runs. -pub(super) fn enforce_configured_chains( - module: &str, +/// Refuse any subscription naming a chain absent from `[chains]` or carrying +/// an unparseable chain-log filter, before any guest code runs. +pub(super) fn enforce_subscriptions( + role: Role, + name: &str, loaded: &LoadedManifest, chains: &ConfiguredChains, ) -> Result<()> { @@ -124,16 +127,37 @@ pub(super) fn enforce_configured_chains( continue; }; if !chains.contains(*chain_id) { - return Err(unconfigured_chain(module, *chain_id, chains)); + return Err(unconfigured_chain(role, name, *chain_id, chains)); + } + if let Subscription::ChainLog { + address, + event_signature, + .. + } = sub + { + build_alloy_filter(address.as_deref(), event_signature.as_deref()).with_context( + || { + format!( + "{} {name} declares an invalid chain-log filter on chain {chain_id}", + role.claim_role(), + ) + }, + )?; } } Ok(()) } -pub(super) fn unconfigured_chain(module: &str, chain_id: u64, chains: &ConfiguredChains) -> Error { +pub(super) fn unconfigured_chain( + role: Role, + name: &str, + chain_id: u64, + chains: &ConfiguredChains, +) -> Error { + let noun = role.claim_role(); if chains.defaulted { return anyhow!( - "module {module} subscribes to chain {chain_id} but no engine.toml was found \ + "{noun} {name} subscribes to chain {chain_id} but no engine.toml was found \ (running on defaults, no chains configured); create engine.toml with a \ [chains.{chain_id}] entry" ); @@ -149,7 +173,7 @@ pub(super) fn unconfigured_chain(module: &str, chain_id: u64, chains: &Configure .join(", ") }; anyhow!( - "module {module} subscribes to chain {chain_id} but engine.toml declares no \ + "{noun} {name} subscribes to chain {chain_id} but engine.toml declares no \ [chains.{chain_id}] entry; configured chains: {configured}" ) } @@ -174,7 +198,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R &provider_registry, RolePass { role: Role::Adapter, - chains: None, + chains: &configured_chains, }, &mut ledger, )?; @@ -186,7 +210,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R registry, RolePass { role: Role::Module, - chains: Some(&configured_chains), + chains: &configured_chains, }, &mut ledger, )?; @@ -198,7 +222,7 @@ pub(super) fn run(engine_cfg: &EngineConfig, registry: &CapabilityRegistry) -> R struct RolePass<'a> { role: Role, - chains: Option<&'a ConfiguredChains>, + chains: &'a ConfiguredChains, } /// In declaration order. @@ -214,10 +238,8 @@ fn load_role_manifests<'a>( .with_context(|| format!("{} {}", pass.role.load_context(), path.display()))?; let namespace = manifest_namespace(&loaded); claim_namespace(ledger, &namespace, pass.role.claim_role(), path)?; - if let Some(chains) = pass.chains { - enforce_configured_chains(&namespace, &loaded, chains) - .with_context(|| format!("{} {}", pass.role.load_context(), path.display()))?; - } + enforce_subscriptions(pass.role, &namespace, &loaded, pass.chains) + .with_context(|| format!("{} {}", pass.role.load_context(), path.display()))?; manifests.push(loaded); } Ok(manifests) diff --git a/crates/nexum-runtime/src/supervisor/subscriptions.rs b/crates/nexum-runtime/src/supervisor/subscriptions.rs index 4662e5b..c50bcb7 100644 --- a/crates/nexum-runtime/src/supervisor/subscriptions.rs +++ b/crates/nexum-runtime/src/supervisor/subscriptions.rs @@ -4,7 +4,6 @@ use std::collections::BTreeSet; use alloy_chains::Chain; -use tracing::warn; use super::Supervisor; use super::cursors::{chainlog_cursor_key, read_chain_log_cursor}; @@ -42,42 +41,34 @@ impl Supervisor { max_lookback, } = sub { - match build_alloy_filter(address.as_deref(), event_signature.as_deref()) { - Ok(filter) => { - let chain = Chain::from_id(*chain_id); - // A `resume` subscription reads its durable cursor - // once here at boot; others start at head. - let (cursor_key, initial_cursor) = if *resume { - let key = chainlog_cursor_key( - chain, - address.as_deref(), - event_signature.as_deref(), - ); - let seed = read_chain_log_cursor( - &self.shared.components.store, - module.name.as_str(), - &key, - ); - (Some(key), seed) - } else { - (None, None) - }; - out.push(ChainLogSub { - module: module.name.clone(), - chain, - filter, - cursor_key, - initial_cursor, - max_lookback: *max_lookback, - }); - } - Err(err) => warn!( - module = %module.name, - chain_id, - error = %err, - "invalid chain-log subscription - skipping", - ), - } + let filter = build_alloy_filter(address.as_deref(), event_signature.as_deref()) + .expect("chain-log filters are validated at load"); + let chain = Chain::from_id(*chain_id); + // A `resume` subscription reads its durable cursor + // once here at boot; others start at head. + let (cursor_key, initial_cursor) = if *resume { + let key = chainlog_cursor_key( + chain, + address.as_deref(), + event_signature.as_deref(), + ); + let seed = read_chain_log_cursor( + &self.shared.components.store, + module.name.as_str(), + &key, + ); + (Some(key), seed) + } else { + (None, None) + }; + out.push(ChainLogSub { + module: module.name.clone(), + chain, + filter, + cursor_key, + initial_cursor, + max_lookback: *max_lookback, + }); } } } diff --git a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs index 31e8163..a088f71 100644 --- a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs +++ b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs @@ -57,6 +57,54 @@ async fn boot_single_refuses_a_subscription_on_an_unconfigured_chain() { .lacks("compile"); } +/// The gate covers `[[adapters]]` entries too: a provider manifest cannot +/// subscribe past the operator's `[chains]` set. +#[tokio::test] +async fn boot_refuses_an_adapter_subscription_on_an_unconfigured_chain() { + BootScenario::over(mock_components()) + .extensions(acme_extensions()) + .adapter( + TestManifest::new("feed") + .kind("acme-adapter") + .cap("chain") + .block_sub(424_242), + ) + .expect_refusal() + .await + .names("load provider") + .names("adapter feed subscribes to chain 424242") + .names("[chains.424242]") + .lacks("compile"); +} + +/// Filter values fail closed at load: an unparseable address or topic +/// refuses the boot instead of skipping the subscription at collection. +#[tokio::test] +async fn boot_refuses_an_invalid_chain_log_filter() { + for (manifest, detail) in [ + ( + TestManifest::new("example") + .cap("logging") + .chain_log_sub_filtered(1, Some("0xabc"), None), + "invalid chain-log address", + ), + ( + TestManifest::new("example") + .cap("logging") + .chain_log_sub_filtered(1, None, Some("not-a-topic")), + "invalid topic", + ), + ] { + BootScenario::new() + .module(manifest) + .expect_refusal() + .await + .names("module example declares an invalid chain-log filter on chain 1") + .names(detail) + .lacks("compile"); + } +} + #[tokio::test] async fn boot_admits_a_block_subscription_on_a_configured_chain_past_the_chain_gate() { BootScenario::new() @@ -119,7 +167,8 @@ fn configured_chains_normalise_named_and_numeric_spellings() { #[test] fn unconfigured_chain_message_says_none_when_engine_toml_declares_no_chains() { let chains = ConfiguredChains::from_config(&EngineConfig::default()); - let msg = crate::supervisor::unconfigured_chain("example", 424_242, &chains).to_string(); + let msg = crate::supervisor::unconfigured_chain(Role::Module, "example", 424_242, &chains) + .to_string(); assert!(msg.contains("configured chains: none"), "{msg}"); assert!(!msg.contains("no engine.toml was found"), "{msg}"); } From 9f2cbcbae1e2b1903dd55d1f0a7d550445eeb525 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:24:42 +0000 Subject: [PATCH 5/6] refactor(supervisor): read the boot_single manifest role from the Role table AI Assistance: Claude Opus 5 used for review and the fix --- crates/nexum-runtime/src/supervisor/mod.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index f9a5ab5..7e7fdf3 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -147,8 +147,12 @@ impl Supervisor { // Provider kinds come only from `engine.toml`, so none register here. let shared = wire_extensions(engine, components, extensions, clocks, false)?; let registry = capability_registry(&shared.extensions); - let loaded_manifest = - load_required_manifest(&entry.path, entry.manifest.as_deref(), ®istry, "module")?; + let loaded_manifest = load_required_manifest( + &entry.path, + entry.manifest.as_deref(), + ®istry, + Role::Module.manifest_role(), + )?; enforce_subscriptions( Role::Module, &manifest_namespace(&loaded_manifest), From 2d089e760f3dbe12647f3788268e053e4b741b18 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 04:24:43 +0000 Subject: [PATCH 6/6] test(supervisor): pin the chain-log filter rebuild and the pre-compile refusal point AI Assistance: Claude Opus 5 used for review and the tests --- .../src/supervisor/tests/chain_gate.rs | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs index a088f71..4e8a5cb 100644 --- a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs +++ b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs @@ -74,6 +74,7 @@ async fn boot_refuses_an_adapter_subscription_on_an_unconfigured_chain() { .names("load provider") .names("adapter feed subscribes to chain 424242") .names("[chains.424242]") + .lacks("read component") .lacks("compile"); } @@ -101,10 +102,51 @@ async fn boot_refuses_an_invalid_chain_log_filter() { .await .names("module example declares an invalid chain-log filter on chain 1") .names(detail) + .lacks("read component") .lacks("compile"); } } +/// The load-time filter check and the collection-time rebuild read the same +/// manifest values, so the collection rebuild cannot fail. +#[tokio::test] +async fn a_validated_chain_log_filter_survives_to_the_collected_subscription() { + let Some(wasm) = example_wasm_or_skip() else { + return; + }; + let address = "0xC92E8bdf79f0507f65a392b0ab4667716BFE0110"; + let topic = "0x237e158222e3e6968b72b9db0d8043aacf074ad9f650f0d1606b4d82ee432c00"; + let booted = BootScenario::new() + .wasm(wasm) + .module( + TestManifest::new("example") + .cap("logging") + .chain_log_sub_filtered(1, Some(address), Some(topic)), + ) + .boot() + .await + .expect("the example boots alive"); + + let subs = booted.supervisor.chain_log_subscriptions(); + assert_eq!( + subs.len(), + 1, + "the alive module contributes its subscription" + ); + assert_eq!(subs[0].module.as_str(), "example"); + assert_eq!(subs[0].chain.id(), 1); + assert!(subs[0].cursor_key.is_none(), "resume defaults to off"); + // alloy `Filter` exposes no getter; assert through its serialisation. + let serialised = serde_json::to_value(&subs[0].filter).unwrap().to_string(); + assert!( + serialised + .to_lowercase() + .contains(&address.to_lowercase()[2..]), + "{serialised}", + ); + assert!(serialised.contains(&topic[2..]), "{serialised}"); +} + #[tokio::test] async fn boot_admits_a_block_subscription_on_a_configured_chain_past_the_chain_gate() { BootScenario::new()