fix: do not exit process on sync failure

Author: ncitron

core/src/client/api.rs

@@ -15,7 +15,7 @@ use helios_common::{
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 pub trait HeliosApi<N: NetworkSpec>: Send + Sync + 'static {
     // node management
-    async fn wait_synced(&self);
+    async fn wait_synced(&self) -> Result<()>;
     async fn shutdown(&self);
     // state fetch
     async fn get_balance(&self, address: Address, block_id: BlockId) -> Result<U256>;

core/src/client/node.rs

@@ -1,6 +1,5 @@
 use std::marker::PhantomData;
 use std::sync::Arc;
-use std::time::Duration;
 
 use alloy::consensus::BlockHeader;
 use alloy::eips::{BlockId, BlockNumberOrTag};
@@ -27,7 +26,7 @@ use helios_common::{
 use crate::consensus::Consensus;
 use crate::errors::ClientError;
 use crate::execution::filter_state::{FilterState, FilterType};
-use crate::time::{interval, SystemTime, UNIX_EPOCH};
+use crate::time::{SystemTime, UNIX_EPOCH};
 
 use super::api::HeliosApi;
 
@@ -169,14 +168,8 @@ impl<N: NetworkSpec, C: Consensus<N::BlockResponse>, E: ExecutionProvider<N>> He
         }
     }
 
-    async fn wait_synced(&self) {
-        let mut interval = interval(Duration::from_millis(100));
-        loop {
-            interval.tick().await;
-            if let Ok(SyncStatus::None) = self.syncing().await {
-                break;
-            }
-        }
+    async fn wait_synced(&self) -> Result<()> {
+        self.consensus.wait_synced().await
     }
 
     async fn call(&self, tx: &N::TransactionRequest, block_id: BlockId) -> Result<Bytes> {

core/src/consensus.rs

@@ -1,8 +1,10 @@
 use alloy::network::{primitives::HeaderResponse, BlockResponse, TransactionResponse};
+use async_trait::async_trait;
 use eyre::Result;
 use serde::Serialize;
 use tokio::sync::{mpsc, watch};
 
+#[async_trait]
 pub trait Consensus<
     B: BlockResponse<Transaction: TransactionResponse, Header: HeaderResponse> + Serialize,
 >: Sync + Send + 'static
@@ -12,4 +14,5 @@ pub trait Consensus<
     fn expected_highest_block(&self) -> u64;
     fn chain_id(&self) -> u64;
     fn shutdown(&self) -> Result<()>;
+    async fn wait_synced(&self) -> Result<()>;
 }

ethereum/src/consensus.rs

@@ -1,5 +1,4 @@
 use std::marker::PhantomData;
-use std::process;
 use std::sync::Arc;
 
 use alloy::consensus::proofs::{calculate_transaction_root, calculate_withdrawals_root};
@@ -20,7 +19,7 @@ use url::Url;
 use tokio::sync::mpsc::channel;
 use tokio::sync::mpsc::Receiver;
 use tokio::sync::mpsc::Sender;
-use tokio::sync::watch;
+use tokio::sync::{watch, Mutex};
 
 use helios_consensus_core::{
     apply_bootstrap, apply_finality_update, apply_update, calc_sync_period,
@@ -40,10 +39,18 @@ use crate::constants::MAX_REQUEST_LIGHT_CLIENT_UPDATES;
 use crate::database::Database;
 use crate::rpc::ConsensusRpc;
 
+#[derive(Debug, Clone)]
+pub enum ConsensusSyncStatus {
+    Syncing,
+    Synced,
+    Error(String),
+}
+
 pub struct ConsensusClient<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> {
     pub block_recv: Option<Receiver<Block<Transaction>>>,
     pub finalized_block_recv: Option<watch::Receiver<Option<Block<Transaction>>>>,
     pub checkpoint_recv: watch::Receiver<Option<B256>>,
+    sync_status_recv: Mutex<watch::Receiver<ConsensusSyncStatus>>,
     shutdown_send: watch::Sender<bool>,
     genesis_time: u64,
     config: Arc<Config>,
@@ -62,6 +69,7 @@ pub struct Inner<S: ConsensusSpec, R: ConsensusRpc<S>> {
     phantom: PhantomData<S>,
 }
 
+#[async_trait::async_trait]
 impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> Consensus<Block>
     for ConsensusClient<S, R, DB>
 {
@@ -85,13 +93,30 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> Consensus<Block>
         self.shutdown_send.send(true)?;
         Ok(())
     }
+
+    async fn wait_synced(&self) -> Result<()> {
+        let mut sync_status_recv = self.sync_status_recv.lock().await;
+
+        loop {
+            let status = sync_status_recv.borrow().clone();
+            match status {
+                ConsensusSyncStatus::Synced => return Ok(()),
+                ConsensusSyncStatus::Error(err) => return Err(eyre!("sync failed: {}", err)),
+                ConsensusSyncStatus::Syncing => {
+                    // Wait for status to change
+                    sync_status_recv.changed().await?;
+                }
+            }
+        }
+    }
 }
 
 impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, DB> {
     pub fn new(rpc: &Url, config: Arc<Config>) -> Result<ConsensusClient<S, R, DB>> {
         let (block_send, block_recv) = channel(256);
         let (finalized_block_send, finalized_block_recv) = watch::channel(None);
         let (checkpoint_send, checkpoint_recv) = watch::channel(None);
+        let (sync_status_send, sync_status_recv) = watch::channel(ConsensusSyncStatus::Syncing);
         let (shutdown_send, shutdown_recv) = watch::channel(false);
 
         let config_clone = config.clone();
@@ -125,20 +150,24 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, D
                     let res = sync_all_fallbacks(&mut inner, config.chain.chain_id).await;
                     if let Err(err) = res {
                         error!(target: "helios::consensus", err = %err, "sync failed");
-                        process::exit(1);
+                        _ = sync_status_send.send(ConsensusSyncStatus::Error(err.to_string()));
+                        return;
                     }
                 } else if let Some(fallback) = &config.fallback {
                     let res = sync_fallback(&mut inner, fallback).await;
                     if let Err(err) = res {
                         error!(target: "helios::consensus", err = %err, "sync failed");
-                        process::exit(1);
+                        _ = sync_status_send.send(ConsensusSyncStatus::Error(err.to_string()));
+                        return;
                     }
                 } else {
                     error!(target: "helios::consensus", err = %err, "sync failed");
-                    process::exit(1);
+                    _ = sync_status_send.send(ConsensusSyncStatus::Error(err.to_string()));
+                    return;
                 }
             }
 
+            _ = sync_status_send.send(ConsensusSyncStatus::Synced);
             _ = inner.send_blocks().await;
 
             let start = Instant::now() + inner.duration_until_next_update().to_std().unwrap();
@@ -180,6 +209,7 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, D
             block_recv: Some(block_recv),
             finalized_block_recv: Some(finalized_block_recv),
             checkpoint_recv,
+            sync_status_recv: Mutex::new(sync_status_recv),
             shutdown_send,
             genesis_time,
             config: config_clone,

linea/src/consensus.rs

@@ -71,6 +71,7 @@ impl ConsensusClient {
     }
 }
 
+#[async_trait::async_trait]
 impl Consensus<Block<Transaction>> for ConsensusClient {
     fn chain_id(&self) -> u64 {
         self.chain_id
@@ -91,6 +92,11 @@ impl Consensus<Block<Transaction>> for ConsensusClient {
     fn expected_highest_block(&self) -> u64 {
         u64::MAX
     }
+
+    async fn wait_synced(&self) -> eyre::Result<()> {
+        // Linea consensus doesn't have a sync process, so immediately return Ok
+        Ok(())
+    }
 }
 
 #[allow(dead_code)]

opstack/src/consensus.rs

@@ -87,6 +87,7 @@ impl ConsensusClient {
     }
 }
 
+#[async_trait::async_trait]
 impl Consensus<Block<Transaction>> for ConsensusClient {
     fn chain_id(&self) -> u64 {
         self.chain_id
@@ -107,6 +108,11 @@ impl Consensus<Block<Transaction>> for ConsensusClient {
     fn expected_highest_block(&self) -> u64 {
         u64::MAX
     }
+
+    async fn wait_synced(&self) -> eyre::Result<()> {
+        // OpStack consensus doesn't have a sync process, so immediately return Ok
+        Ok(())
+    }
 }
 
 #[allow(dead_code)]