refactor

Author: frisitano

crates/chain-orchestrator/src/event.rs

@@ -61,4 +61,6 @@ pub enum ChainOrchestratorEvent {
     /// An L2 block has been committed returning the [`L2BlockInfoWithL1Messages`] and an
     /// optional [`BatchInfo`] if the block is associated with a committed batch.
     L2ChainCommitted(L2BlockInfoWithL1Messages, Option<BatchInfo>, bool),
+    /// An L2 consolidated block has been committed returning the [`L2BlockInfoWithL1Messages`].
+    L2ConsolidatedBlockCommitted(L2BlockInfoWithL1Messages),
 }

crates/chain-orchestrator/src/lib.rs

@@ -202,7 +202,7 @@ impl<
         let database = ctx.database.clone();
         let block_info: L2BlockInfoWithL1Messages = (&block_with_peer.block).into();
         Self::do_handle_block_from_peer(ctx, block_with_peer).await?;
-        database.insert_block(block_info.clone(), None).await?;
+        database.update_l1_messages_with_l2_block(block_info.clone()).await?;
         Ok(ChainOrchestratorEvent::L2ChainCommitted(block_info, None, true))
     }
 
@@ -412,12 +412,32 @@ impl<
         }
     }
 
-    /// Inserts an L2 block in the database.
-    pub fn consolidate_l2_blocks(
+    /// Persist L1 consolidate blocks in the database.
+    pub fn persist_l1_consolidated_blocks(
         &mut self,
-        block_info: Vec<L2BlockInfoWithL1Messages>,
-        batch_info: Option<BatchInfo>,
+        block_infos: Vec<L2BlockInfoWithL1Messages>,
+        batch_info: BatchInfo,
     ) {
+        let database = self.database.clone();
+        let fut = self.handle_metered(
+            ChainOrchestratorItem::InsertConsolidatedL2Blocks,
+            Box::pin(async move {
+                let head = block_infos.last().expect("block info must not be empty").clone();
+                for block in block_infos {
+                    database.insert_block(block, batch_info).await?;
+                }
+                Result::<_, ChainOrchestratorError>::Ok(Some(
+                    ChainOrchestratorEvent::L2ConsolidatedBlockCommitted(head),
+                ))
+            }),
+        );
+
+        self.pending_futures.push_back(ChainOrchestratorFuture::HandleDerivedBlock(fut));
+        self.waker.wake();
+    }
+
+    /// Consolidates L2 blocks from the network which have been validated
+    pub fn consolidate_validated_l2_blocks(&mut self, block_info: Vec<L2BlockInfoWithL1Messages>) {
         let database = self.database.clone();
         let l1_synced = self.l1_synced;
         let optimistic_mode = self.optimistic_mode.clone();
@@ -449,10 +469,10 @@ impl<
 
                 // Insert the blocks into the database.
                 let head = block_info.last().expect("block info must not be empty").clone();
-                database.insert_blocks(block_info, batch_info).await?;
+                database.update_l1_messages_from_l2_blocks(block_info).await?;
 
                 Result::<_, ChainOrchestratorError>::Ok(Some(
-                    ChainOrchestratorEvent::L2ChainCommitted(head, batch_info, consolidated),
+                    ChainOrchestratorEvent::L2ChainCommitted(head, None, consolidated),
                 ))
             }),
         );
@@ -962,7 +982,7 @@ mod test {
     /// A headers+bodies client that stores the headers and bodies in memory, with an artificial
     /// soft bodies response limit that is set to 20 by default.
     ///
-    /// This full block client can be [Cloned] and shared between multiple tasks.
+    /// This full block client can be [Clone]d and shared between multiple tasks.
     #[derive(Clone, Debug)]
     struct TestScrollFullBlockClient {
         headers: Arc<Mutex<HashMap<B256, Header>>>,
@@ -1230,13 +1250,13 @@ mod test {
             l1_messages: vec![],
         };
 
-        chain_orchestrator.consolidate_l2_blocks(vec![block_1.clone()], Some(batch_1_info));
+        chain_orchestrator.persist_l1_consolidated_blocks(vec![block_1.clone()], batch_1_info);
         chain_orchestrator.next().await.unwrap().unwrap();
 
-        chain_orchestrator.consolidate_l2_blocks(vec![block_2.clone()], Some(batch_2_info));
+        chain_orchestrator.persist_l1_consolidated_blocks(vec![block_2.clone()], batch_2_info);
         chain_orchestrator.next().await.unwrap().unwrap();
 
-        chain_orchestrator.consolidate_l2_blocks(vec![block_3.clone()], Some(batch_2_info));
+        chain_orchestrator.persist_l1_consolidated_blocks(vec![block_3.clone()], batch_2_info);
         chain_orchestrator.next().await.unwrap().unwrap();
 
         // Now simulate a batch revert by submitting a new batch with index 101
@@ -1513,7 +1533,12 @@ mod test {
             } else {
                 None
             };
-            indexer.consolidate_l2_blocks(vec![l2_block.clone()], batch_info);
+            if let Some(batch_info) = batch_info {
+                indexer.persist_l1_consolidated_blocks(vec![l2_block.clone()], batch_info);
+            } else {
+                indexer.consolidate_validated_l2_blocks(vec![l2_block.clone()]);
+            }
+
             indexer.next().await.unwrap().unwrap();
             blocks.push(l2_block);
         }

crates/chain-orchestrator/src/metrics.rs

@@ -7,6 +7,8 @@ use strum::EnumIter;
 pub enum ChainOrchestratorItem {
     /// Handle a block received from the network.
     NewBlock,
+    /// Insert consolidated L2 blocks into the database.
+    InsertConsolidatedL2Blocks,
     /// L2 block.
     InsertL2Block,
     /// L1 reorg.
@@ -26,6 +28,7 @@ impl ChainOrchestratorItem {
     pub const fn as_str(&self) -> &'static str {
         match self {
             Self::NewBlock => "new_block",
+            Self::InsertConsolidatedL2Blocks => "insert_consolidated_l2_blocks",
             Self::InsertL2Block => "l2_block",
             Self::L1Reorg => "l1_reorg",
             Self::L1Finalization => "l1_finalization",

crates/database/db/src/db.rs

@@ -168,7 +168,7 @@ mod test {
                 },
                 l1_messages: vec![],
             };
-            db.insert_block(block_info, batch_info.into()).await.unwrap();
+            db.insert_block(block_info, batch_info).await.unwrap();
             block_number += 1;
         }
 
@@ -212,7 +212,7 @@ mod test {
                 },
                 l1_messages: vec![],
             };
-            db.insert_block(block_info, first_batch_info.into()).await.unwrap();
+            db.insert_block(block_info, first_batch_info).await.unwrap();
             block_number += 1;
         }
 
@@ -402,7 +402,7 @@ mod test {
             let block_info = BlockInfo { number: i, hash: B256::arbitrary(&mut u).unwrap() };
             let l2_block = L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] };
             block_infos.push(block_info);
-            db.insert_block(l2_block, Some(batch_info)).await.unwrap();
+            db.insert_block(l2_block, batch_info).await.unwrap();
         }
 
         // Test getting existing blocks
@@ -443,23 +443,14 @@ mod test {
 
         db.insert_block(
             L2BlockInfoWithL1Messages { block_info: safe_block_1, l1_messages: vec![] },
-            Some(batch_info),
+            batch_info,
         )
         .await
         .unwrap();
 
         db.insert_block(
             L2BlockInfoWithL1Messages { block_info: safe_block_2, l1_messages: vec![] },
-            Some(batch_info),
-        )
-        .await
-        .unwrap();
-
-        // Insert block without batch info (unsafe block)
-        let unsafe_block = BlockInfo { number: 202, hash: B256::arbitrary(&mut u).unwrap() };
-        db.insert_block(
-            L2BlockInfoWithL1Messages { block_info: unsafe_block, l1_messages: vec![] },
-            None,
+            batch_info,
         )
         .await
         .unwrap();
@@ -486,7 +477,7 @@ mod test {
 
             db.insert_block(
                 L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] },
-                Some(batch_info),
+                batch_info,
             )
             .await
             .unwrap();
@@ -537,15 +528,7 @@ mod test {
             let block_info = BlockInfo { number: 500 + i, hash: B256::arbitrary(&mut u).unwrap() };
             let l2_block = L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] };
 
-            db.insert_block(l2_block, Some(batch_info)).await.unwrap();
-        }
-
-        // Insert some blocks without batch index (should not be deleted)
-        for i in 0..3 {
-            let block_info = BlockInfo { number: 600 + i, hash: B256::arbitrary(&mut u).unwrap() };
-            let l2_block = L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] };
-
-            db.insert_block(l2_block, None).await.unwrap();
+            db.insert_block(l2_block, batch_info).await.unwrap();
         }
 
         // Delete L2 blocks with batch index > 105
@@ -601,7 +584,8 @@ mod test {
             L2BlockInfoWithL1Messages { block_info, l1_messages: l1_message_hashes.clone() };
 
         // Insert block
-        db.insert_block(l2_block, Some(batch_info)).await.unwrap();
+        db.insert_block(l2_block.clone(), batch_info).await.unwrap();
+        db.update_l1_messages_with_l2_block(l2_block).await.unwrap();
 
         // Verify block was inserted
         let retrieved_block = db.get_l2_block_info_by_number(500).await.unwrap();
@@ -637,7 +621,7 @@ mod test {
         let block_info = BlockInfo { number: 600, hash: B256::arbitrary(&mut u).unwrap() };
         let l2_block = L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] };
 
-        db.insert_block(l2_block, Some(batch_info_1)).await.unwrap();
+        db.insert_block(l2_block, batch_info_1).await.unwrap();
 
         // Verify initial insertion
         let retrieved_block = db.get_l2_block_info_by_number(600).await.unwrap();
@@ -650,15 +634,15 @@ mod test {
             .await
             .unwrap()
             .unwrap();
-        let (initial_block_info, initial_batch_info): (BlockInfo, Option<BatchInfo>) =
+        let (initial_block_info, initial_batch_info): (BlockInfo, BatchInfo) =
             initial_l2_block_model.into();
         assert_eq!(initial_block_info, block_info);
-        assert_eq!(initial_batch_info, Some(batch_info_1));
+        assert_eq!(initial_batch_info, batch_info_1);
 
         // Update the same block with different batch info (upsert)
         let updated_l2_block = L2BlockInfoWithL1Messages { block_info, l1_messages: vec![] };
 
-        db.insert_block(updated_l2_block, Some(batch_info_2)).await.unwrap();
+        db.insert_block(updated_l2_block, batch_info_2).await.unwrap();
 
         // Verify the block still exists and was updated
         let retrieved_block = db.get_l2_block_info_by_number(600).await.unwrap().unwrap();
@@ -671,9 +655,9 @@ mod test {
             .await
             .unwrap()
             .unwrap();
-        let (updated_block_info, updated_batch_info): (BlockInfo, Option<BatchInfo>) =
+        let (updated_block_info, updated_batch_info): (BlockInfo, BatchInfo) =
             updated_l2_block_model.into();
         assert_eq!(updated_block_info, block_info);
-        assert_eq!(updated_batch_info, Some(batch_info_2));
+        assert_eq!(updated_batch_info, batch_info_2);
     }
 }

crates/database/db/src/models/l2_block.rs

@@ -9,21 +9,17 @@ pub struct Model {
     #[sea_orm(primary_key)]
     block_number: i64,
     block_hash: Vec<u8>,
-    batch_index: Option<i64>,
-    batch_hash: Option<Vec<u8>>,
+    batch_index: i64,
+    batch_hash: Vec<u8>,
 }
 
 impl Model {
     pub(crate) fn block_info(&self) -> BlockInfo {
         BlockInfo { number: self.block_number as u64, hash: B256::from_slice(&self.block_hash) }
     }
 
-    pub(crate) fn batch_info(&self) -> Option<BatchInfo> {
-        self.batch_hash.as_ref().map(|hash| BatchInfo {
-            index: self.batch_index.expect("batch index must be present if batch hash is present")
-                as u64,
-            hash: B256::from_slice(hash),
-        })
+    pub(crate) fn batch_info(&self) -> BatchInfo {
+        BatchInfo { index: self.batch_index as u64, hash: B256::from_slice(&self.batch_hash) }
     }
 }
 
@@ -50,32 +46,32 @@ impl Related<super::batch_commit::Entity> for Entity {
 /// The active model behavior for the batch input model.
 impl ActiveModelBehavior for ActiveModel {}
 
-impl From<(BlockInfo, Option<BatchInfo>)> for ActiveModel {
-    fn from((block_info, batch_info): (BlockInfo, Option<BatchInfo>)) -> Self {
+impl From<(BlockInfo, BatchInfo)> for ActiveModel {
+    fn from((block_info, batch_info): (BlockInfo, BatchInfo)) -> Self {
         Self {
             block_number: ActiveValue::Set(
                 block_info.number.try_into().expect("block number should fit in i64"),
             ),
             block_hash: ActiveValue::Set(block_info.hash.to_vec()),
             batch_index: ActiveValue::Set(
-                batch_info.map(|x| x.index.try_into().expect("index should fit in i64")),
+                batch_info.index.try_into().expect("index should fit in i64"),
             ),
-            batch_hash: ActiveValue::Set(batch_info.map(|x| x.hash.to_vec())),
+            batch_hash: ActiveValue::Set(batch_info.hash.to_vec()),
         }
     }
 }
 
-impl From<Model> for (BlockInfo, Option<BatchInfo>) {
+impl From<Model> for (BlockInfo, BatchInfo) {
     fn from(value: Model) -> Self {
         (
             BlockInfo {
                 number: value.block_number as u64,
                 hash: B256::from_slice(&value.block_hash),
             },
-            value.batch_hash.map(|b| BatchInfo {
-                index: value.batch_index.unwrap() as u64,
-                hash: B256::from_slice(&b),
-            }),
+            BatchInfo {
+                index: value.batch_index as u64,
+                hash: B256::from_slice(&value.batch_hash),
+            },
         )
     }
 }