From 2bf6b34cb9ffb07a34e4c4b8cbfef101aaad9c85 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 27 Aug 2026 16:40:38 +0400 Subject: [PATCH] feat(nucleus): extend shutdown coordination --- nucleus/README.md | 2 ++ nucleus/src/shutdown.rs | 66 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 61 insertions(+), 7 deletions(-) diff --git a/nucleus/README.md b/nucleus/README.md index 219947f7..eeb70264 100644 --- a/nucleus/README.md +++ b/nucleus/README.md @@ -12,6 +12,8 @@ zero when the system clock predates the epoch. Its default feature set is empty. - `shutdown`: ordered cancellation, service handles, and termination reporting. The pacemaker quiesces execution and terminally syncs the ledger before the sequencer and appender tier; remaining backing services stop afterward. + Termination returns the strongest reason observed while draining so a later + service failure cannot be hidden by an earlier clean report. Dropping the manager cancels every tier without waiting for services to stop. - `notifier`: the one-shot, non-resetting `EventNotifier` latch. - `ledger`: shared block-boundary metadata, including each block's locally diff --git a/nucleus/src/shutdown.rs b/nucleus/src/shutdown.rs index 3dd2396d..7de4772b 100644 --- a/nucleus/src/shutdown.rs +++ b/nucleus/src/shutdown.rs @@ -9,7 +9,7 @@ use std::{ error::Error, - io, + io, mem, time::{Duration, Instant}, }; @@ -27,6 +27,20 @@ type HandleFuture = BoxFuture<'static, (Service, ShutdownTier, ShutdownReason)>; /// Background service tracked by the shutdown manager. #[derive(Clone, Copy, Debug)] pub enum Service { + /// Leader JSON-RPC and WebSocket ingress. + Rpc, + /// Process metrics endpoint. + Metrics, + /// Leader base-chain startup setup. + OnchainSetup, + /// Program-scheduled task service. + TaskScheduler, + /// Scheduled base-chain intent execution service. + IntentExecution, + /// Observed undelegation request service. + UndelegationRequests, + /// Periodic validator fee claiming service. + FeeClaim, /// Ledger append worker. LedgerAppender, /// Ledger read worker. @@ -54,11 +68,13 @@ impl Service { fn tier(&self) -> ShutdownTier { use Service::*; match self { - ReplicationClient => ShutdownTier::One, + Rpc | OnchainSetup | TaskScheduler | IntentExecution | UndelegationRequests + | FeeClaim | ReplicationClient => ShutdownTier::One, PaceMaker => ShutdownTier::Two, // The pacemaker drains the sequencer and sends the appender's final // sync before either service reaches this tier. Sequencer | LedgerAppender => ShutdownTier::Three, + Metrics => ShutdownTier::Four, _ => ShutdownTier::Four, } } @@ -113,23 +129,27 @@ impl ShutdownManager { /// /// Each tier gets `TIMEOUT` to report before the next tier is /// cancelled. Already terminated services are skipped by their tier. - pub async fn terminate(&mut self) { + pub async fn terminate(&mut self) -> ShutdownReason { info!("initiating graceful shutdown of the engine"); let start = Instant::now(); let mut timers = [start; ShutdownTier::COUNT]; + let mut outcome = ShutdownReason::Signalled; for tier in ShutdownTier::ORDER { timers[tier as usize] = Instant::now(); self.tokens[tier as usize].cancel(); if self.pending[tier as usize] == 0 { continue; } - if timeout(TIMEOUT, self.drain(tier, &timers)).await.is_err() { + if let Err(e) = timeout(TIMEOUT, self.drain(tier, &timers, &mut outcome)).await { let remaining = self.pending[tier as usize]; let elapsed = timers[tier as usize].elapsed(); warn!(?tier, remaining, ?elapsed, "shutdown tier timed out"); + let error = Box::new(io::Error::from(e)); + outcome = outcome.combine(ShutdownReason::Error(error)); } } info!(elapsed = ?start.elapsed(), "engine shutdown complete"); + outcome } /// Register a service and return its cancellation handle. @@ -148,7 +168,12 @@ impl ShutdownManager { } } - async fn drain(&mut self, tier: ShutdownTier, timers: &[Instant]) { + async fn drain( + &mut self, + tier: ShutdownTier, + timers: &[Instant], + outcome: &mut ShutdownReason, + ) { while self.pending[tier as usize] != 0 { let Some((service, tier, reason)) = self.handles.next().await else { return; @@ -156,7 +181,8 @@ impl ShutdownManager { // Another tier may finish while this one drains; debit its own pending count. self.pending(tier, -1); let elapsed = timers[tier as usize].elapsed(); - Self::log(service, reason, elapsed); + Self::log(service, &reason, elapsed); + *outcome = mem::take(outcome).combine(reason); } } @@ -164,7 +190,7 @@ impl ShutdownManager { self.pending[tier as usize] += op; } - fn log(service: Service, reason: ShutdownReason, elapsed: Duration) { + fn log(service: Service, reason: &ShutdownReason, elapsed: Duration) { match reason { ShutdownReason::Unexpected => { warn!(?service, ?elapsed, "terminated unexpectedly") @@ -223,6 +249,32 @@ pub enum ShutdownReason { RestartRequired, } +impl ShutdownReason { + /// Combines independently observed shutdown reasons into one process outcome. + /// + /// The first concrete error is retained ahead of an unexpected exit, a + /// requested restart, or a clean signal. This lets callers wait for the + /// first terminating service, drain every remaining service, and decide the + /// process outcome without losing a later failure. + pub fn combine(self, next: Self) -> Self { + if next.exit_code() > self.exit_code() { next } else { self } + } + + /// Returns the stable process exit code for this shutdown reason. + /// + /// `0` denotes a clean signal, `1` requests a restart, `2` denotes an + /// unexpected service exit, and `3` preserves a concrete service error. + /// [`Self::combine`] also uses this ordering to retain the strongest reason. + pub fn exit_code(&self) -> u8 { + match self { + Self::Signalled => 0, + Self::RestartRequired => 1, + Self::Unexpected => 2, + Self::Error(_) => 3, + } + } +} + impl ShutdownHandle { /// Request engine shutdown and report this service's termination reason. pub fn terminate(&mut self, reason: ShutdownReason) {