From b5c4bbaee8716a947018663504f75435b846f12e Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 17 Aug 2026 23:57:59 +0000 Subject: [PATCH] refactor(runtime)!: rename the host plan and stream layer to sources What the host opens, reconnects and backfills is a source; what a module declares is a trigger, and the two sides no longer share a word. Rename trigger_plan, TriggerPlan, event_triggers and EventTrigger to source_plan, SourcePlan, event_sources and EventSource, and move supervisor/triggers.rs to sources.rs. ProviderPool::subscribe_blocks and watch_chain_logs become open_block_source and open_event_source, honest about transport in neither direction. enforce_triggers becomes enforce_trigger_chains. LogSource becomes LogChannel, with the tracing field channel, so source cannot mean guest stdout one layer up. SourcePlan.demanded_extension_kinds shares one spelling with SourceContext, and the test-only ManifestSource becomes ManifestInput, completing the source reservation in ADR 0019. Every operator string on the layer now reads one reconnect in one vocabulary, block source or event source, in the messages and the tracing fields alike. A vocabulary guard test allowlists the anchored upstream and nexum-tasks tokens, exempts an allowlisted token only for itself rather than its whole line, and pins exactly one call site for each of the two alloy watch methods. The operator handbook follows the renamed channel field and the overcommit wording. Unchanged: BlockStream and CanonicalLogStream (a stream is the transport a source is delivered over), the nexum_runtime_stream_reconnects_total metric with its labels, and the persisted store keys. Closes #241 AI Assistance: Claude Fable 5 used for the implementation. --- crates/nexum-runtime/Cargo.toml | 2 +- crates/nexum-runtime/src/builder.rs | 8 +- crates/nexum-runtime/src/host/error.rs | 4 +- crates/nexum-runtime/src/host/extension.rs | 2 +- .../nexum-runtime/src/host/impls/logging.rs | 4 +- crates/nexum-runtime/src/host/logs/mod.rs | 36 ++--- crates/nexum-runtime/src/host/logs/stdio.rs | 62 ++++---- crates/nexum-runtime/src/host/logs/store.rs | 4 +- .../nexum-runtime/src/host/provider_pool.rs | 22 +-- crates/nexum-runtime/src/manifest/types.rs | 2 +- crates/nexum-runtime/src/metrics.rs | 2 +- .../nexum-runtime/src/runtime/event_loop.rs | 143 +++++++++--------- .../nexum-runtime/src/supervisor/cursors.rs | 4 +- .../nexum-runtime/src/supervisor/dispatch.rs | 6 +- crates/nexum-runtime/src/supervisor/mod.rs | 8 +- .../nexum-runtime/src/supervisor/prepass.rs | 6 +- .../supervisor/{triggers.rs => sources.rs} | 32 ++-- crates/nexum-runtime/src/supervisor/store.rs | 6 +- .../src/supervisor/tests/boot_refusals.rs | 4 +- .../src/supervisor/tests/chain_gate.rs | 18 +-- .../nexum-runtime/src/supervisor/tests/e2e.rs | 15 +- .../src/supervisor/tests/lifecycle.rs | 22 +-- .../nexum-runtime/src/supervisor/tests/mod.rs | 6 +- .../nexum-runtime/src/test_utils/harness.rs | 18 +-- .../nexum-runtime/src/test_utils/manifest.rs | 12 +- crates/nexum-runtime/src/test_utils/mod.rs | 10 +- .../nexum-runtime/src/test_utils/scenario.rs | 8 +- docs/02-modules-triggers-packaging.md | 2 +- docs/production.md | 4 +- 29 files changed, 238 insertions(+), 234 deletions(-) rename crates/nexum-runtime/src/supervisor/{triggers.rs => sources.rs} (89%) diff --git a/crates/nexum-runtime/Cargo.toml b/crates/nexum-runtime/Cargo.toml index da837837..3c405c7a 100644 --- a/crates/nexum-runtime/Cargo.toml +++ b/crates/nexum-runtime/Cargo.toml @@ -26,7 +26,7 @@ thiserror.workspace = true # Newtype boilerplate (`Display`, `AsRef`, `From`) for identity wrappers. derive_more.workspace = true # `strum::IntoStaticStr`: the snake_case variant name is the tracing -# `source` field (`LogSource`) and the boot-refusal `error_kind` label. +# `channel` field (`LogChannel`) and the boot-refusal `error_kind` label. strum.workspace = true # Full-semver interface ids and their compatibility tracks. semver.workspace = true diff --git a/crates/nexum-runtime/src/builder.rs b/crates/nexum-runtime/src/builder.rs index 4c5e38f3..94eb41b6 100644 --- a/crates/nexum-runtime/src/builder.rs +++ b/crates/nexum-runtime/src/builder.rs @@ -4,7 +4,7 @@ //! component builders, add-ons) through a type-state chain; //! [`ReadyBuilder::launch`] opens the backends and hands off to //! [`AssembledRuntime::launch`], which installs add-ons, builds the engine and -//! linker, boots the supervisor, opens the trigger sources, spawns the event +//! linker, boots the supervisor, opens the sources, spawns the event //! loop, and returns a [`RuntimeHandle`]. [`RuntimeBuilder::runtime`] binds a //! [`Runtime`] preset for the common case. @@ -296,7 +296,7 @@ impl AssembledRuntime { }; let alive = supervisor.alive_count(); - let plan = supervisor.trigger_plan(); + let plan = supervisor.source_plan(); info!( modules = supervisor.module_count(), alive, @@ -347,7 +347,7 @@ impl AssembledRuntime { { let mut sources = SourceContext::new( engine_cfg, - &plan.extension_kinds, + &plan.demanded_extension_kinds, &executor, &mut reconnect_tasks, ); @@ -387,7 +387,7 @@ impl AssembledRuntime { ); let chain_log_streams = event_loop::open_chain_log_streams( &components.chain, - plan.event_triggers, + plan.event_sources, &executor, &mut reconnect_tasks, ); diff --git a/crates/nexum-runtime/src/host/error.rs b/crates/nexum-runtime/src/host/error.rs index 10dc6090..0a8ded72 100644 --- a/crates/nexum-runtime/src/host/error.rs +++ b/crates/nexum-runtime/src/host/error.rs @@ -395,12 +395,12 @@ mod tests { } let sink = Sink::default(); - let subscriber = tracing_subscriber::fmt() + let collector = tracing_subscriber::fmt() .with_max_level(tracing::Level::DEBUG) .with_ansi(false) .with_writer(sink.clone()) .finish(); - tracing::subscriber::with_default(subscriber, || { + tracing::subscriber::with_default(collector, || { let _ = store_fault( "mod-a", "set", diff --git a/crates/nexum-runtime/src/host/extension.rs b/crates/nexum-runtime/src/host/extension.rs index fcd043c5..5cab2d1b 100644 --- a/crates/nexum-runtime/src/host/extension.rs +++ b/crates/nexum-runtime/src/host/extension.rs @@ -1,5 +1,5 @@ //! Extension seam: what one extension contributes to the host (namespace, -//! capabilities, linker hook, trigger sources, and manifest-section install +//! capabilities, linker hook, sources, and manifest-section install //! predicates). use std::collections::BTreeSet; diff --git a/crates/nexum-runtime/src/host/impls/logging.rs b/crates/nexum-runtime/src/host/impls/logging.rs index 12197013..b77e0294 100644 --- a/crates/nexum-runtime/src/host/impls/logging.rs +++ b/crates/nexum-runtime/src/host/impls/logging.rs @@ -5,7 +5,7 @@ use tracing_core::Level; use crate::bindings::nexum; use crate::host::component::RuntimeTypes; -use crate::host::logs::{LogRecord, LogSource}; +use crate::host::logs::{LogChannel, LogRecord}; use crate::host::state::HostState; impl nexum::host::logging::Host for HostState { @@ -22,7 +22,7 @@ impl nexum::host::logging::Host for HostState { }; self.log_router.record(LogRecord::now( self.run.clone(), - LogSource::HostInterface, + LogChannel::HostInterface, level, message, )); diff --git a/crates/nexum-runtime/src/host/logs/mod.rs b/crates/nexum-runtime/src/host/logs/mod.rs index e2d1ee50..1b885baa 100644 --- a/crates/nexum-runtime/src/host/logs/mod.rs +++ b/crates/nexum-runtime/src/host/logs/mod.rs @@ -6,7 +6,7 @@ //! event and the retention store. [`LogPipeline`] is the shared handle, //! carrying the write side and the store's read side. //! -//! One guest panic yields three records distinguished by [`LogSource`] +//! One guest panic yields three records distinguished by [`LogChannel`] //! (stderr, host logging call, supervisor death), redundancy covering //! channels that survive different failure modes. @@ -47,11 +47,11 @@ impl RunId { } /// Which capture point produced a record; the snake_case name is the tracing -/// `source` field. +/// `channel` field. #[derive(Debug, Clone, Copy, PartialEq, Eq, IntoStaticStr)] #[strum(serialize_all = "snake_case")] #[non_exhaustive] -pub enum LogSource { +pub enum LogChannel { /// The `nexum:host/logging` glue: an explicit guest `log` call. HostInterface, /// A line captured from the guest's stdout pipe. @@ -70,7 +70,7 @@ pub struct LogRecord { /// Wall-clock capture time. pub ts: SystemTime, /// Capture point of origin. - pub source: LogSource, + pub channel: LogChannel, /// Line severity. pub level: Level, /// The line text. @@ -79,11 +79,11 @@ pub struct LogRecord { impl LogRecord { /// Record stamped at the current instant. - pub fn now(run: RunId, source: LogSource, level: Level, message: String) -> Self { + pub fn now(run: RunId, channel: LogChannel, level: Level, message: String) -> Self { Self { run, ts: SystemTime::now(), - source, + channel, level, message, } @@ -133,18 +133,18 @@ impl LogRouter { fn emit_tracing(record: &LogRecord) { let module = record.run.module.as_str(); let run = record.run.seq; - let source: &'static str = record.source.into(); + let channel: &'static str = record.channel.into(); let message = record.message.as_str(); if record.level == Level::TRACE { - tracing::trace!(module, run, source, "{message}"); + tracing::trace!(module, run, channel, "{message}"); } else if record.level == Level::DEBUG { - tracing::debug!(module, run, source, "{message}"); + tracing::debug!(module, run, channel, "{message}"); } else if record.level == Level::INFO { - tracing::info!(module, run, source, "{message}"); + tracing::info!(module, run, channel, "{message}"); } else if record.level == Level::WARN { - tracing::warn!(module, run, source, "{message}"); + tracing::warn!(module, run, channel, "{message}"); } else { - tracing::error!(module, run, source, "{message}"); + tracing::error!(module, run, channel, "{message}"); } } @@ -225,14 +225,14 @@ mod tests { let router = LogRouter::new(store.clone()); router.record(LogRecord::now( RunId::new(test_module_id(), 0), - LogSource::HostInterface, + LogChannel::HostInterface, Level::INFO, "hello".to_owned(), )); let appended = store.appended.lock().unwrap(); assert_eq!(appended.len(), 1, "retention consumer saw the record"); assert_eq!(appended[0].message, "hello"); - assert_eq!(appended[0].source, LogSource::HostInterface); + assert_eq!(appended[0].channel, LogChannel::HostInterface); } #[test] @@ -245,7 +245,7 @@ mod tests { let run = RunId::new(test_module_id(), 0); pipeline.router().record(LogRecord::now( run.clone(), - LogSource::Stdout, + LogChannel::Stdout, Level::INFO, "line".to_owned(), )); @@ -257,10 +257,10 @@ mod tests { } #[test] - fn source_names_are_snake_case_for_the_tracing_field() { - let s: &'static str = LogSource::HostInterface.into(); + fn channel_names_are_snake_case_for_the_tracing_field() { + let s: &'static str = LogChannel::HostInterface.into(); assert_eq!(s, "host_interface"); - let s: &'static str = LogSource::Panic.into(); + let s: &'static str = LogChannel::Panic.into(); assert_eq!(s, "panic"); } } diff --git a/crates/nexum-runtime/src/host/logs/stdio.rs b/crates/nexum-runtime/src/host/logs/stdio.rs index 020b8f60..157db841 100644 --- a/crates/nexum-runtime/src/host/logs/stdio.rs +++ b/crates/nexum-runtime/src/host/logs/stdio.rs @@ -1,6 +1,6 @@ //! Per-store stdout/stderr capture: a [`StdoutStream`] line-buffering guest //! output and routing each line as a [`LogRecord`] tagged with its run and -//! source. +//! channel. use std::io; use std::pin::Pin; @@ -12,27 +12,27 @@ use wasmtime_wasi::cli::{IsTerminal, StdoutStream}; use tracing_core::Level; -use super::{LogRecord, LogRouter, LogSource, RunId}; +use super::{LogChannel, LogRecord, LogRouter, RunId}; /// Cap on an unterminated in-flight line; crossing it force-flushes the /// buffer as one record. const MAX_LINE_BYTES: usize = 1 << 20; /// Per-store stdout or stderr sink; each [`StdoutStream::async_stream`] yields -/// a line-splitting writer bound to the run and source. +/// a line-splitting writer bound to the run and channel. pub struct StdioStream { router: Arc, run: RunId, - source: LogSource, + channel: LogChannel, } impl StdioStream { - /// Sink routing `source` lines for `run` through `router`. - pub fn new(router: Arc, run: RunId, source: LogSource) -> Self { + /// Sink routing `channel` lines for `run` through `router`. + pub fn new(router: Arc, run: RunId, channel: LogChannel) -> Self { Self { router, run, - source, + channel, } } } @@ -48,7 +48,7 @@ impl StdoutStream for StdioStream { Box::new(LineWriter { router: self.router.clone(), run: self.run.clone(), - source: self.source, + channel: self.channel, buf: Vec::new(), }) } @@ -59,7 +59,7 @@ impl StdoutStream for StdioStream { struct LineWriter { router: Arc, run: RunId, - source: LogSource, + channel: LogChannel, buf: Vec, } @@ -72,13 +72,13 @@ impl LineWriter { route_line( &self.router, &self.run, - self.source, + self.channel, &line[..line.len() - 1], ); } if self.buf.len() > MAX_LINE_BYTES { let chunk = std::mem::take(&mut self.buf); - route_line(&self.router, &self.run, self.source, &chunk); + route_line(&self.router, &self.run, self.channel, &chunk); } } @@ -89,20 +89,20 @@ impl LineWriter { return; } let rest = std::mem::take(&mut self.buf); - route_line(&self.router, &self.run, self.source, &rest); + route_line(&self.router, &self.run, self.channel, &rest); } } /// Level for a captured line: stdout INFO, stderr WARN. -fn level_for(source: LogSource) -> Level { - match source { - LogSource::Stderr => Level::WARN, +fn level_for(channel: LogChannel) -> Level { + match channel { + LogChannel::Stderr => Level::WARN, _ => Level::INFO, } } /// Decode and route one line, dropping a trailing `\r` and skipping empties. -fn route_line(router: &LogRouter, run: &RunId, source: LogSource, bytes: &[u8]) { +fn route_line(router: &LogRouter, run: &RunId, channel: LogChannel, bytes: &[u8]) { let bytes = bytes.strip_suffix(b"\r").unwrap_or(bytes); if bytes.is_empty() { return; @@ -110,8 +110,8 @@ fn route_line(router: &LogRouter, run: &RunId, source: LogSource, bytes: &[u8]) let message = String::from_utf8_lossy(bytes).into_owned(); router.record(LogRecord::now( run.clone(), - source, - level_for(source), + channel, + level_for(channel), message, )); } @@ -153,7 +153,7 @@ mod tests { use tokio::io::AsyncWriteExt; use super::*; - use crate::host::logs::{LogPipeline, LogRecord, LogSource, RunId, RunLogStore}; + use crate::host::logs::{LogChannel, LogPipeline, LogRecord, RunId, RunLogStore}; /// Store recording every appended message for assertions. #[derive(Default)] @@ -173,7 +173,7 @@ mod tests { } } - fn setup(source: LogSource) -> (LineWriter, Arc) { + fn setup(channel: LogChannel) -> (LineWriter, Arc) { let store = Arc::new(CaptureStore::default()); let pipeline = LogPipeline::new(store.clone()); let writer = LineWriter { @@ -182,7 +182,7 @@ mod tests { crate::module_id::ModuleId::parse("m").expect("valid module name"), 0, ), - source, + channel, buf: Vec::new(), }; (writer, store) @@ -200,14 +200,14 @@ mod tests { #[tokio::test] async fn splits_on_newlines() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(b"alpha\nbeta\n").await.unwrap(); assert_eq!(messages(&store), ["alpha", "beta"]); } #[tokio::test] async fn buffers_a_partial_line_until_the_newline_arrives() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(b"partial").await.unwrap(); assert!(messages(&store).is_empty(), "no newline yet"); w.write_all(b" line\n").await.unwrap(); @@ -219,7 +219,7 @@ mod tests { // The euro sign is three bytes; splitting mid-code-point across // two writes must not corrupt the decoded line. let euro = "\u{20ac}".as_bytes(); - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(&euro[..1]).await.unwrap(); w.write_all(&euro[1..]).await.unwrap(); w.write_all(b"\n").await.unwrap(); @@ -228,7 +228,7 @@ mod tests { #[tokio::test] async fn interleaved_writes_accumulate_into_one_line() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); for chunk in [&b"a"[..], b"b", b"c", b"\n", b"d", b"e", b"\n"] { w.write_all(chunk).await.unwrap(); } @@ -237,7 +237,7 @@ mod tests { #[tokio::test] async fn final_unterminated_line_is_flushed_on_drop() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(b"no trailing newline").await.unwrap(); assert!(messages(&store).is_empty(), "buffered, not yet flushed"); drop(w); @@ -246,30 +246,30 @@ mod tests { #[tokio::test] async fn empty_lines_are_skipped() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(b"\n\nkept\n\n").await.unwrap(); assert_eq!(messages(&store), ["kept"]); } #[tokio::test] async fn trailing_carriage_return_is_trimmed() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); w.write_all(b"crlf\r\n").await.unwrap(); assert_eq!(messages(&store), ["crlf"]); } #[tokio::test] async fn stderr_lines_carry_the_warn_level() { - let (mut w, store) = setup(LogSource::Stderr); + let (mut w, store) = setup(LogChannel::Stderr); w.write_all(b"oops\n").await.unwrap(); let records = store.records.lock().unwrap(); - assert_eq!(records[0].source, LogSource::Stderr); + assert_eq!(records[0].channel, LogChannel::Stderr); assert_eq!(records[0].level, Level::WARN); } #[tokio::test] async fn over_long_unterminated_line_is_force_flushed() { - let (mut w, store) = setup(LogSource::Stdout); + let (mut w, store) = setup(LogChannel::Stdout); let flood = vec![b'x'; MAX_LINE_BYTES + 1]; w.write_all(&flood).await.unwrap(); // The force-flush bounds host memory without waiting for a newline. diff --git a/crates/nexum-runtime/src/host/logs/store.rs b/crates/nexum-runtime/src/host/logs/store.rs index 71676d26..88b4cd53 100644 --- a/crates/nexum-runtime/src/host/logs/store.rs +++ b/crates/nexum-runtime/src/host/logs/store.rs @@ -216,7 +216,7 @@ mod tests { use tracing_core::Level; use super::*; - use crate::host::logs::{LogSource, RECORD_OVERHEAD}; + use crate::host::logs::{LogChannel, RECORD_OVERHEAD}; fn limits(bytes_per_run: usize, runs_retained: usize) -> LogRetentionLimits { LogRetentionLimits { @@ -232,7 +232,7 @@ mod tests { fn record(run: &RunId, message: &str) -> LogRecord { LogRecord::now( run.clone(), - LogSource::Stdout, + LogChannel::Stdout, Level::INFO, message.to_owned(), ) diff --git a/crates/nexum-runtime/src/host/provider_pool.rs b/crates/nexum-runtime/src/host/provider_pool.rs index 7b569add..c554c732 100644 --- a/crates/nexum-runtime/src/host/provider_pool.rs +++ b/crates/nexum-runtime/src/host/provider_pool.rs @@ -161,9 +161,9 @@ impl ProviderPool { }) } - /// Follow canonical block headers on `chain`: WS via - /// `eth_subscribe(newHeads)`, HTTP by polling at the chain's block time. - pub async fn subscribe_blocks(&self, chain: Chain) -> Result { + /// Canonical block headers on `chain`: WS via `eth_subscribe(newHeads)`, + /// HTTP by polling at the chain's block time. + pub async fn open_block_source(&self, chain: Chain) -> Result { let ep = self .providers .get(&chain) @@ -197,7 +197,7 @@ impl ProviderPool { /// Canonical (reorg-aware) log stream on `chain` from `start_block`. Each /// item is one block's batch (possibly with no logs); reorg rollbacks /// carry `removed == true`. - pub fn watch_chain_logs( + pub fn open_event_source( &self, chain: Chain, filter: Filter, @@ -311,11 +311,11 @@ mod tests { } #[tokio::test] - async fn empty_pool_rejects_block_subscribe() { + async fn empty_pool_rejects_open_block_source() { let pool = ProviderPool::empty(); // Can't use .unwrap_err() because BlockStream doesn't impl Debug. assert!(matches!( - pool.subscribe_blocks(Chain::from_id(1)).await, + pool.open_block_source(Chain::from_id(1)).await, Err(PoolError::UnknownChain(c)) if c == Chain::from_id(1) )); } @@ -330,12 +330,12 @@ mod tests { } #[test] - fn empty_pool_rejects_watch_chain_logs() { + fn empty_pool_rejects_open_event_source() { let pool = ProviderPool::empty(); let filter = alloy_rpc_types_eth::Filter::new(); // Can't use .unwrap_err() because CanonicalLogStream doesn't impl Debug. assert!(matches!( - pool.watch_chain_logs(Chain::from_id(1), filter, 0), + pool.open_event_source(Chain::from_id(1), filter, 0), Err(PoolError::UnknownChain(c)) if c == Chain::from_id(1) )); } @@ -521,10 +521,10 @@ mod tests { } #[tokio::test] - async fn http_config_block_subscribe_takes_poll_path() { + async fn http_config_block_source_takes_poll_path() { use wiremock::{Mock, MockServer, ResponseTemplate, matchers::any}; - // An HTTP transport has no pubsub, so `subscribe_blocks` must fall + // An HTTP transport has no pubsub, so `open_block_source` must fall // back to polling rather than erroring. The head fetch // (`eth_blockNumber`) is the only call made at setup - the block // poller stream is lazy - so one mocked response proves the poll @@ -543,7 +543,7 @@ mod tests { // BlockStream doesn't impl Debug, so assert on `is_ok` rather than // unwrapping. assert!( - pool.subscribe_blocks(Chain::from_id(1)).await.is_ok(), + pool.open_block_source(Chain::from_id(1)).await.is_ok(), "http config should open the block poll path without erroring", ); } diff --git a/crates/nexum-runtime/src/manifest/types.rs b/crates/nexum-runtime/src/manifest/types.rs index f4c54eda..5dca38d5 100644 --- a/crates/nexum-runtime/src/manifest/types.rs +++ b/crates/nexum-runtime/src/manifest/types.rs @@ -46,7 +46,7 @@ pub type ExtensionSections = BTreeMap; #[derive(Debug, Clone)] pub enum Trigger { /// A new block; one stream per chain id, fanned out to every module - /// watching that chain. + /// with a block trigger on that chain. Block { /// EVM chain id. chain_id: u64, diff --git a/crates/nexum-runtime/src/metrics.rs b/crates/nexum-runtime/src/metrics.rs index d96d88d1..53e460fe 100644 --- a/crates/nexum-runtime/src/metrics.rs +++ b/crates/nexum-runtime/src/metrics.rs @@ -72,7 +72,7 @@ pub const METRICS: &[Metric] = &[ Metric { name: "nexum_runtime_stream_reconnects_total", kind: Kind::Counter, - help: "Chain subscription reconnects by chain.", + help: "Stream reconnects by kind and chain; kind \"chain-log\" also carries module.", }, ]; diff --git a/crates/nexum-runtime/src/runtime/event_loop.rs b/crates/nexum-runtime/src/runtime/event_loop.rs index 80088984..f2d711b9 100644 --- a/crates/nexum-runtime/src/runtime/event_loop.rs +++ b/crates/nexum-runtime/src/runtime/event_loop.rs @@ -4,12 +4,12 @@ //! and retracts a reorged delivered tail. //! //! `open_block_streams` and `open_chain_log_streams` each spawn one -//! reconnect-aware task per trigger or chain: it opens the stream, pumps items to -//! an mpsc channel, and on drop waits `restart_policy::backoff_for` before -//! reopening, resetting the backoff once the stream has been healthy for -//! `HEALTHY_WINDOW`. The tasks exit with [`TaskExit::ReceiverGone`] when `run` -//! drops the receivers; their handles collect into a [`TaskSet`] the loop -//! drains on shutdown. +//! reconnect-aware task per event source or chain: it opens the stream, +//! pumps items to an mpsc channel, and on drop waits +//! `restart_policy::backoff_for` before reopening, resetting the backoff +//! once the stream has been healthy for `HEALTHY_WINDOW`. The tasks exit +//! with [`TaskExit::ReceiverGone`] when `run` drops the receivers; their +//! handles collect into a [`TaskSet`] the loop drains on shutdown. use std::sync::Arc; use std::time::{Duration, Instant}; @@ -29,7 +29,7 @@ use crate::host::extension::{ExtensionDelivery, ExtensionSource}; use crate::host::provider_pool::ProviderPool; use crate::module_id::ModuleId; use crate::runtime::restart_policy::{backoff_for, jitter_seed}; -use crate::supervisor::{EventTrigger, Supervisor}; +use crate::supervisor::{EventSource, Supervisor}; use nexum_tasks::{TaskExecutor, TaskExit, TaskSet}; /// Uninterrupted-event duration before the backoff counter resets to 0. @@ -37,7 +37,7 @@ const HEALTHY_WINDOW: Duration = Duration::from_secs(60); /// Silence between block events beyond which the next event logs a gap-closed /// line, surfacing an alloy-internal transport reconnect that produced no -/// `stream ended` event. +/// `source ended` event. const BLOCK_GAP_LOG_THRESHOLD: Duration = Duration::from_secs(60); /// Channel buffer for each reconnect task. @@ -46,7 +46,7 @@ const RECONNECT_CHANNEL_BUF: usize = 64; /// Block-gap size at or above which a re-open logs a large-backfill notice. const LARGE_GAP_LOG_THRESHOLD: u64 = 1_000; -/// Open one reconnect-aware block-subscription task per chain, spawned via +/// Open one reconnect-aware block-source task per chain, spawned via /// `executor` with handles pushed into `tasks` for graceful shutdown. pub fn open_block_streams( pool: &ProviderPool, @@ -67,30 +67,30 @@ pub fn open_block_streams( streams } -/// Open one reconnect-aware chain-log task per event trigger; see +/// Open one reconnect-aware chain-log task per event source; see /// [`open_block_streams`]. pub fn open_chain_log_streams( pool: &ProviderPool, - triggers: Vec, + sources: Vec, executor: &TaskExecutor, tasks: &mut TaskSet, ) -> Vec { let mut streams = Vec::new(); - for trigger in triggers { + for source in sources { let (tx, rx) = mpsc::channel::(RECONNECT_CHANNEL_BUF); let pool = pool.clone(); let resume = ChainLogResume { - // The cursor key is constant per trigger and cloned onto every + // The cursor key is constant per source and cloned onto every // log; `Arc` keeps that clone cheap. - cursor_key: trigger.cursor_key.map(Arc::from), - initial_cursor: trigger.initial_cursor, - max_lookback: trigger.max_lookback, + cursor_key: source.cursor_key.map(Arc::from), + initial_cursor: source.initial_cursor, + max_lookback: source.max_lookback, }; tasks.push(executor.spawn(reconnecting_chain_log_task( pool, - trigger.module, - trigger.chain, - trigger.filter, + source.module, + source.chain, + source.filter, resume, tx, ))); @@ -118,7 +118,7 @@ async fn backoff_pause(attempt: &mut u32, seed: u64, log: impl FnOnce(u32, u64)) tokio::time::sleep(backoff).await; } -/// Reconnect-aware loop for one chain's block subscription: re-opens the +/// Reconnect-aware loop for one chain's block source: re-opens the /// stream with exponential backoff after every drop or error. async fn reconnecting_block_task( pool: ProviderPool, @@ -130,12 +130,12 @@ async fn reconnecting_block_task( let mut attempt: u32 = 0; let mut last_event: Option = None; loop { - match pool.subscribe_blocks(chain).await { + match pool.open_block_source(chain).await { Ok(mut inner) => { if attempt == 0 { - info!(chain_id, "block subscription open"); + info!(chain_id, "block source open"); } else { - info!(chain_id, attempt, "block subscription reopened"); + info!(chain_id, attempt, "block source reopened"); metrics::counter!( "nexum_runtime_stream_reconnects_total", "kind" => "block", @@ -148,14 +148,14 @@ async fn reconnecting_block_task( if attempt > 0 && last_event.is_some_and(|t| now.duration_since(t) >= HEALTHY_WINDOW) { - info!(chain_id, "block stream healthy - resetting backoff"); + info!(chain_id, "block source healthy - resetting backoff"); attempt = 0; } // Detect transport-layer reconnects that // alloy handled internally - `inner.next().await` // keeps yielding events but with a long gap. The - // engine's reconnect path (`stream ended` -> wait - // backoff -> `subscription reopened`) does not fire + // engine's reconnect path (`source ended` -> wait + // backoff -> `source reopened`) does not fire // for these, so without this log a soak operator // sees an `alloy_transport_ws::native` ERROR // followed by silence indistinguishable from a @@ -168,7 +168,7 @@ async fn reconnecting_block_task( chain_id, gap_s, kind = "block", - "stream gap closed - first event after silence \ + "source gap closed - first event after silence \ (likely an alloy-internal transport reconnect)" ); } @@ -181,23 +181,23 @@ async fn reconnecting_block_task( return TaskExit::ReceiverGone; } } - warn!(chain_id, "block stream ended (WebSocket dropped?)"); + warn!(chain_id, "block source ended (WebSocket dropped?)"); } Err(err) => { - warn!(chain_id, error = %err, "block subscription failed"); + warn!(chain_id, error = %err, "block source open failed"); } } backoff_pause(&mut attempt, seed, |attempt, backoff_ms| { warn!( chain_id, - attempt, backoff_ms, "reconnecting block subscription after backoff", + attempt, backoff_ms, "reconnecting block source after backoff", ); }) .await; } } -/// Per-trigger resume and backfill knobs for a chain-log task. +/// Per-source resume and backfill knobs for a chain-log task. struct ChainLogResume { /// Durable cursor key; `Some` for a `resume` trigger. cursor_key: Option>, @@ -215,7 +215,7 @@ struct DeliveredTail { logs: Vec, } -/// Poller-backed loop for one (module, chain) event trigger; a +/// Poller-backed loop for one (module, chain) event source; a /// re-open resumes past the scanned range and retracts a reorged tail. async fn reconnecting_chain_log_task( pool: ProviderPool, @@ -250,7 +250,7 @@ async fn reconnecting_chain_log_task( error = %err, attempt, backoff_ms, - "chain-log provider lookup failed - retrying after backoff", + "event source provider lookup failed - retrying after backoff", ); }) .await; @@ -267,7 +267,7 @@ async fn reconnecting_chain_log_task( error = %err, attempt, backoff_ms, - "chain-log head fetch failed - retrying after backoff", + "event source head fetch failed - retrying after backoff", ); }) .await; @@ -288,7 +288,7 @@ async fn reconnecting_chain_log_task( tail_block = t.number, attempt, backoff_ms, - "chain-log tail hash unconfirmed - retrying after backoff", + "event source tail hash unconfirmed - retrying after backoff", ); }) .await; @@ -309,7 +309,7 @@ async fn reconnecting_chain_log_task( chain_id, skipped_from = start_block, skipped_to = floor, - "chain-log gap exceeds max_lookback - skipping the oldest missed blocks", + "event source gap exceeds max_lookback - skipping the oldest missed blocks", ); start_block = floor; } @@ -323,20 +323,20 @@ async fn reconnecting_chain_log_task( from = start_block, to = head, blocks = head.saturating_sub(start_block), - "chain-log poller backfilling a large gap" + "event source backfilling a large gap" ); } - match pool.watch_chain_logs(chain, filter.clone(), start_block) { + match pool.open_event_source(chain, filter.clone(), start_block) { Ok(mut inner) => { if attempt == 0 { - info!(module = %module, chain_id, start_block, "chain-log poller open"); + info!(module = %module, chain_id, start_block, "event source open"); } else { info!( module = %module, chain_id, attempt, start_block, - "chain-log poller reopened" + "event source reopened" ); metrics::counter!( "nexum_runtime_stream_reconnects_total", @@ -357,7 +357,7 @@ async fn reconnecting_chain_log_task( module = %module, chain_id, tail_block = t.number, - "chain-log tail reorged while disconnected - retracting its logs", + "event source tail reorged while disconnected - retracting its logs", ); for mut log in t.logs { log.removed = true; @@ -375,7 +375,7 @@ async fn reconnecting_chain_log_task( info!( module = %module, chain_id, - "chain-log stream healthy - resetting backoff" + "event source healthy - resetting backoff" ); attempt = 0; } @@ -419,20 +419,20 @@ async fn reconnecting_chain_log_task( module = %module, chain_id, error = %err, - "chain-log poller error - reopening" + "event source error - reopening" ); break; } } } - warn!(module = %module, chain_id, "chain-log poller stream ended - reopening"); + warn!(module = %module, chain_id, "event source ended - reopening"); } Err(err) => { warn!( module = %module, chain_id, error = %err, - "chain-log poller open failed" + "event source open failed" ); } } @@ -442,7 +442,7 @@ async fn reconnecting_chain_log_task( chain_id, attempt, backoff_ms, - "reconnecting chain-log poller after backoff", + "reconnecting event source after backoff", ); }) .await; @@ -460,7 +460,7 @@ pub type TaggedBlockStream = std::pin::Pin< >; /// `(module, chain, log, cursor_key)`; `cursor_key` is `Some` for `resume`. pub type TaggedChainLog = (ModuleId, Chain, alloy_rpc_types_eth::Log, Option>); -/// Stream of [`TaggedChainLog`], merged across every subscribed chain. +/// Stream of [`TaggedChainLog`], merged across every open event source. pub type TaggedChainLogStream = std::pin::Pin + Send>>; /// Drive the supervisor with triggers until `shutdown` resolves. @@ -470,7 +470,7 @@ pub type TaggedChainLogStream = /// supervisor's stop probe between the per-module calls of one trigger. The /// in-flight call finishes before the loop exits; the guard `shutdown` /// yields is held until return, so the drain covers that call and its -/// cursor commit. Returns the `(blocks, chain_logs)` dispatch tally. +/// cursor commit. Returns the `(blocks, events)` dispatch tally. pub async fn run( supervisor: &mut Supervisor, block_streams: Vec, @@ -504,7 +504,7 @@ pub async fn run( }; let mut shutdown = Box::pin(shutdown); let mut dispatched_blocks: u64 = 0; - let mut dispatched_chain_logs: u64 = 0; + let mut dispatched_events: u64 = 0; let mut dispatched_extension_triggers: u64 = 0; let started = Instant::now(); loop { @@ -538,7 +538,7 @@ pub async fn run( timestamp: header.timestamp.saturating_mul(1000), }), Some(Err((chain, err))) => { - warn!(chain_id = chain.id(), error = %err, "block stream error - continuing"); + warn!(chain_id = chain.id(), error = %err, "block source error - continuing"); continue; } None => NextTrigger::StreamPanic("block"), @@ -547,7 +547,7 @@ pub async fn run( Some((module, chain, log, cursor_key)) => { NextTrigger::Event(module, chain, Box::new(log), cursor_key) } - None => NextTrigger::StreamPanic("chain-log"), + None => NextTrigger::StreamPanic("event"), }, next = extension_deliveries.next() => match next { Some(delivery) => NextTrigger::Extension(delivery), @@ -565,7 +565,7 @@ pub async fn run( supervisor .dispatch_event(&module, chain, *log, cursor_key.as_deref()) .await; - dispatched_chain_logs += 1; + dispatched_events += 1; } NextTrigger::Extension(delivery) => { supervisor.dispatch_extension_trigger(delivery).await; @@ -582,13 +582,13 @@ pub async fn run( tasks.shutdown().await; info!( dispatched_blocks, - dispatched_chain_logs, + dispatched_events, dispatched_extension_triggers, uptime_secs = started.elapsed().as_secs(), "graceful shutdown complete", ); drop(guard); - return (dispatched_blocks, dispatched_chain_logs); + return (dispatched_blocks, dispatched_events); } NextTrigger::StreamPanic(kind) => { // Reconnect tasks should loop forever. @@ -602,7 +602,7 @@ pub async fn run( kind, "reconnect task ended unexpectedly - shutting down for engine restart" ); - return (dispatched_blocks, dispatched_chain_logs); + return (dispatched_blocks, dispatched_events); } } } @@ -725,7 +725,7 @@ mod tests { tasks: &mut TaskSet, initial_cursor: Option, ) -> TaggedChainLogStream { - let triggers = vec![EventTrigger { + let sources = vec![EventSource { module: ModuleId::parse("mod").expect("valid module name"), chain: alloy_chains::Chain::mainnet(), filter: alloy_rpc_types_eth::Filter::default(), @@ -733,9 +733,9 @@ mod tests { initial_cursor, max_lookback: None, }]; - open_chain_log_streams(pool, triggers, executor, tasks) + open_chain_log_streams(pool, sources, executor, tasks) .pop() - .expect("one stream per trigger") + .expect("one stream per source") } async fn recv(stream: &mut TaggedChainLogStream) -> Log { @@ -1089,16 +1089,16 @@ mod tests { tasks.shutdown().await; } - /// `open_chain_log_streams` spawns one reconnect task per event trigger. + /// `open_chain_log_streams` spawns one reconnect task per event source. #[tokio::test] - async fn open_chain_log_streams_opens_one_task_per_trigger() { + async fn open_chain_log_streams_opens_one_task_per_source() { let rpc = MockRpc::new(); let pool = pool_for(&rpc); let manager = TaskManager::new(); let executor = manager.executor(); let mut tasks = TaskSet::new(); - let triggers = vec![ - EventTrigger { + let sources = vec![ + EventSource { module: ModuleId::parse("mod-a").expect("valid module name"), chain: alloy_chains::Chain::mainnet(), filter: alloy_rpc_types_eth::Filter::default(), @@ -1106,7 +1106,7 @@ mod tests { initial_cursor: None, max_lookback: None, }, - EventTrigger { + EventSource { module: ModuleId::parse("mod-b").expect("valid module name"), chain: alloy_chains::Chain::mainnet(), filter: alloy_rpc_types_eth::Filter::default(), @@ -1115,8 +1115,8 @@ mod tests { max_lookback: None, }, ]; - let streams = open_chain_log_streams(&pool, triggers, &executor, &mut tasks); - assert_eq!(streams.len(), 2, "one stream per trigger"); + let streams = open_chain_log_streams(&pool, sources, &executor, &mut tasks); + assert_eq!(streams.len(), 2, "one stream per source"); tasks.shutdown().await; } @@ -1151,7 +1151,7 @@ mod tests { } #[tokio::test(start_paused = true)] - async fn block_subscription_reopens_after_a_failed_open() { + async fn block_source_reopens_after_a_failed_open() { let rpc = MockRpc::new(); rpc.push_script(vec![rpc_err("node down at boot")]); let pool = pool_for(&rpc); @@ -1177,7 +1177,7 @@ mod tests { } }) .await - .expect("the reopened subscription delivers"); + .expect("the reopened source delivers"); assert_eq!(header.number, 5); tasks.shutdown().await; } @@ -1353,7 +1353,7 @@ mod tests { log_node.push_chain_log(alloy_rpc_types_eth::Log::default()); let block_streams = open_block_streams(&pool, &[Chain::mainnet()], &executor, &mut tasks); - let event_triggers = vec![crate::supervisor::EventTrigger { + let event_sources = vec![crate::supervisor::EventSource { module: ModuleId::parse("test-module").expect("valid module name"), chain: Chain::from_id(100), filter: Filter::default(), @@ -1361,13 +1361,12 @@ mod tests { initial_cursor: None, max_lookback: None, }]; - let chain_log_streams = - open_chain_log_streams(&pool, event_triggers, &executor, &mut tasks); + let chain_log_streams = open_chain_log_streams(&pool, event_sources, &executor, &mut tasks); // 500 ms only bounds wall time; the assertion is on the tally, so a // miss means a broken select arm, not a slow scheduler. let shutdown = tokio::time::sleep(Duration::from_millis(500)); - let (blocks, chain_logs) = tokio::time::timeout( + let (blocks, events) = tokio::time::timeout( Duration::from_secs(10), run( &mut booted.supervisor, @@ -1382,7 +1381,7 @@ mod tests { .expect("run() must return once shutdown fires"); assert_eq!(blocks, 1, "the queued block must be drained and dispatched"); assert_eq!( - chain_logs, 1, + events, 1, "the queued chain-log must be drained and dispatched", ); } diff --git a/crates/nexum-runtime/src/supervisor/cursors.rs b/crates/nexum-runtime/src/supervisor/cursors.rs index c2879d4a..ab0a10f4 100644 --- a/crates/nexum-runtime/src/supervisor/cursors.rs +++ b/crates/nexum-runtime/src/supervisor/cursors.rs @@ -78,14 +78,14 @@ pub(super) fn commit_chain_log_cursor( warn!( module = %module, error = %e, - "failed to persist chain-log cursor", + "failed to persist event source cursor", ); } } Err(e) => warn!( module = %module, error = %e, - "failed to open module store for chain-log cursor", + "failed to open module store for event source cursor", ), } } diff --git a/crates/nexum-runtime/src/supervisor/dispatch.rs b/crates/nexum-runtime/src/supervisor/dispatch.rs index 63984a6b..f5a1408d 100644 --- a/crates/nexum-runtime/src/supervisor/dispatch.rs +++ b/crates/nexum-runtime/src/supervisor/dispatch.rs @@ -15,7 +15,7 @@ use super::lifecycle::{revive_one, sweep}; use crate::bindings::nexum; use crate::host::component::RuntimeTypes; use crate::host::extension::ExtensionDelivery; -use crate::host::logs::{LogRecord, LogSource}; +use crate::host::logs::{LogChannel, LogRecord}; use crate::manifest::Trigger; use crate::module_id::ModuleId; @@ -133,7 +133,7 @@ impl Supervisor { let block_number = log.block_number; let removed = log.removed; - let trigger = nexum::host::types::Trigger::Event(super::triggers::wit_log(&log, chain)); + let trigger = nexum::host::types::Trigger::Event(super::sources::wit_log(&log, chain)); let ok = matches!( self.dispatch_to( idx, @@ -334,7 +334,7 @@ impl Supervisor { // trap already went to host tracing. router.record(LogRecord::now( module.live.run.clone(), - LogSource::Panic, + LogChannel::Panic, Level::ERROR, format!("run terminated abnormally: {}", trap.root_cause()), )); diff --git a/crates/nexum-runtime/src/supervisor/mod.rs b/crates/nexum-runtime/src/supervisor/mod.rs index 8c117031..cbc74ef1 100644 --- a/crates/nexum-runtime/src/supervisor/mod.rs +++ b/crates/nexum-runtime/src/supervisor/mod.rs @@ -8,13 +8,13 @@ mod dispatch; mod lifecycle; pub(crate) mod load; pub(crate) mod prepass; +mod sources; mod store; -mod triggers; pub use load::LoadRefusal; pub use prepass::{BootRefusal, ConfiguredChains}; +pub use sources::{EventSource, SourcePlan, Viability}; pub use store::{WasiClockOverride, build_linker}; -pub use triggers::{EventTrigger, TriggerPlan, Viability}; use std::sync::Arc; @@ -32,7 +32,7 @@ use crate::runtime::poison_policy::PoisonPolicy; use admission::{capability_registry, enforce_extension_uniqueness}; use cursors::ChainLogCursors; use load::LoadedModule; -use prepass::{enforce_triggers, load_required_manifest, manifest_namespace}; +use prepass::{enforce_trigger_chains, load_required_manifest, manifest_namespace}; /// Owns every loaded module. pub struct Supervisor { @@ -128,7 +128,7 @@ impl Supervisor { let registry = capability_registry(&shared.extensions); let loaded_manifest = load_required_manifest(&entry.path, entry.manifest.as_deref(), ®istry)?; - enforce_triggers( + enforce_trigger_chains( manifest_namespace(&loaded_manifest).as_str(), &loaded_manifest, &env.configured_chains, diff --git a/crates/nexum-runtime/src/supervisor/prepass.rs b/crates/nexum-runtime/src/supervisor/prepass.rs index dedb8abd..93e64ace 100644 --- a/crates/nexum-runtime/src/supervisor/prepass.rs +++ b/crates/nexum-runtime/src/supervisor/prepass.rs @@ -94,7 +94,7 @@ pub enum BootRefusal { /// The chains engine.toml declares. configured: BTreeSet, }, - /// N in-ceiling components can still oversubscribe the host together; + /// N in-ceiling components can still overcommit the host together; /// `[policy.total]` bounds the declared sum. #[error( "component {id} takes the summed memory reservation to {sum} bytes, \ @@ -219,7 +219,7 @@ impl ConfiguredChains { /// Refuse any trigger naming a chain absent from `[chains]`, before any /// guest code runs. -pub(super) fn enforce_triggers( +pub(super) fn enforce_trigger_chains( name: &str, loaded: &LoadedManifest, chains: &ConfiguredChains, @@ -293,7 +293,7 @@ pub(super) fn run( .with_refusal_context(|| format!("load module {}", entry.path.display()))?; let namespace = manifest_namespace(&loaded); claim_namespace(&mut ledger, namespace.as_str(), &entry.path)?; - enforce_triggers(namespace.as_str(), &loaded, &configured_chains) + enforce_trigger_chains(namespace.as_str(), &loaded, &configured_chains) .with_refusal_context(|| format!("load module {}", entry.path.display()))?; let limits = resolve_module_limits( &entry.id, diff --git a/crates/nexum-runtime/src/supervisor/triggers.rs b/crates/nexum-runtime/src/supervisor/sources.rs similarity index 89% rename from crates/nexum-runtime/src/supervisor/triggers.rs rename to crates/nexum-runtime/src/supervisor/sources.rs index 86267d65..f9e5dca0 100644 --- a/crates/nexum-runtime/src/supervisor/triggers.rs +++ b/crates/nexum-runtime/src/supervisor/sources.rs @@ -1,4 +1,4 @@ -//! Project loaded modules' triggers into what the event loop opens; +//! Project loaded modules' triggers into the sources the host opens; //! dead modules are excluded so no stream opens for an unreachable module. use std::collections::BTreeSet; @@ -15,10 +15,10 @@ use crate::module_id::ModuleId; impl Supervisor { /// One pass, one health filter: a dead module contributes to no field, /// so no stream of any kind opens for it. - pub fn trigger_plan(&self) -> TriggerPlan { + pub fn source_plan(&self) -> SourcePlan { let mut block_chains: Vec = Vec::new(); - let mut event_triggers = Vec::new(); - let mut extension_kinds = BTreeSet::new(); + let mut event_sources = Vec::new(); + let mut demanded_extension_kinds = BTreeSet::new(); let mut dead_hold_triggers = false; for module in &self.modules { if !module.health.dispatchable() { @@ -52,7 +52,7 @@ impl Supervisor { } else { (None, None) }; - event_triggers.push(EventTrigger { + event_sources.push(EventSource { module: module.name.clone(), chain, filter, @@ -62,7 +62,7 @@ impl Supervisor { }); } Trigger::Extension { extension_kind, .. } => { - extension_kinds.insert(extension_kind.clone()); + demanded_extension_kinds.insert(extension_kind.clone()); } Trigger::Schedule { .. } => {} } @@ -70,33 +70,33 @@ impl Supervisor { } block_chains.sort_by_key(|c| c.id()); block_chains.dedup(); - TriggerPlan { + SourcePlan { block_chains, - event_triggers, - extension_kinds, + event_sources, + demanded_extension_kinds, dead_hold_triggers, } } } /// Everything the launch path opens, projected once from the live modules. -pub struct TriggerPlan { +pub struct SourcePlan { /// Sorted by numeric id and deduped. pub block_chains: Vec, /// The stream tags every log with the owning module for routing. - pub event_triggers: Vec, + pub event_sources: Vec, /// An extension opens a source only for kinds appearing here. - pub extension_kinds: BTreeSet, + pub demanded_extension_kinds: BTreeSet, /// A dead module declares at least one trigger. pub dead_hold_triggers: bool, } -impl TriggerPlan { +impl SourcePlan { /// 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.event_triggers.is_empty() + || !self.event_sources.is_empty() || open_extension_sources > 0 { Viability::Live @@ -119,8 +119,8 @@ pub enum Viability { Live, } -/// One module's declared interest in a chain's logs, resolved at boot. -pub struct EventTrigger { +/// One chain-log source to open, resolved from a module's event trigger. +pub struct EventSource { /// Also the module's store namespace. pub module: ModuleId, /// Chain the filter runs against; it must have an `engine.toml` entry. diff --git a/crates/nexum-runtime/src/supervisor/store.rs b/crates/nexum-runtime/src/supervisor/store.rs index e95f1ac7..1492e348 100644 --- a/crates/nexum-runtime/src/supervisor/store.rs +++ b/crates/nexum-runtime/src/supervisor/store.rs @@ -14,7 +14,7 @@ use crate::engine_config::{OutboundHttpLimits, PolicyCeilings}; use crate::host::component::{RuntimeTypes, StateHandle, StateStore}; use crate::host::extension::Extension; use crate::host::http::HttpGate; -use crate::host::logs::{LogSource, RunId, StdioStream}; +use crate::host::logs::{LogChannel, RunId, StdioStream}; use crate::host::state::HostState; use crate::host_pattern::HostPattern; use crate::manifest::ResourceSection; @@ -181,12 +181,12 @@ fn build( .stdout(StdioStream::new( router.clone(), run.clone(), - LogSource::Stdout, + LogChannel::Stdout, )) .stderr(StdioStream::new( router.clone(), run.clone(), - LogSource::Stderr, + LogChannel::Stderr, )); if let Some(clocks) = &shared.clocks { builder.wall_clock(SharedWallClock(clocks.wall.clone())); diff --git a/crates/nexum-runtime/src/supervisor/tests/boot_refusals.rs b/crates/nexum-runtime/src/supervisor/tests/boot_refusals.rs index 3a980533..ed7dd7da 100644 --- a/crates/nexum-runtime/src/supervisor/tests/boot_refusals.rs +++ b/crates/nexum-runtime/src/supervisor/tests/boot_refusals.rs @@ -72,7 +72,7 @@ async fn boot_refuses_a_component_without_a_manifest() { let scenario = BootScenario::new(); let orphan = scenario.dir().join("orphan.wasm"); scenario - .module(Entry::new(ManifestSource::Beside).wasm(orphan)) + .module(Entry::new(ManifestInput::Beside).wasm(orphan)) .expect_refusal() .await .variant::(|e| { @@ -230,7 +230,7 @@ async fn a_component_policy_row_overrides_the_global_capability_set() { /// Two in-ceiling components still refuse together when their declared /// reservations cross `[policy.total]`; the refusal names the second. #[tokio::test] -async fn boot_refuses_an_oversubscribed_component_set() { +async fn boot_refuses_an_overcommitted_component_set() { BootScenario::new() .policy(PolicySection { total: TotalPolicy { diff --git a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs index 2f005d42..bf180cb4 100644 --- a/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs +++ b/crates/nexum-runtime/src/supervisor/tests/chain_gate.rs @@ -3,14 +3,14 @@ use super::*; #[tokio::test] -async fn empty_supervisor_returns_no_triggers() { +async fn empty_supervisor_returns_an_empty_source_plan() { let booted = BootScenario::over(mock_components()) .boot() .await .expect("an empty scenario boots"); - let plan = booted.supervisor.trigger_plan(); + let plan = booted.supervisor.source_plan(); assert!(plan.block_chains.is_empty()); - assert!(plan.event_triggers.is_empty()); + assert!(plan.event_sources.is_empty()); assert_eq!(plan.viability(0), Viability::Nothing); assert_eq!(booted.supervisor.module_count(), 0); } @@ -136,13 +136,13 @@ async fn a_validated_event_filter_survives_to_the_collected_stream() { .await .expect("the example boots alive"); - let triggers = booted.supervisor.trigger_plan().event_triggers; - assert_eq!(triggers.len(), 1, "the alive module contributes its stream"); - assert_eq!(triggers[0].module.as_str(), "example"); - assert_eq!(triggers[0].chain.id(), 1); - assert!(triggers[0].cursor_key.is_none(), "resume defaults to off"); + let sources = booted.supervisor.source_plan().event_sources; + assert_eq!(sources.len(), 1, "the alive module contributes its stream"); + assert_eq!(sources[0].module.as_str(), "example"); + assert_eq!(sources[0].chain.id(), 1); + assert!(sources[0].cursor_key.is_none(), "resume defaults to off"); // alloy `Filter` exposes no getter; assert through its serialization. - let serialized = serde_json::to_value(&triggers[0].filter) + let serialized = serde_json::to_value(&sources[0].filter) .unwrap() .to_string(); assert!( diff --git a/crates/nexum-runtime/src/supervisor/tests/e2e.rs b/crates/nexum-runtime/src/supervisor/tests/e2e.rs index b28165ef..f1200f35 100644 --- a/crates/nexum-runtime/src/supervisor/tests/e2e.rs +++ b/crates/nexum-runtime/src/supervisor/tests/e2e.rs @@ -306,7 +306,7 @@ async fn host_interface_records_are_retrievable_after_a_run() { assert!( page.records .iter() - .all(|r| r.source == LogSource::HostInterface), + .all(|r| r.channel == LogChannel::HostInterface), "the example module logs only through the host interface", ); assert!( @@ -344,7 +344,7 @@ async fn dying_run_leaves_a_panic_record() { let panic = page .records .iter() - .find(|r| r.source == LogSource::Panic) + .find(|r| r.channel == LogChannel::Panic) .expect("a panic record on the dead run"); assert_eq!(panic.level, Level::ERROR); assert!(panic.message.contains("terminated")); @@ -376,17 +376,18 @@ async fn facade_panic_leaves_stderr_host_interface_and_panic_records() { let runs = booted.logs().list_runs("panic-bomb"); assert_eq!(runs.len(), 1); let page = booted.logs().read(&runs[0].run, 0); - let find = |source: LogSource, needle: &str| { + let find = |channel: LogChannel, needle: &str| { page.records .iter() - .find(|r| r.source == source && r.message.contains(needle)) + .find(|r| r.channel == channel && r.message.contains(needle)) }; - let stderr = find(LogSource::Stderr, "detonated").expect("the hook's stderr line was captured"); + let stderr = + find(LogChannel::Stderr, "detonated").expect("the hook's stderr line was captured"); assert_eq!(stderr.level, Level::WARN, "stderr copy is warn"); let host = - find(LogSource::HostInterface, "detonated").expect("the hook's sink call was captured"); + find(LogChannel::HostInterface, "detonated").expect("the hook's sink call was captured"); assert_eq!(host.level, Level::ERROR, "sink copy is error"); let death = - find(LogSource::Panic, "terminated").expect("the supervisor synthesized the death record"); + find(LogChannel::Panic, "terminated").expect("the supervisor synthesized the death record"); assert_eq!(death.level, Level::ERROR, "death record is error"); } diff --git a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs index 1ee8ea2a..ca7a24cd 100644 --- a/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs +++ b/crates/nexum-runtime/src/supervisor/tests/lifecycle.rs @@ -62,13 +62,13 @@ async fn init_failure_marks_module_dead_excluding_dispatch_and_triggers() { 0, "no live module declares chain 11155111 blocks", ); - let plan = booted.supervisor.trigger_plan(); + let plan = booted.supervisor.source_plan(); assert!( plan.block_chains.is_empty(), "dead module must not contribute block chains", ); assert!( - plan.event_triggers.is_empty(), + plan.event_sources.is_empty(), "dead module must not contribute chain-log streams", ); assert_eq!( @@ -78,9 +78,9 @@ async fn init_failure_marks_module_dead_excluding_dispatch_and_triggers() { ); } -/// Positive control: the alive module's triggers survive the filter. +/// Positive control: the alive module's source survives the filter. #[tokio::test] -async fn alive_module_triggers_survive_alongside_dead_module() { +async fn alive_module_source_survives_alongside_dead_module() { let Some(price_alert_wasm) = module_wasm_or_skip("price-alert") else { return; }; @@ -103,7 +103,7 @@ async fn alive_module_triggers_survive_alongside_dead_module() { 1, "only the example is alive" ); - let plan = booted.supervisor.trigger_plan(); + let plan = booted.supervisor.source_plan(); assert_eq!( plan.block_chains.iter().map(|c| c.id()).collect::>(), vec![1], @@ -165,9 +165,9 @@ async fn dead_module_extension_kind_is_excluded_from_the_plan() { .await .expect("both modules load; only price-alert's init fails"); - let plan = booted.supervisor.trigger_plan(); + let plan = booted.supervisor.source_plan(); assert_eq!( - plan.extension_kinds + plan.demanded_extension_kinds .iter() .map(String::as_str) .collect::>(), @@ -207,8 +207,12 @@ async fn a_declared_extension_kind_alone_is_not_viable() { .await .expect("the example boots alive"); - let plan = booted.supervisor.trigger_plan(); - assert_eq!(plan.extension_kinds.len(), 1, "the live kind is declared"); + let plan = booted.supervisor.source_plan(); + assert_eq!( + plan.demanded_extension_kinds.len(), + 1, + "the live kind is declared" + ); assert_eq!( plan.viability(0), Viability::Nothing, diff --git a/crates/nexum-runtime/src/supervisor/tests/mod.rs b/crates/nexum-runtime/src/supervisor/tests/mod.rs index 72598e75..0f5a4ce8 100644 --- a/crates/nexum-runtime/src/supervisor/tests/mod.rs +++ b/crates/nexum-runtime/src/supervisor/tests/mod.rs @@ -24,20 +24,20 @@ use super::dispatch::with_dispatch_deadline; use super::prepass::{ NamespaceLedger, claim_namespace, enforce_total_reservation, unconfigured_chain, }; +use super::sources::{build_alloy_filter, wit_log}; use super::store::resolve_module_limits; -use super::triggers::{build_alloy_filter, wit_log}; use super::*; use crate::bindings::nexum; use crate::digest::{ContentDigest, DigestMismatch}; use crate::engine_config::{ ComponentPolicy, ModuleLimits, PolicyCeilings, PolicySection, ResolvedModuleLimits, TotalPolicy, }; -use crate::host::logs::LogSource; +use crate::host::logs::LogChannel; use crate::host::provider_pool::ProviderPool; use crate::manifest::{self, CapabilityError, CapabilityRegistry, ParseError, ResourceSection}; use crate::preset::CoreRuntime; use crate::test_utils::{ - BootScenario, Entry, ManifestSource, Refusal, TestManifest, example_wasm_or_skip, + BootScenario, Entry, ManifestInput, Refusal, TestManifest, example_wasm_or_skip, mock_components, module_wasm_or_skip, test_wasmtime_engine, }; diff --git a/crates/nexum-runtime/src/test_utils/harness.rs b/crates/nexum-runtime/src/test_utils/harness.rs index 9d78ac57..3b03d95e 100644 --- a/crates/nexum-runtime/src/test_utils/harness.rs +++ b/crates/nexum-runtime/src/test_utils/harness.rs @@ -21,7 +21,7 @@ use alloy_chains::Chain; use alloy_rpc_types_eth::{Header, Log}; use super::clock::ManualClock; -use super::manifest::ManifestSource; +use super::manifest::ManifestInput; use super::rpc::FakeNode; use super::scenario::{BootScenario, Booted, Entry}; use super::{HARNESS_POLL_INTERVAL, MockStateStore, MockTypes, Prebuilt}; @@ -35,7 +35,7 @@ use crate::host::logs::{LogPipeline, LogRecord}; /// backends. A manifest is mandatory. pub struct TestRuntimeBuilder { wasm: PathBuf, - manifest: ManifestSource, + manifest: ManifestInput, extensions: Vec>>, limits: ModuleLimits, chain: FakeNode, @@ -49,7 +49,7 @@ impl TestRuntime { pub fn builder(wasm: impl Into) -> TestRuntimeBuilder { TestRuntimeBuilder { wasm: wasm.into(), - manifest: ManifestSource::Beside, + manifest: ManifestInput::Beside, extensions: Vec::new(), limits: ModuleLimits::default(), chain: FakeNode::new(), @@ -63,13 +63,13 @@ impl TestRuntime { impl TestRuntimeBuilder { /// Load the manifest from an existing file. pub fn manifest_path(mut self, path: impl Into) -> Self { - self.manifest = ManifestSource::Path(path.into()); + self.manifest = ManifestInput::Path(path.into()); self } /// Write `toml` to a temp file at launch and load the module from it. pub fn manifest_inline(mut self, toml: impl Into) -> Self { - self.manifest = ManifestSource::Toml(toml.into()); + self.manifest = ManifestInput::Toml(toml.into()); self } @@ -324,8 +324,8 @@ mod tests { .await .expect("the on_trigger log line lands after dispatch"); assert_eq!( - record.source, - crate::host::logs::LogSource::HostInterface, + record.channel, + crate::host::logs::LogChannel::HostInterface, "the example module logs through the host interface", ); @@ -795,8 +795,8 @@ mod tests { // The line is a host-interface log carrying exactly the pinned // seconds, parsed back to guard against a substring false positive. assert_eq!( - record.source, - crate::host::logs::LogSource::HostInterface, + record.channel, + crate::host::logs::LogChannel::HostInterface, "the fixture logs through the host interface", ); let logged: u64 = record diff --git a/crates/nexum-runtime/src/test_utils/manifest.rs b/crates/nexum-runtime/src/test_utils/manifest.rs index 34a9bc0f..1c906201 100644 --- a/crates/nexum-runtime/src/test_utils/manifest.rs +++ b/crates/nexum-runtime/src/test_utils/manifest.rs @@ -5,7 +5,7 @@ use std::path::{Path, PathBuf}; /// How a test supplies the manifest: the three shapes the loader must /// handle, including the absent one. #[derive(Debug, Clone, derive_more::From)] -pub enum ManifestSource { +pub enum ManifestInput { /// No explicit path; the loader falls back to discovery beside the component. Beside, /// A path handed to the loader as-is, existing or not. @@ -16,7 +16,7 @@ pub enum ManifestSource { Toml(String), } -impl ManifestSource { +impl ManifestInput { /// Materialize inline text at `path`; [`Beside`](Self::Beside) resolves to nothing. pub fn resolve(&self, path: &Path) -> Option { match self { @@ -30,7 +30,7 @@ impl ManifestSource { } } -impl From for ManifestSource { +impl From for ManifestInput { fn from(manifest: TestManifest) -> Self { Self::Toml(manifest.to_toml()) } @@ -417,18 +417,18 @@ on = "event" let dir = tempfile::tempdir().expect("tempdir"); let at = dir.path().join("component.toml"); - assert_eq!(ManifestSource::Beside.resolve(&at), None); + assert_eq!(ManifestInput::Beside.resolve(&at), None); assert!(!at.exists(), "a discovered manifest writes nothing"); let explicit = dir.path().join("absent.toml"); assert_eq!( - ManifestSource::from(explicit.clone()).resolve(&at), + ManifestInput::from(explicit.clone()).resolve(&at), Some(explicit), "an explicit path passes through untouched", ); assert!(!at.exists(), "an explicit path writes nothing"); - let inline = ManifestSource::from(TestManifest::new("inline").cap("logging")); + let inline = ManifestInput::from(TestManifest::new("inline").cap("logging")); assert_eq!(inline.resolve(&at).as_deref(), Some(at.as_path())); assert_eq!(load_path(&at).name.as_str(), "inline"); } diff --git a/crates/nexum-runtime/src/test_utils/mod.rs b/crates/nexum-runtime/src/test_utils/mod.rs index 060f2157..784e4b10 100644 --- a/crates/nexum-runtime/src/test_utils/mod.rs +++ b/crates/nexum-runtime/src/test_utils/mod.rs @@ -45,7 +45,7 @@ pub mod wasm; pub use builders::Prebuilt; pub use harness::{TestRuntime, TestRuntimeBuilder}; -pub use manifest::{ManifestSource, TestManifest, manifest}; +pub use manifest::{ManifestInput, TestManifest, manifest}; pub use scenario::{BootScenario, Booted, Entry, Refusal}; pub use store::{MockStateHandle, MockStateStore}; pub use types::MockTypes; @@ -168,14 +168,14 @@ mod tests { } #[tokio::test] - async fn subscribe_blocks_yields_pushed_headers() { + async fn open_block_source_yields_pushed_headers() { let node = FakeNode::new(); let pool = node.pool(&[Chain::from_id(1)], HARNESS_POLL_INTERVAL); let mut header: alloy_rpc_types_eth::Header = alloy_rpc_types_eth::Header::default(); header.inner.number = 7; node.push_block(header); let mut stream = pool - .subscribe_blocks(Chain::from_id(1)) + .open_block_source(Chain::from_id(1)) .await .expect("block stream"); let item = stream @@ -187,12 +187,12 @@ mod tests { } #[tokio::test] - async fn watch_chain_logs_yields_pushed_logs() { + async fn open_event_source_yields_pushed_logs() { let node = FakeNode::new(); let pool = node.pool(&[Chain::from_id(1)], HARNESS_POLL_INTERVAL); node.push_chain_log(alloy_rpc_types_eth::Log::default()); let mut stream = pool - .watch_chain_logs(Chain::from_id(1), Default::default(), 1) + .open_event_source(Chain::from_id(1), Default::default(), 1) .expect("chain-log poller stream"); let batch = stream .next() diff --git a/crates/nexum-runtime/src/test_utils/scenario.rs b/crates/nexum-runtime/src/test_utils/scenario.rs index de43fed2..54559398 100644 --- a/crates/nexum-runtime/src/test_utils/scenario.rs +++ b/crates/nexum-runtime/src/test_utils/scenario.rs @@ -8,7 +8,7 @@ use alloy_chains::Chain; use derive_more::From; use tempfile::TempDir; -use super::manifest::{ManifestSource, TestManifest}; +use super::manifest::{ManifestInput, TestManifest}; use super::{in_memory_logs, test_chain_configs}; use crate::digest::ContentDigest; use crate::engine_config::{ChainConfig, EngineConfig, ModuleEntry, ModuleLimits, PolicySection}; @@ -25,13 +25,13 @@ use crate::test_utils::wasm::test_wasmtime_engine; pub struct Entry { id: Option, wasm: Option, - manifest: ManifestSource, + manifest: ManifestInput, digest: Option, } impl Entry { /// An entry loading `manifest` on the scenario-wide component. - pub fn new(manifest: impl Into) -> Self { + pub fn new(manifest: impl Into) -> Self { Self { id: None, wasm: None, @@ -594,7 +594,7 @@ mod tests { let scenario = BootScenario::new(); let orphan = scenario.dir().join("orphan.wasm"); scenario - .module(Entry::new(ManifestSource::Beside).wasm(orphan)) + .module(Entry::new(ManifestInput::Beside).wasm(orphan)) .expect_refusal() .await .variant::(|e| { diff --git a/docs/02-modules-triggers-packaging.md b/docs/02-modules-triggers-packaging.md index bf4f7527..2e4bbcd8 100644 --- a/docs/02-modules-triggers-packaging.md +++ b/docs/02-modules-triggers-packaging.md @@ -67,7 +67,7 @@ Key design points: - **`[[trigger]]` tables are declarative.** A component does not open its own sources imperatively. The runtime loads each component and runs its `init` first, then derives the plan from the booted supervisor and opens the sources. - `call_init` runs during load in `crates/nexum-runtime/src/supervisor/load.rs`, and `trigger_plan` reads the already-booted supervisor in `crates/nexum-runtime/src/supervisor/triggers.rs`. + `call_init` runs during load in `crates/nexum-runtime/src/supervisor/load.rs`, and `source_plan` reads the already-booted supervisor in `crates/nexum-runtime/src/supervisor/sources.rs`. - **`[dependencies]` drives what the runtime links.** Each key names a host capability, and its table carries the attributes that qualify it. A component that declares `http` imports `wasi:http/outgoing-handler`, the SDK's `http::fetch` helper wraps it, and the host checks every outgoing request against the `hosts` list on the `http` dependency. diff --git a/docs/production.md b/docs/production.md index 5928ceb8..7b051697 100644 --- a/docs/production.md +++ b/docs/production.md @@ -157,7 +157,7 @@ A `resume = true` trigger then replays the in-flight log at the next start; a bl The engine emits JSON `tracing` events on stdout, one flat object per line. `--pretty-logs` switches to the human format. Every event carries `timestamp`, `level`, `target` (the crate and module path), and `message`. -A guest log line is mirrored into host tracing at the guest's own level with `module`, `run`, and `source` fields, and guest stdout and stderr are captured line by line. +A guest log line is mirrored into host tracing at the guest's own level with `module`, `run`, and `channel` fields, and guest stdout and stderr are captured line by line. Production should not see `ERROR` from `nexum_runtime::*`. `RUST_LOG` wins over `[engine] log_level`, which is itself a full `EnvFilter` directive rather than a bare level. @@ -269,7 +269,7 @@ The chain interface has no batch verb; the guest SDK lowers a batch of RPC reque `nexum_runtime_chain_request_total{outcome="err"}` is the degradation signal. Resource ceilings live in `engine.toml` `[policy]` and apply to every component; a `[policy.component.]` row, keyed on `[[modules]].id`, overrides them for one. -`[policy.total].max_memory_bytes` bounds the summed reservations, and an oversubscribed set refuses at boot naming the entry that crossed it. +`[policy.total].max_memory_bytes` bounds the summed reservations, and an overcommitted set refuses at boot naming the entry that crossed it. A `[component.resources]` field in a manifest narrows a ceiling for one component and can never widen it. A component that consistently traps on fuel exhaustion is a bug, not a tuning miss.