test: add test cases for handling l1 events and l1 reorg

Cargo.toml

@@ -129,11 +129,13 @@ alloy-consensus = { version = "1.0.37", default-features = false }
 alloy-eips = { version = "1.0.37", default-features = false }
 alloy-json-rpc = { version = "1.0.37", default-features = false }
 alloy-network = { version = "1.0.37", default-features = false }
+alloy-node-bindings = { version = "1.0.37", default-features = false }
 alloy-primitives = { version = "1.4.1", default-features = false }
 alloy-provider = { version = "1.0.37", default-features = false }
 alloy-rpc-client = { version = "1.0.37", default-features = false }
 alloy-rpc-types-engine = { version = "1.0.37", default-features = false }
 alloy-rpc-types-eth = { version = "1.0.37", default-features = false }
+alloy-rpc-types-anvil = { version = "1.0.37", default-features = false }
 alloy-sol-types = { version = "1.4.1", default-features = false }
 alloy-signer = { version = "1.0.37", default-features = false }
 alloy-signer-local = { version = "1.0.37", default-features = false }
@@ -150,9 +152,11 @@ scroll-alloy-rpc-types-engine = { git = "https://github.com/scroll-tech/reth.git
 
 # reth
 reth-chainspec = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
+reth-db = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
 reth-e2e-test-utils = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4" }
 reth-eth-wire = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
 reth-eth-wire-types = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
+reth-fs-util = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
 reth-network = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
 reth-network-api = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }
 reth-network-p2p = { git = "https://github.com/scroll-tech/reth.git", tag = "scroll-v91.4", default-features = false }

Makefile

@@ -64,7 +64,7 @@ lint: fmt lint-toml clippy udeps codespell zepter
 
 .PHONY: test
 test:
-	cargo nextest run \
+	cargo +nightly-2026-01-05 nextest run \
 	--workspace \
 	--locked \
 	--all-features \

crates/chain-orchestrator/src/event.rs

@@ -44,7 +44,7 @@ pub enum ChainOrchestratorEvent {
         l1_block_number: u64,
     },
     /// A batch has been finalized returning a list of finalized batches.
-    BatchFinalized {
+    BatchFinalizeIndexed {
         /// 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.
@@ -119,4 +119,6 @@ pub enum ChainOrchestratorEvent {
     },
     /// The head of the fork choice state has been updated in the engine driver.
     FcsHeadUpdated(BlockInfo),
+    /// The chain orchestrator is shutting down.
+    Shutdown,
 }

crates/chain-orchestrator/src/lib.rs

@@ -189,6 +189,7 @@ impl<
                 biased;
 
                 _guard = &mut shutdown => {
+                    self.notify(ChainOrchestratorEvent::Shutdown);
                     break;
                 }
                 Some(command) = self.handle_rx.recv() => {
@@ -838,7 +839,7 @@ impl<
             self.derivation_pipeline.push_batch(*batch, BatchStatus::Finalized).await;
         }
 
-        Ok(Some(ChainOrchestratorEvent::BatchFinalized { l1_block_info, triggered_batches }))
+        Ok(Some(ChainOrchestratorEvent::BatchFinalizeIndexed { l1_block_info, triggered_batches }))
     }
 
     /// Handles a batch revert event by updating the database.

crates/node/Cargo.toml

@@ -77,10 +77,14 @@ aws-sdk-kms = "1.76.0"
 
 # test-utils
 alloy-eips = { workspace = true, optional = true }
+anvil = { git = "https://github.com/foundry-rs/foundry.git", rev = "2c84e1c970d11ef5023a77d8002a1cb70b143888", default-features = false, optional = true }
 alloy-rpc-types-eth = { workspace = true, optional = true }
 alloy-rpc-types-engine = { workspace = true, optional = true }
+alloy-rpc-types-anvil = { workspace = true, optional = true }
+reth-db = { workspace = true, optional = true, features = ["test-utils"] }
 reth-e2e-test-utils = { workspace = true, optional = true }
 reth-engine-local = { workspace = true, optional = true }
+reth-fs-util = { workspace = true, optional = true }
 reth-provider = { workspace = true, optional = true }
 reth-rpc-layer = { workspace = true, optional = true }
 reth-rpc-server-types = { workspace = true, optional = true }
@@ -110,6 +114,7 @@ console-subscriber = "0.5.0"
 alloy-chains.workspace = true
 alloy-eips.workspace = true
 futures.workspace = true
+reth-db = { workspace = true, features = ["test-utils"] }
 reth-e2e-test-utils.workspace = true
 reth-node-core.workspace = true
 reth-provider.workspace = true
@@ -130,7 +135,9 @@ alloy-rpc-types-eth = { workspace = true }
 [features]
 js-tracer = ["reth-scroll-node/js-tracer", "reth-scroll-rpc/js-tracer"]
 test-utils = [
+    "reth-db",
     "reth-engine-local",
+    "reth-fs-util",
     "reth-trie-db/test-utils",
     "reth-chainspec/test-utils",
     "reth-evm/test-utils",
@@ -156,6 +163,11 @@ test-utils = [
     "rollup-node-chain-orchestrator/test-utils",
     "scroll-network/test-utils",
     "alloy-eips",
+    "anvil",
     "reth-storage-api",
     "alloy-rpc-types-eth",
+    "alloy-rpc-types-anvil",
+    "alloy-provider/anvil-api",
+    "alloy-provider/anvil-node",
+    "reth-db?/test-utils",
 ]

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::{L1Watcher, L1WatcherHandle};
+use rollup_node_watcher::{L1Watcher, L1WatcherCommand};
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_alloy_provider::{ScrollAuthApiEngineClient, ScrollEngineApi};
@@ -53,12 +53,23 @@ use scroll_network::ScrollNetworkManager;
 use scroll_wire::ScrollWireEvent;
 use tokio::sync::mpsc::UnboundedReceiver;
 
-/// A struct that represents the arguments for the rollup node.
-#[derive(Debug, Clone, clap::Args)]
-pub struct ScrollRollupNodeConfig {
+/// Test-related configuration arguments.
+#[derive(Debug, Clone, Default, clap::Args)]
+pub struct TestArgs {
     /// Whether the rollup node should be run in test mode.
     #[arg(long)]
     pub test: bool,
+    /// Test mode: skip L1 watcher Synced notifications.
+    #[arg(long, default_value = "false")]
+    pub skip_l1_synced: bool,
+}
+
+/// A struct that represents the arguments for the rollup node.
+#[derive(Debug, Clone, clap::Args)]
+pub struct ScrollRollupNodeConfig {
+    /// Test-related arguments
+    #[command(flatten)]
+    pub test_args: TestArgs,
     /// Consensus args
     #[command(flatten)]
     pub consensus_args: ConsensusArgs,
@@ -136,6 +147,7 @@ impl ScrollRollupNodeConfig {
     ) -> eyre::Result<()> {
         // Instantiate the database
         let db_path = node_config.datadir().db();
+
         let database_path = if let Some(database_path) = &self.database_args.rn_db_path {
             database_path.to_string_lossy().to_string()
         } else {
@@ -235,7 +247,7 @@ impl ScrollRollupNodeConfig {
         // Run the database migrations
         if let Some(named) = chain_spec.chain().named() {
             named
-                .migrate(db.inner().get_connection(), self.test)
+                .migrate(db.inner().get_connection(), self.test_args.test)
                 .await
                 .expect("failed to perform migration");
         } else {
@@ -349,49 +361,45 @@ impl ScrollRollupNodeConfig {
         };
         let consensus = self.consensus_args.consensus(authorized_signer)?;
 
-        // 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");
-                (
-                    None,
-                    Some(
-                        L1Watcher::spawn(
-                            provider,
-                            l1_block_startup_info,
-                            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
+        let is_anvil_provider = self.blob_provider_args.anvil_url.is_some();
+
+        let (_l1_notification_tx, _l1_command_rx, l1_watcher_handle): (_, _, _) = if let Some(
+            provider,
+        ) =
+            l1_provider.filter(|_| !self.test_args.test || is_anvil_provider)
+        {
+            tracing::info!(target: "scroll::node::args", ?l1_block_startup_info, "Starting L1 watcher");
+
+            #[cfg(feature = "test-utils")]
+            let skip_synced = self.test_args.test && self.test_args.skip_l1_synced;
+
+            let (notification_tx, handle) = L1Watcher::spawn(
+                provider,
+                l1_block_startup_info,
+                node_config,
+                self.l1_provider_args.logs_query_block_range,
                 #[cfg(feature = "test-utils")]
-                {
-                    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"))]
-                {
-                    (None, None)
-                }
-            };
+                skip_synced,
+            )
+            .await;
+            (Some(notification_tx), None::<UnboundedReceiver<L1WatcherCommand>>, Some(handle))
+        } else {
+            // Create a channel for L1 notifications that we can use to inject L1 messages for
+            // testing
+            #[cfg(feature = "test-utils")]
+            {
+                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);
+
+                (Some(notification_tx), Some(command_rx), Some(handle))
+            }
+
+            #[cfg(not(feature = "test-utils"))]
+            {
+                (None, None, None)
+            }
+        };
 
         // Construct the l1 provider.
         let l1_messages_provider = db.clone();
@@ -433,7 +441,7 @@ impl ScrollRollupNodeConfig {
         let signer = if let Some(configured_signer) = self.signer_args.signer(chain_id).await? {
             // Use the signer configured by SignerArgs
             Some(rollup_node_signer::Signer::spawn(configured_signer))
-        } else if self.test {
+        } else if self.test_args.test {
             // Use a random private key signer for testing
             Some(rollup_node_signer::Signer::spawn(PrivateKeySigner::random()))
         } else {
@@ -465,7 +473,8 @@ impl ScrollRollupNodeConfig {
         )
         .await;
 
-        let (chain_orchestrator, handle) = ChainOrchestrator::new(
+        #[cfg_attr(not(feature = "test-utils"), allow(unused_mut))]
+        let (chain_orchestrator, mut handle) = ChainOrchestrator::new(
             db,
             config,
             Arc::new(block_client),
@@ -481,7 +490,14 @@ impl ScrollRollupNodeConfig {
         .await?;
 
         #[cfg(feature = "test-utils")]
-        let handle = handle.with_l1_watcher_mock(_l1_watcher_mock);
+        {
+            let command_rx = _l1_command_rx.map(|rx| Arc::new(tokio::sync::Mutex::new(rx)));
+            let l1_watcher_mock = rollup_node_watcher::test_utils::L1WatcherMock {
+                command_rx,
+                notification_tx: _l1_notification_tx.expect("L1 notification sender should be set"),
+            };
+            handle = handle.with_l1_watcher_mock(Some(l1_watcher_mock));
+        }
 
         Ok((chain_orchestrator, handle))
     }
@@ -919,7 +935,7 @@ mod tests {
     #[test]
     fn test_validate_sequencer_enabled_without_any_signer_fails() {
         let config = ScrollRollupNodeConfig {
-            test: false,
+            test_args: TestArgs::default(),
             sequencer_args: SequencerArgs { sequencer_enabled: true, ..Default::default() },
             signer_args: SignerArgs { key_file: None, aws_kms_key_id: None, private_key: None },
             database_args: RollupNodeDatabaseArgs::default(),
@@ -947,7 +963,7 @@ mod tests {
     #[test]
     fn test_validate_sequencer_enabled_with_both_signers_fails() {
         let config = ScrollRollupNodeConfig {
-            test: false,
+            test_args: TestArgs::default(),
             sequencer_args: SequencerArgs { sequencer_enabled: true, ..Default::default() },
             signer_args: SignerArgs {
                 key_file: Some(PathBuf::from("/path/to/key")),
@@ -977,7 +993,7 @@ mod tests {
     #[test]
     fn test_validate_sequencer_enabled_with_key_file_succeeds() {
         let config = ScrollRollupNodeConfig {
-            test: false,
+            test_args: TestArgs::default(),
             sequencer_args: SequencerArgs { sequencer_enabled: true, ..Default::default() },
             signer_args: SignerArgs {
                 key_file: Some(PathBuf::from("/path/to/key")),
@@ -1002,7 +1018,7 @@ mod tests {
     #[test]
     fn test_validate_sequencer_enabled_with_aws_kms_succeeds() {
         let config = ScrollRollupNodeConfig {
-            test: false,
+            test_args: TestArgs::default(),
             sequencer_args: SequencerArgs { sequencer_enabled: true, ..Default::default() },
             signer_args: SignerArgs {
                 key_file: None,
@@ -1027,7 +1043,7 @@ mod tests {
     #[test]
     fn test_validate_sequencer_disabled_without_any_signer_succeeds() {
         let config = ScrollRollupNodeConfig {
-            test: false,
+            test_args: TestArgs::default(),
             sequencer_args: SequencerArgs { sequencer_enabled: false, ..Default::default() },
             signer_args: SignerArgs { key_file: None, aws_kms_key_id: None, private_key: None },
             database_args: RollupNodeDatabaseArgs::default(),

crates/node/src/test_utils/block_builder.rs

@@ -92,7 +92,8 @@ impl<'a> BlockBuilder<'a> {
 
     /// Build the block and validate against expectations.
     pub async fn build_and_await_block(self) -> eyre::Result<ScrollBlock> {
-        let sequencer_node = &self.fixture.nodes[0];
+        let sequencer_node =
+            self.fixture.nodes[0].as_ref().expect("sequencer node has been shutdown");
 
         // Get the sequencer from the rollup manager handle
         let handle = &sequencer_node.rollup_manager_handle;