feat: finalize all batches up to received index (#288)

* feat: finalize all batches up to received index

* test: fix finalized block number

Author: greged93

crates/chain-orchestrator/src/lib.rs

@@ -721,8 +721,8 @@ impl<
         l1_block_number: Arc<AtomicU64>,
         l2_block_number: Arc<AtomicU64>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        // finalized the batch.
-        database.finalize_batch(batch_hash, block_number).await?;
+        // finalize all batches up to `batch_index`.
+        database.finalize_batches_up_to_index(batch_index, block_number).await?;
 
         let mut finalized_block = None;
         let mut finalized_batch = None;

crates/database/db/src/db.rs

@@ -108,33 +108,54 @@ mod test {
     }
 
     #[tokio::test]
-    async fn test_database_finalize_batch_commit() {
+    async fn test_database_finalize_batch_commits() {
         // Set up the test database.
         let db = setup_test_db().await;
 
         // Generate unstructured bytes.
-        let mut bytes = [0u8; 1024];
+        let mut bytes = [0u8; 2048];
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
-        // Generate a random BatchCommitData.
-        let batch_commit = BatchCommitData::arbitrary(&mut u).unwrap();
+        // Generate 10 finalized batches at L1 block 100.
+        for i in 0..10 {
+            let batch_commit = BatchCommitData {
+                index: i,
+                calldata: Arc::new(vec![].into()),
+                finalized_block_number: Some(100),
+                ..Arbitrary::arbitrary(&mut u).unwrap()
+            };
+            db.insert_batch(batch_commit.clone()).await.unwrap();
+        }
+        // Finalize all batches below batch index 10.
+        db.finalize_batches_up_to_index(10, 100).await.unwrap();
 
-        // Store the batch and finalize it.
-        let finalized_block_number = u64::arbitrary(&mut u).unwrap();
-        db.insert_batch(batch_commit.clone()).await.unwrap();
-        db.finalize_batch(batch_commit.hash, finalized_block_number).await.unwrap();
+        // Generate 10 commit batches not finalized.
+        for i in 10..20 {
+            let batch_commit = BatchCommitData {
+                index: i,
+                calldata: Arc::new(vec![].into()),
+                finalized_block_number: None,
+                ..Arbitrary::arbitrary(&mut u).unwrap()
+            };
+            db.insert_batch(batch_commit.clone()).await.unwrap();
+        }
+
+        // Finalize all batches below batch index 15.
+        db.finalize_batches_up_to_index(15, 200).await.unwrap();
 
         // Verify the finalized_block_number is correctly updated.
-        let finalized_block_number_from_db = models::batch_commit::Entity::find()
-            .filter(models::batch_commit::Column::Hash.eq(batch_commit.hash.to_vec()))
-            .one(db.get_connection())
-            .await
-            .unwrap()
-            .unwrap()
-            .finalized_block_number
-            .unwrap();
-        assert_eq!(finalized_block_number, finalized_block_number_from_db as u64);
+        let batches = db.get_batches().await.unwrap().collect::<Vec<Result<_, _>>>().await;
+        for batch in batches {
+            let batch = batch.unwrap();
+            if batch.index < 10 {
+                assert_eq!(batch.finalized_block_number, Some(100));
+            } else if batch.index <= 15 {
+                assert_eq!(batch.finalized_block_number, Some(200));
+            } else {
+                assert_eq!(batch.finalized_block_number, None);
+            }
+        }
     }
 
     #[tokio::test]
@@ -281,7 +302,7 @@ mod test {
             // Finalize batch up to block number 110.
             if block_number <= 110 {
                 finalized_batches_hashes.push(hash);
-                db.finalize_batch(hash, block_number).await.unwrap();
+                db.finalize_batches_up_to_index(batch_index, block_number).await.unwrap();
             }
 
             block_number += 1;

crates/database/db/src/operations.rs

@@ -9,8 +9,7 @@ use rollup_node_primitives::{
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use sea_orm::{
     sea_query::{Expr, OnConflict},
-    ActiveModelTrait, ColumnTrait, Condition, DbErr, EntityTrait, QueryFilter, QueryOrder,
-    QuerySelect, Set,
+    ColumnTrait, Condition, DbErr, EntityTrait, QueryFilter, QueryOrder, QuerySelect,
 };
 use std::fmt;
 
@@ -39,34 +38,27 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
             .map(|_| ())?)
     }
 
-    /// Finalize a [`BatchCommitData`] with the provided `batch_hash` in the database and set the
+    /// Finalizes all [`BatchCommitData`] up to the provided `batch_index` by setting their
     /// finalized block number to the provided block number.
-    ///
-    /// Errors if the [`BatchCommitData`] associated with the provided `batch_hash` is not found in
-    /// the database, this method logs and returns an error.
-    async fn finalize_batch(
+    async fn finalize_batches_up_to_index(
         &self,
-        batch_hash: B256,
+        batch_index: u64,
         block_number: u64,
     ) -> Result<(), DatabaseError> {
-        if let Some(batch) = models::batch_commit::Entity::find()
-            .filter(models::batch_commit::Column::Hash.eq(batch_hash.to_vec()))
-            .one(self.get_connection())
-            .await?
-        {
-            tracing::trace!(target: "scroll::db", batch_hash = ?batch_hash, block_number, "Finalizing batch commit in database.");
-            let mut batch: models::batch_commit::ActiveModel = batch.into();
-            batch.finalized_block_number = Set(Some(block_number as i64));
-            batch.update(self.get_connection()).await?;
-        } else {
-            tracing::error!(
-                target: "scroll::db",
-                batch_hash = ?batch_hash,
-                block_number,
-                "Batch not found in DB when trying to finalize."
-            );
-            return Err(DatabaseError::BatchNotFound(batch_hash));
-        }
+        tracing::trace!(target: "scroll::db", ?batch_index, block_number, "Finalizing batch commits in database up to index.");
+
+        models::batch_commit::Entity::update_many()
+            .filter(
+                models::batch_commit::Column::Index
+                    .lte(batch_index)
+                    .and(models::batch_commit::Column::FinalizedBlockNumber.is_null()),
+            )
+            .col_expr(
+                models::batch_commit::Column::FinalizedBlockNumber,
+                Expr::value(Some(block_number as i64)),
+            )
+            .exec(self.get_connection())
+            .await?;
 
         Ok(())
     }

crates/watcher/src/lib.rs

@@ -263,7 +263,7 @@ where
             }
 
             // send all notifications on the channel.
-            self.notify_all(notifications).await;
+            self.notify_all(notifications).await?;
 
             // update the latest block the l1 watcher has indexed.
             self.update_current_block(&latest);
@@ -587,12 +587,13 @@ where
     }
 
     /// Send all notifications on the channel.
-    async fn notify_all(&self, notifications: Vec<L1Notification>) {
+    async fn notify_all(&self, notifications: Vec<L1Notification>) -> L1WatcherResult<()> {
         for notification in notifications {
             self.metrics.process_l1_notification(&notification);
             tracing::trace!(target: "scroll::watcher", %notification, "sending l1 notification");
-            let _ = self.notify(notification).await;
+            self.notify(notification).await?;
         }
+        Ok(())
     }
 
     /// Send the notification in the channel.
@@ -654,6 +655,8 @@ where
         // block.
         filter = filter.from_block(self.current_block_number + 1).to_block(to_block);
 
+        tracing::trace!(target: "scroll::watcher", ?filter, "fetching logs");
+
         Ok(self.execution_provider.get_logs(&filter).await?)
     }
 }