apollo_l1_provider: go to bootstrap when getting commit_block, not on startup (#9840)

Author: guy-starkware

crates/apollo_l1_provider/src/bootstrapper.rs

@@ -2,30 +2,28 @@ 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 indexmap::IndexSet;
 use starknet_api::block::BlockNumber;
 use starknet_api::transaction::TransactionHash;
-use tokio::sync::OnceCell;
-use tracing::{debug, error, info};
+use tracing::debug;
 
-pub type LazyCatchUpHeight = Arc<OnceCell<BlockNumber>>;
+// When the Provider gets a commit_block that is too high, it starts bootstrapping.
+// The commit is rejected by the provider, so it must use sync to catch up to the height of the
+// commit, including that height. The sync task continues until reaching the target height,
+// inclusive, and only after the commit_block (from sync) causes the Provider's current height to be
+// one above the target height, is the backlog applied. Once done with the sync+backlog, the current
+// height should be one above the last commit in the backlog, which makes it ready for the next
+// commit_block from the batcher.
 
 /// Caches 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 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 catch_up_height: BlockNumber,
     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,
@@ -39,16 +37,14 @@ impl Bootstrapper {
 
     pub fn new(
         l1_provider_client: SharedL1ProviderClient,
-        batcher_client: SharedBatcherClient,
         sync_client: SharedStateSyncClient,
         sync_retry_interval: Duration,
-        catch_up_height: LazyCatchUpHeight,
+        catch_up_height: BlockNumber,
     ) -> Self {
         Self {
             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(),
@@ -57,12 +53,8 @@ impl Bootstrapper {
     }
 
     /// Check if the caller has caught up with the bootstrapper.
-    /// 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 = match self.catch_up_height() {
-            Some(catch_up_height) => current_provider_height > catch_up_height,
-            None => current_provider_height == BlockNumber(0),
-        };
+        let is_caught_up = current_provider_height > self.catch_up_height;
 
         self.sync_task_health_check(is_caught_up);
 
@@ -88,25 +80,29 @@ impl Bootstrapper {
 
     /// Spawns async task that produces and sends commit block messages to the provider, according
     /// 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) {
+    pub fn start_l2_sync(
+        &mut self,
+        current_provider_height: BlockNumber,
+        catch_up_height: BlockNumber,
+    ) {
+        self.catch_up_height = catch_up_height;
         // 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(),
+            catch_up_height,
             self.sync_retry_interval,
         ));
 
         self.sync_task_handle = SyncTaskHandle::Started(sync_task_handle.into());
     }
 
-    pub fn catch_up_height(&self) -> Option<BlockNumber> {
-        self.catch_up_height.get().copied()
+    pub fn catch_up_height(&self) -> BlockNumber {
+        self.catch_up_height
     }
 
     pub fn sync_started(&self) -> bool {
@@ -155,36 +151,17 @@ impl std::fmt::Debug for Bootstrapper {
 
 async fn l2_sync_task(
     l1_provider_client: SharedL1ProviderClient,
-    batcher_client: SharedBatcherClient,
     sync_client: SharedStateSyncClient,
     mut current_height: BlockNumber,
-    catch_up_height: LazyCatchUpHeight,
+    catch_up_height: BlockNumber,
     retry_interval: Duration,
 ) {
-    info!("Try fetching batcher height to initialize catch up point");
-    while !catch_up_height.initialized() {
-        let Ok(GetHeightResponse { height: batcher_height }) = batcher_client.get_height().await
-        else {
-            error!("Batcher height request failed. Retrying...");
-            tokio::time::sleep(retry_interval).await;
-            continue;
-        };
-
-        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");
-
     while current_height <= catch_up_height {
         // TODO(Gilad): add tracing instrument.
-        debug!("Try syncing L1Provider with L2 height: {}", current_height);
+        debug!(
+            "Syncing L1Provider with L2 height: {} to target height: {}",
+            current_height, catch_up_height
+        );
         let block = sync_client.get_block(current_height).await.inspect_err(|err| debug!("{err}"));
 
         match block {

crates/apollo_l1_provider/src/l1_provider.rs

@@ -68,15 +68,13 @@ impl L1Provider {
             );
         };
 
-        // The provider now goes into bootstrap state.
-        // TODO(guyn): in the future, this will happen when batcher calls the provider with a height
-        // bigger than the current height. This would happen either in Uninitialized or in other
-        // non-bootstrap states. That means we will move from Uninitialized to Pending (but
-        // on a height much lower than the batcher's).
+        // The provider now goes into Pending state.
+        // The current_height is set to a very old height, that doesn't include any of the events
+        // sent now, or to be scraped in the future. The provider will begin bootstrapping when the
+        // batcher calls commit_block with a height above the current height.
         self.start_height = Some(historic_l2_height);
         self.current_height = historic_l2_height;
-        self.state = ProviderState::Bootstrap;
-        self.bootstrapper.start_l2_sync(self.current_height).await;
+        self.state = ProviderState::Pending;
         self.add_events(events)?;
 
         Ok(())
@@ -284,16 +282,43 @@ impl L1Provider {
 
         // If not historical height and not bootstrapping, must go into bootstrap state upon getting
         // wrong height.
-        // TODO(guyn): for now, we go into bootstrap using panic. We should improve this.
-        self.check_height_with_panic(height);
-        self.apply_commit_block(committed_txs, rejected_txs);
-
-        self.state = self.state.transition_to_pending();
-        Ok(())
+        match self.check_height_with_error(height) {
+            Ok(_) => {
+                self.apply_commit_block(committed_txs, rejected_txs);
+                self.state = self.state.transition_to_pending();
+                Ok(())
+            }
+            Err(err) => {
+                // We are returning an error -> not accepting the block with this height. In order
+                // to to be able to serve future requests, we must catch up to it, and finish
+                // catching up when the provider has synced this height.
+                if self.state.is_uninitialized() {
+                    warn!(
+                        "Provider received a block height ({height}) while it is uninitialized. \
+                         Cannot start bootstrapping until getting the historic_height from the \
+                         scraper during the initialize call."
+                    );
+                } else {
+                    info!(
+                        "Provider received a block_height ({height}) that is higher than the \
+                         current height ({}), starting bootstrapping.",
+                        self.current_height
+                    );
+                    self.start_bootstrapping(height);
+                }
+                Err(err)
+            }
+        }
     }
 
     // Functions called internally.
 
+    /// Go from current state to Bootstrap state and start the L2 sync.
+    pub fn start_bootstrapping(&mut self, target_height: BlockNumber) {
+        self.state = ProviderState::Bootstrap;
+        self.bootstrapper.start_l2_sync(self.current_height, target_height);
+    }
+
     /// Commit the given transactions, and increment the current height.
     fn apply_commit_block(
         &mut self,
@@ -323,7 +348,6 @@ impl L1Provider {
             "Bootstrapper processing commit-block at height: {new_height}, current height is \
              {current_height}"
         );
-
         match new_height.cmp(&current_height) {
             // This is likely a bug in the batcher/sync, it should never be _behind_ the provider.
             Less => {
@@ -369,6 +393,8 @@ impl L1Provider {
         };
 
         // If caught up, apply the backlog and transition to Pending.
+        // Note that at this point self.current_height is already incremented to the next height, it
+        // is one more than the latest block that was committed.
         if self.bootstrapper.is_caught_up(self.current_height) {
             info!(
                 "Bootstrapper sync completed, provider height is now {}, processing backlog...",
@@ -408,21 +434,6 @@ impl L1Provider {
         Ok(())
     }
 
-    fn check_height_with_panic(&mut self, height: BlockNumber) {
-        if height > self.current_height {
-            // TODO(shahak): Add a way to move to bootstrap mode from any point and move to
-            // bootstrap here instead of panicking.
-            panic!(
-                "Batcher surpassed l1 provider. Panicking in order to restart the provider and \
-                 bootstrap again. l1 provider height: {}, batcher height: {}",
-                self.current_height, height
-            );
-        }
-        if height < self.current_height {
-            panic!("Unexpected height: expected >= {}, got {}", self.current_height, height);
-        }
-    }
-
     fn check_height_with_error(&mut self, height: BlockNumber) -> L1ProviderResult<()> {
         if height != self.current_height {
             return Err(L1ProviderError::UnexpectedHeight {
@@ -535,6 +546,8 @@ impl L1ProviderBuilder {
             })
             .or(self.startup_height);
 
+        // TODO(guyn): try to remove the input catchup_height entirely (check if it is needed for
+        // tests/Anvil).
         let catchup_height = self
             .config
             .bootstrap_catch_up_height_override
@@ -546,13 +559,10 @@ impl L1ProviderBuilder {
                 );
             })
             .or(self.catchup_height)
-            .map(|catchup_height| Arc::new(catchup_height.into()))
-            // When kept None, this value is fetched from the batcher by the bootstrapper at runtime.
-            .unwrap_or_default();
+            .unwrap_or_default(); // If bootstrapper is not running, the catchup height can be arbitrarily low. 
 
         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,