add l1_watcher_command_rx to addons for testing like l1_watcher_tx

jonastheis · jonastheis · commit 41b3eada7837 · 2025-11-12T14:37:40.000+08:00
diff --git a/crates/chain-orchestrator/src/lib.rs b/crates/chain-orchestrator/src/lib.rs
@@ -2233,179 +2233,3 @@ async fn compute_l1_message_queue_hash(
 //         );
 //     }
 // }
-
-#[cfg(test)]
-mod tests {
-    // use super::*;
-    // use alloy_primitives::B256;
-    // use rollup_node_primitives::BatchCommitData;
-    // use std::sync::Arc;
-
-    // Commented out due to removal of MockL1WatcherHandle
-    // #[tokio::test]
-    // async fn test_gap_recovery() {
-    //     use rollup_node_watcher::MockL1WatcherHandle;
-    //
-    //     // setup a test node
-    //     let (mut nodes, _tasks, _wallet) = setup(1, false).await.unwrap();
-    //     let node = nodes.pop().unwrap();
-    //
-    //     // create a fork choice state
-    //     let genesis_hash = node.inner.chain_spec().genesis_hash();
-    //     let fcs = ForkchoiceState::new(
-    //         BlockInfo { hash: genesis_hash, number: 0 },
-    //         Default::default(),
-    //         Default::default(),
-    //     );
-    //
-    //     // create the engine driver connected to the node
-    //     let auth_client = node.inner.engine_http_client();
-    //     let engine_client = ScrollAuthApiEngineClient::new(auth_client);
-    //     let engine = Engine::new(Arc::new(engine_client), fcs);
-    //
-    //     // create a test database
-    //     let db = Arc::new(setup_test_db().await);
-    //
-    //     // prepare derivation pipeline
-    //     let mock_l1_provider = MockL1Provider { db: db.clone(), blobs: HashMap::new() };
-    //     let derivation_pipeline =
-    //         DerivationPipeline::new(mock_l1_provider, db.clone(), u64::MAX).await;
-    //
-    //     let (scroll_network_manager, scroll_network_handle) =
-    //         scroll_network::ScrollNetworkManager::new(
-    //             node.inner.chain_spec().clone(),
-    //             NetworkConfigBuilder::<ScrollNetworkPrimitives>::with_rng_secret_key()
-    //                 .build_with_noop_provider(node.inner.chain_spec().clone()),
-    //             ScrollWireConfig::new(true),
-    //             None,
-    //             Default::default(),
-    //             None,
-    //         )
-    //         .await;
-    //     tokio::spawn(scroll_network_manager);
-    //
-    //     // create full block client
-    //     let block_client = FullBlockClient::new(
-    //         scroll_network_handle
-    //             .inner()
-    //             .fetch_client()
-    //             .await
-    //             .expect("failed to fetch block client"),
-    //         Arc::new(ScrollBeaconConsensus::new(node.inner.chain_spec())),
-    //     );
-    //
-    //     // create l2 provider
-    //     let client = RpcClient::builder().http(node.rpc_url());
-    //     let l2_provider = ProviderBuilder::<_, _, Scroll>::default().connect_client(client);
-    //     let l2_provider = Arc::new(l2_provider);
-    //
-    //     // prepare L1 notification channel
-    //     let (l1_notification_tx, l1_notification_rx) = mpsc::channel(100);
-    //
-    //     // create mock L1 watcher handle for testing gap recovery
-    //     let mock_l1_watcher_handle = MockL1WatcherHandle::new();
-    //
-    //     // initialize database state
-    //     db.set_latest_l1_block_number(0).await.unwrap();
-    //
-    //     let (chain_orchestrator, _handle) = ChainOrchestrator::new(
-    //         db.clone(),
-    //         ChainOrchestratorConfig::new(node.inner.chain_spec().clone(), 0, 0),
-    //         Arc::new(block_client),
-    //         l2_provider,
-    //         l1_notification_rx,
-    //         Some(mock_l1_watcher_handle.clone()),
-    //         scroll_network_handle.into_scroll_network().await,
-    //         Box::new(NoopConsensus::default()),
-    //         engine,
-    //         Some(Sequencer::new(
-    //             Arc::new(MockL1Provider { db: db.clone(), blobs: HashMap::new() }),
-    //             SequencerConfig {
-    //                 chain_spec: node.inner.chain_spec(),
-    //                 fee_recipient: Address::random(),
-    //                 auto_start: false,
-    //                 payload_building_config: PayloadBuildingConfig {
-    //                     block_gas_limit: 15_000_000,
-    //                     max_l1_messages_per_block: 4,
-    //                     l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
-    //                 },
-    //                 block_time: 1,
-    //                 payload_building_duration: 0,
-    //                 allow_empty_blocks: false,
-    //             },
-    //         )),
-    //         None,
-    //         derivation_pipeline,
-    //     )
-    //     .await
-    //     .unwrap();
-    //
-    //     // Spawn a task that constantly polls chain orchestrator to process L1 notifications
-    //     let (_signal, shutdown) = shutdown_signal();
-    //     tokio::spawn(async {
-    //         let (_signal, inner) = shutdown_signal();
-    //         let chain_orchestrator = chain_orchestrator.run_until_shutdown(inner);
-    //         tokio::select! {
-    //             biased;
-    //
-    //             _ = shutdown => {},
-    //             _ = chain_orchestrator => {},
-    //         }
-    //     });
-    //
-    //     let genesis_batch = create_test_batch(1, 100);
-    //     l1_notification_tx
-    //         .send(Arc::new(L1Notification::BatchCommit(genesis_batch)))
-    //         .await
-    //         .unwrap();
-    //     tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-    //
-    //     let batch_with_gap = create_test_batch(3, 102);
-    //     l1_notification_tx
-    //         .send(Arc::new(L1Notification::BatchCommit(batch_with_gap)))
-    //         .await
-    //         .unwrap();
-    //     tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
-    //
-    //     mock_l1_watcher_handle.assert_reset_to(100);
-    //
-    //     // Insert first L1 message
-    //     // let l1_msg_0 = create_test_l1_message(0);
-    //     // l1_notification_tx.send(Arc::new(L1Notification::L1Message {
-    //     //     message: l1_msg_0,
-    //     //     block_number: 105,
-    //     //     block_timestamp: 0,
-    //     // })).await.unwrap();
-    //     // tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-    //     //
-    //     // let l1_msg_with_gap = create_test_l1_message(2);
-    //     // l1_notification_tx.send(Arc::new(L1Notification::L1Message {
-    //     //     message: l1_msg_with_gap,
-    //     //     block_number: 107,
-    //     //     block_timestamp: 0,
-    //     // })).await.unwrap();
-    //     // tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
-    //     //
-    //     // // Verify that reset was triggered to block 105 (last known L1 message)
-    //     // mock_l1_watcher_handle.assert_reset_to(105);
-    // }
-
-    // Helper function to create a simple test batch commit
-    // fn create_test_batch(index: u64, block_number: u64) -> BatchCommitData {
-    //     use alloy_primitives::Bytes;
-    //     BatchCommitData {
-    //         index,
-    //         hash: B256::random(),
-    //         block_number,
-    //         block_timestamp: 0,
-    //         calldata: Arc::new(Bytes::new()),
-    //         blob_versioned_hash: None,
-    //         finalized_block_number: None,
-    //     }
-    // }
-
-    // Helper function to create a simple test L1 message
-    // fn create_test_l1_message(queue_index: u64) -> TxL1Message {
-    //     TxL1Message { queue_index, ..Default::default() }
-    // }
-}
diff --git a/crates/node/src/add_ons/handle.rs b/crates/node/src/add_ons/handle.rs
@@ -5,6 +5,10 @@ use reth_rpc_eth_api::EthApiTypes;
 use reth_scroll_node::ScrollNetworkPrimitives;
 use rollup_node_chain_orchestrator::ChainOrchestratorHandle;
 #[cfg(feature = "test-utils")]
+use tokio::sync::mpsc::UnboundedReceiver;
+#[cfg(feature = "test-utils")]
+use tokio::sync::Mutex;
+#[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.
@@ -20,6 +24,10 @@ pub struct ScrollAddOnsHandle<
     /// An optional channel used to send `L1Watcher` notifications to the `RollupNodeManager`.
     #[cfg(feature = "test-utils")]
     pub l1_watcher_tx: Option<Sender<Arc<L1Notification>>>,
+    /// An optional channel used to receive commands from the `RollupNodeManager` to the
+    /// `L1Watcher`.
+    #[cfg(feature = "test-utils")]
+    pub l1_watcher_command_rx: Arc<Mutex<UnboundedReceiver<rollup_node_watcher::L1WatcherCommand>>>,
 }
 
 impl<
diff --git a/crates/node/src/add_ons/mod.rs b/crates/node/src/add_ons/mod.rs
@@ -2,6 +2,7 @@
 
 use super::args::ScrollRollupNodeConfig;
 use crate::constants;
+use std::sync::Arc;
 
 use reth_evm::{ConfigureEngineEvm, EvmFactory, EvmFactoryFor};
 use reth_network::NetworkProtocols;
@@ -37,7 +38,7 @@ pub use rpc::{RollupNodeExtApiClient, RollupNodeExtApiServer, RollupNodeRpcExt};
 mod rollup;
 pub use rollup::IsDevChain;
 use rollup::RollupManagerAddOn;
-use tokio::sync::mpsc::UnboundedReceiver;
+use tokio::sync::{mpsc::UnboundedReceiver, Mutex};
 
 /// Add-ons for the Scroll follower node.
 #[derive(Debug)]
@@ -137,7 +138,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, l1_watcher_tx, l1_watcher_command_rx) =
             rollup_node_manager_addon.launch(ctx.clone(), rpc_handle.clone()).await?;
 
         tx.send(rollup_manager_handle.clone())
@@ -148,6 +149,10 @@ where
             rpc_handle,
             #[cfg(feature = "test-utils")]
             l1_watcher_tx,
+            #[cfg(feature = "test-utils")]
+            l1_watcher_command_rx: Arc::new(Mutex::new(
+                l1_watcher_command_rx.expect("l1_watcher_command_rx must exist in test utils"),
+            )),
         })
     }
 }
diff --git a/crates/node/src/add_ons/rollup.rs b/crates/node/src/add_ons/rollup.rs
@@ -9,7 +9,7 @@ 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 rollup_node_watcher::{L1Notification, L1WatcherCommand};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_wire::ScrollWireEvent;
 use std::sync::Arc;
@@ -55,13 +55,17 @@ impl RollupManagerAddOn {
         self,
         ctx: AddOnsContext<'_, N>,
         rpc: RpcHandle<N, EthApi>,
-    ) -> eyre::Result<(ChainOrchestratorHandle<N::Network>, Option<Sender<Arc<L1Notification>>>)>
+    ) -> eyre::Result<(
+        ChainOrchestratorHandle<N::Network>,
+        Option<Sender<Arc<L1Notification>>>,
+        Option<UnboundedReceiver<L1WatcherCommand>>,
+    )>
     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, l1_notification_tx, l1_watcher_command_rx) = self
             .config
             .build((&ctx).into(), self.scroll_wire_event, rpc.rpc_server_handles)
             .await?;
@@ -70,6 +74,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, l1_notification_tx, l1_watcher_command_rx))
     }
 }
diff --git a/crates/node/src/args.rs b/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, L1WatcherHandle};
+use rollup_node_watcher::{L1Notification, L1Watcher, L1WatcherCommand, L1WatcherHandle};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_alloy_provider::{ScrollAuthApiEngineClient, ScrollEngineApi};
@@ -167,6 +167,7 @@ impl ScrollRollupNodeConfig {
         >,
         ChainOrchestratorHandle<N>,
         Option<Sender<Arc<L1Notification>>>,
+        Option<UnboundedReceiver<L1WatcherCommand>>,
     )>
     where
         N: FullNetwork<Primitives = ScrollNetworkPrimitives> + NetworkProtocols,
@@ -340,9 +341,11 @@ impl ScrollRollupNodeConfig {
         };
         let consensus = self.consensus_args.consensus(authorized_signer)?;
 
-        let (l1_notification_tx, l1_watcher_handle): (
+        #[allow(clippy::type_complexity)]
+        let (l1_notification_tx, l1_watcher_handle, l1_watcher_command_rx): (
             Option<Sender<Arc<L1Notification>>>,
             Option<L1WatcherHandle>,
+            Option<UnboundedReceiver<L1WatcherCommand>>,
         ) = if let Some(provider) = l1_provider.filter(|_| !self.test) {
             tracing::info!(target: "scroll::node::args", ?l1_block_startup_info, "Starting L1 watcher");
             let handle = L1Watcher::spawn(
@@ -352,24 +355,23 @@ impl ScrollRollupNodeConfig {
                 self.l1_provider_args.logs_query_block_range,
             )
             .await;
-            (None, Some(handle))
+            (None, Some(handle), None)
         } 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);
 
-                // TODO: expose command_rx to allow for tests to assert commands sent to the watcher
-                let (command_tx, _command_rx) = tokio::sync::mpsc::unbounded_channel();
+                let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
                 let handle = L1WatcherHandle::new(command_tx, rx);
 
-                (Some(tx), Some(handle))
+                (Some(tx), Some(handle), Some(command_rx))
             }
 
             #[cfg(not(feature = "test-utils"))]
             {
-                (None, None)
+                (None, None, None)
             }
         };
 
@@ -460,7 +462,7 @@ impl ScrollRollupNodeConfig {
         )
         .await?;
 
-        Ok((chain_orchestrator, handle, l1_notification_tx))
+        Ok((chain_orchestrator, handle, l1_notification_tx, l1_watcher_command_rx))
     }
 }
 
diff --git a/crates/node/tests/e2e.rs b/crates/node/tests/e2e.rs
@@ -839,7 +839,7 @@ async fn shutdown_consolidates_most_recent_batch_on_startup() -> eyre::Result<()
     config.hydrate(node.inner.config.clone()).await?;
 
     let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
-    let (chain_orchestrator, handle, l1_notification_tx) = config
+    let (chain_orchestrator, handle, l1_notification_tx, _) = config
         .clone()
         .build(
             RollupNodeContext::new(
@@ -989,7 +989,7 @@ async fn shutdown_consolidates_most_recent_batch_on_startup() -> eyre::Result<()
 
     // Start the RNM again.
     let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
-    let (chain_orchestrator, handle, l1_notification_tx) = config
+    let (chain_orchestrator, handle, l1_notification_tx, _) = config
         .clone()
         .build(
             RollupNodeContext::new(
@@ -1119,7 +1119,7 @@ async fn graceful_shutdown_sets_fcs_to_latest_signed_block_in_db_on_start_up() -
     config.hydrate(node.inner.config.clone()).await?;
 
     let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
-    let (rnm, handle, l1_watcher_tx) = config
+    let (rnm, handle, l1_watcher_tx, _) = config
         .clone()
         .build(
             RollupNodeContext::new(
@@ -1192,7 +1192,7 @@ async fn graceful_shutdown_sets_fcs_to_latest_signed_block_in_db_on_start_up() -
 
     // Start the RNM again.
     let (_, events) = ScrollWireProtocolHandler::new(ScrollWireConfig::new(true));
-    let (rnm, handle, _) = config
+    let (rnm, handle, _, _) = config
         .clone()
         .build(
             RollupNodeContext::new(
diff --git a/crates/watcher/src/handle/command.rs b/crates/watcher/src/handle/command.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use tokio::sync::mpsc;
 
 /// Commands that can be sent to the L1 Watcher.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum L1WatcherCommand {
     /// Reset the watcher to a specific L1 block number.
     ///
