feat: chain-replication testing (#225)

Author: thedevbirb

.gitignore

@@ -32,12 +32,12 @@ main_node/data
 follower_node/data
 follower_node/config
 follower_node/.env
-.local/
-.local_gateway_and_follower
-.local_main_node
-.local_main_node2
+
+.local*
+
 based/.envrc
 .envrc
+.env
 
 based/data
 based/genesis.json

based/Cargo.toml

@@ -31,7 +31,9 @@ opt-level = 3
 alloy-consensus = "0.14.0"
 alloy-eips = "0.14.0"
 alloy-network = "0.14.0"
-alloy-primitives = { version = "1.0.0", default-features = false, features = ["getrandom"] }
+alloy-primitives = { version = "1.0.0", default-features = false, features = [
+  "getrandom",
+] }
 alloy-provider = "0.14.0"
 alloy-rlp = "0.3.11"
 alloy-rpc-types = { version = "0.14.0", features = ["engine"] }
@@ -71,11 +73,15 @@ kona-rpc = { git = "https://github.com/op-rs/kona", rev = "88c80aa46975d196c711d
 mio = { features = ["net", "os-poll"], version = "1.0.4" }
 mio_httpc = { features = ["native"], version = "0.10.6" }
 moka = "0.12.10"
-op-alloy-consensus = { version = "=0.14.1", default-features = false, features = ["k256"] }
+op-alloy-consensus = { version = "=0.14.1", default-features = false, features = [
+  "k256",
+] }
 op-alloy-network = "0.14.1"
 op-alloy-rpc-types = "0.14.1"
 op-alloy-flz = { version = "0.13.0", default-features = false }
-op-alloy-rpc-types-engine = { version = "0.14.1", default-features = false, features = ["serde"] }
+op-alloy-rpc-types-engine = { version = "0.14.1", default-features = false, features = [
+  "serde",
+] }
 op-revm = "3.0.2"
 parking_lot = "0.12.3"
 paste = "0.1.18"
@@ -91,7 +97,9 @@ reth-db-api = { path = "../reth/crates/storage/db-api" }
 reth-db-common = { path = "../reth/crates/storage/db-common" }
 reth-evm = { path = "../reth/crates/evm" }
 reth-execution-errors = { path = "../reth/crates/evm/execution-errors" }
-reth-node-ethereum = { path = "../reth/crates/ethereum/node", features = ["test-utils"] }
+reth-node-ethereum = { path = "../reth/crates/ethereum/node", features = [
+  "test-utils",
+] }
 reth-node-types = { path = "../reth/crates/node/types" }
 reth-optimism-chainspec = { path = "../reth/crates/optimism/chainspec" }
 reth-optimism-cli = { path = "../reth/crates/optimism/cli" }
@@ -103,22 +111,32 @@ reth-optimism-primitives = { path = "../reth/crates/optimism/primitives" }
 reth-optimism-payload-builder = { path = "../reth/crates/optimism/payload" }
 reth-primitives = { path = "../reth/crates/primitives" }
 reth-primitives-traits = { path = "../reth/crates/primitives-traits" }
-reth-provider = { path = "../reth/crates/storage/provider", features = ["test-utils"] }
+reth-provider = { path = "../reth/crates/storage/provider", features = [
+  "test-utils",
+] }
 reth-revm = { path = "../reth/crates/revm" }
 reth-rpc-builder = { path = "../reth/crates/rpc/rpc-builder" }
 reth-rpc-layer = { path = "../reth/crates/rpc/rpc-layer" }
 reth-stages-types = { path = "../reth/crates/stages/types" }
 reth-storage-api = { path = "../reth/crates/storage/storage-api" }
 reth-storage-errors = { path = "../reth/crates/storage/errors" }
 reth-trie = { path = "../reth/crates/trie/trie" }
-reth-trie-common = { path = "../reth/crates/trie/common", features = ["test-utils"] }
+reth-trie-common = { path = "../reth/crates/trie/common", features = [
+  "test-utils",
+] }
 reth-trie-db = { path = "../reth/crates/trie/db" }
 reth-trie-parallel = { path = "../reth/crates/trie/parallel" }
-revm = { version = "22.0.1", features = ["optional_balance_check", "secp256k1", "std"], default-features = false }
+revm = { version = "22.0.1", features = [
+  "optional_balance_check",
+  "secp256k1",
+  "std",
+], default-features = false }
 revm-handler = "3.0.0"
 revm-inspector = "3.0.0"
 revm-interpreter = "18.0.0"
-revm-primitives = { version = "18.0.0", features = ["std"], default-features = false }
+revm-primitives = { version = "18.0.0", features = [
+  "std",
+], default-features = false }
 rustc-hash = "2.0.0"
 serde = { version = "1.0.217", features = ["derive"] }
 serde_json = "1.0.137"

based/bin/gateway/src/main.rs

@@ -10,7 +10,7 @@ use bop_common::{
     time::Duration,
     utils::{init_tracing, wait_for_signal},
 };
-use bop_db::{DatabaseRead, init_database};
+use bop_db::{DatabaseRead, DatabaseWrite as _, init_database};
 use bop_metrics::MetricsConsumer;
 use bop_rpc::{gossiper::Gossiper, start_rpc};
 use bop_sequencer::{
@@ -20,7 +20,7 @@ use bop_sequencer::{
 use clap::Parser;
 use revm_primitives::B256;
 use tokio::runtime::Runtime;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 fn main() {
     if std::env::var_os("RUST_BACKTRACE").is_none() {
@@ -67,6 +67,15 @@ fn run(args: GatewayArgs) -> eyre::Result<()> {
     let evm_config = sequencer_config.evm_config.clone();
     let (frag_broadcast_tx, _) = tokio::sync::broadcast::channel(10_000);
 
+    if let Some(ref range) = args.replay_blocks_range {
+        warn!(?range, "Replay range found. Performing DB rollback if necessary");
+        // Example: if start is 1000, then we should rollback to 998
+        while db_bop.head_block_number()? > range.start().saturating_sub(2) {
+            db_bop.roll_back_head()?;
+        }
+        info!(db_block = db_bop.head_block_number()?, db_hash = ?db_bop.head_block_hash()?, "New database head");
+    }
+
     let sequencer_vsync_window = Duration::from_micros(args.vsync_window_us as u64);
     let simulator_vsync_window = Duration::from_micros(args.vsync_window_us as u64);
 

based/crates/common/src/communication/messages.rs

@@ -166,6 +166,7 @@ pub enum EngineApi {
     NewPayloadV4 { payload: OpExecutionPayloadV4, versioned_hashes: Vec<B256>, parent_beacon_block_root: B256 },
     GetPayloadV4 { payload_id: PayloadId, res: oneshot::Sender<OpExecutionPayloadEnvelopeV4Patch> },
 }
+
 impl EngineApi {
     pub fn messages_from_block(
         block: &BlockSyncMessage,

based/crates/common/src/config.rs

@@ -1,4 +1,4 @@
-use std::{net::Ipv4Addr, path::PathBuf, sync::Arc, str::FromStr};
+use std::{net::Ipv4Addr, ops::RangeInclusive, path::PathBuf, str::FromStr, sync::Arc};
 
 use alloy_rpc_types::engine::JwtSecret;
 use clap::Parser;
@@ -40,9 +40,6 @@ pub struct GatewayArgs {
     pub rpc_port_ws: u16,
     #[arg(long = "rpc.jwt")]
     pub rpc_jwt: String,
-    /// PortalApi RPC URL
-    #[arg(long = "portal_rpc.url", default_value = "http://localhost:9998")]
-    pub portal_rpc_url: Url,
     /// Url to an L2 eth api rpc
     #[arg(long = "eth_client.url", default_value = "http://localhost:8545")]
     pub eth_client_url: Url,
@@ -77,6 +74,11 @@ pub struct GatewayArgs {
     /// Mock mode
     #[arg(long = "mock")]
     pub mock: Option<MockMode>,
+    /// For chain replication testing, it expresses the range of blocks we want to replay. In
+    /// particular, assuming the local eth client is at height N <= replay_target, the
+    /// chain-replication will happen for range N..=replay_target.
+    #[clap(long = "chain-replay.blocks-range", value_parser = range_inclusive_from_str)]
+    pub replay_blocks_range: Option<RangeInclusive<u64>>,
     /// Enable DEBUG logging
     #[arg(long = "debug")]
     pub debug: bool,
@@ -202,3 +204,22 @@ impl From<&GatewayArgs> for LoggingConfig {
         }
     }
 }
+
+fn range_inclusive_from_str(s: &str) -> Result<RangeInclusive<u64>, &'static str> {
+    let parts = s.split("..=").collect::<Vec<_>>();
+    if parts.len() != 2 {
+        return Err("Range must be in the format 'start..=end'");
+    }
+
+    let Ok(start) = parts[0].parse::<u64>() else {
+        return Err("Invalid start of range");
+    };
+    let Ok(end) = parts[1].parse::<u64>() else {
+        return Err("Invalid end of range");
+    };
+    if start > end {
+        return Err("Start of range must be less than or equal to end");
+    }
+
+    Ok(start..=end)
+}

based/crates/rpc/src/gossiper.rs

@@ -35,15 +35,16 @@ impl Gossiper {
         let signed = msg.to_signed(&self.signer);
         let payload = signed.to_json();
 
-        if matches!(signed.message, p2p::VersionedMessage::FragV0(_)) {
-            if let Err(e) = self.frag_broadcast.send(signed) {
-                tracing::debug!(" broadcast of frag failed {e}")
-            }
+        if matches!(signed.message, p2p::VersionedMessage::FragV0(_)) && self.frag_broadcast.send(signed).is_err() {
+            tracing::debug!("broadcast of frag failed")
         }
 
-        let Ok(res) = self.client.post(self.target_rpc.clone()).json(&payload).send() else {
-            tracing::error!("couldn't send {}", payload);
-            return;
+        let res = match self.client.post(self.target_rpc.clone()).json(&payload).send() {
+            Ok(res) => res,
+            Err(e) => {
+                tracing::error!(?e, target = ?self.target_rpc.clone(), "couldn't send payload: {}", payload);
+                return;
+            }
         };
 
         let code = res.status();

based/crates/sequencer/src/lib.rs

@@ -10,7 +10,7 @@ use bop_common::{
     actor::Actor,
     communication::{
         Connections, ReceiversSpine, SendersSpine, SpineConnections, TrackedSenders,
-        messages::{self, BlockFetch, BlockSyncError, EngineApi, SimulatorToSequencer, SimulatorToSequencerMsg},
+        messages::{BlockFetch, BlockSyncError, EngineApi, SimulatorToSequencer, SimulatorToSequencerMsg},
     },
     custom_v4::OpExecutionPayloadEnvelopeV4Patch,
     db::DatabaseWrite,
@@ -133,7 +133,7 @@ where
         });
 
         // handle engine API messages from rpc
-        connections.receive_for(Duration::from_millis(10), |msg: messages::EngineApi, senders| {
+        connections.receive_for(Duration::from_millis(10), |msg, senders| {
             let state = std::mem::take(&mut self.state);
             let cur_seq_state = telemetry::system::SequencerState::from(&state);
             self.state = state.handle_engine_api(msg, &mut self.data, senders);

chain-replay/.env.example

@@ -0,0 +1,47 @@
+BOP_REPLAY_BLOCKS_RANGE=2000..=2500
+BOP_REPLAY_CHAIN_NAME=based-op-sepolia
+BOP_REPLAY_FOLLOWER_NODES_COUNT=2
+
+BOP_REPLAY_GATEWAY_PORT=29997
+BOP_REPLAY_GATEWAY_AUTH_URL=http://127.0.0.1:29997
+BOP_REPLAY_GATEWAY_AUTH_JWT=bfea64567a978df0bcd973699554671aa20b20820bc40ba45dfed46025e2fa10
+BOP_REPLAY_GATEWAY_URL=http://127.0.0.1:29998
+BOP_REPLAY_GATEWAY_PORT_NO_AUTH=29998
+BOP_REPLAY_GATEWAY_SEQUENCING_KEY=42241483bfabbedc7c9d0d466c22c0820c40c886ddbe5b10a60308a032baac58
+
+BOP_REPLAY_KONA_SEQUENCER_GOSSIP_PORT=22303
+BOP_REPLAY_KONA_SEQUENCER_DISCOVERY_PORT=22304
+BOP_REPLAY_P2P_SEQUENCER_KEY=0x377d4db65cb871eeca771ce5c9db54e90e7413234d2f094acc13ac2d428dbf35
+BOP_REPLAY_ROLLUP_FILE_PATH=../../.local_gateway_and_follower_based-op-sepolia/config/rollup.json
+
+BOP_REPLAY_L1_BEACON_RPC_URL=http://34.194.193.217:5052
+BOP_REPLAY_L1_RPC_URL=http://34.194.193.217:8545
+BOP_REPLAY_L2_EL_VERIFIER_URL=http://18.185.199.51:8545
+
+BOP_REPLAY_BASED_OP_GETH_0_ENGINE_RPC_PORT=28651
+BOP_REPLAY_BASED_OP_GETH_0_GOSSIP_PORT=21303
+BOP_REPLAY_BASED_OP_GETH_0_RPC_PORT=28645
+BOP_REPLAY_BASED_OP_GETH_0_WS_PORT=28646
+BOP_REPLAY_BASED_OP_GETH_0_METRICS_PORT=28010
+BOP_REPLAY_BASED_OP_NODE_0_GOSSIP_PORT=29103
+BOP_REPLAY_BASED_OP_NODE_0_RPC_PORT=28547
+BOP_REPLAY_BASED_OP_NODE_0_METRICS_PORT=28011
+
+BOP_REPLAY_BASED_OP_GETH_1_ENGINE_RPC_PORT=28751
+BOP_REPLAY_BASED_OP_GETH_1_GOSSIP_PORT=21403
+BOP_REPLAY_BASED_OP_GETH_1_RPC_PORT=28745
+BOP_REPLAY_BASED_OP_GETH_1_WS_PORT=28746
+BOP_REPLAY_BASED_OP_GETH_1_METRICS_PORT=28110
+BOP_REPLAY_BASED_OP_NODE_1_GOSSIP_PORT=29203
+BOP_REPLAY_BASED_OP_NODE_1_RPC_PORT=28647
+BOP_REPLAY_BASED_OP_NODE_1_METRICS_PORT=28111
+
+BOP_REPLAY_METRICS_EXPORTER_PORT=29000
+BOP_REPLAY_PROMETHEUS_PORT=29090
+BOP_REPLAY_GRAFANA_PORT=23000
+
+BOP_REPLAY_REGISTRY_PORT=28082
+BOP_REPLAY_PORTAL=http://127.0.0.1:28081
+
+BOP_REPLAY_RUST_LOG=bop_common=debug,bop_db=debug,bop_metrics=debug,bop_pool=debug,bop_rpc=debug,bop_sequencer=debug,bop_gateway=debug,bop_metrics_exporter=debug,gateway_registry=debug
+RUST_LOG=info,chain_replay=debug,discv5::service=error