feat: l1 reorg (#409)

* l1 event handling

* l1 event handling

* clean up

* add test cases

* cleanup

* address feedback

* use alloc String

* use alloc ToString

* fix L1 watcher revert batches

* add revert batch signatures to watcher filter

* fix database revert batch operation

Author: frisitano

Cargo.lock

@@ -1,7 +1,9 @@
 use super::ChainOrchestratorError;
 use alloy_provider::Provider;
 use futures::{stream::FuturesOrdered, TryStreamExt};
-use rollup_node_primitives::{BatchConsolidationOutcome, BatchInfo, L2BlockInfoWithL1Messages};
+use rollup_node_primitives::{
+    BatchConsolidationOutcome, BatchInfo, BatchStatus, L2BlockInfoWithL1Messages,
+};
 use scroll_alloy_network::Scroll;
 use scroll_derivation_pipeline::{BatchDerivationResult, DerivedAttributes};
 use scroll_engine::{block_matches_attributes, ForkchoiceState};
@@ -53,7 +55,11 @@ pub(crate) async fn reconcile_batch<L2P: Provider<Scroll>>(
     }
 
     let actions: Vec<BlockConsolidationAction> = futures.try_collect().await?;
-    Ok(BatchReconciliationResult { batch_info: batch.batch_info, actions })
+    Ok(BatchReconciliationResult {
+        batch_info: batch.batch_info,
+        actions,
+        target_status: batch.target_status,
+    })
 }
 
 /// The result of reconciling a batch with the L2 chain.
@@ -63,6 +69,8 @@ pub(crate) struct BatchReconciliationResult {
     pub batch_info: BatchInfo,
     /// The actions that must be performed on the L2 chain to consolidate the batch.
     pub actions: Vec<BlockConsolidationAction>,
+    /// The target status of the batch after consolidation.
+    pub target_status: BatchStatus,
 }
 
 impl BatchReconciliationResult {
@@ -93,7 +101,8 @@ impl BatchReconciliationResult {
         self,
         reorg_results: Vec<L2BlockInfoWithL1Messages>,
     ) -> Result<BatchConsolidationOutcome, ChainOrchestratorError> {
-        let mut consolidate_chain = BatchConsolidationOutcome::new(self.batch_info);
+        let mut consolidate_chain =
+            BatchConsolidationOutcome::new(self.batch_info, self.target_status);
 
         // First append all non-reorg results to the consolidated chain.
         self.actions.into_iter().filter(|action| !action.is_reorg()).for_each(|action| {

crates/chain-orchestrator/src/consolidation.rs

@@ -42,11 +42,21 @@ pub enum ChainOrchestratorEvent {
         batch_info: BatchInfo,
         /// The L1 block number in which the batch was committed.
         l1_block_number: u64,
-        /// The safe L2 block info.
-        safe_head: Option<BlockInfo>,
     },
     /// A batch has been finalized returning a list of finalized batches.
-    BatchFinalized(u64, Vec<BatchInfo>),
+    BatchFinalized {
+        /// The L1 block info at which the batch finalization event was received.
+        l1_block_info: BlockInfo,
+        /// The list of batches that have been triggered for the derivation pipeline.
+        triggered_batches: Vec<BatchInfo>,
+    },
+    /// A batch has been reverted returning the batch info and the new safe head.
+    BatchReverted {
+        /// The latest batch info after the revert.
+        batch_info: BatchInfo,
+        /// The new safe head after the revert.
+        safe_head: BlockInfo,
+    },
     /// A new L1 block has been received returning the L1 block number.
     NewL1Block(u64),
     /// An L1 block has been finalized returning the L1 block number and the list of finalized

crates/chain-orchestrator/src/event.rs

@@ -30,6 +30,9 @@ pub enum ChainOrchestratorCommand<N: FullNetwork<Primitives = ScrollNetworkPrimi
     /// Enable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     SetGossip((bool, oneshot::Sender<()>)),
+    /// Returns a database handle for direct database access.
+    #[cfg(feature = "test-utils")]
+    DatabaseHandle(oneshot::Sender<std::sync::Arc<scroll_db::Database>>),
 }
 
 /// The database queries that can be sent to the rollup manager.

crates/chain-orchestrator/src/handle/command.rs

@@ -110,4 +110,14 @@ impl<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> ChainOrchestratorHand
         self.send_command(ChainOrchestratorCommand::SetGossip((enabled, tx)));
         rx.await
     }
+
+    /// Sends a command to the rollup manager to get a database handle for direct database access.
+    #[cfg(feature = "test-utils")]
+    pub async fn get_database_handle(
+        &self,
+    ) -> Result<std::sync::Arc<scroll_db::Database>, oneshot::error::RecvError> {
+        let (tx, rx) = oneshot::channel();
+        self.send_command(ChainOrchestratorCommand::DatabaseHandle(tx));
+        rx.await
+    }
 }