test(l1): add reorg testing framework

.github/workflows/pr-main_l1.yaml

@@ -223,7 +223,7 @@
 
   # The purpose of this job is to add it as a required check in GitHub so that we don't have to add every individual job as a required check
   all-tests:
-    # "Integration Test" is a required check, don't change the name
+    # "Integration Test" is a required check, don't change the name
     name: Integration Test
     runs-on: ubuntu-latest
     needs: [run-assertoor, run-hive]
@@ -241,3 +241,20 @@
             echo "Job Hive failed"
             exit 1
           fi
+
+  reorg-tests:
+    name: Reorg Tests
+    runs-on: ubuntu-latest
+    if: ${{ github.event_name != 'merge_group' }}
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v4
+
+      - name: Setup Rust Environment
+        uses: ./.github/actions/setup-rust
+
+      - name: Compile ethrex binary
+        run: cargo build --bin ethrex
+
+      - name: Run reorg tests
+        run: cd tooling/reorgs && cargo run

.gitignore

@@ -15,6 +15,8 @@ tooling/ef_tests/state/vectors
 tooling/ef_tests/state/runner_v2/failure_report.txt
 tooling/ef_tests/state/runner_v2/success_report.txt
 
+tooling/reorgs/data
+
 # Repos checked out by make target
 /hive/
 ethereum-package/

Cargo.toml

@@ -33,6 +33,7 @@ members = [
   "tooling/archive_sync",
   "tooling/replayer",
   "crates/common/config",
+  "tooling/reorgs",
 ]
 resolver = "2"
 

cmd/ethrex/cli.rs

@@ -34,7 +34,7 @@ pub struct CLI {
     pub command: Option<Subcommand>,
 }
 
-#[derive(ClapParser, Debug)]
+#[derive(ClapParser, Debug, Clone)]
 pub struct Options {
     #[arg(
         long = "network",

crates/networking/rpc/types/fork_choice.rs

@@ -2,7 +2,7 @@ use super::payload::PayloadStatus;
 use ethrex_common::{Address, H256, serde_utils, types::Withdrawal};
 use serde::{Deserialize, Serialize};
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ForkChoiceState {
     #[allow(unused)]

crates/networking/rpc/types/payload.rs

@@ -174,7 +174,7 @@ pub struct PayloadStatus {
     pub validation_error: Option<String>,
 }
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "UPPERCASE")]
 pub enum PayloadValidationStatus {
     Valid,

tooling/reorgs/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "reorgs"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+ethrex.workspace = true
+ethrex-common.workspace = true
+ethrex-blockchain.workspace = true
+ethrex-rpc.workspace = true
+ethrex-config.workspace = true
+ethrex-l2-common.workspace = true
+ethrex-l2-rpc.workspace = true
+
+tokio.workspace = true
+tokio-util.workspace = true
+tracing.workspace = true
+rand.workspace = true
+sha2.workspace = true
+hex.workspace = true
+nix = { version = "0.30", features = ["signal"] }
+secp256k1.workspace = true

tooling/reorgs/README.md

@@ -0,0 +1,23 @@
+# Reorg integration tests
+
+This directory contains tests for chain reorganization.
+
+## How to run
+
+First, compile the `ethrex` binary if you haven't already:
+
+```bash
+cargo build --workspace --bin ethrex
+```
+
+Then, run the reorg tests using:
+
+```bash
+cargo run
+```
+
+You can run a custom binary by specifying the path:
+
+```bash
+cargo run -- /path/to/your/binary
+```

tooling/reorgs/src/main.rs

@@ -0,0 +1,219 @@
+use std::{
+    path::{Path, PathBuf},
+    process::Command,
+    sync::Arc,
+};
+
+use ethrex::{cli::Options, initializers::init_tracing};
+use ethrex_l2_rpc::signer::{LocalSigner, Signer};
+use tokio::sync::Mutex;
+use tracing::{error, info, warn};
+
+use crate::simulator::Simulator;
+
+mod simulator;
+
+#[tokio::main]
+async fn main() {
+    // Setup logging
+    init_tracing(&Options::default_l1());
+
+    // Fetch the path to the ethrex binary from the command line arguments
+    // If not provided, use the default path
+    let cmd_path: PathBuf = std::env::args()
+        .nth(1)
+        .map(|o| o.parse().unwrap())
+        .unwrap_or_else(|| "../../target/debug/ethrex".parse().unwrap());
+
+    let version = get_ethrex_version(&cmd_path).await;
+
+    info!(%version, binary_path = %cmd_path.display(), "Fetched ethrex binary version");
+    info!("Starting test run");
+    info!("");
+
+    run_test(&cmd_path, test_one_block_reorg_and_back).await;
+
+    // TODO: this test is failing
+    // run_test(&cmd_path, test_many_blocks_reorg).await;
+}
+
+async fn get_ethrex_version(cmd_path: &Path) -> String {
+    let version_output = Command::new(cmd_path)
+        .arg("--version")
+        .output()
+        .expect("failed to get ethrex version");
+    String::from_utf8(version_output.stdout).expect("failed to parse version output")
+}
+
+async fn run_test<F, Fut>(cmd_path: &Path, test_fn: F)
+where
+    F: Fn(Arc<Mutex<Simulator>>) -> Fut,
+    Fut: Future<Output = ()> + Send + 'static,
+{
+    let test_name = std::any::type_name::<F>();
+    let start = std::time::Instant::now();
+
+    info!(test=%test_name, "Running test");
+    let simulator = Arc::new(Mutex::new(Simulator::new(cmd_path.to_path_buf())));
+
+    // Run in another task to clean up properly on panic
+    let result = tokio::spawn(test_fn(simulator.clone())).await;
+
+    simulator.lock_owned().await.stop();
+    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+
+    match result {
+        Ok(_) => info!(test=%test_name, elapsed=?start.elapsed(), "test completed successfully"),
+        Err(err) if err.is_panic() => {
+            error!(test=%test_name, %err, "test panicked");
+            std::process::exit(1);
+        }
+        Err(err) => {
+            warn!(test=%test_name, %err, "test task was cancelled");
+        }
+    }
+    // Add a blank line after each test for readability
+    info!("");
+}
+
+async fn test_one_block_reorg_and_back(simulator: Arc<Mutex<Simulator>>) {
+    let mut simulator = simulator.lock().await;
+    let signer: Signer = LocalSigner::new(
+        "941e103320615d394a55708be13e45994c7d93b932b064dbcb2b511fe3254e2e"
+            .parse()
+            .unwrap(),
+    )
+    .into();
+    // Some random address
+    let recipient = "941e103320615d394a55708be13e45994c7d93b0".parse().unwrap();
+    let transfer_amount = 1000000;
+
+    let node0 = simulator.start_node().await;
+    let node1 = simulator.start_node().await;
+
+    // Create a chain with a few empty blocks
+    let mut base_chain = simulator.get_base_chain();
+    for _ in 0..10 {
+        let extended_base_chain = node0.build_payload(base_chain).await;
+        node0.notify_new_payload(&extended_base_chain).await;
+        node0.update_forkchoice(&extended_base_chain).await;
+
+        node1.notify_new_payload(&extended_base_chain).await;
+        node1.update_forkchoice(&extended_base_chain).await;
+        base_chain = extended_base_chain;
+    }
+
+    let initial_balance = node0.get_balance(recipient).await;
+
+    // Fork the chain
+    let side_chain = base_chain.fork();
+
+    // Mine a new block in the base chain
+    let base_chain = node0.build_payload(base_chain).await;
+    node0.notify_new_payload(&base_chain).await;
+    node0.update_forkchoice(&base_chain).await;
+
+    // Mine a new block in the base chain (but don't announce it yet)
+    let extended_base_chain = node0.build_payload(base_chain).await;
+
+    // In parallel, mine a block in the side chain, with an ETH transfer
+    node1
+        .send_eth_transfer(&signer, recipient, transfer_amount)
+        .await;
+
+    let side_chain = node1.build_payload(side_chain).await;
+    node1.notify_new_payload(&side_chain).await;
+    node1.update_forkchoice(&side_chain).await;
+
+    // Sanity check: balance hasn't changed
+    let same_balance = node0.get_balance(recipient).await;
+    assert_eq!(same_balance, initial_balance);
+
+    // Notify the first node of the side chain block, it should reorg
+    node0.notify_new_payload(&side_chain).await;
+    node0.update_forkchoice(&side_chain).await;
+
+    // Check the transfer has been processed
+    let new_balance = node0.get_balance(recipient).await;
+    assert_eq!(new_balance, initial_balance + transfer_amount);
+
+    // Finally, move to the extended base chain, it should reorg back
+    node0.notify_new_payload(&extended_base_chain).await;
+    node0.update_forkchoice(&extended_base_chain).await;
+
+    // Check the transfer has been reverted
+    let new_balance = node0.get_balance(recipient).await;
+    assert_eq!(new_balance, initial_balance);
+}
+
+#[expect(unused)]
+async fn test_many_blocks_reorg(simulator: Arc<Mutex<Simulator>>) {
+    let mut simulator = simulator.lock().await;
+    let signer: Signer = LocalSigner::new(
+        "941e103320615d394a55708be13e45994c7d93b932b064dbcb2b511fe3254e2e"
+            .parse()
+            .unwrap(),
+    )
+    .into();
+    // Some random address
+    let recipient = "941e103320615d394a55708be13e45994c7d93b0".parse().unwrap();
+    let transfer_amount = 1000000;
+
+    let node0 = simulator.start_node().await;
+    let node1 = simulator.start_node().await;
+
+    // Create a chain with a few empty blocks
+    let mut base_chain = simulator.get_base_chain();
+    for _ in 0..10 {
+        let extended_base_chain = node0.build_payload(base_chain).await;
+        node0.notify_new_payload(&extended_base_chain).await;
+        node0.update_forkchoice(&extended_base_chain).await;
+
+        node1.notify_new_payload(&extended_base_chain).await;
+        node1.update_forkchoice(&extended_base_chain).await;
+        base_chain = extended_base_chain;
+    }
+
+    let initial_balance = node0.get_balance(recipient).await;
+
+    // Fork the chain
+    let mut side_chain = base_chain.fork();
+
+    // Create a side chain with multiple blocks only known to node0
+    for _ in 0..10 {
+        side_chain = node0.build_payload(side_chain).await;
+        node0.notify_new_payload(&side_chain).await;
+        node0.update_forkchoice(&side_chain).await;
+    }
+
+    // Sanity check: balance hasn't changed
+    let same_balance = node0.get_balance(recipient).await;
+    assert_eq!(same_balance, initial_balance);
+
+    // Advance the base chain with multiple blocks only known to node1
+    for _ in 0..10 {
+        base_chain = node1.build_payload(base_chain).await;
+        node1.notify_new_payload(&base_chain).await;
+        node1.update_forkchoice(&base_chain).await;
+    }
+
+    // Sanity check: balance hasn't changed
+    let same_balance = node0.get_balance(recipient).await;
+    assert_eq!(same_balance, initial_balance);
+
+    // Advance the side chain with one more block and an ETH transfer
+    node1
+        .send_eth_transfer(&signer, recipient, transfer_amount)
+        .await;
+    base_chain = node1.build_payload(base_chain).await;
+    node1.notify_new_payload(&base_chain).await;
+    node1.update_forkchoice(&base_chain).await;
+
+    // Bring node0 again to the base chain, it should reorg
+    node0.notify_new_payload(&base_chain).await;
+    node0.update_forkchoice(&base_chain).await;
+
+    // Check the transfer has been processed
+    let new_balance = node0.get_balance(recipient).await;
+    assert_eq!(new_balance, initial_balance + transfer_amount);
+}