Sync error handling

anchor/eth/Cargo.toml

@@ -12,6 +12,7 @@ path = "execution.rs"
 alloy = { workspace = true }
 base64 = { workspace = true }
 database = { workspace = true }
+fastrand = "2.3.0"
 futures = { workspace = true }
 hex = { workspace = true }
 indexmap = { workspace = true }

anchor/eth/src/lib.rs

@@ -1,4 +1,4 @@
-pub use sync::{Config, SsvEventSyncer};
+pub use sync::{Config, SsvEventSyncer, OPERATIONAL_STATUS};
 mod error;
 mod event_parser;
 mod event_processor;

anchor/eth/src/sync.rs

@@ -8,9 +8,9 @@ use alloy::sol_types::SolEvent;
 use database::NetworkDatabase;
 use futures::future::{try_join_all, Future};
 use futures::StreamExt;
-use rand::Rng;
 use ssv_network_config::SsvNetworkConfig;
 use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, LazyLock};
 use tokio::sync::oneshot::Sender;
 use tokio::time::Duration;
@@ -38,14 +38,19 @@ static SSV_EVENTS: LazyLock<Vec<&str>> = LazyLock::new(|| {
     ]
 });
 
+/// Current operational status of sync. If there is an issue with the rpc endpoint or the ws
+/// endpoing, the status is considered down. Otherwise, it is up
+pub static OPERATIONAL_STATUS: AtomicBool = AtomicBool::new(true);
+
 /// Batch size for log fetching
 const BATCH_SIZE: u64 = 10000;
 
 /// Batch size for task groups
 const GROUP_SIZE: usize = 50;
 
-/// Retry information for log fetching
-const MAX_RETRIES: i32 = 5;
+/// Exponential backoff constants
+const INITIAL_BACKOFF_MS: u64 = 100; // Start with 100ms delay
+const MAX_BACKOFF_MS: u64 = 30_000; // Don't wait longer than 30 seconds
 
 // Block follow distance
 const FOLLOW_DISTANCE: u64 = 8;
@@ -162,10 +167,14 @@ impl SsvEventSyncer {
         let mut start_block = std::cmp::max(deployment_block, last_processed_block);
 
         loop {
-            let current_block = self.rpc_client.get_block_number().await.map_err(|e| {
-                error!(?e, "Failed to fetch block number");
-                ExecutionError::RpcError(format!("Unable to fetch block number {}", e))
-            })?;
+            let current_block = match self.rpc_client.get_block_number().await {
+                Ok(block) => block,
+                Err(e) => {
+                    error!(?e, "Failed to fetch block number");
+                    self.troubleshoot_rpc().await;
+                    continue;
+                }
+            };
 
             // Basic verification
             if current_block < FOLLOW_DISTANCE {
@@ -184,7 +193,7 @@ impl SsvEventSyncer {
                 break;
             }
 
-            // Here, we have a start..endblock that we need to sync the logs from. This range gets
+            // Here, we have a start..end block that we need to sync the logs from. This range gets
             // broken up into individual ranges of BATCH_SIZE where the logs are fetches from. The
             // individual ranges are further broken up into a set of batches that are sequentually
             // processes. This makes it so we dont have a ton of logs that all have to be processed
@@ -274,7 +283,7 @@ impl SsvEventSyncer {
         from_block: u64,
         to_block: u64,
         deployment_address: Address,
-    ) -> impl Future<Output = Result<Vec<Log>, ExecutionError>> {
+    ) -> impl Future<Output = Result<Vec<Log>, ExecutionError>> + use<'_> {
         // Setup filter and rpc client
         let rpc_client = self.rpc_client.clone();
         let filter = Filter::new()
@@ -286,35 +295,50 @@ impl SsvEventSyncer {
         // Try to fetch logs with a retry upon error. Try up to MAX_RETRIES times and error if we
         // exceed this as we can assume there is some underlying connection issue
         async move {
-            let mut retry_cnt = 0;
             loop {
                 match rpc_client.get_logs(&filter).await {
                     Ok(logs) => {
                         debug!(log_count = logs.len(), "Successfully fetched logs");
                         return Ok(logs);
                     }
                     Err(e) => {
-                        if retry_cnt > MAX_RETRIES {
-                            error!(?e, retry_cnt, "Max retries exceeded while fetching logs");
-                            return Err(ExecutionError::RpcError(
-                                "Unable to fetch logs".to_string(),
-                            ));
-                        }
-
-                        warn!(?e, retry_cnt, "Error fetching logs, retrying");
-
-                        // increment retry_count and jitter retry duration
-                        let jitter = rand::thread_rng().gen_range(0..=100);
-                        let sleep_duration = Duration::from_millis(jitter);
-                        tokio::time::sleep(sleep_duration).await;
-                        retry_cnt += 1;
-                        continue;
+                        warn!(?e, "Error fetching logs");
+                        self.troubleshoot_rpc().await;
                     }
                 }
             }
         }
     }
 
+    // When we encounter a rpc error, keep polling until success
+    async fn troubleshoot_rpc(&self) {
+        OPERATIONAL_STATUS.store(false, Ordering::Relaxed);
+
+        let mut retry_count = 0;
+        let mut current_backoff_ms = INITIAL_BACKOFF_MS;
+
+        while let Err(e) = self.rpc_client.get_block_number().await {
+            // Calculate next backoff with some jitter
+            let jitter = fastrand::u64(0..=50); // Random 0-50ms
+            current_backoff_ms = (current_backoff_ms * 2) // Exponential growth
+                .min(MAX_BACKOFF_MS) // Don't exceed max backoff
+                .saturating_add(jitter); // Add jitter safely
+
+            warn!(
+                error = ?e,
+                retry_count,
+                backoff_ms = current_backoff_ms,
+                "RPC error, backing off before retry"
+            );
+            retry_count += 1;
+
+            tokio::time::sleep(Duration::from_millis(current_backoff_ms)).await;
+        }
+
+        // Success! We can exit the retry loop
+        OPERATIONAL_STATUS.store(true, Ordering::Relaxed);
+    }
+
     // Once caught up with the chain, start live sync which will stream in live blocks from the
     // network. The events will be processed and duties will be created in response to network
     // actions
@@ -345,7 +369,9 @@ impl SsvEventSyncer {
                         // Historical sync any missed blocks while down, can pass 0 as deployment
                         // block since it will use last_processed_block from DB anyways
                         self.historical_sync(contract_address, 0).await?;
+                        OPERATIONAL_STATUS.store(true, Ordering::Relaxed);
                     } else {
+                        OPERATIONAL_STATUS.store(false, Ordering::Relaxed);
                         tokio::time::sleep(Duration::from_secs(1)).await;
                     }
                     None