feat(runtime): project subscriptions into health-filtered plan

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<T: RuntimeTypes> 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,17 @@ impl<T: RuntimeTypes> 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();
         // 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 +281,39 @@ impl<T: RuntimeTypes> 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)"
-                );
+        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 \
+                 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,
+                });
             }
-            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 => {}
         }
 
         // 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,
         );

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<T: RuntimeTypes> Supervisor<T> {
             .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()

crates/nexum-runtime/src/supervisor/subscriptions.rs

@@ -13,81 +13,118 @@ use crate::manifest::Subscription;
 use crate::module_id::ModuleId;
 
 impl<T: RuntimeTypes> Supervisor<T> {
-    /// Alive modules only; sorted by numeric id and deduped.
-    pub fn block_chains(&self) -> Vec<Chain> {
-        let mut out: Vec<Chain> = 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<Chain> = Vec::new();
+        let mut chain_log_subs = Vec::new();
+        let mut extension_kinds = BTreeSet::new();
+        let mut dead_subscribers = false;
+        for module in &self.modules {
+            if !module.health.dispatchable() {
+                dead_subscribers |= !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<ChainLogSub> {
-        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();
+        SubscriptionPlan {
+            block_chains,
+            chain_log_subs,
+            extension_kinds,
+            dead_subscribers,
+        }
     }
+}
+
+/// Everything the launch path opens, projected once from the live modules.
+pub struct SubscriptionPlan {
+    /// Sorted by numeric id and deduped.
+    pub block_chains: Vec<Chain>,
+    /// The stream tags every log with the owning module for routing.
+    pub chain_log_subs: Vec<ChainLogSub>,
+    /// An extension opens an event source only for kinds appearing here.
+    pub extension_kinds: BTreeSet<String>,
+    /// A dead module declares at least one subscription.
+    pub dead_subscribers: bool,
+}
 
-    /// An extension opens an event source only when its kind appears here.
-    pub fn extension_subscription_kinds(&self) -> BTreeSet<String> {
-        self.modules
-            .iter()
-            .flat_map(|m| m.subscriptions.iter())
-            .filter_map(|s| match s {
-                Subscription::Extension { kind, .. } => Some(kind.clone()),
-                _ => None,
-            })
-            .collect()
+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.
+#[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,
+    /// At least one event source drives the engine.
+    Live,
+}
+
 pub struct ChainLogSub {
     /// Also the module's store namespace.
     pub module: ModuleId,

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.viability(0), 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,