feat: retry-safe Base RPC failover transport with chain-id validation (#869)

crates/api/src/main.rs

@@ -20040,8 +20040,8 @@ mod tests {
             stripe_payment_method_configuration: None,
             store: None,
             base_rpc_urls: BaseRpcUrlConfig {
-                base_sepolia: Some(base_sepolia_rpc_url),
-                base_mainnet: None,
+                base_sepolia: vec![base_sepolia_rpc_url],
+                base_mainnet: Vec::new(),
             },
             base_broadcast_enabled: true,
             operator_api_token: None,

crates/chain-base/src/lib.rs

@@ -16,7 +16,6 @@ use sha2::Sha256;
 use sha3::{Digest, Keccak256};
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
-    env,
 };
 use thiserror::Error;
 use uuid::Uuid;
@@ -2128,10 +2127,231 @@ pub struct AutonomousFactorySafeObservation {
     pub evidence_boundary: String,
 }
 
+/// Ordered list of HTTPS Base RPC endpoints for failover.
+/// Comma-separated env var values are split and each URL is trimmed.
+/// The first endpoint that validates chain ID 8453 is used.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct BaseRpcUrlConfig {
-    pub base_sepolia: Option<String>,
-    pub base_mainnet: Option<String>,
+    pub base_sepolia: Vec<String>,
+    pub base_mainnet: Vec<String>,
+}
+
+/// Retry configuration for Base RPC failover transport.
+#[derive(Debug, Clone, Copy)]
+pub struct FailoverRetryConfig {
+    /// Maximum HTTP-level retries per endpoint (429, 5xx only)
+    pub max_retries: u32,
+    /// Base backoff in milliseconds (doubles each attempt)
+    pub base_backoff_ms: u64,
+    /// Maximum backoff in milliseconds
+    pub max_backoff_ms: u64,
+}
+
+impl Default for FailoverRetryConfig {
+    fn default() -> Self {
+        Self {
+            max_retries: 3,
+            base_backoff_ms: 200,
+            max_backoff_ms: 5_000,
+        }
+    }
+}
+
+impl FailoverRetryConfig {
+    pub fn deterministic_backoff_ms(&self, attempt: u32) -> u64 {
+        let base = self.base_backoff_ms * (1u64 << attempt.min(10));
+        base.min(self.max_backoff_ms)
+    }
+}
+
+/// A JSON-RPC transport that tries each endpoint in order, validates chain ID,
+/// and retries only on bounded transport errors, HTTP 429, and HTTP 5xx.
+/// Never retries confirmed JSON-RPC execution errors or logs credentials.
+#[derive(Debug, Clone)]
+pub struct FailoverJsonRpcTransport<T: JsonRpcTransport = ReqwestJsonRpcTransport> {
+    inner: T,
+    endpoints: Vec<String>,
+    expected_chain_id: u64,
+    retry: FailoverRetryConfig,
+}
+
+/// Whether an HTTP status is a bounded, retriable transport failure.
+pub fn is_retriable_http_status(status: u16) -> bool {
+    status == 429 || (500..=599).contains(&status)
+}
+
+/// Whether an error may be retried on the same endpoint and then fail over.
+fn is_retriable_error(error: &ChainBaseError) -> bool {
+    match error {
+        ChainBaseError::RpcTransport(_) => true,
+        ChainBaseError::RpcHttpStatus(status) => is_retriable_http_status(*status),
+        ChainBaseError::InvalidRpcResponse(_) => true,
+        _ => false,
+    }
+}
+
+impl FailoverJsonRpcTransport<ReqwestJsonRpcTransport> {
+    pub fn new(endpoints: Vec<String>, expected_chain_id: u64, retry: FailoverRetryConfig) -> Self {
+        Self::with_transport(
+            ReqwestJsonRpcTransport::default(),
+            endpoints,
+            expected_chain_id,
+            retry,
+        )
+    }
+}
+
+impl<T: JsonRpcTransport> FailoverJsonRpcTransport<T> {
+    /// Build a failover transport over an injected inner transport (used by offline tests).
+    pub fn with_transport(
+        inner: T,
+        endpoints: Vec<String>,
+        expected_chain_id: u64,
+        retry: FailoverRetryConfig,
+    ) -> Self {
+        Self {
+            inner,
+            endpoints,
+            expected_chain_id,
+            retry,
+        }
+    }
+
+    /// Validate a single endpoint's chain ID, returning a mismatch on wrong chain.
+    async fn validate_endpoint(&self, endpoint: &str) -> Result<(), ChainBaseError> {
+        let value = self
+            .inner
+            .post_json_value(
+                endpoint,
+                &serde_json::json!({
+                    "jsonrpc": "2.0",
+                    "id": 0,
+                    "method": "eth_chainId",
+                    "params": []
+                }),
+            )
+            .await?;
+        let chain_id_hex = value.get("result").and_then(Value::as_str).unwrap_or("");
+        let observed =
+            u64::from_str_radix(chain_id_hex.trim_start_matches("0x"), 16).map_err(|_| {
+                ChainBaseError::RelayerChainMismatch {
+                    expected: self.expected_chain_id,
+                    observed: 0,
+                }
+            })?;
+        if observed != self.expected_chain_id {
+            return Err(ChainBaseError::RelayerChainMismatch {
+                expected: self.expected_chain_id,
+                observed,
+            });
+        }
+        Ok(())
+    }
+
+    /// Post a single request across endpoints: validate chain, retry bounded failures,
+    /// then advance to the next endpoint. Never retries a confirmed execution error.
+    async fn post_value_with_failover(&self, request: &Value) -> Result<Value, ChainBaseError> {
+        let mut last_error: Option<ChainBaseError> = None;
+        for endpoint in &self.endpoints {
+            match self.validate_endpoint(endpoint).await {
+                Ok(()) => {}
+                Err(
+                    error @ (ChainBaseError::RelayerChainMismatch { .. }
+                    | ChainBaseError::RpcTransport(_)
+                    | ChainBaseError::RpcHttpStatus(_)
+                    | ChainBaseError::InvalidRpcResponse(_)),
+                ) => {
+                    last_error = Some(error);
+                    continue;
+                }
+                Err(error) => return Err(error),
+            }
+            for attempt in 0..=self.retry.max_retries {
+                if attempt > 0 {
+                    let delay = self.retry.deterministic_backoff_ms(attempt - 1);
+                    tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
+                }
+                match self.inner.post_json_value(endpoint, request).await {
+                    Ok(value) => return Ok(value),
+                    Err(error) => {
+                        if !is_retriable_error(&error) {
+                            return Err(error);
+                        }
+                        last_error = Some(error);
+                        if attempt == self.retry.max_retries {
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        Err(last_error.unwrap_or_else(|| {
+            ChainBaseError::RpcTransport("no Base RPC endpoints configured".to_string())
+        }))
+    }
+
+    /// Batch variant of `post_value_with_failover`.
+    async fn post_values_with_failover(
+        &self,
+        requests: &[Value],
+    ) -> Result<Vec<Value>, ChainBaseError> {
+        let mut last_error: Option<ChainBaseError> = None;
+        for endpoint in &self.endpoints {
+            match self.validate_endpoint(endpoint).await {
+                Ok(()) => {}
+                Err(
+                    error @ (ChainBaseError::RelayerChainMismatch { .. }
+                    | ChainBaseError::RpcTransport(_)
+                    | ChainBaseError::RpcHttpStatus(_)
+                    | ChainBaseError::InvalidRpcResponse(_)),
+                ) => {
+                    last_error = Some(error);
+                    continue;
+                }
+                Err(error) => return Err(error),
+            }
+            for attempt in 0..=self.retry.max_retries {
+                if attempt > 0 {
+                    let delay = self.retry.deterministic_backoff_ms(attempt - 1);
+                    tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
+                }
+                match self.inner.post_json_values(endpoint, requests).await {
+                    Ok(values) => return Ok(values),
+                    Err(error) => {
+                        if !is_retriable_error(&error) {
+                            return Err(error);
+                        }
+                        last_error = Some(error);
+                        if attempt == self.retry.max_retries {
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        Err(last_error.unwrap_or_else(|| {
+            ChainBaseError::RpcTransport("no Base RPC endpoints configured".to_string())
+        }))
+    }
+}
+
+#[async_trait::async_trait]
+impl<T: JsonRpcTransport> JsonRpcTransport for FailoverJsonRpcTransport<T> {
+    async fn post_json_value(
+        &self,
+        _rpc_url: &str,
+        request: &Value,
+    ) -> Result<Value, ChainBaseError> {
+        self.post_value_with_failover(request).await
+    }
+
+    async fn post_json_values(
+        &self,
+        _rpc_url: &str,
+        requests: &[Value],
+    ) -> Result<Vec<Value>, ChainBaseError> {
+        self.post_values_with_failover(requests).await
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -2412,29 +2632,72 @@ pub fn base_network_descriptor(network: &str) -> Result<BaseNetworkDescriptor, C
 impl BaseRpcUrlConfig {
     pub fn from_env() -> Self {
         Self {
-            base_sepolia: non_empty_env("BASE_SEPOLIA_RPC_URL"),
-            base_mainnet: non_empty_env("BASE_MAINNET_RPC_URL"),
+            base_sepolia: split_env_list("BASE_SEPOLIA_RPC_URL"),
+            base_mainnet: split_env_list("BASE_MAINNET_RPC_URL"),
         }
     }
 
-    pub fn resolve(
+    /// Resolve the list of all configured endpoints for a network.
+    /// Returns the descriptor and a non-empty list of trimmed URLs.
+    pub fn resolve_endpoint_list(
         &self,
         network: &str,
-    ) -> Result<(BaseNetworkDescriptor, String), ChainBaseError> {
+    ) -> Result<(BaseNetworkDescriptor, Vec<String>), ChainBaseError> {
         let descriptor = base_network_descriptor(network)?;
-        let url = match descriptor.rpc_url_env.as_str() {
+        let urls = match descriptor.rpc_url_env.as_str() {
             "BASE_SEPOLIA_RPC_URL" => self.base_sepolia.clone(),
             "BASE_MAINNET_RPC_URL" => self.base_mainnet.clone(),
-            _ => None,
+            _ => Vec::new(),
+        };
+        if urls.is_empty() {
+            return Err(ChainBaseError::MissingRpcUrl {
+                network: descriptor.name.clone(),
+                env_var: descriptor.rpc_url_env.clone(),
+            });
         }
-        .ok_or_else(|| ChainBaseError::MissingRpcUrl {
-            network: descriptor.name.clone(),
-            env_var: descriptor.rpc_url_env.clone(),
-        })?;
-        Ok((descriptor, url))
+        Ok((descriptor, urls))
+    }
+
+    /// Resolve the first configured endpoint for a network (backward-compatible).
+    pub fn resolve(
+        &self,
+        network: &str,
+    ) -> Result<(BaseNetworkDescriptor, String), ChainBaseError> {
+        let (descriptor, urls) = self.resolve_endpoint_list(network)?;
+        Ok((descriptor, urls.into_iter().next().unwrap()))
+    }
+
+    /// Build a failover transport from the configured endpoint list for the given
+    /// Base network. Validates chain ID before using any endpoint and retries only
+    /// on bounded transport errors, HTTP 429, and HTTP 5xx.
+    pub fn failover_transport(
+        &self,
+        network: &str,
+    ) -> Result<FailoverJsonRpcTransport, ChainBaseError> {
+        let (descriptor, urls) = self.resolve_endpoint_list(network)?;
+        Ok(FailoverJsonRpcTransport::new(
+            urls,
+            descriptor.chain_id,
+            FailoverRetryConfig::default(),
+        ))
     }
 }
 
+/// Split a comma-separated environment variable into a list of trimmed non-empty URLs.
+fn split_env_list(env_var: &str) -> Vec<String> {
+    std::env::var(env_var)
+        .ok()
+        .map(|value| {
+            value
+                .split(',')
+                .map(str::trim)
+                .filter(|s| !s.is_empty())
+                .map(String::from)
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
 pub fn parse_eth_get_logs_response(value: Value) -> Result<EthGetLogsResponse, ChainBaseError> {
     let envelope: EthGetLogsEnvelope = serde_json::from_value(value)
         .map_err(|error| ChainBaseError::InvalidRpcResponse(error.to_string()))?;
@@ -3347,13 +3610,6 @@ where
     )
 }
 
-fn non_empty_env(key: &str) -> Option<String> {
-    env::var(key)
-        .ok()
-        .map(|value| value.trim().to_string())
-        .filter(|value| !value.is_empty())
-}
-
 impl BaseContractLogQuery {
     pub fn new(
         contract: impl Into<String>,
@@ -9135,8 +9391,8 @@ mod tests {
     #[test]
     fn base_rpc_url_config_resolves_only_configured_networks() {
         let config = BaseRpcUrlConfig {
-            base_sepolia: Some("https://sepolia.example".to_string()),
-            base_mainnet: None,
+            base_sepolia: vec!["https://sepolia.example".to_string()],
+            base_mainnet: Vec::new(),
         };
 
         let (network, url) = config.resolve("base-sepolia").unwrap();
@@ -9151,6 +9407,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn base_rpc_url_config_supports_comma_separated_endpoint_lists() {
+        let config = BaseRpcUrlConfig {
+            base_sepolia: Vec::new(),
+            base_mainnet: vec![
+                "https://rpc1.example".to_string(),
+                "https://rpc2.example".to_string(),
+            ],
+        };
+        let (network, urls) = config.resolve_endpoint_list("base-mainnet").unwrap();
+        assert_eq!(network.name, "Base");
+        assert_eq!(urls.len(), 2);
+        assert_eq!(urls[0], "https://rpc1.example");
+        assert_eq!(urls[1], "https://rpc2.example");
+    }
+
+    #[test]
+    fn failover_retry_config_deterministic_backoff() {
+        let retry = FailoverRetryConfig::default();
+        assert_eq!(retry.deterministic_backoff_ms(0), 200);
+        assert_eq!(retry.deterministic_backoff_ms(1), 400);
+        assert_eq!(retry.deterministic_backoff_ms(2), 800);
+        assert_eq!(retry.deterministic_backoff_ms(3), 1600);
+        // Should cap at max_backoff_ms
+        assert_eq!(retry.deterministic_backoff_ms(10), 5000);
+    }
+
     #[test]
     fn rejects_reversed_log_query_range() {
         let error = BaseContractLogQuery::new(