fix(batcher): stop listening to blocks when one of the rpcs disconnects (#1961)

Author: MarcosNicolau

batcher/aligned-batcher/src/lib.rs

@@ -41,7 +41,7 @@ use aws_sdk_s3::client::Client as S3Client;
 use eth::payment_service::{BatcherPaymentService, CreateNewTaskFeeParams, SignerMiddlewareT};
 use ethers::prelude::{Middleware, Provider};
 use ethers::types::{Address, Signature, TransactionReceipt, U256};
-use futures_util::{future, SinkExt, StreamExt, TryStreamExt};
+use futures_util::{future, join, SinkExt, StreamExt, TryStreamExt};
 use lambdaworks_crypto::merkle_tree::merkle::MerkleTree;
 use lambdaworks_crypto::merkle_tree::traits::IsMerkleTreeBackend;
 use log::{debug, error, info, warn};
@@ -322,39 +322,80 @@ impl Batcher {
     pub async fn listen_new_blocks_retryable(
         self: Arc<Self>,
     ) -> Result<(), RetryError<BatcherError>> {
-        let eth_ws_provider = Provider::connect(&self.eth_ws_url).await.map_err(|e| {
-            warn!("Failed to instantiate Ethereum websocket provider");
-            RetryError::Transient(BatcherError::EthereumSubscriptionError(e.to_string()))
-        })?;
+        // Try to connect at least to one of the nodes (main or fallback)
+        let eth_ws_provider = Provider::connect(&self.eth_ws_url).await.ok();
+        let eth_ws_provider_fallback = Provider::connect(&self.eth_ws_url_fallback).await.ok();
+        if eth_ws_provider.is_none() {
+            warn!("Failed to instantiate Ethereum main websocket provider");
+        }
+        if eth_ws_provider_fallback.is_none() {
+            warn!("Failed to instantiate fallback Ethereum websocket provider");
+        }
+        if eth_ws_provider.is_none() && eth_ws_provider_fallback.is_none() {
+            return Err(RetryError::Transient(
+                BatcherError::EthereumSubscriptionError(
+                    "Both Ethereum websocket providers failed to connect".to_string(),
+                ),
+            ));
+        }
 
-        let eth_ws_provider_fallback =
-            Provider::connect(&self.eth_ws_url_fallback)
-                .await
-                .map_err(|e| {
-                    warn!("Failed to instantiate fallback Ethereum websocket provider");
-                    RetryError::Transient(BatcherError::EthereumSubscriptionError(e.to_string()))
-                })?;
-
-        let mut stream = eth_ws_provider.subscribe_blocks().await.map_err(|e| {
-            warn!("Error subscribing to blocks.");
-            RetryError::Transient(BatcherError::EthereumSubscriptionError(e.to_string()))
-        })?;
-
-        let mut stream_fallback =
-            eth_ws_provider_fallback
-                .subscribe_blocks()
-                .await
-                .map_err(|e| {
-                    warn!("Error subscribing to blocks.");
-                    RetryError::Transient(BatcherError::EthereumSubscriptionError(e.to_string()))
-                })?;
+        // Try to connect to one stream (main or fallback)
+        let mut stream = match &eth_ws_provider {
+            Some(provider) => match provider.subscribe_blocks().await {
+                Ok(s) => Some(s),
+                Err(e) => {
+                    warn!("Error subscribing to blocks on primary provider: {:?}", e);
+                    None
+                }
+            },
+            None => None,
+        };
+        let mut stream_fallback = match &eth_ws_provider_fallback {
+            Some(provider) => match provider.subscribe_blocks().await {
+                Ok(s) => Some(s),
+                Err(e) => {
+                    warn!("Error subscribing to blocks on fallback provider: {:?}", e);
+                    None
+                }
+            },
+            None => None,
+        };
+        if stream.is_none() && stream_fallback.is_none() {
+            return Err(RetryError::Transient(
+                BatcherError::EthereumSubscriptionError(
+                    "Both Ethereum block subscriptions failed".to_string(),
+                ),
+            ));
+        }
 
         let last_seen_block = Mutex::<u64>::new(0);
 
-        while let Some(block) = tokio::select! {
-            block = stream.next() => block,
-            block = stream_fallback.next() => block,
-        } {
+        loop {
+            // Wait for both responses
+            let (block_main, block_fallback) = join!(
+                async {
+                    match stream.as_mut() {
+                        Some(s) => s.next().await,
+                        None => None,
+                    }
+                },
+                async {
+                    match stream_fallback.as_mut() {
+                        Some(s) => s.next().await,
+                        None => None,
+                    }
+                }
+            );
+
+            let block = if let Some(block) = block_main {
+                block
+            } else if let Some(block) = block_fallback {
+                block
+            } else {
+                // Both rpc failed to respond, break and try to reconnect
+                break;
+            };
+
             let batcher = self.clone();
             let block_number = block.number.unwrap_or_default();
             let block_number = u64::try_from(block_number).unwrap_or_default();
@@ -371,10 +412,10 @@ impl Batcher {
             tokio::spawn(async move {
                 if let Err(e) = batcher.handle_new_block(block_number).await {
                     error!("Error when handling new block: {:?}", e);
-                };
+                }
             });
         }
-        error!("Failed to fetch blocks");
+        error!("Both main and fallback Ethereum WS clients subscriptions have disconnected, will try to reconnect...");
 
         Err(RetryError::Transient(
             BatcherError::EthereumSubscriptionError("Could not get new blocks".to_string()),

batcher/aligned-sdk/src/common/constants.rs

@@ -41,7 +41,7 @@ pub const DEFAULT_MAX_FEE_BATCH_SIZE: usize = 10;
 pub const ETHEREUM_CALL_MIN_RETRY_DELAY: u64 = 500; // milliseconds
 pub const ETHEREUM_CALL_MAX_RETRIES: usize = 5;
 pub const ETHEREUM_CALL_BACKOFF_FACTOR: f32 = 2.0;
-pub const ETHEREUM_CALL_MAX_RETRY_DELAY: u64 = 3600; // seconds
+pub const ETHEREUM_CALL_MAX_RETRY_DELAY: u64 = 60; // seconds
 
 /// Ethereum transaction retry constants
 pub const BUMP_MIN_RETRY_DELAY: u64 = 500; // milliseconds