Add RPC provider failover for block ingestor

chain/ethereum/Cargo.toml

@@ -16,6 +16,7 @@ tiny-keccak = "1.5.0"
 hex = "0.4.3"
 semver = { workspace = true }
 thiserror = { workspace = true }
+futures = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tower = { workspace = true }

chain/ethereum/src/adapter.rs

@@ -1174,6 +1174,11 @@ pub trait EthereumAdapter: Send + Sync + 'static {
         address: Address,
         block_ptr: BlockPtr,
     ) -> Result<alloy::primitives::Bytes, EthereumRpcError>;
+
+    /// Returns a boolean indicating whether the adapter can reach
+    /// the RPC provider it is configured to use.
+    /// This is used to determine if a provider should be considered healthy.
+    async fn is_reachable(&self) -> bool;
 }
 
 #[cfg(test)]

chain/ethereum/src/ethereum_adapter.rs

@@ -57,7 +57,7 @@ use std::convert::TryFrom;
 use std::iter::FromIterator;
 use std::pin::Pin;
 use std::sync::Arc;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use tokio::sync::RwLock;
 use tokio::time::timeout;
 
@@ -1246,7 +1246,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
         let alloy = self.alloy.clone();
         retry("eth_getBlockByNumber(latest) no txs RPC call", logger)
             .redact_log_urls(true)
-            .no_limit()
+            .limit(ENV_VARS.request_retries)
             .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs())
             .run(move || {
                 let alloy = alloy.cheap_clone();
@@ -1270,6 +1270,14 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .await
     }
 
+    async fn is_reachable(&self) -> bool {
+        let alloy = self.alloy.clone();
+        tokio::time::timeout(Duration::from_secs(10), alloy.get_block_number())
+            .await
+            .map(|r| r.is_ok())
+            .unwrap_or(false)
+    }
+
     async fn block_by_hash(
         &self,
         logger: &Logger,