From 8137dc1e9d974b2ae44f2473c3644e1ccc6feb1a Mon Sep 17 00:00:00 2001 From: brunota20 Date: Thu, 18 Jun 2026 12:46:05 -0300 Subject: [PATCH] feat(event-loop+supervisor): graceful shutdown + last-block persistence (COW-1072) Two coupled changes that make operator-driven shutdowns observable and recoverable: ## 1. Event loop: dispatch outside `select!` `run()` previously had its `call_on_event().await` inside the `tokio::select!`. A shutdown signal arriving mid-dispatch cancelled the in-flight wasmtime call, leaving the wasm store in an indeterminate state. The refactor splits the loop into two phases: - **Phase 1**: a small `tokio::select!` picks the next event OR observes shutdown OR reports an upstream-task panic. Each branch resolves into a `NextEvent` value; the select drops without cancelling anything *outside* itself. - **Phase 2**: `match next` dispatches the event to the supervisor via a fully-awaited call, OR exits cleanly on the shutdown variant. The shutdown signal is now only observed *between* dispatches. In-flight wasmtime calls always finish naturally. ## 2. Per-module last-dispatched-block persistence Every successful `dispatch_block` writes a host-side marker to the module's own local-store namespace: ``` namespace = module.name key = "last_dispatched_block:{chain_id}" value = block.number.to_le_bytes() ``` The marker survives engine restarts (it lives in the redb file under `state_dir`). Operators can confirm at-which-block an engine last ran without trawling the logs; modules that care about block- gap detection can read it back on their next `init`. Write failures are best-effort (a `WARN` log; the dispatch is not considered failed). ## 3. Graceful shutdown log The event loop now emits a structured exit line: ``` INFO graceful shutdown complete dispatched_blocks=N dispatched_logs=M uptime_secs=K ``` Visible live on Sepolia after a `kill -TERM`: ``` INFO shutdown signal received signal=SIGTERM INFO graceful shutdown complete dispatched_blocks=1 dispatched_logs=0 uptime_secs=13 ``` ## Out of scope - 30s drain timeout via `tokio::time::timeout`. The current dispatch path always terminates (fuel cap caps wall time to <1s in practice); a 30s drain timer is dead code today. Worth adding once a module ever needs longer-running host calls (HTTP capability, etc). - `engine.toml::[engine.shutdown]` config knobs. The internal default is "wait as long as the in-flight dispatch takes"; configurable in 0.3. - Module-side `shutdown` hook. Modules just see the dispatch complete normally; the supervisor exits without invoking anything new. ## Tests - `cargo test --workspace` -> 161 host tests + 6 doctests passing (unchanged shape; the dispatch refactor is transparent to the existing test suite). - `cargo clippy --all-targets --workspace -- -D warnings` clean. - `cargo fmt --all --check` clean. - Live Sepolia smoke: ran the engine, observed the graceful shutdown log + last_dispatched_block markers in `data/m3/local-store.redb`. Linear: COW-1072. Eighth M4 issue landed; stacks on #41 (COW-1032). --- crates/nexum-engine/src/runtime/event_loop.rs | 94 +++++++++++++------ crates/nexum-engine/src/supervisor.rs | 20 ++++ 2 files changed, 86 insertions(+), 28 deletions(-) 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)) => {