feat: update batch finalization logic

crates/chain-orchestrator/src/event.rs

@@ -2,9 +2,7 @@ use alloy_consensus::Header;
 use alloy_primitives::{Signature, B256};
 use reth_network_peers::PeerId;
 use reth_scroll_primitives::ScrollBlock;
-use rollup_node_primitives::{
-    BatchInfo, BlockInfo, ChainImport, L2BlockInfoWithL1Messages, WithFinalizedBlockNumber,
-};
+use rollup_node_primitives::{BatchInfo, BlockInfo, ChainImport, L2BlockInfoWithL1Messages};
 
 /// An event emitted by the `ChainOrchestrator`.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -40,9 +38,8 @@ pub enum ChainOrchestratorEvent {
         /// The safe L2 block info.
         safe_head: Option<BlockInfo>,
     },
-    /// A batch has been finalized returning an optional finalized L2 block. Also returns a
-    /// [`BatchInfo`] if the finalized event occurred in a finalized L1 block.
-    BatchFinalized(Option<WithFinalizedBlockNumber<BatchInfo>>, Option<BlockInfo>),
+    /// A batch has been finalized returning a list of finalized batches.
+    BatchFinalized(u64, Vec<BatchInfo>),
     /// An L1 block has been finalized returning the L1 block number and the list of finalized
     /// batches.
     L1BlockFinalized(u64, Vec<BatchInfo>),

crates/chain-orchestrator/src/lib.rs

@@ -556,6 +556,7 @@ impl<
                         self.database.clone(),
                         index,
                         block_number,
+                        self.l1_finalized_block_number.clone(),
                     )),
                 ))
             }
@@ -695,10 +696,20 @@ impl<
         database: Arc<Database>,
         batch_index: u64,
         block_number: u64,
+        finalized_block_number: Arc<AtomicU64>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
         // finalize all batches up to `batch_index`.
         database.finalize_batches_up_to_index(batch_index, block_number).await?;
 
+        // Get all unprocessed batches that have been finalized by this L1 block finalization.
+        let finalized_block_number = finalized_block_number.load(Ordering::Relaxed);
+        if finalized_block_number >= block_number {
+            let finalized_batches = database
+                .fetch_and_update_unprocessed_finalized_batches(finalized_block_number)
+                .await?;
+            return Ok(Some(ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches)))
+        }
+
         Ok(None)
     }
 }

crates/manager/src/manager/mod.rs

@@ -274,17 +274,16 @@ where
                 // // push the batch info into the derivation pipeline.
                 // self.derivation_pipeline.push_batch(batch_info, l1_block_number);
             }
-            ChainOrchestratorEvent::BatchFinalized(batch_info, ..) => {
+            ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches) => {
                 // Uncomment once we implement issue #273.
                 // // update the fcs on new finalized block.
                 // if let Some(finalized_block) = finalized_block {
                 //     self.engine.set_finalized_block_info(finalized_block);
                 // }
                 // Remove once we implement issue #273.
                 // Update the derivation pipeline on new finalized batch.
-                #[allow(clippy::collapsible_match)]
-                if let Some(batch_info) = batch_info {
-                    self.derivation_pipeline.push_batch(batch_info.inner, batch_info.number);
+                for batch_info in finalized_batches {
+                    self.derivation_pipeline.push_batch(batch_info, block_number);
                 }
             }
             ChainOrchestratorEvent::L1BlockFinalized(l1_block_number, finalized_batches, ..) => {