fix(apollo_l1_provider): change catch up height from sync to batcher height (#6120)

Author: noamsp-starkware

Cargo.lock

@@ -51,16 +51,16 @@
   "class_manager_config.class_storage_config.class_hash_storage_config.path_prefix": "/data/node_0/executable_0/class_manager/class_hash_storage",
   "class_manager_config.class_storage_config.persistent_root": "/data/node_0/executable_0/class_manager/classes",
   "compiler_config.max_bytecode_size": 81920,
-  "components.batcher.execution_mode": "Disabled",
+  "components.batcher.execution_mode": "Remote",
   "components.batcher.ip": "0.0.0.0",
   "components.batcher.local_server_config.channel_buffer_size": 32,
   "components.batcher.max_concurrency": 10,
-  "components.batcher.port": 0,
-  "components.batcher.remote_client_config.idle_connections": 18446744073709551615,
+  "components.batcher.port": 15000,
+  "components.batcher.remote_client_config.idle_connections": 5,
   "components.batcher.remote_client_config.idle_timeout": 90,
   "components.batcher.remote_client_config.retries": 3,
   "components.batcher.remote_client_config.retry_interval": 3,
-  "components.batcher.url": "localhost",
+  "components.batcher.url": "sequencer-batcher-service",
   "components.class_manager.execution_mode": "Disabled",
   "components.class_manager.ip": "0.0.0.0",
   "components.class_manager.local_server_config.channel_buffer_size": 32,

config/sequencer/testing/app_configs/distributed/deployment_test_distributed/l1.json

@@ -114,6 +114,7 @@ impl GetComponentConfigs for DistributedNodeServiceName {
                     l1_gas_price_provider.local(),
                     l1_provider.local(),
                     state_sync.remote(),
+                    batcher.remote(),
                 ),
                 DistributedNodeServiceName::Mempool => get_mempool_component_config(
                     mempool.local(),
@@ -583,13 +584,16 @@ fn get_l1_component_config(
     l1_gas_price_provider_local_config: ReactiveComponentExecutionConfig,
     l1_provider_local_config: ReactiveComponentExecutionConfig,
     state_sync_remote_config: ReactiveComponentExecutionConfig,
+    batcher_remote_config: ReactiveComponentExecutionConfig,
 ) -> ComponentConfig {
     let mut config = ComponentConfig::disabled();
+
     config.l1_gas_price_provider = l1_gas_price_provider_local_config;
     config.l1_gas_price_scraper = ActiveComponentExecutionConfig::enabled();
     config.l1_provider = l1_provider_local_config;
     config.l1_scraper = ActiveComponentExecutionConfig::enabled();
     config.state_sync = state_sync_remote_config;
     config.monitoring_endpoint = ActiveComponentExecutionConfig::enabled();
+    config.batcher = batcher_remote_config;
     config
 }

crates/apollo_deployments/src/deployments/distributed.rs

@@ -6,9 +6,10 @@ repository.workspace = true
 license.workspace = true
 
 [features]
-testing = ["pretty_assertions", "starknet_api/testing"]
+testing = ["apollo_batcher_types/testing", "pretty_assertions", "starknet_api/testing"]
 
 [dependencies]
+apollo_batcher_types.workspace = true
 apollo_config.workspace = true
 apollo_infra.workspace = true
 apollo_l1_provider_types.workspace = true
@@ -29,6 +30,7 @@ validator.workspace = true
 
 [dev-dependencies]
 alloy.workspace = true
+apollo_batcher_types = { workspace = true, features = ["testing"] }
 apollo_l1_provider_types = { workspace = true, features = ["testing"] }
 apollo_state_sync_types = { workspace = true, features = ["testing"] }
 assert_matches.workspace = true

crates/apollo_l1_provider/Cargo.toml

@@ -3,26 +3,29 @@ use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
+use apollo_batcher_types::batcher_types::GetHeightResponse;
+use apollo_batcher_types::communication::SharedBatcherClient;
 use apollo_l1_provider_types::SharedL1ProviderClient;
 use apollo_state_sync_types::communication::SharedStateSyncClient;
 use starknet_api::block::BlockNumber;
 use starknet_api::transaction::TransactionHash;
 use tokio::sync::OnceCell;
-use tracing::{debug, info};
+use tracing::{debug, error, info};
 
 pub type LazyCatchUpHeight = Arc<OnceCell<BlockNumber>>;
 
 /// Cache's commits to be applied later. This flow is only relevant while the node is starting up.
 #[derive(Clone)]
 pub struct Bootstrapper {
-    /// The catch-up height for the bootstrapper is the sync height (unless overridden explicitly).
-    /// This value, due to infra constraints as of now, is only fetchable _after_ the provider is
-    /// running, and not during its initialization, hence we are forced to lazily fetch it at
-    /// runtime.
+    /// The catch-up height for the bootstrapper is the batcher height (unless overridden
+    /// explicitly). This value, due to infra constraints as of now, is only fetchable _after_
+    /// the provider is running, and not during its initialization, hence we are forced to
+    /// lazily fetch it at runtime.
     pub catch_up_height: LazyCatchUpHeight,
     pub sync_retry_interval: Duration,
     pub commit_block_backlog: Vec<CommitBlockBacklog>,
     pub l1_provider_client: SharedL1ProviderClient,
+    pub batcher_client: SharedBatcherClient,
     pub sync_client: SharedStateSyncClient,
     // Keep track of sync task for health checks and logging status.
     pub sync_task_handle: SyncTaskHandle,
@@ -36,6 +39,7 @@ impl Bootstrapper {
 
     pub fn new(
         l1_provider_client: SharedL1ProviderClient,
+        batcher_client: SharedBatcherClient,
         sync_client: SharedStateSyncClient,
         sync_retry_interval: Duration,
         catch_up_height: LazyCatchUpHeight,
@@ -44,6 +48,7 @@ impl Bootstrapper {
             sync_retry_interval,
             commit_block_backlog: Default::default(),
             l1_provider_client,
+            batcher_client,
             sync_client,
             sync_task_handle: SyncTaskHandle::NotStartedYet,
             n_sync_health_check_failures: Default::default(),
@@ -52,11 +57,12 @@ impl Bootstrapper {
     }
 
     /// Check if the caller has caught up with the bootstrapper.
-    /// If catch_up_height is unset, the sync isn't even ready yet.
+    /// If catch_up_height is unset, the batcher isn't even ready yet.
     pub fn is_caught_up(&self, current_provider_height: BlockNumber) -> bool {
-        let is_caught_up = self
-            .catch_up_height()
-            .is_some_and(|catch_up_height| current_provider_height > catch_up_height);
+        let is_caught_up = match self.catch_up_height() {
+            Some(catch_up_height) => current_provider_height > catch_up_height,
+            None => current_provider_height == BlockNumber(0),
+        };
 
         self.sync_task_health_check(is_caught_up);
 
@@ -80,14 +86,15 @@ impl Bootstrapper {
     }
 
     /// Spawns async task that produces and sends commit block messages to the provider, according
-    /// to information from the sync client, until the provider is caught up.
+    /// to information from the batcher and sync clients, until the provider is caught up.
     pub async fn start_l2_sync(&mut self, current_provider_height: BlockNumber) {
         // FIXME: spawning a task like this is evil.
         // However, we aren't using the task executor, so no choice :(
         // Once we start using a centralized threadpool, spawn through it instead of the
         // tokio runtime.
         let sync_task_handle = tokio::spawn(l2_sync_task(
             self.l1_provider_client.clone(),
+            self.batcher_client.clone(),
             self.sync_client.clone(),
             current_provider_height,
             self.catch_up_height.clone(),
@@ -145,27 +152,34 @@ impl std::fmt::Debug for Bootstrapper {
     }
 }
 
+// TODO(noamsp): fix catch up height to use batcher height and not the latest block number in
+// storage.
 async fn l2_sync_task(
     l1_provider_client: SharedL1ProviderClient,
+    batcher_client: SharedBatcherClient,
     sync_client: SharedStateSyncClient,
     mut current_height: BlockNumber,
     catch_up_height: LazyCatchUpHeight,
     retry_interval: Duration,
 ) {
-    // Currently infra doesn't support starting up the provider only after sync is ready.
-    info!("Try fetching sync height to initialize catch up point");
+    info!("Try fetching batcher height to initialize catch up point");
     while !catch_up_height.initialized() {
-        let Some(sync_height) = sync_client
-            .get_latest_block_number()
-            .await
-            .expect("network error handling not supported yet")
+        let Ok(GetHeightResponse { height: batcher_height }) = batcher_client.get_height().await
         else {
-            debug!("Sync state not ready yet, trying again later");
+            error!("Batcher height request failed. Retrying...");
             tokio::time::sleep(retry_interval).await;
             continue;
         };
-        info!("Catch up height set: {sync_height}");
-        catch_up_height.set(sync_height).expect("This is the only write-point, cannot fail")
+
+        let Some(batcher_latest_block_number) = batcher_height.prev() else {
+            info!("Batcher height is 0, no need to catch up. exiting...");
+            return;
+        };
+
+        info!("Catch up height set: {batcher_latest_block_number}");
+        catch_up_height
+            .set(batcher_latest_block_number)
+            .expect("This is the only write-point, cannot fail")
     }
     let catch_up_height = *catch_up_height.get().expect("Initialized above");
 

crates/apollo_l1_provider/src/bootstrapper.rs

@@ -2,6 +2,7 @@ use std::cmp::Ordering::{Equal, Greater, Less};
 use std::collections::HashSet;
 use std::sync::Arc;
 
+use apollo_batcher_types::communication::SharedBatcherClient;
 use apollo_infra::component_definitions::ComponentStarter;
 use apollo_l1_provider_types::errors::L1ProviderError;
 use apollo_l1_provider_types::{
@@ -264,6 +265,7 @@ impl ComponentStarter for L1Provider {}
 pub struct L1ProviderBuilder {
     pub config: L1ProviderConfig,
     pub l1_provider_client: SharedL1ProviderClient,
+    pub batcher_client: SharedBatcherClient,
     pub state_sync_client: SharedStateSyncClient,
     startup_height: Option<BlockNumber>,
     catchup_height: Option<BlockNumber>,
@@ -273,11 +275,13 @@ impl L1ProviderBuilder {
     pub fn new(
         config: L1ProviderConfig,
         l1_provider_client: SharedL1ProviderClient,
+        batcher_client: SharedBatcherClient,
         state_sync_client: SharedStateSyncClient,
     ) -> Self {
         Self {
             config,
             l1_provider_client,
+            batcher_client,
             state_sync_client,
             startup_height: None,
             catchup_height: None,
@@ -320,16 +324,17 @@ impl L1ProviderBuilder {
                 warn!(
                     "OVERRIDE L1 provider catch-up height: {catch_up_height_override}. WARNING: \
                      this MUST be greater or equal to the default non-overridden value, which is \
-                     the (runtime fetched) sync height, or the sync will never complete!"
+                     the (runtime fetched) batcher height, or the sync will never complete!"
                 );
             })
             .or(self.catchup_height)
             .map(|catchup_height| Arc::new(catchup_height.into()))
-            // When kept None, this value is fetched from the sync by the bootstrapper at runtime.
+            // When kept None, this value is fetched from the batcher by the bootstrapper at runtime.
             .unwrap_or_default();
 
         let bootstrapper = Bootstrapper::new(
             self.l1_provider_client,
+            self.batcher_client,
             self.state_sync_client,
             self.config.startup_sync_sleep_retry_interval_seconds,
             catchup_height,

crates/apollo_l1_provider/src/l1_provider.rs

@@ -2,6 +2,7 @@ use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
 
+use apollo_batcher_types::communication::MockBatcherClient;
 use apollo_l1_provider_types::errors::L1ProviderError;
 use apollo_l1_provider_types::SessionState::{
     self,
@@ -56,16 +57,17 @@ macro_rules! bootstrapper {
                     height: BlockNumber($height),
                     committed_txs: vec![$(tx_hash!($tx)),*]
                 }),*
-                ].into_iter().collect(),
-                catch_up_height: Arc::new(BlockNumber($catch).into()),
-                l1_provider_client: Arc::new(FakeL1ProviderClient::default()),
-                sync_client: Arc::new(MockStateSyncClient::default()),
-                sync_task_handle: SyncTaskHandle::default(),
-                n_sync_health_check_failures: Default::default(),
-                sync_retry_interval: Duration::from_millis(10)
-            }
-        }};
-    }
+            ].into_iter().collect(),
+            catch_up_height: Arc::new(BlockNumber($catch).into()),
+            l1_provider_client: Arc::new(FakeL1ProviderClient::default()),
+            batcher_client: Arc::new(MockBatcherClient::default()),
+            sync_client: Arc::new(MockStateSyncClient::default()),
+            sync_task_handle: SyncTaskHandle::default(),
+            n_sync_health_check_failures: Default::default(),
+            sync_retry_interval: Duration::from_millis(10)
+        }
+    }};
+}
 
 fn l1_handler_event(tx_hash: TransactionHash) -> Event {
     Event::L1HandlerTransaction(executable_l1_handler_tx(L1HandlerTxArgs {