fix(chain): avoid global shutdown on IBD failure (#43)

* fix(chain): avoid global shutdown on IBD failure

* fix(chain): propagate IBD failure through system signal

Cargo.lock

@@ -4273,6 +4273,7 @@ dependencies = [
  "logos-blockchain-network-service",
  "logos-blockchain-services-utils",
  "logos-blockchain-storage-service",
+ "logos-blockchain-system-sig-service",
  "logos-blockchain-time-service",
  "logos-blockchain-tracing",
  "logos-blockchain-tx-service",
@@ -5077,6 +5078,7 @@ dependencies = [
  "async-trait",
  "logos-blockchain-log-targets",
  "overwatch",
+ "tokio",
  "tracing",
 ]
 

services/chain/chain-network/Cargo.toml

@@ -25,6 +25,7 @@ lb-ledger               = { workspace = true }
 lb-network-service      = { workspace = true }
 lb-services-utils       = { workspace = true }
 lb-storage-service      = { workspace = true }
+lb-system-sig-service   = { workspace = true }
 lb-time-service         = { workspace = true }
 lb-tracing              = { workspace = true }
 lb-tx-service           = { workspace = true }

services/chain/chain-network/src/lib.rs

@@ -29,6 +29,7 @@ pub use lb_ledger::EpochState;
 use lb_network_service::NetworkService;
 use lb_services_utils::wait_until_services_are_ready;
 use lb_storage_service::StorageService;
+use lb_system_sig_service::{SystemSig, SystemSigMessage};
 use lb_time_service::{TimeService, TimeServiceMessage};
 use lb_tx_service::{
     TxMempoolService, backend::RecoverableMempool,
@@ -230,6 +231,7 @@ where
             TxMempoolService<MempoolNetAdapter, Mempool, Mempool::Storage, RuntimeServiceId>,
         >
         + AsServiceId<TimeService<TimeBackend, RuntimeServiceId>>,
+    RuntimeServiceId: AsServiceId<SystemSig<RuntimeServiceId>>,
 {
     fn init(
         service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
@@ -252,6 +254,11 @@ where
             &self.service_resources_handle,
         )
         .await;
+        let system_sig_relay = self
+            .service_resources_handle
+            .overwatch_handle
+            .relay::<SystemSig<RuntimeServiceId>>()
+            .await?;
 
         let ChainNetworkSettings {
             network: network_config,
@@ -304,17 +311,13 @@ where
             }
             Err(e) => {
                 error!(
-                    "Initial Block Download failed: {e:?}. Initiating graceful shutdown. Retry with different bootstrap peers"
+                    "Initial Block Download failed: {e:?}. Chain network service will stop; retry with different bootstrap peers"
                 );
-                if let Err(shutdown_err) = self
-                    .service_resources_handle
-                    .overwatch_handle
-                    .shutdown()
-                    .await
+                if let Err((error, SystemSigMessage::Shutdown)) =
+                    system_sig_relay.send(SystemSigMessage::Shutdown).await
                 {
-                    error!("Failed to shutdown overwatch: {shutdown_err:?}");
+                    error!("Failed to request top-level shutdown after IBD failure: {error:?}");
                 }
-
                 return Err(DynError::from(format!(
                     "Initial Block Download failed: {e:?}"
                 )));

services/system-sig/Cargo.toml

@@ -17,4 +17,5 @@ async-ctrlc    = { workspace = true }
 async-trait    = { workspace = true }
 lb-log-targets = { workspace = true }
 overwatch      = { workspace = true }
+tokio          = { workspace = true }
 tracing        = { workspace = true }

services/system-sig/src/lib.rs

@@ -16,6 +16,11 @@ pub struct SystemSig<RuntimeServiceId> {
     service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
 }
 
+#[derive(Debug)]
+pub enum SystemSigMessage {
+    Shutdown,
+}
+
 impl<RuntimeServiceId> SystemSig<RuntimeServiceId>
 where
     RuntimeServiceId: Debug + Display + Sync,
@@ -31,13 +36,13 @@ impl<RuntimeServiceId> ServiceData for SystemSig<RuntimeServiceId> {
     type Settings = ();
     type State = NoState<Self::Settings>;
     type StateOperator = NoOperator<Self::State>;
-    type Message = ();
+    type Message = SystemSigMessage;
 }
 
 #[async_trait::async_trait]
 impl<RuntimeServiceId> ServiceCore<RuntimeServiceId> for SystemSig<RuntimeServiceId>
 where
-    RuntimeServiceId: Debug + Display + Sync + Send + AsServiceId<Self>,
+    RuntimeServiceId: Debug + Display + Sync + Send + Clone + AsServiceId<Self>,
 {
     fn init(
         service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
@@ -49,21 +54,26 @@ where
     }
 
     async fn run(self) -> Result<(), DynError> {
-        let Self {
-            service_resources_handle,
-        } = self;
+        let service_resources_handle = self.service_resources_handle;
+        let overwatch_handle = service_resources_handle.overwatch_handle.clone();
+        let status_updater = service_resources_handle.status_updater;
+        let mut inbound_relay = service_resources_handle.inbound_relay;
         let ctrl_c = async_ctrlc::CtrlC::new()?;
 
-        service_resources_handle.status_updater.notify_ready();
+        status_updater.notify_ready();
         tracing::info!(
             target: LOG_TARGET,
             "Service '{}' is ready.",
             <RuntimeServiceId as AsServiceId<Self>>::SERVICE_ID
         );
 
-        // Wait for the Ctrl-C signal
-        ctrl_c.await;
-        Self::ctrl_c_signal_received(&service_resources_handle.overwatch_handle).await;
+        tokio::select! {
+            () = ctrl_c => Self::ctrl_c_signal_received(&overwatch_handle).await,
+            Some(SystemSigMessage::Shutdown) = inbound_relay.recv() => {
+                tracing::debug!(target: LOG_TARGET, "Shutdown requested by a service failure");
+                drop(overwatch_handle.shutdown().await);
+            }
+        }
 
         Ok(())
     }