feat: defensive indexing (#301)

* throw errors if L1 messages and batch commits are not sequential

* add comment

* fix tests

* fix tests

Author: frisitano

crates/chain-orchestrator/src/error.rs

@@ -29,6 +29,10 @@ pub enum ChainOrchestratorError {
     /// An L1 message was not found in the database.
     #[error("L1 message not found at {0}")]
     L1MessageNotFound(L1MessageStart),
+    /// A gap was detected in the L1 message queue: the previous message before index {0} is
+    /// missing.
+    #[error("L1 message queue gap detected at index {0}, previous L1 message not found")]
+    L1MessageQueueGap(u64),
     /// An inconsistency was detected when trying to consolidate the chain.
     #[error("Chain inconsistency detected")]
     ChainInconsistency,
@@ -38,6 +42,9 @@ pub enum ChainOrchestratorError {
         /// The hash of the block header that was requested.
         hash: B256,
     },
+    /// A gap was detected in batch commit events: the previous batch before index {0} is missing.
+    #[error("Batch commit gap detected at index {0}, previous batch commit not found")]
+    BatchCommitGap(u64),
     /// An error occurred while making a network request.
     #[error("Network request error: {0}")]
     NetworkRequestError(#[from] reth_network_p2p::error::RequestError),

crates/chain-orchestrator/src/lib.rs

@@ -636,6 +636,19 @@ impl<
             compute_l1_message_queue_hash(&database, &l1_message, l1_v2_message_queue_start_index)
                 .await?;
         let l1_message = L1MessageEnvelope::new(l1_message, l1_block_number, None, queue_hash);
+
+        // Perform a consistency check to ensure the previous L1 message exists in the database.
+        if l1_message.transaction.queue_index > 0 &&
+            database
+                .get_l1_message_by_index(l1_message.transaction.queue_index - 1)
+                .await?
+                .is_none()
+        {
+            return Err(ChainOrchestratorError::L1MessageQueueGap(
+                l1_message.transaction.queue_index,
+            ))
+        }
+
         database.insert_l1_message(l1_message).await?;
         Ok(Some(event))
     }
@@ -648,6 +661,11 @@ impl<
         let txn = database.tx().await?;
         let prev_batch_index = batch.index - 1;
 
+        // Perform a consistency check to ensure the previous commit batch exists in the database.
+        if txn.get_batch_by_index(prev_batch_index).await?.is_none() {
+            return Err(ChainOrchestratorError::BatchCommitGap(batch.index))
+        }
+
         // remove any batches with an index greater than the previous batch.
         let affected = txn.delete_batches_gt_batch_index(prev_batch_index).await?;
 
@@ -1152,24 +1170,28 @@ mod test {
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
-        let batch_commit = BatchCommitData::arbitrary(&mut u).unwrap();
-        chain_orchestrator
-            .handle_l1_notification(L1Notification::BatchCommit(batch_commit.clone()));
+        // Insert a batch commit in the database to satisfy the chain orchestrator consistency
+        // checks
+        let batch_0 = BatchCommitData { index: 0, ..Arbitrary::arbitrary(&mut u).unwrap() };
+        db.insert_batch(batch_0).await.unwrap();
+
+        let batch_1 = BatchCommitData { index: 1, ..Arbitrary::arbitrary(&mut u).unwrap() };
+        chain_orchestrator.handle_l1_notification(L1Notification::BatchCommit(batch_1.clone()));
 
         let event = chain_orchestrator.next().await.unwrap().unwrap();
 
         // Verify the event structure
         match event {
             ChainOrchestratorEvent::BatchCommitIndexed { batch_info, safe_head, .. } => {
-                assert_eq!(batch_info.index, batch_commit.index);
-                assert_eq!(batch_info.hash, batch_commit.hash);
+                assert_eq!(batch_info.index, batch_1.index);
+                assert_eq!(batch_info.hash, batch_1.hash);
                 assert_eq!(safe_head, None); // No safe head since no batch revert
             }
             _ => panic!("Expected BatchCommitIndexed event"),
         }
 
-        let batch_commit_result = db.get_batch_by_index(batch_commit.index).await.unwrap().unwrap();
-        assert_eq!(batch_commit, batch_commit_result);
+        let batch_commit_result = db.get_batch_by_index(batch_1.index).await.unwrap().unwrap();
+        assert_eq!(batch_1, batch_commit_result);
     }
 
     #[tokio::test]
@@ -1182,6 +1204,15 @@ mod test {
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
+        // Insert batch 0 into the database to satisfy the consistency conditions in the chain
+        // orchestrator
+        let batch_0 = BatchCommitData {
+            index: 99,
+            calldata: Arc::new(vec![].into()),
+            ..Arbitrary::arbitrary(&mut u).unwrap()
+        };
+        db.insert_batch(batch_0).await.unwrap();
+
         // Create sequential batches
         let batch_1 = BatchCommitData {
             index: 100,
@@ -1305,22 +1336,33 @@ mod test {
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
-        let message = TxL1Message {
+        // Insert an initial message in the database to satisfy the consistency checks in the chain
+        // orchestrator.
+        let message_0 = L1MessageEnvelope {
+            transaction: TxL1Message {
+                queue_index: TEST_L1_MESSAGE_QUEUE_INDEX_BOUNDARY - 2,
+                ..Arbitrary::arbitrary(&mut u).unwrap()
+            },
+            ..Arbitrary::arbitrary(&mut u).unwrap()
+        };
+        db.insert_l1_message(message_0).await.unwrap();
+
+        let message_1 = TxL1Message {
             queue_index: TEST_L1_MESSAGE_QUEUE_INDEX_BOUNDARY - 1,
             ..Arbitrary::arbitrary(&mut u).unwrap()
         };
         let block_number = u64::arbitrary(&mut u).unwrap();
         chain_orchestrator.handle_l1_notification(L1Notification::L1Message {
-            message: message.clone(),
+            message: message_1.clone(),
             block_number,
             block_timestamp: 0,
         });
 
         let _ = chain_orchestrator.next().await;
 
         let l1_message_result =
-            db.get_l1_message_by_index(message.queue_index).await.unwrap().unwrap();
-        let l1_message = L1MessageEnvelope::new(message, block_number, None, None);
+            db.get_l1_message_by_index(message_1.queue_index).await.unwrap().unwrap();
+        let l1_message = L1MessageEnvelope::new(message_1, block_number, None, None);
 
         assert_eq!(l1_message, l1_message_result);
     }
@@ -1330,6 +1372,19 @@ mod test {
         // Instantiate chain orchestrator and db
         let (mut chain_orchestrator, db) = setup_test_chain_orchestrator().await;
 
+        // Insert the previous l1 message in the database to satisfy the chain orchestrator
+        // consistency checks.
+        let message = L1MessageEnvelope {
+            transaction: TxL1Message {
+                queue_index: TEST_L1_MESSAGE_QUEUE_INDEX_BOUNDARY - 1,
+                ..Default::default()
+            },
+            l1_block_number: 1475587,
+            l2_block_number: None,
+            queue_hash: None,
+        };
+        db.insert_l1_message(message).await.unwrap();
+
         // insert the previous L1 message in database.
         chain_orchestrator.handle_l1_notification(L1Notification::L1Message {
             message: TxL1Message {
@@ -1377,29 +1432,45 @@ mod test {
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
+        // Insert batch 0 into the database to satisfy the consistency checks in the chain
+        // orchestrator
+        let batch_0 =
+            BatchCommitData { index: 0, block_number: 0, ..Arbitrary::arbitrary(&mut u).unwrap() };
+        db.insert_batch(batch_0).await.unwrap();
+
+        // Insert l1 message into the database to satisfy the consistency checks in the chain
+        // orchestrator
+        let l1_message = L1MessageEnvelope {
+            queue_hash: None,
+            l1_block_number: 0,
+            l2_block_number: None,
+            transaction: TxL1Message { queue_index: 0, ..Arbitrary::arbitrary(&mut u).unwrap() },
+        };
+        db.insert_l1_message(l1_message).await.unwrap();
+
         // Generate a 3 random batch inputs and set their block numbers
         let mut batch_commit_block_1 = BatchCommitData::arbitrary(&mut u).unwrap();
         batch_commit_block_1.block_number = 1;
         batch_commit_block_1.index = 1;
         let batch_commit_block_1 = batch_commit_block_1;
 
-        let mut batch_commit_block_20 = BatchCommitData::arbitrary(&mut u).unwrap();
-        batch_commit_block_20.block_number = 20;
-        batch_commit_block_20.index = 20;
-        let batch_commit_block_20 = batch_commit_block_20;
+        let mut batch_commit_block_2 = BatchCommitData::arbitrary(&mut u).unwrap();
+        batch_commit_block_2.block_number = 2;
+        batch_commit_block_2.index = 2;
+        let batch_commit_block_2 = batch_commit_block_2;
 
-        let mut batch_commit_block_30 = BatchCommitData::arbitrary(&mut u).unwrap();
-        batch_commit_block_30.block_number = 30;
-        batch_commit_block_30.index = 30;
-        let batch_commit_block_30 = batch_commit_block_30;
+        let mut batch_commit_block_3 = BatchCommitData::arbitrary(&mut u).unwrap();
+        batch_commit_block_3.block_number = 3;
+        batch_commit_block_3.index = 3;
+        let batch_commit_block_3 = batch_commit_block_3;
 
         // Index batch inputs
         chain_orchestrator
             .handle_l1_notification(L1Notification::BatchCommit(batch_commit_block_1.clone()));
         chain_orchestrator
-            .handle_l1_notification(L1Notification::BatchCommit(batch_commit_block_20.clone()));
+            .handle_l1_notification(L1Notification::BatchCommit(batch_commit_block_2.clone()));
         chain_orchestrator
-            .handle_l1_notification(L1Notification::BatchCommit(batch_commit_block_30.clone()));
+            .handle_l1_notification(L1Notification::BatchCommit(batch_commit_block_3.clone()));
 
         // Generate 3 random L1 messages and set their block numbers
         let l1_message_block_1 = L1MessageEnvelope {
@@ -1408,15 +1479,15 @@ mod test {
             l2_block_number: None,
             transaction: TxL1Message { queue_index: 1, ..Arbitrary::arbitrary(&mut u).unwrap() },
         };
-        let l1_message_block_20 = L1MessageEnvelope {
+        let l1_message_block_2 = L1MessageEnvelope {
             queue_hash: None,
-            l1_block_number: 20,
+            l1_block_number: 2,
             l2_block_number: None,
             transaction: TxL1Message { queue_index: 2, ..Arbitrary::arbitrary(&mut u).unwrap() },
         };
-        let l1_message_block_30 = L1MessageEnvelope {
+        let l1_message_block_3 = L1MessageEnvelope {
             queue_hash: None,
-            l1_block_number: 30,
+            l1_block_number: 3,
             l2_block_number: None,
             transaction: TxL1Message { queue_index: 3, ..Arbitrary::arbitrary(&mut u).unwrap() },
         };
@@ -1428,18 +1499,18 @@ mod test {
             block_timestamp: 0,
         });
         chain_orchestrator.handle_l1_notification(L1Notification::L1Message {
-            message: l1_message_block_20.clone().transaction,
-            block_number: l1_message_block_20.clone().l1_block_number,
+            message: l1_message_block_2.clone().transaction,
+            block_number: l1_message_block_2.clone().l1_block_number,
             block_timestamp: 0,
         });
         chain_orchestrator.handle_l1_notification(L1Notification::L1Message {
-            message: l1_message_block_30.clone().transaction,
-            block_number: l1_message_block_30.clone().l1_block_number,
+            message: l1_message_block_3.clone().transaction,
+            block_number: l1_message_block_3.clone().l1_block_number,
             block_timestamp: 0,
         });
 
-        // Reorg at block 20
-        chain_orchestrator.handle_l1_notification(L1Notification::Reorg(20));
+        // Reorg at block 2
+        chain_orchestrator.handle_l1_notification(L1Notification::Reorg(2));
 
         for _ in 0..7 {
             chain_orchestrator.next().await.unwrap().unwrap();
@@ -1451,7 +1522,7 @@ mod test {
 
         assert_eq!(3, batch_commits.len());
         assert!(batch_commits.contains(&batch_commit_block_1));
-        assert!(batch_commits.contains(&batch_commit_block_20));
+        assert!(batch_commits.contains(&batch_commit_block_2));
 
         // check that the L1 message at block 30 is deleted
         let l1_messages = db
@@ -1461,9 +1532,9 @@ mod test {
             .map(|res| res.unwrap())
             .collect::<Vec<_>>()
             .await;
-        assert_eq!(2, l1_messages.len());
+        assert_eq!(3, l1_messages.len());
         assert!(l1_messages.contains(&l1_message_block_1));
-        assert!(l1_messages.contains(&l1_message_block_20));
+        assert!(l1_messages.contains(&l1_message_block_2));
     }
 
     // We ignore this test for now as it requires a more complex setup which leverages an L2 node

crates/manager/src/manager/mod.rs

@@ -349,9 +349,11 @@ where
     fn handle_chain_orchestrator_error(&self, err: &ChainOrchestratorError) {
         error!(
             target: "scroll::node::manager",
-            ?err,
+            error = ?err,
+            msg = %err,
             "Error occurred in the chain orchestrator"
         );
+
         match err {
             ChainOrchestratorError::L1MessageMismatch { expected, actual } => {
                 if let Some(event_sender) = self.event_sender.as_ref() {