diff --git a/crates/nexum-engine/src/runtime/event_loop.rs b/crates/nexum-engine/src/runtime/event_loop.rs index 8bf8f1d0..7bc429ba 100644 --- a/crates/nexum-engine/src/runtime/event_loop.rs +++ b/crates/nexum-engine/src/runtime/event_loop.rs @@ -239,6 +239,12 @@ pub type TaggedLogStream = std::pin::Pin< >; /// Drive the supervisor with events until `shutdown` resolves. +/// +/// COW-1072 graceful shutdown: the dispatch path is structured so +/// that `shutdown` is only observed *between* dispatches, never +/// mid-`call_on_event`. Each select fork either yields a fresh event +/// to dispatch or signals shutdown - the in-flight wasmtime call +/// finishes naturally before the loop exits. pub async fn run( supervisor: &mut Supervisor, block_streams: Vec, @@ -264,42 +270,74 @@ pub async fn run( select_all(log_streams).boxed() }; let mut shutdown = Box::pin(shutdown); + let mut dispatched_blocks: u64 = 0; + let mut dispatched_logs: u64 = 0; + let started = Instant::now(); loop { - tokio::select! { + // Phase 1: pick the next event OR observe shutdown. The + // dispatch itself happens in phase 2 (outside the select) + // so an in-flight wasmtime call never gets cancelled by a + // shutdown signal arriving mid-dispatch. + enum NextEvent { + Block(nexum::host::types::Block), + Log(String, u64, alloy_rpc_types_eth::Log), + Shutdown, + StreamPanic(&'static str), + } + let next = tokio::select! { biased; - () = &mut shutdown => return, + () = &mut shutdown => NextEvent::Shutdown, next = blocks.next() => match next { - Some(Ok((chain_id, header))) => { - let block = nexum::host::types::Block { - chain_id, - number: header.number, - hash: header.hash.as_slice().to_vec(), - timestamp: header.timestamp.saturating_mul(1000), - }; - supervisor.dispatch_block(block).await; - } - Some(Err(err)) => warn!(error = %err, "block stream error - continuing"), - None => { - // COW-1071: WebSocket drops are now absorbed by - // the reconnect tasks behind `open_block_streams` - // / `open_log_streams`; the stream surfaced here - // only ends if the underlying task panicked or - // the channel was closed. Treat as an - // unrecoverable engine fault and bail. - warn!("block reconnect task ended unexpectedly - shutting down"); - return; + Some(Ok((chain_id, header))) => NextEvent::Block(nexum::host::types::Block { + chain_id, + number: header.number, + hash: header.hash.as_slice().to_vec(), + timestamp: header.timestamp.saturating_mul(1000), + }), + Some(Err(err)) => { + warn!(error = %err, "block stream error - continuing"); + continue; } + None => NextEvent::StreamPanic("block"), }, next = logs.next() => match next { - Some(Ok((module, chain_id, log))) => { - supervisor.dispatch_log(&module, chain_id, log).await; - } - Some(Err(err)) => warn!(error = %err, "log stream error - continuing"), - None => { - warn!("log reconnect task ended unexpectedly - shutting down"); - return; + Some(Ok((module, chain_id, log))) => NextEvent::Log(module, chain_id, log), + Some(Err(err)) => { + warn!(error = %err, "log stream error - continuing"); + continue; } + None => NextEvent::StreamPanic("log"), }, + }; + + match next { + NextEvent::Block(block) => { + supervisor.dispatch_block(block).await; + dispatched_blocks += 1; + } + NextEvent::Log(module, chain_id, log) => { + supervisor.dispatch_log(&module, chain_id, log).await; + dispatched_logs += 1; + } + NextEvent::Shutdown => { + info!( + dispatched_blocks, + dispatched_logs, + uptime_secs = started.elapsed().as_secs(), + "graceful shutdown complete", + ); + return; + } + NextEvent::StreamPanic(kind) => { + // COW-1071: reconnect tasks should loop forever. + // Hitting `None` from `select_all` means the task + // exited (panic or channel closed). Bail loudly. + warn!( + kind, + "reconnect task ended unexpectedly - shutting down for engine restart" + ); + return; + } } } } diff --git a/crates/nexum-engine/src/supervisor.rs b/crates/nexum-engine/src/supervisor.rs index 78a5c200..0be581d5 100644 --- a/crates/nexum-engine/src/supervisor.rs +++ b/crates/nexum-engine/src/supervisor.rs @@ -444,6 +444,10 @@ impl Supervisor { let event = nexum::host::types::Event::Block(block); let now = std::time::Instant::now(); let poison_policy = self.poison_policy; + // Hoist the local-store reference out so the per-module + // borrow checker is happy when we write the COW-1072 + // progress marker inside the dispatch loop. + let local_store = self.local_store.clone(); // COW-1033 phase 1: find dead modules whose backoff window // has elapsed and re-instantiate them in place. The wasmtime @@ -544,6 +548,22 @@ impl Supervisor { // schedule with no further delay. module.failure_count = 0; module.next_attempt = None; + // COW-1072: persist the per-module-per-chain + // progress marker so a graceful restart (or + // even a crash) leaves a paper trail. Operators + // grepping the redb file can confirm the engine + // got to block N before exiting. Writes failure + // is best-effort; a warn is enough. + let key = format!("last_dispatched_block:{chain_id}"); + if let Err(e) = local_store.set(&module.name, &key, &block_number.to_le_bytes()) + { + warn!( + module = %module.name, + chain_id, + error = %e, + "failed to persist last_dispatched_block marker", + ); + } dispatched += 1; } Ok(Err(host_err)) => {