implement gap and skip detection for revert events

Author: jonastheis

crates/chain-orchestrator/src/event.rs

@@ -66,7 +66,15 @@ pub enum ChainOrchestratorEvent {
         /// The new safe head after the revert.
         safe_head: BlockInfo,
     },
-    // TODO: revert events
+    /// A gap has been detected in the reverted batches.
+    BatchRevertGap {
+        /// The missing batch index.
+        missing_index: u64,
+        /// The latest known L1 block number to reset to before the gap.
+        l1_block_number_reset: u64,
+    },
+    /// A duplicate batch revert has been detected.
+    BatchRevertDuplicate(u64),
     /// A new L1 block has been received returning the L1 block number.
     NewL1Block(u64),
     /// An L1 block has been finalized returning the L1 block number and the list of finalized

crates/chain-orchestrator/src/lib.rs

@@ -26,7 +26,10 @@ use scroll_alloy_consensus::{ScrollTxEnvelope, TxL1Message};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_alloy_provider::ScrollEngineApi;
-use scroll_db::{Database, DatabaseError, DatabaseReadOperations, DatabaseWriteOperations, L1MessageKey, TXMut, UnwindResult};
+use scroll_db::{
+    Database, DatabaseError, DatabaseReadOperations, DatabaseWriteOperations, L1MessageKey, TXMut,
+    UnwindResult,
+};
 use scroll_derivation_pipeline::{BatchDerivationResult, DerivationPipeline};
 use scroll_engine::Engine;
 use scroll_network::{
@@ -61,7 +64,8 @@ pub use sync::{SyncMode, SyncState};
 
 mod status;
 use crate::ChainOrchestratorEvent::{
-    BatchCommitDuplicate, BatchCommitGap, L1MessageDuplicate, L1MessageGap,
+    BatchCommitDuplicate, BatchCommitGap, BatchRevertDuplicate, BatchRevertGap, L1MessageDuplicate,
+    L1MessageGap,
 };
 pub use status::ChainOrchestratorStatus;
 
@@ -755,7 +759,14 @@ impl<
 
                     // Perform a consistency check to ensure the previous commit batch exists in the
                     // database.
-                    if tx.get_batch_by_index(prev_batch_index).await?.is_none() {
+                    if tx
+                        .get_batch_by_index(prev_batch_index)
+                        .await?
+                        .iter()
+                        .filter(|x| x.reverted_block_number.is_none())
+                        .count() ==
+                        0
+                    {
                         // Query database for the L1 block of the last known batch
                         let reset_block = tx.get_last_batch_commit_l1_block().await?.unwrap_or(0);
 
@@ -766,16 +777,20 @@ impl<
                     }
 
                     // Check if batch already exists in DB.
-                    if let Some(existing_batch) = tx.get_batch_by_index(batch.index).await? {
+                    for existing_batch in tx.get_batch_by_index(batch.index).await? {
                         if existing_batch.hash == batch.hash {
                             // This means we have already processed this batch commit, we will skip
                             // it.
                             return Ok(Some(BatchCommitDuplicate(existing_batch.index)));
+                        } else if existing_batch.reverted_block_number.is_none() {
+                            // This means we have received a different batch commit at the same
+                            // index which has not been reverted yet. ->
+                            // we missed a revert a event
+                            return Ok(Some(BatchRevertGap {
+                                missing_index: batch.index,
+                                l1_block_number_reset: existing_batch.block_number,
+                            }));
                         }
-                        // TODO: once batch reverts are implemented, we need to handle this
-                        // case.
-                        // If we have a batch at the same index in the DB this means we have
-                        // missed a batch revert event.
                     }
 
                     let event = ChainOrchestratorEvent::BatchCommitIndexed {
@@ -804,6 +819,16 @@ impl<
                 );
                 self.l1_watcher_handle.trigger_gap_recovery(l1_block_number_reset).await;
             }
+            Some(BatchRevertGap { missing_index, l1_block_number_reset }) => {
+                tracing::warn!(
+                    target: "scroll::chain_orchestrator",
+                    "Batch revert gap detected at index {}, last known batch at L1 block {}",
+                    missing_index,
+                    l1_block_number_reset
+                );
+                // TODO: getting channel closed here
+                // self.l1_watcher_handle.trigger_gap_recovery(l1_block_number_reset).await;
+            }
             Some(BatchCommitDuplicate(index)) => {
                 tracing::info!(
                     target: "scroll::chain_orchestrator",
@@ -867,6 +892,33 @@ impl<
         end_index: u64,
         l1_block_info: BlockInfo,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
+        let event = self
+            .database
+            .tx(move |tx| async move {
+                // Check if we received this revert already.
+                // If any of the batches with same index is reverted with the same L1 block then the
+                // event is duplicate
+                for existing_batch in tx.get_batch_by_index(end_index).await? {
+                    if existing_batch.reverted_block_number == Some(l1_block_info.number) {
+                        return Ok::<_, ChainOrchestratorError>(Some(BatchRevertDuplicate(
+                            existing_batch.index,
+                        )));
+                    }
+                }
+
+                Ok(None)
+            })
+            .await?;
+
+        if let Some(BatchRevertDuplicate(index)) = event {
+            tracing::info!(
+                target: "scroll::chain_orchestrator",
+                "Duplicate batch revert detected at {:?}, skipping",
+                index
+            );
+            return Ok(event);
+        }
+
         let (safe_block_info, batch_info) = self
             .database
             .tx_mut(move |tx| async move {
@@ -895,8 +947,7 @@ impl<
         l1_message: TxL1Message,
         l1_block_info: BlockInfo,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let l1_v2_message_queue_start_index =
-            self.config.l1_v2_message_queue_start_index();
+        let l1_v2_message_queue_start_index = self.config.l1_v2_message_queue_start_index();
 
         let event = self.database
             .tx_mut(move |tx| {

crates/database/db/src/db.rs

@@ -523,7 +523,7 @@ impl DatabaseReadOperations for Database {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
-    ) -> Result<Option<BatchCommitData>, DatabaseError> {
+    ) -> Result<Vec<BatchCommitData>, DatabaseError> {
         metered!(
             DatabaseOperation::GetBatchByIndex,
             self,
@@ -881,8 +881,7 @@ mod test {
 
         // Round trip the BatchCommitData through the database.
         db.insert_batch(batch_commit.clone()).await.unwrap();
-        let batch_commit_from_db =
-            db.get_batch_by_index(batch_commit.index).await.unwrap().unwrap();
+        let batch_commit_from_db = db.get_batch_by_hash(batch_commit.hash).await.unwrap().unwrap();
 
         assert_eq!(batch_commit, batch_commit_from_db);
     }
@@ -1402,7 +1401,7 @@ mod test {
 
         // Insert L2 blocks with different batch indices
         for i in 100..110 {
-            let batch_data = db.get_batch_by_index(i).await.unwrap().unwrap();
+            let batch_data = db.get_batch_by_index(i).await.unwrap().first().unwrap().clone();
             let batch_info: BatchInfo = batch_data.into();
             let block_info = BlockInfo { number: 500 + i, hash: B256::arbitrary(&mut u).unwrap() };
 
@@ -1579,9 +1578,9 @@ mod test {
         db.set_finalized_l1_block_number(21).await.unwrap();
 
         // Verify the batches and blocks were inserted correctly
-        let retrieved_batch_1 = db.get_batch_by_index(1).await.unwrap().unwrap();
-        let retrieved_batch_2 = db.get_batch_by_index(2).await.unwrap().unwrap();
-        let retrieved_batch_3 = db.get_batch_by_index(3).await.unwrap().unwrap();
+        let retrieved_batch_1 = db.get_batch_by_index(1).await.unwrap().first().unwrap().clone();
+        let retrieved_batch_2 = db.get_batch_by_index(2).await.unwrap().first().unwrap().clone();
+        let retrieved_batch_3 = db.get_batch_by_index(3).await.unwrap().first().unwrap().clone();
         let retried_block_1 = db.get_l2_block_info_by_number(1).await.unwrap().unwrap();
         let retried_block_2 = db.get_l2_block_info_by_number(2).await.unwrap().unwrap();
         let retried_block_3 = db.get_l2_block_info_by_number(3).await.unwrap().unwrap();

crates/database/db/src/operations.rs

@@ -915,7 +915,7 @@ pub trait DatabaseReadOperations {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
-    ) -> Result<Option<BatchCommitData>, DatabaseError>;
+    ) -> Result<Vec<BatchCommitData>, DatabaseError>;
 
     /// Get a [`BatchCommitData`] from the database by its batch hash.
     async fn get_batch_by_hash(
@@ -1010,16 +1010,15 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
-    ) -> Result<Option<BatchCommitData>, DatabaseError> {
+    ) -> Result<Vec<BatchCommitData>, DatabaseError> {
         Ok(models::batch_commit::Entity::find()
             .filter(
                 models::batch_commit::Column::Index
-                    .eq(TryInto::<i64>::try_into(batch_index).expect("index should fit in i64"))
-                    .and(models::batch_commit::Column::RevertedBlockNumber.is_null()),
+                    .eq(TryInto::<i64>::try_into(batch_index).expect("index should fit in i64")),
             )
-            .one(self.get_connection())
+            .all(self.get_connection())
             .await
-            .map(|x| x.map(Into::into))?)
+            .map(|x| x.into_iter().map(Into::into).collect())?)
     }
 
     async fn get_batch_by_hash(
@@ -1151,6 +1150,7 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
 
     async fn get_last_batch_commit_l1_block(&self) -> Result<Option<u64>, DatabaseError> {
         Ok(models::batch_commit::Entity::find()
+            .filter(models::batch_commit::Column::RevertedBlockNumber.is_null())
             .order_by_desc(models::batch_commit::Column::BlockNumber)
             .select_only()
             .column(models::batch_commit::Column::BlockNumber)

crates/derivation-pipeline/src/lib.rs

@@ -218,7 +218,7 @@ where
 
             // get the batch commit data.
             let batch = db
-                .get_batch_by_index(batch_info.index)
+                .get_batch_by_hash(batch_info.hash)
                 .await
                 .map_err(|err| (request.clone(), err.into()))?
                 .ok_or((