feat: implement self-healing recovery mechanism for gaps in L1 data

crates/chain-orchestrator/Cargo.toml

@@ -69,10 +69,12 @@ alloy-transport.workspace = true
 # rollup-node
 scroll-db = { workspace = true, features = ["test-utils"] }
 rollup-node-primitives = { workspace = true, features = ["arbitrary"] }
+rollup-node-watcher = { workspace = true, features = ["test-utils"] }
 
 # scroll
 reth-scroll-chainspec.workspace = true
 reth-scroll-forks.workspace = true
+reth-scroll-node = { workspace = true, features = ["test-utils"] }
 
 # reth
 reth-eth-wire-types.workspace = true

crates/chain-orchestrator/src/error.rs

@@ -40,6 +40,9 @@ pub enum ChainOrchestratorError {
     /// missing.
     #[error("L1 message queue gap detected at index {0}, previous L1 message not found")]
     L1MessageQueueGap(u64),
+    /// A duplicate L1 message was detected at index {0}.
+    #[error("Duplicate L1 message detected at index {0}")]
+    DuplicateL1Message(u64),
     /// An inconsistency was detected when trying to consolidate the chain.
     #[error("Chain inconsistency detected")]
     ChainInconsistency,
@@ -60,6 +63,9 @@ pub enum ChainOrchestratorError {
     /// 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),
+    /// A duplicate batch commit was detected at index {0}.
+    #[error("Duplicate batch commit detected at {0}")]
+    DuplicateBatchCommit(BatchInfo),
     /// An error occurred while making a network request.
     #[error("Network request error: {0}")]
     NetworkRequestError(#[from] reth_network_p2p::error::RequestError),
@@ -92,6 +98,9 @@ pub enum ChainOrchestratorError {
     /// An error occurred while handling rollup node primitives.
     #[error("An error occurred while handling rollup node primitives: {0}")]
     RollupNodePrimitiveError(rollup_node_primitives::RollupNodePrimitiveError),
+    /// An error occurred during gap reset.
+    #[error("Gap reset error: {0}")]
+    GapResetError(String),
 }
 
 impl CanRetry for ChainOrchestratorError {

crates/database/db/src/db.rs

@@ -413,11 +413,12 @@ impl DatabaseReadOperations for Database {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
+        processed: Option<bool>,
     ) -> Result<Option<BatchCommitData>, DatabaseError> {
         metered!(
             DatabaseOperation::GetBatchByIndex,
             self,
-            tx(move |tx| async move { tx.get_batch_by_index(batch_index).await })
+            tx(move |tx| async move { tx.get_batch_by_index(batch_index, processed).await })
         )
     }
 
@@ -453,6 +454,22 @@ impl DatabaseReadOperations for Database {
         )
     }
 
+    async fn get_last_batch_commit_l1_block(&self) -> Result<Option<u64>, DatabaseError> {
+        metered!(
+            DatabaseOperation::GetLastBatchCommitL1Block,
+            self,
+            tx(|tx| async move { tx.get_last_batch_commit_l1_block().await })
+        )
+    }
+
+    async fn get_last_l1_message_l1_block(&self) -> Result<Option<u64>, DatabaseError> {
+        metered!(
+            DatabaseOperation::GetLastL1MessageL1Block,
+            self,
+            tx(|tx| async move { tx.get_last_l1_message_l1_block().await })
+        )
+    }
+
     async fn get_n_l1_messages(
         &self,
         start: Option<L1MessageKey>,
@@ -719,7 +736,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();
+            db.get_batch_by_index(batch_commit.index, None).await.unwrap().unwrap();
 
         assert_eq!(batch_commit, batch_commit_from_db);
     }
@@ -1233,7 +1250,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, None).await.unwrap().unwrap();
             let batch_info: BatchInfo = batch_data.into();
             let block_info = BlockInfo { number: 500 + i, hash: B256::arbitrary(&mut u).unwrap() };
 
@@ -1402,9 +1419,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, None).await.unwrap().unwrap();
+        let retrieved_batch_2 = db.get_batch_by_index(2, None).await.unwrap().unwrap();
+        let retrieved_batch_3 = db.get_batch_by_index(3, None).await.unwrap().unwrap();
         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();
@@ -1425,9 +1442,9 @@ mod test {
         assert_eq!(result, (Some(block_2), Some(11)));
 
         // Verify that batches 2 and 3 are deleted
-        let batch_1 = db.get_batch_by_index(1).await.unwrap();
-        let batch_2 = db.get_batch_by_index(2).await.unwrap();
-        let batch_3 = db.get_batch_by_index(3).await.unwrap();
+        let batch_1 = db.get_batch_by_index(1, None).await.unwrap();
+        let batch_2 = db.get_batch_by_index(2, None).await.unwrap();
+        let batch_3 = db.get_batch_by_index(3, None).await.unwrap();
         assert!(batch_1.is_some());
         assert!(batch_2.is_none());
         assert!(batch_3.is_none());

crates/database/db/src/metrics.rs

@@ -47,6 +47,8 @@ pub(crate) enum DatabaseOperation {
     GetFinalizedL1BlockNumber,
     GetProcessedL1BlockNumber,
     GetL2HeadBlockNumber,
+    GetLastBatchCommitL1Block,
+    GetLastL1MessageL1Block,
     GetNL1Messages,
     GetNL2BlockDataHint,
     GetL2BlockAndBatchInfoByHash,
@@ -92,6 +94,8 @@ impl DatabaseOperation {
             Self::GetFinalizedL1BlockNumber => "get_finalized_l1_block_number",
             Self::GetProcessedL1BlockNumber => "get_processed_l1_block_number",
             Self::GetL2HeadBlockNumber => "get_l2_head_block_number",
+            Self::GetLastBatchCommitL1Block => "get_last_batch_commit_l1_block",
+            Self::GetLastL1MessageL1Block => "get_last_l1_message_l1_block",
             Self::GetNL1Messages => "get_n_l1_messages",
             Self::GetNL2BlockDataHint => "get_n_l2_block_data_hint",
             Self::GetL2BlockAndBatchInfoByHash => "get_l2_block_and_batch_info_by_hash",

crates/database/db/src/operations.rs

@@ -374,7 +374,8 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
             .map(|(_, batch_info)| batch_info)
             .filter(|b| b.index > 1)
         {
-            let batch = self.get_batch_by_index(batch_info.index).await?.expect("batch must exist");
+            let batch =
+                self.get_batch_by_index(batch_info.index, None).await?.expect("batch must exist");
             self.delete_batches_gt_block_number(batch.block_number.saturating_sub(1)).await?;
         };
 
@@ -383,7 +384,8 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
         else {
             return Ok((None, None));
         };
-        let batch = self.get_batch_by_index(batch_info.index).await?.expect("batch must exist");
+        let batch =
+            self.get_batch_by_index(batch_info.index, None).await?.expect("batch must exist");
         Ok((Some(block_info), Some(batch.block_number.saturating_add(1))))
     }
 
@@ -649,6 +651,7 @@ pub trait DatabaseReadOperations {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
+        processed: Option<bool>,
     ) -> Result<Option<BatchCommitData>, DatabaseError>;
 
     /// Get the latest L1 block number from the database.
@@ -663,6 +666,12 @@ pub trait DatabaseReadOperations {
     /// Get the latest L2 head block info.
     async fn get_l2_head_block_number(&self) -> Result<u64, DatabaseError>;
 
+    /// Get the L1 block number of the last batch commit in the database.
+    async fn get_last_batch_commit_l1_block(&self) -> Result<Option<u64>, DatabaseError>;
+
+    /// Get the L1 block number of the last L1 message in the database.
+    async fn get_last_l1_message_l1_block(&self) -> Result<Option<u64>, DatabaseError>;
+
     /// Get a vector of n [`L1MessageEnvelope`]s in the database starting from the provided `start`
     /// point.
     async fn get_n_l1_messages(
@@ -721,13 +730,21 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
     async fn get_batch_by_index(
         &self,
         batch_index: u64,
+        processed: Option<bool>,
     ) -> Result<Option<BatchCommitData>, DatabaseError> {
-        Ok(models::batch_commit::Entity::find_by_id(
-            TryInto::<i64>::try_into(batch_index).expect("index should fit in i64"),
-        )
-        .one(self.get_connection())
-        .await
-        .map(|x| x.map(Into::into))?)
+        let query = if let Some(p) = processed {
+            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::Processed.eq(p)),
+            )
+        } else {
+            models::batch_commit::Entity::find_by_id(
+                TryInto::<i64>::try_into(batch_index).expect("index should fit in i64"),
+            )
+        };
+
+        Ok(query.one(self.get_connection()).await.map(|x| x.map(Into::into))?)
     }
 
     async fn get_latest_l1_block_number(&self) -> Result<u64, DatabaseError> {
@@ -782,6 +799,28 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
             .expect("l2_head_block should always be a valid u64"))
     }
 
+    async fn get_last_batch_commit_l1_block(&self) -> Result<Option<u64>, DatabaseError> {
+        Ok(models::batch_commit::Entity::find()
+            .order_by_desc(models::batch_commit::Column::BlockNumber)
+            .select_only()
+            .column(models::batch_commit::Column::BlockNumber)
+            .into_tuple::<i64>()
+            .one(self.get_connection())
+            .await?
+            .map(|block_number| block_number as u64))
+    }
+
+    async fn get_last_l1_message_l1_block(&self) -> Result<Option<u64>, DatabaseError> {
+        Ok(models::l1_message::Entity::find()
+            .order_by_desc(models::l1_message::Column::L1BlockNumber)
+            .select_only()
+            .column(models::l1_message::Column::L1BlockNumber)
+            .into_tuple::<i64>()
+            .one(self.get_connection())
+            .await?
+            .map(|block_number| block_number as u64))
+    }
+
     async fn get_n_l1_messages(
         &self,
         start: Option<L1MessageKey>,

crates/derivation-pipeline/src/lib.rs

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

crates/node/src/args.rs

@@ -342,35 +342,34 @@ impl ScrollRollupNodeConfig {
         };
         let consensus = self.consensus_args.consensus(authorized_signer)?;
 
-        let (l1_notification_tx, l1_notification_rx): (Option<Sender<Arc<L1Notification>>>, _) =
-            if let Some(provider) = l1_provider.filter(|_| !self.test) {
-                tracing::info!(target: "scroll::node::args", ?l1_start_block_number, "Starting L1 watcher");
-                (
-                    None,
-                    Some(
-                        L1Watcher::spawn(
-                            provider,
-                            l1_start_block_number,
-                            node_config,
-                            self.l1_provider_args.logs_query_block_range,
-                        )
-                        .await,
-                    ),
-                )
-            } else {
-                // Create a channel for L1 notifications that we can use to inject L1 messages for
-                // testing
-                #[cfg(feature = "test-utils")]
-                {
-                    let (tx, rx) = tokio::sync::mpsc::channel(1000);
-                    (Some(tx), Some(rx))
-                }
-
-                #[cfg(not(feature = "test-utils"))]
-                {
-                    (None, None)
-                }
-            };
+        let (l1_notification_tx, l1_notification_rx, l1_watcher_handle): (
+            Option<Sender<Arc<L1Notification>>>,
+            _,
+            Option<rollup_node_watcher::L1WatcherHandle>,
+        ) = if let Some(provider) = l1_provider.filter(|_| !self.test) {
+            tracing::info!(target: "scroll::node::args", ?l1_start_block_number, "Starting L1 watcher");
+            let (rx, handle) = L1Watcher::spawn(
+                provider,
+                l1_start_block_number,
+                node_config,
+                self.l1_provider_args.logs_query_block_range,
+            )
+            .await;
+            (None, Some(rx), Some(handle))
+        } else {
+            // Create a channel for L1 notifications that we can use to inject L1 messages for
+            // testing
+            #[cfg(feature = "test-utils")]
+            {
+                let (tx, rx) = tokio::sync::mpsc::channel(1000);
+                (Some(tx), Some(rx), None)
+            }
+
+            #[cfg(not(feature = "test-utils"))]
+            {
+                (None, None, None)
+            }
+        };
 
         // Construct the l1 provider.
         let l1_messages_provider = db.clone();
@@ -450,6 +449,7 @@ impl ScrollRollupNodeConfig {
             Arc::new(block_client),
             l2_provider,
             l1_notification_rx.expect("L1 notification receiver should be set"),
+            l1_watcher_handle,
             scroll_network_handle.into_scroll_network().await,
             consensus,
             engine,

crates/node/tests/e2e.rs

@@ -5,7 +5,7 @@ use alloy_primitives::{address, b256, Address, Bytes, Signature, B256, U256};
 use alloy_rpc_types_eth::Block;
 use alloy_signer::Signer;
 use alloy_signer_local::PrivateKeySigner;
-use eyre::Ok;
+use eyre::{bail, Ok};
 use futures::{task::noop_waker_ref, FutureExt, StreamExt};
 use reth_chainspec::EthChainSpec;
 use reth_e2e_test_utils::{NodeHelperType, TmpDB};
@@ -48,7 +48,7 @@ use std::{
     task::{Context, Poll},
     time::Duration,
 };
-use tokio::{sync::Mutex, time};
+use tokio::{select, sync::Mutex, time};
 use tracing::trace;
 
 #[tokio::test]
@@ -1025,20 +1025,28 @@ async fn shutdown_consolidates_most_recent_batch_on_startup() -> eyre::Result<()
     // Lets finalize the second batch.
     l1_notification_tx.send(Arc::new(L1Notification::Finalized(batch_1_data.block_number))).await?;
 
+    let mut l2_block = None;
     // Lets fetch the first consolidated block event - this should be the first block of the batch.
-    let l2_block = loop {
-        if let Some(ChainOrchestratorEvent::BlockConsolidated(consolidation_outcome)) =
-            rnm_events.next().await
-        {
-            break consolidation_outcome.block_info().clone();
+    select! {
+        _ = tokio::time::sleep(Duration::from_secs(5)) => {
+            bail!("Timed out waiting for first consolidated block after RNM restart");
         }
-    };
+
+        evt = rnm_events.next() => {
+            if let Some(ChainOrchestratorEvent::BlockConsolidated(consolidation_outcome)) = evt {
+                l2_block = Some(consolidation_outcome.block_info().clone());
+            } else {
+                println!("Received unexpected event: {:?}", evt);
+            }
+        }
+    }
 
     // One issue #273 is completed, we will again have safe blocks != finalized blocks, and this
     // should be changed to 1. Assert that the consolidated block is the first block that was not
     // previously processed of the batch.
     assert_eq!(
-        l2_block.block_info.number, 41,
+        l2_block.unwrap().block_info.number,
+        41,
         "Consolidated block number does not match expected number"
     );
 

crates/watcher/src/handle/command.rs

@@ -0,0 +1,19 @@
+use crate::L1Notification;
+use std::sync::Arc;
+use tokio::sync::{mpsc, oneshot};
+
+/// Commands that can be sent to the L1 Watcher.
+#[derive(Debug)]
+pub enum L1WatcherCommand {
+    /// Reset the watcher to a specific L1 block number.
+    ///
+    /// This is used for gap recovery when the chain orchestrator detects missing L1 events.
+    ResetToBlock {
+        /// The L1 block number to reset to (last known good state)
+        block: u64,
+        /// New sender to replace the current notification channel
+        new_sender: mpsc::Sender<Arc<L1Notification>>,
+        /// Oneshot sender to signal completion of the reset operation
+        response_sender: oneshot::Sender<()>,
+    },
+}