papyrus_base_layer: add time-based retry of the primary L1 endpoint (#14521)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: asaf-sw

crates/apollo_base_layer_tests/src/anvil_base_layer.rs

@@ -267,6 +267,15 @@ impl BaseLayerContract for AnvilBaseLayer {
     async fn cycle_provider_url(&mut self) -> Result<(), Self::Error> {
         unimplemented!("Anvil base layer is tied to a an Anvil server, url is fixed.")
     }
+
+    async fn reset_provider_url_to_primary(&mut self) -> Result<(), Self::Error> {
+        unimplemented!("Anvil base layer is tied to a an Anvil server, url is fixed.")
+    }
+
+    // Anvil is tied to a single fixed URL, so it is always on its only (primary) endpoint.
+    async fn is_at_primary(&self) -> Result<bool, Self::Error> {
+        Ok(true)
+    }
 }
 
 /// Converts a given [L1 handler transaction](starknet_api::transaction::L1HandlerTransaction)

crates/apollo_deployments/resources/app_configs/base_layer_config.json

@@ -2,6 +2,7 @@
   "base_layer_config.bpo1_start_block_number": 9456501,
   "base_layer_config.bpo2_start_block_number": 9504747,
   "base_layer_config.fusaka_no_bpo_start_block_number": 9408577,
+  "base_layer_config.retry_primary_interval_seconds": 60,
   "base_layer_config.starknet_contract_address": "",
   "base_layer_config.timeout_millis": 1000
 }

crates/apollo_deployments/resources/app_configs/replacer_base_layer_config.json

@@ -2,6 +2,7 @@
   "base_layer_config.bpo1_start_block_number": "$$$_BASE_LAYER_CONFIG-BPO1_START_BLOCK_NUMBER_$$$",
   "base_layer_config.bpo2_start_block_number": "$$$_BASE_LAYER_CONFIG-BPO2_START_BLOCK_NUMBER_$$$",
   "base_layer_config.fusaka_no_bpo_start_block_number": "$$$_BASE_LAYER_CONFIG-FUSAKA_NO_BPO_START_BLOCK_NUMBER_$$$",
+  "base_layer_config.retry_primary_interval_seconds": 60,
   "base_layer_config.starknet_contract_address": "$$$_BASE_LAYER_CONFIG-STARKNET_CONTRACT_ADDRESS_$$$",
   "base_layer_config.timeout_millis": 1000
 }

crates/apollo_node/resources/config_schema.json

@@ -24,6 +24,11 @@
     "privacy": "Private",
     "value": "https://mainnet.infura.io/v3/YOUR_INFURA_API_KEY"
   },
+  "base_layer_config.retry_primary_interval_seconds": {
+    "description": "The interval (seconds) after which the next base-layer access retries the primary (first) endpoint.",
+    "privacy": "Public",
+    "value": 60
+  },
   "base_layer_config.starknet_contract_address": {
     "description": "Starknet contract address in ethereum.",
     "privacy": "Public",

crates/apollo_node/src/components.rs

@@ -314,7 +314,10 @@ pub async fn create_node_components(
                 .get_l1_gas_price_shared_client()
                 .expect("L1 gas price client should be available");
             let base_layer = EthereumBaseLayerContract::new(base_layer_config.clone());
-            let cyclic_base_layer_wrapper = CyclicBaseLayerWrapper::new(base_layer);
+            let cyclic_base_layer_wrapper = CyclicBaseLayerWrapper::new(
+                base_layer,
+                base_layer_config.retry_primary_interval_seconds,
+            );
 
             Some(L1GasPriceScraper::new(
                 l1_gas_price_scraper_config.clone(),
@@ -390,7 +393,10 @@ pub async fn create_node_components(
                 .expect("L1 Events Scraper config should be set");
             let l1_events_provider_client = clients.get_l1_events_provider_shared_client().unwrap();
             let base_layer = EthereumBaseLayerContract::new(base_layer_config.clone());
-            let cyclic_base_layer_wrapper = CyclicBaseLayerWrapper::new(base_layer);
+            let cyclic_base_layer_wrapper = CyclicBaseLayerWrapper::new(
+                base_layer,
+                base_layer_config.retry_primary_interval_seconds,
+            );
 
             Some(
                 L1EventsScraper::new(

crates/papyrus_base_layer/Cargo.toml

@@ -36,3 +36,4 @@ pretty_assertions.workspace = true
 rstest.workspace = true
 starknet-types-core.workspace = true
 starknet_api = { workspace = true, features = ["testing"] }
+tokio = { workspace = true, features = ["test-util"] }

crates/papyrus_base_layer/src/base_layer_test.rs

@@ -134,6 +134,35 @@ async fn test_cycle_wraps_to_primary_through_full_list() {
     assert_eq!(base_layer.get_url().await.unwrap().expose_secret(), primary_url);
 }
 
+#[tokio::test]
+async fn test_reset_provider_url_to_primary_repoints_live_provider() {
+    let primary_url = Url::parse("http://primary-endpoint.test/").unwrap();
+    let secondary_url = Url::parse("http://secondary-endpoint.test/").unwrap();
+    let tertiary_url = Url::parse("http://tertiary-endpoint.test/").unwrap();
+    let config = EthereumBaseLayerConfig {
+        ordered_l1_endpoint_urls: vec![
+            primary_url.clone().into(),
+            secondary_url.clone().into(),
+            tertiary_url.clone().into(),
+        ],
+        ..Default::default()
+    };
+    let mut base_layer = EthereumBaseLayerContract::new(config);
+
+    // Cycle twice to land on the tertiary endpoint.
+    base_layer.cycle_provider_url().await.unwrap();
+    base_layer.cycle_provider_url().await.unwrap();
+    assert_eq!(base_layer.get_url().await.unwrap().expose_secret(), tertiary_url);
+
+    // Reset to primary.
+    base_layer.reset_provider_url_to_primary().await.unwrap();
+    assert_eq!(base_layer.get_url().await.unwrap().expose_secret(), primary_url);
+
+    // Calling reset again when already on primary is a no-op.
+    base_layer.reset_provider_url_to_primary().await.unwrap();
+    assert_eq!(base_layer.get_url().await.unwrap().expose_secret(), primary_url);
+}
+
 #[test]
 fn create_l1_event_data_rejects_out_of_range_inputs() {
     let oversized = felt_max_u256() + U256::from(1_u8);

crates/papyrus_base_layer/src/cyclic_base_layer_wrapper.rs

@@ -1,4 +1,5 @@
 use std::ops::RangeInclusive;
+use std::time::Duration;
 
 use apollo_config::secrets::Sensitive;
 use async_trait::async_trait;
@@ -15,11 +16,26 @@ pub mod cyclic_base_layer_wrapper_test;
 #[derive(Debug)]
 pub struct CyclicBaseLayerWrapper<B: BaseLayerContract + Send + Sync> {
     base_layer: B,
+    retry_primary_interval: Duration,
+    last_primary_retry: tokio::time::Instant,
 }
 
 impl<B: BaseLayerContract + Send + Sync> CyclicBaseLayerWrapper<B> {
-    pub fn new(base_layer: B) -> Self {
-        Self { base_layer }
+    pub fn new(base_layer: B, retry_primary_interval: Duration) -> Self {
+        Self { base_layer, retry_primary_interval, last_primary_retry: tokio::time::Instant::now() }
+    }
+
+    // Retries the primary endpoint once the interval has elapsed since we left it. Does nothing
+    // while already on the primary, so the timer is untouched until a failover moves us off it.
+    async fn retry_primary_if_due(&mut self) -> Result<(), B::Error> {
+        if self.base_layer.is_at_primary().await? {
+            return Ok(());
+        }
+        if self.last_primary_retry.elapsed() >= self.retry_primary_interval {
+            self.last_primary_retry = tokio::time::Instant::now();
+            self.base_layer.reset_provider_url_to_primary().await?;
+        }
+        Ok(())
     }
 
     // Check the result of a function call to the base layer. If it fails, cycle the URL and signal
@@ -38,11 +54,21 @@ impl<B: BaseLayerContract + Send + Sync> CyclicBaseLayerWrapper<B> {
         let Ok(current_url) = current_url_result else {
             return Some(Err(current_url_result.expect_err("result is checked at let-else")));
         };
+        // Record whether we are about to leave the primary, before cycling away from it.
+        let is_at_primary_result = self.base_layer.is_at_primary().await;
+        let Ok(was_at_primary) = is_at_primary_result else {
+            return Some(Err(is_at_primary_result.expect_err("result is checked at let-else")));
+        };
         // Otherwise, cycle the URL so we can try again. Return error in case it fails to cycle.
         let cycle_url_result = self.base_layer.cycle_provider_url().await;
         let Ok(()) = cycle_url_result else {
             return Some(Err(cycle_url_result.expect_err("result is checked at let-else")));
         };
+        // Restart the retry-primary clock only when this failover leaves the primary, so the wait
+        // is measured from when we left it; cycling between backups must not push the retry out.
+        if was_at_primary {
+            self.last_primary_retry = tokio::time::Instant::now();
+        }
         // Get the new URL (return error in case it fails to get it).
         let new_url_result = self.base_layer.get_url().await;
         let Ok(new_url) = new_url_result else {
@@ -69,6 +95,7 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
         &mut self,
         l1_block: L1BlockNumber,
     ) -> Result<BlockHashAndNumber, Self::Error> {
+        self.retry_primary_if_due().await?;
         let start_url = self.base_layer.get_url().await?;
         loop {
             let result = self.base_layer.get_proved_block_at(l1_block).await;
@@ -79,6 +106,7 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
     }
 
     async fn latest_l1_block_number(&mut self) -> Result<L1BlockNumber, Self::Error> {
+        self.retry_primary_if_due().await?;
         let start_url = self.base_layer.get_url().await?;
         loop {
             let result = self.base_layer.latest_l1_block_number().await;
@@ -92,6 +120,7 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
         &mut self,
         block_number: L1BlockNumber,
     ) -> Result<Option<L1BlockReference>, Self::Error> {
+        self.retry_primary_if_due().await?;
         let start_url = self.base_layer.get_url().await?;
         loop {
             let result = self.base_layer.l1_block_at(block_number).await;
@@ -106,6 +135,7 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
         block_range: RangeInclusive<L1BlockNumber>,
         event_identifiers: &'a [&'a str],
     ) -> Result<Vec<L1Event>, Self::Error> {
+        self.retry_primary_if_due().await?;
         let start_url = self.base_layer.get_url().await?;
         loop {
             let result = self.base_layer.events(block_range.clone(), event_identifiers).await;
@@ -119,6 +149,7 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
         &mut self,
         block_number: L1BlockNumber,
     ) -> Result<Option<L1BlockHeader>, Self::Error> {
+        self.retry_primary_if_due().await?;
         let start_url = self.base_layer.get_url().await?;
         loop {
             let result = self.base_layer.get_block_header(block_number).await;
@@ -128,6 +159,8 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
         }
     }
 
+    // Takes &self so it cannot cycle or retry endpoints; callers needing resilience use the &mut
+    // self methods.
     async fn get_block_header_immutable(
         &self,
         block_number: L1BlockNumber,
@@ -146,4 +179,12 @@ impl<B: BaseLayerContract + Send + Sync> BaseLayerContract for CyclicBaseLayerWr
     async fn cycle_provider_url(&mut self) -> Result<(), Self::Error> {
         self.base_layer.cycle_provider_url().await
     }
+
+    async fn reset_provider_url_to_primary(&mut self) -> Result<(), Self::Error> {
+        self.base_layer.reset_provider_url_to_primary().await
+    }
+
+    async fn is_at_primary(&self) -> Result<bool, Self::Error> {
+        self.base_layer.is_at_primary().await
+    }
 }