chain/ethereum: Add RPC provider failover for block ingestor

When do_poll() fails, the ingestor now probes the current provider
before switching. If the current provider is unreachable, all
alternatives are probed in parallel and the first healthy one is
selected. This avoids unnecessary switches on non-RPC failures
(e.g. DB errors, chain reorgs).

Also limits latest_block_ptr retries to ENV_VARS.request_retries
so failures surface to the failover logic instead of retrying
indefinitely.

Author: dimitrovmaksim

Cargo.lock

@@ -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/Cargo.toml

@@ -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/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,