add test cases

Author: frisitano

crates/chain-orchestrator/src/event.rs

@@ -54,7 +54,7 @@ pub enum ChainOrchestratorEvent {
     },
     /// A batch has been reverted returning the batch info and the new safe head.
     BatchReverted {
-        /// The batch info of the reverted batch.
+        /// The latest batch info after the revert.
         batch_info: BatchInfo,
         /// The new safe head after the revert.
         safe_head: BlockInfo,

crates/chain-orchestrator/src/handle/command.rs

@@ -30,6 +30,9 @@ pub enum ChainOrchestratorCommand<N: FullNetwork<Primitives = ScrollNetworkPrimi
     /// Enable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     SetGossip((bool, oneshot::Sender<()>)),
+    /// Returns a database handle for direct database access.
+    #[cfg(feature = "test-utils")]
+    DatabaseHandle(oneshot::Sender<std::sync::Arc<scroll_db::Database>>),
 }
 
 /// The database queries that can be sent to the rollup manager.

crates/chain-orchestrator/src/handle/mod.rs

@@ -110,4 +110,14 @@ impl<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> ChainOrchestratorHand
         self.send_command(ChainOrchestratorCommand::SetGossip((enabled, tx)));
         rx.await
     }
+
+    /// Sends a command to the rollup manager to get a database handle for direct database access.
+    #[cfg(feature = "test-utils")]
+    pub async fn get_database_handle(
+        &self,
+    ) -> Result<std::sync::Arc<scroll_db::Database>, oneshot::error::RecvError> {
+        let (tx, rx) = oneshot::channel();
+        self.send_command(ChainOrchestratorCommand::DatabaseHandle(tx));
+        rx.await
+    }
 }

crates/chain-orchestrator/src/lib.rs

@@ -390,6 +390,10 @@ impl<
                 self.network.handle().set_gossip(enabled).await;
                 let _ = tx.send(());
             }
+            #[cfg(feature = "test-utils")]
+            ChainOrchestratorCommand::DatabaseHandle(tx) => {
+                let _ = tx.send(self.database.clone());
+            }
         }
 
         Ok(())
@@ -708,7 +712,7 @@ impl<
             .await?;
 
         if self.sync_state.is_synced() {
-            self.derivation_pipeline.push_batch(batch_info, BatchStatus::Committed).await;
+            self.derivation_pipeline.push_batch(batch_info, BatchStatus::Consolidated).await;
         }
 
         Ok(event)
@@ -763,12 +767,12 @@ impl<
 
     /// Handles a batch revert event by updating the database.
     async fn handle_batch_revert(
-        &self,
+        &mut self,
         start_index: u64,
         end_index: u64,
         l1_block_info: BlockInfo,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let event = self
+        let (safe_block_info, batch_info) = self
             .database
             .tx_mut(move |tx| async move {
                 tx.insert_l1_block_info(l1_block_info).await?;
@@ -780,15 +784,14 @@ impl<
                 .await?;
 
                 // handle the case of a batch revert.
-                let (safe_head, batch_info) = tx.get_latest_safe_l2_info().await?;
-
-                let event = ChainOrchestratorEvent::BatchReverted { batch_info, safe_head };
-
-                Ok::<_, ChainOrchestratorError>(Some(event))
+                Ok::<_, ChainOrchestratorError>(tx.get_latest_safe_l2_info().await?)
             })
             .await?;
 
-        Ok(event)
+        // Update the forkchoice state to the new safe block.
+        self.engine.update_fcs(None, Some(safe_block_info), None).await?;
+
+        Ok(Some(ChainOrchestratorEvent::BatchReverted { batch_info, safe_head: safe_block_info }))
     }
 
     /// Handles an L1 message by inserting it into the database.
@@ -1144,7 +1147,7 @@ impl<
     ///
     /// This involves validating the L1 messages in the blocks against the expected L1 messages
     /// synced from L1.
-    async fn consolidate_chain(&self) -> Result<(), ChainOrchestratorError> {
+    async fn consolidate_chain(&mut self) -> Result<(), ChainOrchestratorError> {
         tracing::trace!(target: "scroll::chain_orchestrator", fcs = ?self.engine.fcs(), "Consolidating chain from safe to head");
 
         let safe_block_number = self.engine.fcs().safe_block_info().number;
@@ -1181,6 +1184,14 @@ impl<
         // transactions into the transaction pool.
         self.network.handle().inner().update_sync_state(RethSyncState::Idle);
 
+        // Fetch all unprocessed committed batches and push them to the derivation pipeline as
+        // consolidated.
+        let committed_batches =
+            self.database.fetch_and_update_unprocessed_committed_batches().await?;
+        for batch_commit in committed_batches {
+            self.derivation_pipeline.push_batch(batch_commit, BatchStatus::Consolidated).await;
+        }
+
         self.notify(ChainOrchestratorEvent::ChainConsolidated {
             from: safe_block_number,
             to: head_block_number,

crates/database/db/src/db.rs

@@ -323,6 +323,18 @@ impl DatabaseWriteOperations for Database {
         )
     }
 
+    async fn fetch_and_update_unprocessed_committed_batches(
+        &self,
+    ) -> Result<Vec<BatchInfo>, DatabaseError> {
+        metered!(
+            DatabaseOperation::FetchAndUpdateUnprocessedCommittedBatches,
+            self,
+            tx_mut(
+                move |tx| async move { tx.fetch_and_update_unprocessed_committed_batches().await }
+            )
+        )
+    }
+
     async fn delete_batches_gt_block_number(
         &self,
         block_number: u64,

crates/database/db/src/metrics.rs

@@ -33,6 +33,7 @@ pub(crate) enum DatabaseOperation {
     SetProcessedL1BlockNumber,
     SetL2HeadBlockNumber,
     FetchAndUpdateUnprocessedFinalizedBatches,
+    FetchAndUpdateUnprocessedCommittedBatches,
     DeleteBatchesGtBlockNumber,
     DeleteBatchesGtBatchIndex,
     InsertL1Message,
@@ -98,6 +99,9 @@ impl DatabaseOperation {
             Self::FetchAndUpdateUnprocessedFinalizedBatches => {
                 "fetch_and_update_unprocessed_finalized_batches"
             }
+            Self::FetchAndUpdateUnprocessedCommittedBatches => {
+                "fetch_and_update_unprocessed_committed_batches"
+            }
             Self::DeleteBatchesGtBlockNumber => "delete_batches_gt_block_number",
             Self::DeleteBatchesGtBatchIndex => "delete_batches_gt_batch_index",
             Self::InsertL1Message => "insert_l1_message",

crates/database/db/src/operations.rs

@@ -53,12 +53,17 @@ pub trait DatabaseWriteOperations {
     async fn set_l2_head_block_number(&self, number: u64) -> Result<(), DatabaseError>;
 
     /// Fetches unprocessed batches up to the provided finalized L1 block number and updates their
-    /// status.
+    /// status to processing.
     async fn fetch_and_update_unprocessed_finalized_batches(
         &self,
         finalized_l1_block_number: u64,
     ) -> Result<Vec<BatchInfo>, DatabaseError>;
 
+    /// Fetches unprocessed committed batches and updates their status to processing.
+    async fn fetch_and_update_unprocessed_committed_batches(
+        &self,
+    ) -> Result<Vec<BatchInfo>, DatabaseError>;
+
     /// Delete all [`BatchCommitData`]s with a block number greater than the provided block number.
     async fn delete_batches_gt_block_number(&self, block_number: u64)
         -> Result<u64, DatabaseError>;
@@ -520,6 +525,35 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
         Ok(batches)
     }
 
+    async fn fetch_and_update_unprocessed_committed_batches(
+        &self,
+    ) -> Result<Vec<BatchInfo>, DatabaseError> {
+        let conditions = Condition::all().add(models::batch_commit::Column::Status.eq("committed"));
+
+        let batches = models::batch_commit::Entity::find()
+            .filter(conditions.clone())
+            .order_by_asc(models::batch_commit::Column::Index)
+            .select_only()
+            .column(models::batch_commit::Column::Index)
+            .column(models::batch_commit::Column::Hash)
+            .into_tuple::<(i64, Vec<u8>)>()
+            .all(self.get_connection())
+            .await
+            .map(|x| {
+                x.into_iter()
+                    .map(|(index, hash)| BatchInfo::new(index as u64, B256::from_slice(&hash)))
+                    .collect()
+            })?;
+
+        models::batch_commit::Entity::update_many()
+            .col_expr(models::batch_commit::Column::Status, Expr::value("processing"))
+            .filter(conditions)
+            .exec(self.get_connection())
+            .await?;
+
+        Ok(batches)
+    }
+
     async fn delete_batches_gt_block_number(
         &self,
         block_number: u64,