feat: add revert to l1 block rpc (#463)

* add revert to l1 block rpc

* fix tests

* fix features

* fixes

* update fcu logic

Author: frisitano

crates/chain-orchestrator/src/event.rs

@@ -77,6 +77,8 @@ pub enum ChainOrchestratorEvent {
         /// The L2 safe block info.
         l2_safe_block_info: Option<BlockInfo>,
     },
+    /// The chain has been unwound to the specified L1 block number.
+    UnwoundToL1Block(u64),
     /// The chain orchestrator has synced to the L1 head.
     L1Synced,
     /// An L2 block has been committed returning the [`L2BlockInfoWithL1Messages`] and an

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

@@ -27,6 +27,8 @@ pub enum ChainOrchestratorCommand<N: FullNetwork<Primitives = ScrollNetworkPrimi
     DisableAutomaticSequencing(oneshot::Sender<bool>),
     /// Send a database query to the rollup manager.
     DatabaseQuery(DatabaseQuery),
+    /// Revert the rollup node state to the specified L1 block number.
+    RevertToL1Block((u64, oneshot::Sender<bool>)),
     /// Enable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     SetGossip((bool, oneshot::Sender<()>)),

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

@@ -22,13 +22,32 @@ use metrics::ChainOrchestratorHandleMetrics;
 pub struct ChainOrchestratorHandle<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> {
     /// The channel used to send commands to the rollup manager.
     to_manager_tx: mpsc::UnboundedSender<ChainOrchestratorCommand<N>>,
+    /// The metrics for the handle.
     handle_metrics: ChainOrchestratorHandleMetrics,
+    /// Mock for the L1 Watcher used in tests.
+    #[cfg(feature = "test-utils")]
+    pub l1_watcher_mock: Option<rollup_node_watcher::test_utils::L1WatcherMock>,
 }
 
 impl<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> ChainOrchestratorHandle<N> {
     /// Create a new rollup manager handle.
     pub fn new(to_manager_tx: mpsc::UnboundedSender<ChainOrchestratorCommand<N>>) -> Self {
-        Self { to_manager_tx, handle_metrics: ChainOrchestratorHandleMetrics::default() }
+        Self {
+            to_manager_tx,
+            handle_metrics: ChainOrchestratorHandleMetrics::default(),
+            #[cfg(feature = "test-utils")]
+            l1_watcher_mock: None,
+        }
+    }
+
+    /// Sets the L1 watcher mock for the handle.
+    #[cfg(feature = "test-utils")]
+    pub fn with_l1_watcher_mock(
+        mut self,
+        l1_watcher_mock: Option<rollup_node_watcher::test_utils::L1WatcherMock>,
+    ) -> Self {
+        self.l1_watcher_mock = l1_watcher_mock;
+        self
     }
 
     /// Sends a command to the rollup manager.
@@ -103,6 +122,16 @@ impl<N: FullNetwork<Primitives = ScrollNetworkPrimitives>> ChainOrchestratorHand
         rx.await
     }
 
+    /// Revert the rollup node state to the specified L1 block number.
+    pub async fn revert_to_l1_block(
+        &self,
+        block_number: u64,
+    ) -> Result<bool, oneshot::error::RecvError> {
+        let (tx, rx) = oneshot::channel();
+        self.send_command(ChainOrchestratorCommand::RevertToL1Block((block_number, tx)));
+        rx.await
+    }
+
     /// Sends a command to the rollup manager to enable or disable gossiping of blocks to peers.
     #[cfg(feature = "test-utils")]
     pub async fn set_gossip(&self, enabled: bool) -> Result<(), oneshot::error::RecvError> {

crates/chain-orchestrator/src/lib.rs

@@ -20,7 +20,7 @@ use rollup_node_primitives::{
 use rollup_node_providers::L1MessageProvider;
 use rollup_node_sequencer::{Sequencer, SequencerEvent};
 use rollup_node_signer::{SignatureAsBytes, SignerEvent, SignerHandle};
-use rollup_node_watcher::L1Notification;
+use rollup_node_watcher::{L1Notification, L1WatcherHandle};
 use scroll_alloy_consensus::{ScrollTxEnvelope, TxL1Message};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
@@ -35,7 +35,7 @@ use scroll_network::{
     BlockImportOutcome, NewBlockWithPeer, ScrollNetwork, ScrollNetworkManagerEvent,
 };
 use std::{collections::VecDeque, sync::Arc, time::Instant, vec};
-use tokio::sync::mpsc::{self, Receiver, UnboundedReceiver};
+use tokio::sync::mpsc::{self, UnboundedReceiver};
 
 mod config;
 pub use config::ChainOrchestratorConfig;
@@ -115,8 +115,8 @@ pub struct ChainOrchestrator<
     database: Arc<Database>,
     /// The current sync state of the [`ChainOrchestrator`].
     sync_state: SyncState,
-    /// A receiver for [`L1Notification`]s from the [`rollup_node_watcher::L1Watcher`].
-    l1_notification_rx: Receiver<Arc<L1Notification>>,
+    /// A handle for the [`rollup_node_watcher::L1Watcher`].
+    l1_watcher: L1WatcherHandle,
     /// The network manager that manages the scroll p2p network.
     network: ScrollNetwork<N>,
     /// The consensus algorithm used by the rollup node.
@@ -150,7 +150,7 @@ impl<
         config: ChainOrchestratorConfig<ChainSpec>,
         block_client: Arc<FullBlockClient<<N as BlockDownloaderProvider>::Client>>,
         l2_provider: L2P,
-        l1_notification_rx: Receiver<Arc<L1Notification>>,
+        l1_watcher: L1WatcherHandle,
         network: ScrollNetwork<N>,
         consensus: Box<dyn Consensus + 'static>,
         engine: Engine<EC>,
@@ -167,7 +167,7 @@ impl<
                 database,
                 config,
                 sync_state: SyncState::default(),
-                l1_notification_rx,
+                l1_watcher,
                 network,
                 consensus,
                 engine,
@@ -224,7 +224,7 @@ impl<
                     let res = self.handle_network_event(event).await;
                     self.handle_outcome(res);
                 }
-                Some(notification) = self.l1_notification_rx.recv(), if self.sync_state.l2().is_synced() && self.derivation_pipeline.is_empty() => {
+                Some(notification) = self.l1_watcher.l1_notification_receiver().recv(), if self.sync_state.l2().is_synced() && self.derivation_pipeline.is_empty() => {
                     let res = self.handle_l1_notification(notification).await;
                     self.handle_outcome(res);
                 }
@@ -401,6 +401,30 @@ impl<
                     let _ = sender.send(l1_message);
                 }
             },
+            ChainOrchestratorCommand::RevertToL1Block((block_number, tx)) => {
+                self.sync_state.l1_mut().set_syncing();
+                let unwind_result = self.database.unwind(block_number).await?;
+
+                // Check if the unwind impacts the fcs safe head.
+                if let Some(block_info) = unwind_result.l2_safe_block_info {
+                    // If the new safe head is above the current finalized head, update the fcs safe
+                    // head to the new safe head.
+                    if block_info.number >= self.engine.fcs().finalized_block_info().number {
+                        self.engine.update_fcs(None, Some(block_info), None).await?;
+                    } else {
+                        // Otherwise, update the fcs safe head to the finalized head.
+                        self.engine
+                            .update_fcs(None, Some(*self.engine.fcs().finalized_block_info()), None)
+                            .await?;
+                    }
+                }
+
+                // Revert the L1 watcher to the specified block.
+                self.l1_watcher.revert_to_l1_block(block_number);
+
+                self.notify(ChainOrchestratorEvent::UnwoundToL1Block(block_number));
+                let _ = tx.send(true);
+            }
             #[cfg(feature = "test-utils")]
             ChainOrchestratorCommand::SetGossip((enabled, tx)) => {
                 self.network.handle().set_gossip(enabled).await;

crates/node/src/add_ons/handle.rs

@@ -4,8 +4,6 @@ use reth_node_builder::rpc::{RpcHandle, RpcHandleProvider};
 use reth_rpc_eth_api::EthApiTypes;
 use reth_scroll_node::ScrollNetworkPrimitives;
 use rollup_node_chain_orchestrator::ChainOrchestratorHandle;
-#[cfg(feature = "test-utils")]
-use {rollup_node_watcher::L1Notification, std::sync::Arc, tokio::sync::mpsc::Sender};
 
 /// A handle for scroll addons, which includes handles for the rollup manager and RPC server.
 #[derive(Debug, Clone)]
@@ -17,9 +15,6 @@ pub struct ScrollAddOnsHandle<
     pub rollup_manager_handle: ChainOrchestratorHandle<Node::Network>,
     /// The handle used to send commands to the RPC server.
     pub rpc_handle: RpcHandle<Node, EthApi>,
-    /// An optional channel used to send `L1Watcher` notifications to the `RollupNodeManager`.
-    #[cfg(feature = "test-utils")]
-    pub l1_watcher_tx: Option<Sender<Arc<L1Notification>>>,
 }
 
 impl<

crates/node/src/add_ons/mod.rs

@@ -152,7 +152,7 @@ where
         });
 
         let rpc_handle = rpc_add_ons.launch_add_ons_with(ctx.clone(), |_| Ok(())).await?;
-        let (rollup_manager_handle, l1_watcher_tx) =
+        let rollup_manager_handle =
             rollup_node_manager_addon.launch(ctx.clone(), rpc_handle.clone()).await?;
 
         // Only send handle if RPC is enabled
@@ -161,12 +161,7 @@ where
                 .map_err(|_| eyre::eyre!("failed to send rollup manager handle"))?;
         }
 
-        Ok(ScrollAddOnsHandle {
-            rollup_manager_handle,
-            rpc_handle,
-            #[cfg(feature = "test-utils")]
-            l1_watcher_tx,
-        })
+        Ok(ScrollAddOnsHandle { rollup_manager_handle, rpc_handle })
     }
 }
 

crates/node/src/add_ons/rollup.rs

@@ -9,11 +9,9 @@ use reth_rpc_eth_api::EthApiTypes;
 use reth_scroll_chainspec::{ChainConfig, ScrollChainConfig, ScrollChainSpec};
 use reth_scroll_node::ScrollNetworkPrimitives;
 use rollup_node_chain_orchestrator::ChainOrchestratorHandle;
-use rollup_node_watcher::L1Notification;
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_wire::ScrollWireEvent;
-use std::sync::Arc;
-use tokio::sync::mpsc::{Sender, UnboundedReceiver};
+use tokio::sync::mpsc::UnboundedReceiver;
 
 /// Implementing the trait allows the type to return whether it is configured for dev chain.
 #[auto_impl::auto_impl(Arc)]
@@ -55,13 +53,13 @@ impl RollupManagerAddOn {
         self,
         ctx: AddOnsContext<'_, N>,
         rpc: RpcHandle<N, EthApi>,
-    ) -> eyre::Result<(ChainOrchestratorHandle<N::Network>, Option<Sender<Arc<L1Notification>>>)>
+    ) -> eyre::Result<ChainOrchestratorHandle<N::Network>>
     where
         <<N as FullNodeTypes>::Types as NodeTypes>::ChainSpec:
             ChainConfig<Config = ScrollChainConfig> + ScrollHardforks + IsDevChain,
         N::Network: NetworkProtocols + FullNetwork<Primitives = ScrollNetworkPrimitives>,
     {
-        let (chain_orchestrator, handle, l1_notification_tx) = self
+        let (chain_orchestrator, handle) = self
             .config
             .build((&ctx).into(), self.scroll_wire_event, rpc.rpc_server_handles)
             .await?;
@@ -70,6 +68,6 @@ impl RollupManagerAddOn {
             .spawn_critical_with_shutdown_signal("rollup_node_manager", |shutdown| {
                 chain_orchestrator.run_until_shutdown(shutdown)
             });
-        Ok((handle, l1_notification_tx))
+        Ok(handle)
     }
 }

crates/node/src/add_ons/rpc.rs

@@ -115,6 +115,10 @@ pub trait RollupNodeAdminApi {
     /// Disables automatic sequencing in the rollup node.
     #[method(name = "disableAutomaticSequencing")]
     async fn disable_automatic_sequencing(&self) -> RpcResult<bool>;
+
+    /// Reverts the rollup node state to a specified L1 block number.
+    #[method(name = "revertToL1Block")]
+    async fn revert_to_l1_block(&self, block_number: u64) -> RpcResult<bool>;
 }
 
 #[async_trait]
@@ -220,6 +224,24 @@ where
             )
         })
     }
+
+    async fn revert_to_l1_block(&self, block_number: u64) -> RpcResult<bool> {
+        let handle = self.rollup_manager_handle().await.map_err(|e| {
+            ErrorObjectOwned::owned(
+                error::INTERNAL_ERROR_CODE,
+                format!("Failed to get rollup manager handle: {}", e),
+                None::<()>,
+            )
+        })?;
+
+        handle.revert_to_l1_block(block_number).await.map_err(|e| {
+            ErrorObjectOwned::owned(
+                error::INTERNAL_ERROR_CODE,
+                format!("Failed to revert to L1 block {}: {}", block_number, e),
+                None::<()>,
+            )
+        })
+    }
 }
 
 // Implement RollupNodeApiServer for Arc<RollupNodeRpcExt<N>> to allow shared ownership
@@ -257,4 +279,8 @@ where
     async fn disable_automatic_sequencing(&self) -> RpcResult<bool> {
         (**self).disable_automatic_sequencing().await
     }
+
+    async fn revert_to_l1_block(&self, block_number: u64) -> RpcResult<bool> {
+        (**self).revert_to_l1_block(block_number).await
+    }
 }

crates/node/src/args.rs

@@ -38,7 +38,7 @@ use rollup_node_providers::{
 use rollup_node_sequencer::{
     L1MessageInclusionMode, PayloadBuildingConfig, Sequencer, SequencerConfig,
 };
-use rollup_node_watcher::{L1Notification, L1Watcher};
+use rollup_node_watcher::{L1Watcher, L1WatcherHandle};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_alloy_provider::{ScrollAuthApiEngineClient, ScrollEngineApi};
@@ -51,7 +51,7 @@ use scroll_engine::{Engine, ForkchoiceState};
 use scroll_migration::traits::ScrollMigrator;
 use scroll_network::ScrollNetworkManager;
 use scroll_wire::ScrollWireEvent;
-use tokio::sync::mpsc::{Sender, UnboundedReceiver};
+use tokio::sync::mpsc::UnboundedReceiver;
 
 /// A struct that represents the arguments for the rollup node.
 #[derive(Debug, Clone, clap::Args)]
@@ -166,7 +166,6 @@ impl ScrollRollupNodeConfig {
             impl ScrollEngineApi,
         >,
         ChainOrchestratorHandle<N>,
-        Option<Sender<Arc<L1Notification>>>,
     )>
     where
         N: FullNetwork<Primitives = ScrollNetworkPrimitives> + NetworkProtocols,
@@ -350,7 +349,14 @@ impl ScrollRollupNodeConfig {
         };
         let consensus = self.consensus_args.consensus(authorized_signer)?;
 
-        let (l1_notification_tx, l1_notification_rx): (Option<Sender<Arc<L1Notification>>>, _) =
+        // Define some types to support definitions of return type of following function in no_std.
+        #[cfg(feature = "test-utils")]
+        type L1WatcherMockOpt = Option<rollup_node_watcher::test_utils::L1WatcherMock>;
+
+        #[cfg(not(feature = "test-utils"))]
+        type L1WatcherMockOpt = Option<std::convert::Infallible>;
+
+        let (_l1_watcher_mock, l1_watcher_handle): (L1WatcherMockOpt, Option<L1WatcherHandle>) =
             if let Some(provider) = l1_provider.filter(|_| !self.test) {
                 tracing::info!(target: "scroll::node::args", ?l1_block_startup_info, "Starting L1 watcher");
                 (
@@ -370,8 +376,15 @@ impl ScrollRollupNodeConfig {
                 // testing
                 #[cfg(feature = "test-utils")]
                 {
-                    let (tx, rx) = tokio::sync::mpsc::channel(1000);
-                    (Some(tx), Some(rx))
+                    let (notification_tx, notification_rx) = tokio::sync::mpsc::channel(1000);
+                    let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
+                    let handle =
+                        rollup_node_watcher::L1WatcherHandle::new(command_tx, notification_rx);
+                    let watcher_mock = rollup_node_watcher::test_utils::L1WatcherMock {
+                        command_rx: Arc::new(tokio::sync::Mutex::new(command_rx)),
+                        notification_tx,
+                    };
+                    (Some(watcher_mock), Some(handle))
                 }
 
                 #[cfg(not(feature = "test-utils"))]
@@ -457,7 +470,7 @@ impl ScrollRollupNodeConfig {
             config,
             Arc::new(block_client),
             l2_provider,
-            l1_notification_rx.expect("L1 notification receiver should be set"),
+            l1_watcher_handle.expect("L1 notification receiver should be set"),
             scroll_network_handle.into_scroll_network().await,
             consensus,
             engine,
@@ -467,7 +480,10 @@ impl ScrollRollupNodeConfig {
         )
         .await?;
 
-        Ok((chain_orchestrator, handle, l1_notification_tx))
+        #[cfg(feature = "test-utils")]
+        let handle = handle.with_l1_watcher_mock(_l1_watcher_mock);
+
+        Ok((chain_orchestrator, handle))
     }
 }
 

crates/node/src/test_utils/event_utils.rs

@@ -130,6 +130,15 @@ impl<'a> EventWaiter<'a> {
         Ok(())
     }
 
+    /// Wait for chain unwound event on all specified nodes.
+    pub async fn revert_to_l1_block(self) -> eyre::Result<()> {
+        self.wait_for_event_on_all(|e| {
+            matches!(e, ChainOrchestratorEvent::UnwoundToL1Block(_)).then_some(())
+        })
+        .await?;
+        Ok(())
+    }
+
     /// Wait for block consolidated event on all specified nodes.
     pub async fn block_consolidated(self, target_block: u64) -> eyre::Result<()> {
         self.wait_for_event_on_all(|e| {