diff --git a/crates/nexum-runtime/src/manifest/types.rs b/crates/nexum-runtime/src/manifest/types.rs index 279ca7de..628641d2 100644 --- a/crates/nexum-runtime/src/manifest/types.rs +++ b/crates/nexum-runtime/src/manifest/types.rs @@ -69,6 +69,20 @@ pub enum Subscription { /// subscription matches every event from the address(es). #[serde(default)] event_signature: Option, + /// Resume across engine restarts. When `true` the host persists a + /// durable per-subscription cursor and re-opens the log poller + /// from just after the last dispatched block, instead of at the + /// current head. Delivery is then at-least-once, so the module must + /// tolerate redelivery (the chassis idempotency journal already + /// dedups it). + #[serde(default)] + resume: bool, + /// Optional cap on how far back a `resume` subscription will + /// backfill, in blocks. `None` (the default) backfills the entire + /// gap with no loss; set it only for a consumer that explicitly + /// tolerates dropping the oldest missed blocks. + #[serde(default)] + max_lookback: Option, }, /// Cron-scheduled tick. 0.2 parses but does not dispatch; the /// supervisor emits a warning so the operator knows the diff --git a/crates/nexum-runtime/src/runtime/event_loop.rs b/crates/nexum-runtime/src/runtime/event_loop.rs index ad58aac0..1045e79d 100644 --- a/crates/nexum-runtime/src/runtime/event_loop.rs +++ b/crates/nexum-runtime/src/runtime/event_loop.rs @@ -28,6 +28,7 @@ //! an injectable [`TaskExecutor`] and their handles collected into a //! [`TaskSet`] the loop drains on shutdown. +use std::sync::Arc; use std::time::{Duration, Instant}; use alloy_chains::Chain; @@ -42,7 +43,7 @@ use crate::host::component::{ChainProvider, RuntimeTypes}; use crate::host::provider_pool::ProviderError; use crate::runtime::restart_policy::backoff_for; use crate::runtime::task::{TaskExecutor, TaskExit, TaskSet}; -use crate::supervisor::Supervisor; +use crate::supervisor::{ChainLogSub, Supervisor}; /// Errors carried by the tagged block / chain-log streams that the /// supervisor consumes. Library-side code keeps `anyhow::Error` out @@ -118,7 +119,7 @@ where /// [`open_block_streams`]). pub fn open_chain_log_streams( pool: &C, - subs: Vec<(String, Chain, alloy_rpc_types_eth::Filter)>, + subs: Vec, executor: &dyn TaskExecutor, tasks: &mut TaskSet, ) -> Vec @@ -126,13 +127,18 @@ where C: ChainProvider + Clone + Send + Sync + 'static, { let mut streams = Vec::new(); - for (module, chain, filter) in subs { - let (tx, rx) = mpsc::channel::< - Result<(String, Chain, alloy_rpc_types_eth::Log), StreamError>, - >(RECONNECT_CHANNEL_BUF); + for sub in subs { + let (tx, rx) = mpsc::channel::(RECONNECT_CHANNEL_BUF); let pool = pool.clone(); + let resume = ChainLogResume { + // The cursor key is constant per subscription and cloned onto every + // log; `Arc` keeps that clone cheap. + cursor_key: sub.cursor_key.map(Arc::from), + initial_cursor: sub.initial_cursor, + max_lookback: sub.max_lookback, + }; tasks.push(executor.spawn(Box::pin(reconnecting_chain_log_task( - pool, module, chain, filter, tx, + pool, sub.module, sub.chain, sub.filter, resume, tx, )))); let tagged: TaggedChainLogStream = Box::pin(receiver_stream(rx)); streams.push(tagged); @@ -237,6 +243,18 @@ where } } +/// Per-subscription resume and backfill knobs for a chain-log task. +struct ChainLogResume { + /// Durable cursor key, `Some` for a `resume` subscription; the block + /// under it seeds `initial_cursor`. + cursor_key: Option>, + /// Persisted resume block read at boot; the first open starts here. + initial_cursor: Option, + /// Opt-in cap (in blocks) on how far back the poller backfills; `None` + /// backfills the whole gap. + max_lookback: Option, +} + /// Poller-backed loop for a single (module, chain) chain-log /// subscription. Instead of `eth_subscribe(logs)` - which silently /// drops events emitted during a WebSocket reconnect - it drives @@ -246,24 +264,35 @@ where /// transport's own retries), or a reorg deeper than the poller's /// retained history, ends the poller stream; this loop then re-opens /// from the block after the last one it delivered and backfills the -/// entire missed range (no lookback cap; nothing is skipped), with -/// exponential backoff. +/// entire missed range (nothing skipped) with exponential backoff - +/// unless the subscription set `max_lookback`, which bounds how far back +/// the backfill reaches. async fn reconnecting_chain_log_task( pool: C, module: String, chain: Chain, filter: alloy_rpc_types_eth::Filter, - tx: mpsc::Sender>, + resume: ChainLogResume, + tx: mpsc::Sender, ) -> TaskExit where C: ChainProvider + Send + Sync + 'static, { + let ChainLogResume { + cursor_key, + initial_cursor, + max_lookback, + } = resume; let chain_id = chain.id(); let mut attempt: u32 = 0; let mut last_event: Option = None; // Highest block whose logs we have delivered; the resume point after a // poller re-open, so the missed range is synced back rather than skipped. let mut last_seen_block: Option = None; + // Persisted resume cursor, consumed on the first open: for a `resume` + // subscription the poller starts here (replaying the block in full) + // rather than at head. `None` for a fresh or non-resume subscription. + let mut boot_resume: Option = initial_cursor; loop { let head = match pool.block_number(chain).await { Ok(head) => head, @@ -282,12 +311,37 @@ where continue; } }; - // First open starts at the head - no history replay on boot, and - // the across-restart gap (logs mined before the engine started) is - // out of scope since nothing persists a cursor. A re-open resumes - // just after the last delivered block and backfills the entire gap - // so no event is ever skipped. - let start_block = poller_resume_block(last_seen_block, head); + // Choosing the poller start block: + // - `boot_resume` (persisted cursor, first open only): resume AT the + // cursor block, replaying it in full so a mid-block crash before + // the restart loses nothing. Never past head: a reorg that left + // the cursor ahead of head starts at head and lets the poller + // catch up. + // - otherwise a re-open resumes just after the last delivered block + // (within-process gap-free), or at head on the very first open. + // Either way the whole gap is backfilled with no lower floor, so + // nothing is skipped unless the subscription set `max_lookback`. + let mut start_block = match boot_resume.take() { + Some(resume) => resume.min(head), + None => poller_resume_block(last_seen_block, head), + }; + // Opt-in bound: `max_lookback` caps how far back a resume + // subscription backfills. The default (`None`) backfills fully; a + // set cap clamps the start up to `head - cap` and surfaces the + // dropped oldest blocks. + if let Some(cap) = max_lookback { + let floor = head.saturating_sub(cap); + if start_block < floor { + warn!( + module = %module, + chain_id, + skipped_from = start_block, + skipped_to = floor, + "chain-log gap exceeds max_lookback - skipping the oldest missed blocks", + ); + start_block = floor; + } + } // A large gap is backfilled in full (never skipped); surface it so a long // catch-up is visible rather than looking like a stall. if head.saturating_sub(start_block) >= LARGE_GAP_LOG_THRESHOLD { @@ -344,7 +398,8 @@ where last_seen_block = Some(last_seen_block.map_or(block, |seen| seen.max(block))); } - if tx.send(Ok((module.clone(), chain, log))).await.is_err() { + let tagged = Ok((module.clone(), chain, log, cursor_key.clone())); + if tx.send(tagged).await.is_err() { return TaskExit::ReceiverGone; } } @@ -394,12 +449,14 @@ pub type TaggedBlockStream = std::pin::Pin< + Send, >, >; -pub type TaggedChainLogStream = std::pin::Pin< - Box< - dyn futures::Stream> - + Send, - >, ->; +/// One item on a tagged chain-log stream: `(module, chain, log, +/// cursor_key)` or a stream error. `cursor_key` is `Some` for a `resume` +/// subscription (constant per subscription; `Arc` for a cheap per-log +/// clone) and threads the durable cursor key through to the dispatch site. +pub type TaggedChainLog = + Result<(String, Chain, alloy_rpc_types_eth::Log, Option>), StreamError>; +pub type TaggedChainLogStream = + std::pin::Pin + Send>>; /// Drive the supervisor with events until `shutdown` resolves. /// @@ -446,7 +503,12 @@ pub async fn run( Block(nexum::host::types::Block), // The alloy `Log` is boxed so the `Chain` tag does not push // the enum past the large-variant lint threshold. - ChainLog(String, Chain, Box), + ChainLog( + String, + Chain, + Box, + Option>, + ), Shutdown, StreamPanic(&'static str), } @@ -467,7 +529,9 @@ pub async fn run( None => NextEvent::StreamPanic("block"), }, next = chain_logs.next() => match next { - Some(Ok((module, chain, log))) => NextEvent::ChainLog(module, chain, Box::new(log)), + Some(Ok((module, chain, log, cursor_key))) => { + NextEvent::ChainLog(module, chain, Box::new(log), cursor_key) + } Some(Err(err)) => { warn!(error = %err, "chain-log stream error - continuing"); continue; @@ -481,8 +545,10 @@ pub async fn run( supervisor.dispatch_block(block).await; dispatched_blocks += 1; } - NextEvent::ChainLog(module, chain, log) => { - supervisor.dispatch_chain_log(&module, chain, *log).await; + NextEvent::ChainLog(module, chain, log, cursor_key) => { + supervisor + .dispatch_chain_log(&module, chain, *log, cursor_key.as_deref()) + .await; dispatched_chain_logs += 1; } NextEvent::Shutdown => { diff --git a/crates/nexum-runtime/src/supervisor.rs b/crates/nexum-runtime/src/supervisor.rs index 737a869f..0b428d6b 100644 --- a/crates/nexum-runtime/src/supervisor.rs +++ b/crates/nexum-runtime/src/supervisor.rs @@ -540,7 +540,7 @@ impl Supervisor { /// chain, filter)` triple the event loop opens against the /// matching alloy provider; the resulting stream tags every log /// with `module_name` so `dispatch_chain_log` routes correctly. - pub fn chain_log_subscriptions(&self) -> Vec<(String, Chain, alloy_rpc_types_eth::Filter)> { + pub fn chain_log_subscriptions(&self) -> Vec { let mut out = Vec::new(); for module in &self.modules { for sub in &module.subscriptions { @@ -548,11 +548,35 @@ impl Supervisor { chain_id, address, event_signature, + resume, + max_lookback, } = sub { match build_alloy_filter(address.as_deref(), event_signature.as_deref()) { Ok(filter) => { - out.push((module.name.clone(), Chain::from_id(*chain_id), filter)) + let chain = Chain::from_id(*chain_id); + // A `resume` subscription gets a durable cursor + // key and its persisted resume point, read once + // here at boot; others start at head as before. + let (cursor_key, initial_cursor) = if *resume { + let key = chainlog_cursor_key( + chain, + address.as_deref(), + event_signature.as_deref(), + ); + let seed = self.read_chain_log_cursor(&module.name, &key); + (Some(key), seed) + } else { + (None, None) + }; + out.push(ChainLogSub { + module: module.name.clone(), + chain, + filter, + cursor_key, + initial_cursor, + max_lookback: *max_lookback, + }); } Err(err) => warn!( module = %module.name, @@ -567,6 +591,15 @@ impl Supervisor { out } + /// Read the persisted resume cursor for a chain-log subscription, or + /// `None` when absent / unreadable - both treated as "start at head". + fn read_chain_log_cursor(&self, module: &str, key: &str) -> Option { + let handle = self.components.store.module(module).ok()?; + let bytes = handle.get(key).ok()??; + let arr: [u8; 8] = bytes.try_into().ok()?; + Some(u64::from_le_bytes(arr)) + } + /// Dispatch a block event to every module subscribed to /// `block.chain_id`. Returns the number of modules invoked. /// Modules that trap are marked dead and excluded from future dispatch. @@ -714,6 +747,7 @@ impl Supervisor { module_name: &str, chain: Chain, log: alloy_rpc_types_eth::Log, + cursor_key: Option<&str>, ) -> bool { let now = std::time::Instant::now(); let Some(idx) = self.modules.iter().position(|m| m.name == module_name) else { @@ -744,16 +778,46 @@ impl Supervisor { return false; } - let block_number = log.block_number.unwrap_or_default(); + let block_number = log.block_number; let event = nexum::host::types::Event::ChainLogs(nexum::host::types::ChainLogs { chain_id: chain.id(), logs: vec![nexum::host::types::ChainLog::from(&log)], }); - matches!( - self.dispatch_to(idx, chain, "chain-log", block_number, &event) - .await, + let ok = matches!( + self.dispatch_to( + idx, + chain, + "chain-log", + block_number.unwrap_or_default(), + &event + ) + .await, DispatchOutcome::Ok, - ) + ); + // Persist the resume cursor only after a successful dispatch, so a + // block is never recorded as done before the module processed it. + // Advancing to the highest dispatched block is enough; a re-dispatch + // of the same block after a restart is idempotent (at-least-once). + if ok && let (Some(key), Some(block)) = (cursor_key, block_number) { + let store = self.components.store.clone(); + match store.module(module_name) { + Ok(ms) => { + if let Err(e) = ms.set(key, &block.to_le_bytes()) { + warn!( + module = %module_name, + error = %e, + "failed to persist chain-log cursor", + ); + } + } + Err(e) => warn!( + module = %module_name, + error = %e, + "failed to open module store for chain-log cursor", + ), + } + } + ok } /// Shared per-module dispatch path: refuel, call `on_event`, and @@ -1032,6 +1096,51 @@ fn progress_key(chain: Chain) -> String { format!("last_dispatched_block:{}", chain.id()) } +/// A resolved chain-log subscription for the event loop: the owning +/// module, the chain + alloy `Filter`, and - when the subscription opted +/// into `resume` - the durable cursor key plus the block to resume from +/// (read from the store at boot). +pub struct ChainLogSub { + /// Module that declared the subscription; also its store namespace. + pub module: String, + /// Chain the filter applies to. + pub chain: Chain, + /// Alloy filter the poller opens with. + pub filter: alloy_rpc_types_eth::Filter, + /// `Some` iff `resume = true`: the store key the resume cursor is read + /// and written under. + pub cursor_key: Option, + /// The persisted resume block, read at boot for a `resume` + /// subscription; `None` on first run or when `resume` is off. + pub initial_cursor: Option, + /// Opt-in cap on how far back the poller backfills, in blocks. `None` + /// backfills the whole gap; `Some(cap)` bounds the start to + /// `head - cap`, dropping the oldest missed blocks. + pub max_lookback: Option, +} + +/// Durable resume-cursor key for a chain-log subscription. Derived from +/// the normalized manifest inputs - NOT the alloy `Filter`, whose hash +/// uses a process-randomized `HashSet` and is not reproducible across +/// restarts. Stable and independent of `[[subscription]]` ordering. The +/// module name is the store namespace, so it is not part of the digest. +fn chainlog_cursor_key( + chain: Chain, + address: Option<&str>, + event_signature: Option<&str>, +) -> String { + let normalized = format!( + "{}|{}|{}", + chain.id(), + address.unwrap_or("").to_ascii_lowercase(), + event_signature.unwrap_or("").to_ascii_lowercase(), + ); + format!( + "chainlog_cursor:{:x}", + alloy_primitives::keccak256(normalized.as_bytes()) + ) +} + impl From<&alloy_rpc_types_eth::Log> for nexum::host::types::ChainLog { /// Project an alloy `Log` onto the WIT `chain-log` record, preserving every /// RPC field so the guest reconstructs the alloy log without loss. The chain diff --git a/crates/nexum-runtime/src/supervisor/tests.rs b/crates/nexum-runtime/src/supervisor/tests.rs index b26ec409..5568d259 100644 --- a/crates/nexum-runtime/src/supervisor/tests.rs +++ b/crates/nexum-runtime/src/supervisor/tests.rs @@ -1595,3 +1595,42 @@ fn project_chain_log_leaves_pending_fields_none() { assert!(projected.data.is_empty()); assert!(!projected.removed); } + +#[test] +fn chainlog_cursor_key_is_stable_and_case_insensitive() { + // The durable key must be reproducible across restarts (unlike the + // alloy `Filter` hash, which uses a process-randomized HashSet) and + // must normalise hex case. + let a = chainlog_cursor_key(Chain::from_id(1), Some("0xAbC"), Some("0xDeF")); + let b = chainlog_cursor_key(Chain::from_id(1), Some("0xabc"), Some("0xdef")); + assert_eq!(a, b, "hex case must not change the key"); + assert!( + a.starts_with("chainlog_cursor:"), + "key carries the prefix: {a}" + ); +} + +#[test] +fn chainlog_cursor_key_differs_by_each_input() { + let base = chainlog_cursor_key(Chain::from_id(1), Some("0xabc"), Some("0xdef")); + assert_ne!( + base, + chainlog_cursor_key(Chain::from_id(10), Some("0xabc"), Some("0xdef")), + "chain id is part of the key", + ); + assert_ne!( + base, + chainlog_cursor_key(Chain::from_id(1), Some("0x999"), Some("0xdef")), + "address is part of the key", + ); + assert_ne!( + base, + chainlog_cursor_key(Chain::from_id(1), Some("0xabc"), None), + "topic presence changes the key", + ); + assert_ne!( + base, + chainlog_cursor_key(Chain::from_id(1), None, Some("0xdef")), + "address presence changes the key", + ); +} diff --git a/crates/shepherd-cow-host/tests/cow_boot.rs b/crates/shepherd-cow-host/tests/cow_boot.rs index 2242a363..f3a75d9a 100644 --- a/crates/shepherd-cow-host/tests/cow_boot.rs +++ b/crates/shepherd-cow-host/tests/cow_boot.rs @@ -189,7 +189,12 @@ async fn e2e_ethflow_watcher_log_dispatch() { // alive. let synthetic_log = alloy_rpc_types_eth::Log::default(); let dispatched = supervisor - .dispatch_chain_log("ethflow-watcher", Chain::from_id(SEPOLIA), synthetic_log) + .dispatch_chain_log( + "ethflow-watcher", + Chain::from_id(SEPOLIA), + synthetic_log, + None, + ) .await; assert!(dispatched); assert_eq!(supervisor.alive_count(), 1);