From 94940cdd0081e3aee92f2767984049fc4f1979df Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 06:32:42 +0000 Subject: [PATCH 1/2] refactor(runtime): project subscriptions into one health-filtered plan AI Assistance: Claude (Fable 5) used for implementation and tests --- crates/nexum-runtime/src/builder.rs | 58 +++---- crates/nexum-runtime/src/supervisor/mod.rs | 10 +- .../src/supervisor/subscriptions.rs | 158 +++++++++++------- .../src/supervisor/tests/chain_gate.rs | 8 +- .../src/supervisor/tests/lifecycle.rs | 83 +++++++-- 5 files changed, 200 insertions(+), 117 deletions(-) diff --git a/crates/nexum-runtime/src/builder.rs b/crates/nexum-runtime/src/builder.rs index cc4ffb0..37327d7 100644 --- a/crates/nexum-runtime/src/builder.rs +++ b/crates/nexum-runtime/src/builder.rs @@ -29,7 +29,7 @@ use crate::host::provider_pool::ProviderPool; use crate::preset::Runtime; use crate::runtime::event_loop; pub use crate::supervisor::WasiClockOverride; -use crate::supervisor::{self, Supervisor}; +use crate::supervisor::{self, Supervisor, Viability}; /// Ambient inputs the launcher reads. pub struct LaunchContext<'a> { @@ -215,12 +215,12 @@ impl AssembledRuntime<'_, T> { }; let alive = supervisor.alive_count(); - let block_chains = supervisor.block_chains(); + let plan = supervisor.subscription_plan(); info!( modules = supervisor.module_count(), adapters = supervisor.adapter_count(), alive, - chains = block_chains.len(), + chains = plan.block_chains.len(), "supervisor ready" ); if alive == 0 { @@ -262,19 +262,39 @@ impl AssembledRuntime<'_, T> { // The handle keeps the log read side reachable after launch consumes // the components. let logs = components.logs.clone(); - let chain_log_subs = supervisor.chain_log_subscriptions(); + // A non-viable plan is decided here, before any stream opens. + match plan.viable { + Viability::DeadHoldSubs => anyhow::bail!( + "every declared [[subscription]] belongs to an init-failed module - \ + the engine would idle with nothing to run; fix or remove the \ + failing module(s)" + ), + Viability::Nothing => { + // Nothing to drive: return a handle whose event loop is + // already complete so `wait` resolves immediately. + info!("no [[subscription]] entries - engine has nothing to run; exiting"); + let event_loop = executor.spawn(async { TaskExit::ReceiverGone }); + return Ok(RuntimeHandle { + event_loop, + tasks, + logs, + _add_ons: add_on_handles, + }); + } + Viability::Live => {} + } + // Extension event sources open only for subscription kinds some - // loaded module declares; each extension gates further on its own + // live module declares; each extension gates further on its own // service state and returns no stream when it has nothing to // observe. - let subscribed = supervisor.extension_subscription_kinds(); let mut reconnect_tasks = TaskSet::new(); let mut extension_streams = Vec::new(); { let mut sources = EventSources::new( engine_cfg, supervisor.services(), - &subscribed, + &plan.extension_kinds, &executor, &mut reconnect_tasks, ); @@ -283,38 +303,18 @@ impl AssembledRuntime<'_, T> { } } - // No subscriptions: nothing to drive. Return a handle whose event loop - // is already complete so `wait` resolves immediately. - if block_chains.is_empty() && chain_log_subs.is_empty() && extension_streams.is_empty() { - if supervisor.dead_modules_hold_subscriptions() { - anyhow::bail!( - "every declared [[subscription]] belongs to an init-failed module - \ - the engine would idle with nothing to run; fix or remove the \ - failing module(s)" - ); - } - info!("no [[subscription]] entries - engine has nothing to run; exiting"); - let event_loop = executor.spawn(async { TaskExit::ReceiverGone }); - return Ok(RuntimeHandle { - event_loop, - tasks, - logs, - _add_ons: add_on_handles, - }); - } - // Open per-chain block subscriptions + per-module chain-log // subscriptions through the executor, then drive them in the event // loop until shutdown. let block_streams = event_loop::open_block_streams( &components.chain, - &block_chains, + &plan.block_chains, &executor, &mut reconnect_tasks, ); let chain_log_streams = event_loop::open_chain_log_streams( &components.chain, - chain_log_subs, + plan.chain_log_subs, &executor, &mut reconnect_tasks, ); diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index 708d845..7e4e991 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -14,7 +14,7 @@ mod subscriptions; pub use prepass::ConfiguredChains; pub use store::{WasiClockOverride, build_linker, build_provider_linker}; -pub use subscriptions::ChainLogSub; +pub use subscriptions::{ChainLogSub, SubscriptionPlan, Viability}; use std::sync::Arc; @@ -195,14 +195,6 @@ impl Supervisor { .count() } - /// Distinguishes benign "no subscriptions declared" from "every declared - /// subscription belongs to a dead module" (operator error). - pub fn dead_modules_hold_subscriptions(&self) -> bool { - self.modules - .iter() - .any(|m| !m.health.dispatchable() && !m.subscriptions.is_empty()) - } - pub fn poisoned_count(&self) -> usize { self.modules .iter() diff --git a/crates/nexum-runtime/src/supervisor/subscriptions.rs b/crates/nexum-runtime/src/supervisor/subscriptions.rs index 2fa8cbf..8195988 100644 --- a/crates/nexum-runtime/src/supervisor/subscriptions.rs +++ b/crates/nexum-runtime/src/supervisor/subscriptions.rs @@ -13,79 +13,107 @@ use crate::manifest::Subscription; use crate::module_id::ModuleId; impl Supervisor { - /// Alive modules only; sorted by numeric id and deduped. - pub fn block_chains(&self) -> Vec { - let mut out: Vec = Vec::new(); - for module in self.modules.iter().filter(|m| m.health.dispatchable()) { - for sub in &module.subscriptions { - if let Subscription::Block { chain_id } = sub { - out.push(Chain::from_id(*chain_id)); - } + /// One pass, one health filter: a dead module contributes to no field, + /// so no stream of any kind opens for it. + pub fn subscription_plan(&self) -> SubscriptionPlan { + let mut block_chains: Vec = Vec::new(); + let mut chain_log_subs = Vec::new(); + let mut extension_kinds = BTreeSet::new(); + let mut dead_hold_subs = false; + for module in &self.modules { + if !module.health.dispatchable() { + dead_hold_subs |= !module.subscriptions.is_empty(); + continue; } - } - out.sort_by_key(|c| c.id()); - out.dedup(); - out - } - - /// Alive modules only; the stream tags every log with the module for routing. - pub fn chain_log_subscriptions(&self) -> Vec { - let mut out = Vec::new(); - for module in self.modules.iter().filter(|m| m.health.dispatchable()) { for sub in &module.subscriptions { - if let Subscription::ChainLog { - chain_id, - address, - event_signature, - resume, - max_lookback, - } = sub - { - 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( + match sub { + Subscription::Block { chain_id } => { + block_chains.push(Chain::from_id(*chain_id)); + } + Subscription::ChainLog { + chain_id, + address, + event_signature, + resume, + max_lookback, + } => { + 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) + }; + chain_log_subs.push(ChainLogSub { + module: module.name.clone(), 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, - }); + filter, + cursor_key, + initial_cursor, + max_lookback: *max_lookback, + }); + } + Subscription::Extension { kind, .. } => { + extension_kinds.insert(kind.clone()); + } + Subscription::Cron { .. } => {} } } } - out + block_chains.sort_by_key(|c| c.id()); + block_chains.dedup(); + let viable = + if block_chains.is_empty() && chain_log_subs.is_empty() && extension_kinds.is_empty() { + if dead_hold_subs { + Viability::DeadHoldSubs + } else { + Viability::Nothing + } + } else { + Viability::Live + }; + SubscriptionPlan { + block_chains, + chain_log_subs, + extension_kinds, + viable, + } } +} - /// An extension opens an event source only when its kind appears here. - pub fn extension_subscription_kinds(&self) -> BTreeSet { - self.modules - .iter() - .flat_map(|m| m.subscriptions.iter()) - .filter_map(|s| match s { - Subscription::Extension { kind, .. } => Some(kind.clone()), - _ => None, - }) - .collect() - } +/// Everything the launch path opens, projected once from the live modules. +pub struct SubscriptionPlan { + /// Sorted by numeric id and deduped. + pub block_chains: Vec, + /// The stream tags every log with the owning module for routing. + pub chain_log_subs: Vec, + /// An extension opens an event source only for kinds appearing here. + pub extension_kinds: BTreeSet, + pub viable: Viability, +} + +/// The launch verdict; boot-dead is permanent, so it is final at launch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Viability { + /// No module declares a subscription; the engine has nothing to run. + Nothing, + /// Every declared subscription belongs to a dead module. + DeadHoldSubs, + Live, } pub struct ChainLogSub { diff --git a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs index a868a71..15ab7af 100644 --- a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs +++ b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs @@ -8,8 +8,10 @@ async fn empty_supervisor_returns_no_subscriptions() { .boot() .await .expect("an empty scenario boots"); - assert!(booted.supervisor.block_chains().is_empty()); - assert!(booted.supervisor.chain_log_subscriptions().is_empty()); + let plan = booted.supervisor.subscription_plan(); + assert!(plan.block_chains.is_empty()); + assert!(plan.chain_log_subs.is_empty()); + assert_eq!(plan.viable, Viability::Nothing); assert_eq!(booted.supervisor.module_count(), 0); } @@ -127,7 +129,7 @@ async fn a_validated_chain_log_filter_survives_to_the_collected_subscription() { .await .expect("the example boots alive"); - let subs = booted.supervisor.chain_log_subscriptions(); + let subs = booted.supervisor.subscription_plan().chain_log_subs; assert_eq!( subs.len(), 1, diff --git a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs index c3ede10..ebb153a 100644 --- a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs @@ -63,16 +63,18 @@ async fn init_failure_marks_module_dead_excluding_dispatch_and_subscriptions() { 0, "no live module is subscribed to chain 11155111 blocks", ); + let plan = booted.supervisor.subscription_plan(); assert!( - booted.supervisor.block_chains().is_empty(), - "dead module must not contribute to block_chains()", + plan.block_chains.is_empty(), + "dead module must not contribute block chains", ); assert!( - booted.supervisor.chain_log_subscriptions().is_empty(), - "dead module must not contribute to chain_log_subscriptions()", + plan.chain_log_subs.is_empty(), + "dead module must not contribute chain-log subscriptions", ); - assert!( - booted.supervisor.dead_modules_hold_subscriptions(), + assert_eq!( + plan.viable, + Viability::DeadHoldSubs, "the filtered-out subscriptions must be attributed to the dead module", ); } @@ -101,16 +103,75 @@ async fn alive_module_subscriptions_survive_alongside_dead_module() { 1, "only the example is alive" ); - let chains = booted.supervisor.block_chains(); + let plan = booted.supervisor.subscription_plan(); assert_eq!( - chains.iter().map(|c| c.id()).collect::>(), + plan.block_chains.iter().map(|c| c.id()).collect::>(), vec![1], "the alive module's chain survives; the dead module's does not", ); - assert!( - booted.supervisor.dead_modules_hold_subscriptions(), - "the dead module's dropped subscription is attributable", + assert_eq!( + plan.viable, + Viability::Live, + "one live subscription keeps the plan viable despite the dead module", + ); +} + +/// One health filter covers extension kinds too: a dead module's kind opens +/// no extension event source, while a live module's survives. +#[tokio::test] +async fn dead_module_extension_kind_is_excluded_from_the_plan() { + let Some(price_alert_wasm) = module_wasm_or_skip("price-alert") else { + return; + }; + let Some(example_wasm) = example_wasm_or_skip() else { + return; + }; + struct Ticker; + impl Extension for Ticker { + fn namespace(&self) -> &'static str { + "ticker" + } + fn capabilities(&self) -> manifest::NamespaceCaps { + manifest::NamespaceCaps { + prefix: "test:ticker/", + ifaces: &[], + } + } + fn link(&self, _linker: &mut Linker>) -> anyhow::Result<()> { + Ok(()) + } + fn subscriptions(&self) -> &'static [&'static str] { + &["alarms", "ticks"] + } + } + let booted = BootScenario::new() + .extensions([Arc::new(Ticker) as Arc>]) + .module( + Entry::new(price_alert("not-a-number").extension_sub("alarms", &[])) + .wasm(price_alert_wasm), + ) + .module( + Entry::new( + TestManifest::new("example") + .cap("logging") + .extension_sub("ticks", &[]), + ) + .wasm(example_wasm), + ) + .boot() + .await + .expect("both modules load; only price-alert's init fails"); + + let plan = booted.supervisor.subscription_plan(); + assert_eq!( + plan.extension_kinds + .iter() + .map(String::as_str) + .collect::>(), + vec!["ticks"], + "the dead module's kind is excluded; the alive module's survives", ); + assert_eq!(plan.viable, Viability::Live); } /// Boot and restart share one instantiate-and-init helper, so the verdict on From 0ff4f52169499ebd23f88b6c2c58949e155aeabd Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 6 Aug 2026 06:46:57 +0000 Subject: [PATCH 2/2] fix(runtime): decide viability from the sources that really opened AI Assistance: Claude (Opus 5) used for red-team review, the fix, and tests --- crates/nexum-runtime/src/builder.rs | 41 +++++---- .../src/supervisor/subscriptions.rs | 37 +++++--- .../src/supervisor/tests/chain_gate.rs | 2 +- .../src/supervisor/tests/lifecycle.rs | 85 ++++++++++++++----- 4 files changed, 108 insertions(+), 57 deletions(-) diff --git a/crates/nexum-runtime/src/builder.rs b/crates/nexum-runtime/src/builder.rs index 37327d7..0319c32 100644 --- a/crates/nexum-runtime/src/builder.rs +++ b/crates/nexum-runtime/src/builder.rs @@ -262,8 +262,26 @@ impl AssembledRuntime<'_, T> { // The handle keeps the log read side reachable after launch consumes // the components. let logs = components.logs.clone(); - // A non-viable plan is decided here, before any stream opens. - match plan.viable { + // Extension event sources open only for subscription kinds some + // live module declares; each extension gates further on its own + // service state and returns no stream when it has nothing to + // observe. + let mut reconnect_tasks = TaskSet::new(); + let mut extension_streams = Vec::new(); + { + let mut sources = EventSources::new( + engine_cfg, + supervisor.services(), + &plan.extension_kinds, + &executor, + &mut reconnect_tasks, + ); + for ext in &extensions { + extension_streams.extend(ext.events(&mut sources)?); + } + } + + match plan.viability(extension_streams.len()) { Viability::DeadHoldSubs => anyhow::bail!( "every declared [[subscription]] belongs to an init-failed module - \ the engine would idle with nothing to run; fix or remove the \ @@ -284,25 +302,6 @@ impl AssembledRuntime<'_, T> { Viability::Live => {} } - // Extension event sources open only for subscription kinds some - // live module declares; each extension gates further on its own - // service state and returns no stream when it has nothing to - // observe. - let mut reconnect_tasks = TaskSet::new(); - let mut extension_streams = Vec::new(); - { - let mut sources = EventSources::new( - engine_cfg, - supervisor.services(), - &plan.extension_kinds, - &executor, - &mut reconnect_tasks, - ); - for ext in &extensions { - extension_streams.extend(ext.events(&mut sources)?); - } - } - // Open per-chain block subscriptions + per-module chain-log // subscriptions through the executor, then drive them in the event // loop until shutdown. diff --git a/crates/nexum-runtime/src/supervisor/subscriptions.rs b/crates/nexum-runtime/src/supervisor/subscriptions.rs index 8195988..58561a2 100644 --- a/crates/nexum-runtime/src/supervisor/subscriptions.rs +++ b/crates/nexum-runtime/src/supervisor/subscriptions.rs @@ -19,10 +19,10 @@ impl Supervisor { let mut block_chains: Vec = Vec::new(); let mut chain_log_subs = Vec::new(); let mut extension_kinds = BTreeSet::new(); - let mut dead_hold_subs = false; + let mut dead_subscribers = false; for module in &self.modules { if !module.health.dispatchable() { - dead_hold_subs |= !module.subscriptions.is_empty(); + dead_subscribers |= !module.subscriptions.is_empty(); continue; } for sub in &module.subscriptions { @@ -76,21 +76,11 @@ impl Supervisor { } block_chains.sort_by_key(|c| c.id()); block_chains.dedup(); - let viable = - if block_chains.is_empty() && chain_log_subs.is_empty() && extension_kinds.is_empty() { - if dead_hold_subs { - Viability::DeadHoldSubs - } else { - Viability::Nothing - } - } else { - Viability::Live - }; SubscriptionPlan { block_chains, chain_log_subs, extension_kinds, - viable, + dead_subscribers, } } } @@ -103,7 +93,25 @@ pub struct SubscriptionPlan { pub chain_log_subs: Vec, /// An extension opens an event source only for kinds appearing here. pub extension_kinds: BTreeSet, - pub viable: Viability, + /// A dead module declares at least one subscription. + pub dead_subscribers: bool, +} + +impl SubscriptionPlan { + /// A declared extension kind is not yet a source: the extension gates on + /// its own service state, so the caller passes how many really opened. + pub fn viability(&self, open_extension_sources: usize) -> Viability { + if !self.block_chains.is_empty() + || !self.chain_log_subs.is_empty() + || open_extension_sources > 0 + { + Viability::Live + } else if self.dead_subscribers { + Viability::DeadHoldSubs + } else { + Viability::Nothing + } + } } /// The launch verdict; boot-dead is permanent, so it is final at launch. @@ -113,6 +121,7 @@ pub enum Viability { Nothing, /// Every declared subscription belongs to a dead module. DeadHoldSubs, + /// At least one event source drives the engine. Live, } diff --git a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs index 15ab7af..67c0ed9 100644 --- a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs +++ b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs @@ -11,7 +11,7 @@ async fn empty_supervisor_returns_no_subscriptions() { let plan = booted.supervisor.subscription_plan(); assert!(plan.block_chains.is_empty()); assert!(plan.chain_log_subs.is_empty()); - assert_eq!(plan.viable, Viability::Nothing); + assert_eq!(plan.viability(0), Viability::Nothing); assert_eq!(booted.supervisor.module_count(), 0); } diff --git a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs index ebb153a..68d97e3 100644 --- a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs @@ -73,7 +73,7 @@ async fn init_failure_marks_module_dead_excluding_dispatch_and_subscriptions() { "dead module must not contribute chain-log subscriptions", ); assert_eq!( - plan.viable, + plan.viability(0), Viability::DeadHoldSubs, "the filtered-out subscriptions must be attributed to the dead module", ); @@ -110,12 +110,34 @@ async fn alive_module_subscriptions_survive_alongside_dead_module() { "the alive module's chain survives; the dead module's does not", ); assert_eq!( - plan.viable, + plan.viability(0), Viability::Live, "one live subscription keeps the plan viable despite the dead module", ); } +/// Declares two subscription kinds and opens no event source for either, as +/// an extension whose service is unconfigured does. +struct Ticker; + +impl Extension for Ticker { + fn namespace(&self) -> &'static str { + "ticker" + } + fn capabilities(&self) -> manifest::NamespaceCaps { + manifest::NamespaceCaps { + prefix: "test:ticker/", + ifaces: &[], + } + } + fn link(&self, _linker: &mut Linker>) -> anyhow::Result<()> { + Ok(()) + } + fn subscriptions(&self) -> &'static [&'static str] { + &["alarms", "ticks"] + } +} + /// One health filter covers extension kinds too: a dead module's kind opens /// no extension event source, while a live module's survives. #[tokio::test] @@ -126,24 +148,6 @@ async fn dead_module_extension_kind_is_excluded_from_the_plan() { let Some(example_wasm) = example_wasm_or_skip() else { return; }; - struct Ticker; - impl Extension for Ticker { - fn namespace(&self) -> &'static str { - "ticker" - } - fn capabilities(&self) -> manifest::NamespaceCaps { - manifest::NamespaceCaps { - prefix: "test:ticker/", - ifaces: &[], - } - } - fn link(&self, _linker: &mut Linker>) -> anyhow::Result<()> { - Ok(()) - } - fn subscriptions(&self) -> &'static [&'static str] { - &["alarms", "ticks"] - } - } let booted = BootScenario::new() .extensions([Arc::new(Ticker) as Arc>]) .module( @@ -171,7 +175,46 @@ async fn dead_module_extension_kind_is_excluded_from_the_plan() { vec!["ticks"], "the dead module's kind is excluded; the alive module's survives", ); - assert_eq!(plan.viable, Viability::Live); + assert_eq!( + plan.viability(1), + Viability::Live, + "an opened extension source drives the engine", + ); + assert_eq!( + plan.viability(0), + Viability::DeadHoldSubs, + "with no source opened, the dead module's subscriptions are the only ones left", + ); +} + +/// A declared kind is not a source: an extension that opens none leaves the +/// engine with nothing to drive, and no dead module to blame for it. +#[tokio::test] +async fn a_declared_extension_kind_alone_is_not_viable() { + let Some(example_wasm) = example_wasm_or_skip() else { + return; + }; + let booted = BootScenario::new() + .extensions([Arc::new(Ticker) as Arc>]) + .module( + Entry::new( + TestManifest::new("example") + .cap("logging") + .extension_sub("ticks", &[]), + ) + .wasm(example_wasm), + ) + .boot() + .await + .expect("the example boots alive"); + + let plan = booted.supervisor.subscription_plan(); + assert_eq!(plan.extension_kinds.len(), 1, "the live kind is declared"); + assert_eq!( + plan.viability(0), + Viability::Nothing, + "a declared but unopened kind must not park the engine on an empty select", + ); } /// Boot and restart share one instantiate-and-init helper, so the verdict on