clean up

Author: frisitano

crates/chain-orchestrator/src/lib.rs

@@ -439,7 +439,7 @@ impl<
                     let finalized_block_info = batch_reconciliation_result
                         .target_status
                         .is_finalized()
-                        .then(|| block_info.block_info);
+                        .then_some(block_info.block_info);
                     self.engine
                         .update_fcs(None, Some(block_info.block_info), finalized_block_info)
                         .await?;
@@ -478,7 +478,7 @@ impl<
                     let finalized_block_info = batch_reconciliation_result
                         .target_status
                         .is_finalized()
-                        .then(|| block_info.block_info);
+                        .then_some(block_info.block_info);
                     self.engine
                         .update_fcs(
                             Some(block_info.block_info),
@@ -750,7 +750,7 @@ impl<
             self.engine.update_fcs(None, None, finalized_block_info).await?;
         }
 
-        for batch in triggered_batches.iter() {
+        for batch in &triggered_batches {
             self.derivation_pipeline.push_batch(*batch, BatchStatus::Finalized).await;
         }
 

crates/database/db/src/db.rs

@@ -8,8 +8,8 @@ use crate::{
 };
 use alloy_primitives::{Signature, B256};
 use rollup_node_primitives::{
-    BatchCommitData, BatchConsolidationOutcome, BatchInfo, BlockInfo, L1MessageEnvelope,
-    L2BlockInfoWithL1Messages,
+    BatchCommitData, BatchConsolidationOutcome, BatchInfo, BlockInfo, L1BlockStartupInfo,
+    L1MessageEnvelope, L2BlockInfoWithL1Messages,
 };
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use sea_orm::{
@@ -375,11 +375,11 @@ impl DatabaseWriteOperations for Database {
         )
     }
 
-    async fn prepare_on_startup(&self) -> Result<(Vec<BlockInfo>, Option<u64>), DatabaseError> {
+    async fn prepare_l1_watcher_start_info(&self) -> Result<L1BlockStartupInfo, DatabaseError> {
         metered!(
             DatabaseOperation::PrepareOnStartup,
             self,
-            tx_mut(move |tx| async move { tx.prepare_on_startup().await })
+            tx_mut(move |tx| async move { tx.prepare_l1_watcher_start_info().await })
         )
     }
 
@@ -1572,10 +1572,10 @@ mod test {
         assert_eq!(retried_block_4, block_4);
 
         // Call prepare_on_startup which should not error
-        let result = db.prepare_on_startup().await.unwrap();
+        let result = db.prepare_l1_watcher_start_info().await.unwrap();
 
         // verify the result
-        assert_eq!(result, (vec![l1_block_info_3], Some(l1_block_info_3.number)));
+        assert_eq!(result, L1BlockStartupInfo::UnsafeBlocks(vec![l1_block_info_3]));
     }
 
     #[tokio::test]

crates/database/db/src/operations.rs

@@ -4,7 +4,7 @@ use crate::{ReadConnectionProvider, WriteConnectionProvider};
 use alloy_primitives::{Signature, B256};
 use rollup_node_primitives::{
     BatchCommitData, BatchConsolidationOutcome, BatchInfo, BatchStatus, BlockInfo,
-    L1MessageEnvelope, L2BlockInfoWithL1Messages, Metadata,
+    L1BlockStartupInfo, L1MessageEnvelope, L2BlockInfoWithL1Messages, Metadata,
 };
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use sea_orm::{
@@ -118,16 +118,8 @@ pub trait DatabaseWriteOperations {
         l1_block_number: u64,
     ) -> Result<Vec<L1MessageEnvelope>, DatabaseError>;
 
-    /// Prepare the database on startup and return metadata used for other components in the
-    /// rollup-node.
-    ///
-    /// This method first unwinds the database to the finalized L1 block. It then fetches the batch
-    /// info for the latest safe L2 block. It takes note of the L1 block number at which
-    /// this batch was produced (currently the finalized block for the batch until we implement
-    /// issue #273). It then retrieves the latest block for the previous batch (i.e., the batch
-    /// before the latest safe block). It returns a tuple of this latest fetched block and the
-    /// L1 block number of the batch.
-    async fn prepare_on_startup(&self) -> Result<(Vec<BlockInfo>, Option<u64>), DatabaseError>;
+    /// Returns the L1 block info required to start the L1 watcher on startup.
+    async fn prepare_l1_watcher_start_info(&self) -> Result<L1BlockStartupInfo, DatabaseError>;
 
     /// Delete all L2 blocks with a block number greater than the provided block number.
     async fn delete_l2_blocks_gt_block_number(
@@ -598,7 +590,7 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
         Ok(removed_messages.into_iter().map(Into::into).collect())
     }
 
-    async fn prepare_on_startup(&self) -> Result<(Vec<BlockInfo>, Option<u64>), DatabaseError> {
+    async fn prepare_l1_watcher_start_info(&self) -> Result<L1BlockStartupInfo, DatabaseError> {
         tracing::trace!(target: "scroll::db", "Fetching startup safe block from database.");
 
         // set all batches with processing status back to committed
@@ -608,7 +600,7 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
         let l1_block_infos = self.get_l1_block_info().await?;
         let latest_l1_block_info = self.get_latest_indexed_event_l1_block_number().await?;
 
-        Ok((l1_block_infos, latest_l1_block_info))
+        Ok(L1BlockStartupInfo::new(l1_block_infos, latest_l1_block_info))
     }
 
     async fn delete_l2_blocks_gt_block_number(
@@ -1041,13 +1033,7 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
             .into_tuple::<(Option<i64>, Option<i64>, Option<i64>)>()
             .one(self.get_connection())
             .await?
-            .map(|(block_number, finalized_block_number, reverted_block_number)| {
-                [block_number, finalized_block_number, reverted_block_number]
-                    .into_iter()
-                    .flatten()
-                    .max()
-            })
-            .flatten();
+            .and_then(|tuple| <[Option<i64>; 3]>::from(tuple).into_iter().flatten().max());
 
         let latest_l1_block_number =
             [latest_l1_message, latest_batch_event].into_iter().flatten().max();

crates/derivation-pipeline/benches/pipeline.rs

@@ -127,7 +127,7 @@ fn benchmark_pipeline_derivation_in_file_blobs(c: &mut Criterion) {
                     // commit 253 batches.
                     for index in BATCHES_START_INDEX..=BATCHES_STOP_INDEX {
                         let batch_info = BatchInfo { index, hash: Default::default() };
-                        pipeline.push_batch(batch_info.into(), BatchStatus::Committed).await;
+                        pipeline.push_batch(batch_info, BatchStatus::Committed).await;
                     }
 
                     tx.send(pipeline).unwrap();
@@ -163,7 +163,7 @@ fn benchmark_pipeline_derivation_s3_blobs(c: &mut Criterion) {
                     // commit 15 batches.
                     for index in BATCHES_START_INDEX..=BATCHES_START_INDEX + 15 {
                         let batch_info = BatchInfo { index, hash: Default::default() };
-                        pipeline.push_batch(batch_info.clone(), BatchStatus::Committed).await;
+                        pipeline.push_batch(batch_info, BatchStatus::Committed).await;
                     }
 
                     tx.send(pipeline).unwrap();

crates/derivation-pipeline/src/lib.rs

@@ -515,7 +515,7 @@ mod tests {
         // as long as we don't call `push_batch`, pipeline should not return attributes.
         pipeline
             .push_batch(
-                BatchInfo { index: 12, hash: Default::default() }.into(),
+                BatchInfo { index: 12, hash: Default::default() },
                 BatchStatus::Consolidated,
             )
             .await;

crates/node/src/args.rs

@@ -258,16 +258,16 @@ impl ScrollRollupNodeConfig {
         let mut fcs =
             ForkchoiceState::from_provider(&l2_provider).await.unwrap_or_else(chain_spec_fcs);
 
-        let (l1_start_block_number, mut l2_head_block_number) = db
+        let (l1_block_startup_info, mut l2_head_block_number) = db
             .tx_mut(move |tx| async move {
                 // On startup we replay the latest batch of blocks from the database as such we set
                 // the safe block hash to the latest block hash associated with the
                 // previous consolidated batch in the database.
-                let (_startup_safe_block, l1_start_block_number) = tx.prepare_on_startup().await?;
+                let l1_block_startup_info = tx.prepare_l1_watcher_start_info().await?;
 
                 let l2_head_block_number = tx.get_l2_head_block_number().await?;
 
-                Ok::<_, DatabaseError>((l1_start_block_number, l2_head_block_number))
+                Ok::<_, DatabaseError>((l1_block_startup_info, l2_head_block_number))
             })
             .await?;
 
@@ -342,13 +342,13 @@ impl ScrollRollupNodeConfig {
 
         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");
+                tracing::info!(target: "scroll::node::args", ?l1_block_startup_info, "Starting L1 watcher");
                 (
                     None,
                     Some(
                         L1Watcher::spawn(
                             provider,
-                            l1_start_block_number,
+                            l1_block_startup_info,
                             node_config,
                             self.l1_provider_args.logs_query_block_range,
                         )

crates/primitives/src/batch.rs

@@ -86,11 +86,11 @@ impl BatchStatus {
 impl core::fmt::Display for BatchStatus {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         match self {
-            BatchStatus::Committed => write!(f, "committed"),
-            BatchStatus::Processing => write!(f, "processing"),
-            BatchStatus::Consolidated => write!(f, "consolidated"),
-            BatchStatus::Reverted => write!(f, "reverted"),
-            BatchStatus::Finalized => write!(f, "finalized"),
+            Self::Committed => write!(f, "committed"),
+            Self::Processing => write!(f, "processing"),
+            Self::Consolidated => write!(f, "consolidated"),
+            Self::Reverted => write!(f, "reverted"),
+            Self::Finalized => write!(f, "finalized"),
         }
     }
 }
@@ -100,11 +100,11 @@ impl core::str::FromStr for BatchStatus {
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
-            "committed" => Ok(BatchStatus::Committed),
-            "processing" => Ok(BatchStatus::Processing),
-            "consolidated" => Ok(BatchStatus::Consolidated),
-            "reverted" => Ok(BatchStatus::Reverted),
-            "finalized" => Ok(BatchStatus::Finalized),
+            "committed" => Ok(Self::Committed),
+            "processing" => Ok(Self::Processing),
+            "consolidated" => Ok(Self::Consolidated),
+            "reverted" => Ok(Self::Reverted),
+            "finalized" => Ok(Self::Finalized),
             _ => Err(()),
         }
     }

crates/primitives/src/block.rs

@@ -25,6 +25,31 @@ pub struct BlockInfo {
     pub hash: B256,
 }
 
+/// The startup configuration for the L1 watcher.
+#[derive(Debug, PartialEq, Eq)]
+pub enum L1BlockStartupInfo {
+    /// The L1 block infos of the unsafe blocks stored in the database.
+    UnsafeBlocks(Vec<BlockInfo>),
+    /// The finalized block number to start from.
+    FinalizedBlockNumber(u64),
+    /// No startup information available.
+    None,
+}
+
+impl L1BlockStartupInfo {
+    /// Creates a new [`L1BlockStartupInfo`] from the given unsafe blocks and finalized block
+    /// number.
+    pub fn new(unsafe_blocks: Vec<BlockInfo>, finalized_block_number: Option<u64>) -> Self {
+        if !unsafe_blocks.is_empty() {
+            Self::UnsafeBlocks(unsafe_blocks)
+        } else if let Some(number) = finalized_block_number {
+            Self::FinalizedBlockNumber(number)
+        } else {
+            Self::None
+        }
+    }
+}
+
 impl PartialOrd for BlockInfo {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         self.number.partial_cmp(&other.number)

crates/primitives/src/lib.rs

@@ -9,8 +9,9 @@ pub use attributes::ScrollPayloadAttributesWithBatchInfo;
 
 mod block;
 pub use block::{
-    BlockInfo, L2BlockInfoWithL1Messages, WithBatchInfo, WithBlockNumber, WithCommittedBatchInfo,
-    WithFinalizedBatchInfo, WithFinalizedBlockNumber, DEFAULT_BLOCK_DIFFICULTY,
+    BlockInfo, L1BlockStartupInfo, L2BlockInfoWithL1Messages, WithBatchInfo, WithBlockNumber,
+    WithCommittedBatchInfo, WithFinalizedBatchInfo, WithFinalizedBlockNumber,
+    DEFAULT_BLOCK_DIFFICULTY,
 };
 
 mod batch;

crates/watcher/src/lib.rs

@@ -18,7 +18,8 @@ use alloy_sol_types::SolEvent;
 use error::L1WatcherResult;
 use itertools::Itertools;
 use rollup_node_primitives::{
-    BatchCommitData, BatchInfo, BlockInfo, BoundedVec, ConsensusUpdate, NodeConfig,
+    BatchCommitData, BatchInfo, BlockInfo, BoundedVec, ConsensusUpdate, L1BlockStartupInfo,
+    NodeConfig,
 };
 use rollup_node_providers::SystemContractProvider;
 use scroll_alloy_consensus::TxL1Message;
@@ -192,11 +193,11 @@ where
     /// returning [`L1Notification`] in the returned channel.
     pub async fn spawn(
         execution_provider: EP,
-        start_block: Option<u64>,
+        l1_block_startup_info: L1BlockStartupInfo,
         config: Arc<NodeConfig>,
         log_query_block_range: u64,
     ) -> mpsc::Receiver<Arc<L1Notification>> {
-        tracing::trace!(target: "scroll::watcher", ?start_block, ?config, "spawning L1 watcher");
+        tracing::trace!(target: "scroll::watcher", ?l1_block_startup_info, ?config, "spawning L1 watcher");
 
         let (tx, rx) = mpsc::channel(log_query_block_range as usize);
 
@@ -219,11 +220,39 @@ where
             finalized: fetch_block_info(BlockNumberOrTag::Finalized).await,
         };
 
+        let (reorg, start_block) = match l1_block_startup_info {
+            L1BlockStartupInfo::UnsafeBlocks(blocks) => {
+                let mut reorg = true;
+                let mut start_block =
+                    blocks.first().expect("at least one unsafe block").number.saturating_sub(1);
+                for (i, block) in blocks.into_iter().rev().enumerate() {
+                    let current_block =
+                        fetch_block_info(BlockNumberOrTag::Number(block.number)).await;
+                    if current_block.hash == block.hash {
+                        tracing::info!(target: "scroll::watcher", ?block, "found reorg block from unsafe blocks");
+                        reorg = i != 0;
+                        start_block = current_block.number;
+                    }
+                }
+
+                (reorg, start_block)
+            }
+            L1BlockStartupInfo::FinalizedBlockNumber(number) => {
+                tracing::info!(target: "scroll::watcher", ?number, "starting from finalized block number");
+
+                (false, number)
+            }
+            L1BlockStartupInfo::None => {
+                tracing::info!(target: "scroll::watcher", "no L1 startup info, starting from config start block");
+                (false, config.start_l1_block)
+            }
+        };
+
         // init the watcher.
         let watcher = Self {
             execution_provider,
             unfinalized_blocks: BoundedVec::new(HEADER_CAPACITY),
-            current_block_number: start_block.unwrap_or(config.start_l1_block).saturating_sub(1),
+            current_block_number: start_block.saturating_sub(1),
             l1_state,
             sender: tx,
             config,
@@ -233,6 +262,12 @@ where
         };
 
         // notify at spawn.
+        if reorg {
+            watcher
+                .notify(L1Notification::Reorg(start_block))
+                .await
+                .expect("channel is open in this context");
+        }
         watcher
             .notify(L1Notification::Finalized(watcher.l1_state.finalized))
             .await