feat: add UX improvements on e2e testing (#7804)

Cargo.lock

@@ -6176,6 +6176,7 @@ dependencies = [
  "reth-provider",
  "reth-prune",
  "reth-revm",
+ "reth-rpc",
  "reth-rpc-types",
  "reth-rpc-types-compat",
  "reth-stages",
@@ -6451,16 +6452,19 @@ dependencies = [
  "rand 0.8.5",
  "reth",
  "reth-db",
+ "reth-node-builder",
  "reth-node-core",
  "reth-node-ethereum",
  "reth-payload-builder",
  "reth-primitives",
+ "reth-provider",
  "reth-rpc",
  "reth-tracing",
  "secp256k1 0.27.0",
  "serde_json",
  "tokio",
  "tokio-stream",
+ "tracing",
 ]
 
 [[package]]
@@ -7039,6 +7043,7 @@ dependencies = [
  "reqwest 0.11.27",
  "reth",
  "reth-basic-payload-builder",
+ "reth-beacon-consensus",
  "reth-db",
  "reth-e2e-test-utils",
  "reth-evm",

crates/consensus/beacon/Cargo.toml

@@ -52,6 +52,7 @@ reth-blockchain-tree = { workspace = true, features = ["test-utils"] }
 reth-db = { workspace = true, features = ["test-utils"] }
 reth-provider = { workspace = true, features = ["test-utils"] }
 reth-rpc-types-compat.workspace = true
+reth-rpc.workspace = true
 reth-tracing.workspace = true
 reth-revm.workspace = true
 reth-downloaders.workspace = true
@@ -68,4 +69,6 @@ optimism = [
     "reth-provider/optimism",
     "reth-blockchain-tree/optimism",
     "reth-beacon-consensus-core/optimism",
+    "reth-revm/optimism",
+    "reth-rpc/optimism"
 ]

crates/e2e-test-utils/Cargo.toml

@@ -17,6 +17,8 @@ reth-tracing.workspace = true
 reth-db.workspace = true
 reth-rpc.workspace = true
 reth-payload-builder = { workspace = true, features = ["test-utils"] }
+reth-provider.workspace = true
+reth-node-builder.workspace = true
 
 jsonrpsee.workspace = true
 
@@ -32,3 +34,4 @@ alloy-signer-wallet = { workspace = true, features = ["mnemonic"] }
 alloy-rpc-types.workspace = true
 alloy-network.workspace = true
 alloy-consensus.workspace = true
+tracing.workspace = true

crates/e2e-test-utils/src/engine_api.rs

@@ -3,7 +3,10 @@ use jsonrpsee::http_client::HttpClient;
 use reth::{
     api::{EngineTypes, PayloadBuilderAttributes},
     providers::CanonStateNotificationStream,
-    rpc::{api::EngineApiClient, types::engine::ForkchoiceState},
+    rpc::{
+        api::EngineApiClient,
+        types::engine::{ForkchoiceState, PayloadStatusEnum},
+    },
 };
 use reth_payload_builder::PayloadId;
 use reth_primitives::B256;
@@ -30,6 +33,7 @@ impl<E: EngineTypes + 'static> EngineApiHelper<E> {
         &self,
         payload: E::BuiltPayload,
         payload_builder_attributes: E::PayloadBuilderAttributes,
+        expected_status: PayloadStatusEnum,
     ) -> eyre::Result<B256>
     where
         E::ExecutionPayloadV3: From<E::BuiltPayload> + PayloadEnvelopeExt,
@@ -45,8 +49,10 @@ impl<E: EngineTypes + 'static> EngineApiHelper<E> {
             payload_builder_attributes.parent_beacon_block_root().unwrap(),
         )
         .await?;
-        assert!(submission.is_valid(), "{}", submission);
-        Ok(submission.latest_valid_hash.unwrap())
+
+        assert!(submission.status == expected_status);
+
+        Ok(submission.latest_valid_hash.unwrap_or_default())
     }
 
     /// Sends forkchoice update to the engine api
@@ -64,4 +70,20 @@ impl<E: EngineTypes + 'static> EngineApiHelper<E> {
 
         Ok(())
     }
+
+    /// Sends forkchoice update to the engine api with a zero finalized hash
+    pub async fn update_optimistic_forkchoice(&self, hash: B256) -> eyre::Result<()> {
+        EngineApiClient::<E>::fork_choice_updated_v2(
+            &self.engine_api_client,
+            ForkchoiceState {
+                head_block_hash: hash,
+                safe_block_hash: B256::ZERO,
+                finalized_block_hash: B256::ZERO,
+            },
+            None,
+        )
+        .await?;
+
+        Ok(())
+    }
 }

crates/e2e-test-utils/src/lib.rs

@@ -1,3 +1,22 @@
+use node::NodeHelper;
+use reth::{
+    args::{DiscoveryArgs, NetworkArgs, RpcServerArgs},
+    blockchain_tree::ShareableBlockchainTree,
+    builder::{NodeBuilder, NodeConfig, NodeHandle},
+    revm::EvmProcessorFactory,
+    tasks::TaskManager,
+};
+use reth_db::{test_utils::TempDatabase, DatabaseEnv};
+use reth_node_builder::{
+    components::{NetworkBuilder, PayloadServiceBuilder, PoolBuilder},
+    FullNodeComponentsAdapter, FullNodeTypesAdapter, NodeTypes,
+};
+use reth_primitives::ChainSpec;
+use reth_provider::providers::BlockchainProvider;
+use std::sync::Arc;
+use tracing::{span, Level};
+use wallet::Wallet;
+
 /// Wrapper type to create test nodes
 pub mod node;
 
@@ -15,3 +34,78 @@ mod engine_api;
 
 /// Helper traits
 mod traits;
+
+/// Creates the initial setup with `num_nodes` started and interconnected.
+pub async fn setup<N>(
+    num_nodes: usize,
+    chain_spec: Arc<ChainSpec>,
+    is_dev: bool,
+) -> eyre::Result<(Vec<NodeHelperType<N>>, TaskManager, Wallet)>
+where
+    N: Default + reth_node_builder::Node<TmpNodeAdapter<N>>,
+    N::PoolBuilder: PoolBuilder<TmpNodeAdapter<N>>,
+    N::NetworkBuilder: NetworkBuilder<TmpNodeAdapter<N>, TmpPool<N>>,
+    N::PayloadBuilder: PayloadServiceBuilder<TmpNodeAdapter<N>, TmpPool<N>>,
+{
+    let tasks = TaskManager::current();
+    let exec = tasks.executor();
+
+    let network_config = NetworkArgs {
+        discovery: DiscoveryArgs { disable_discovery: true, ..DiscoveryArgs::default() },
+        ..NetworkArgs::default()
+    };
+
+    // Create nodes and peer them
+    let mut nodes: Vec<NodeHelperType<N>> = Vec::with_capacity(num_nodes);
+
+    for idx in 0..num_nodes {
+        let mut node_config = NodeConfig::test()
+            .with_chain(chain_spec.clone())
+            .with_network(network_config.clone())
+            .with_unused_ports()
+            .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
+
+        if is_dev {
+            node_config = node_config.dev();
+        }
+
+        let span = span!(Level::INFO, "node", idx);
+        let _enter = span.enter();
+        let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
+            .testing_node(exec.clone())
+            .node(Default::default())
+            .launch()
+            .await?;
+
+        let mut node = NodeHelper::new(node).await?;
+
+        // Connect each node in a chain.
+        if let Some(previous_node) = nodes.last_mut() {
+            previous_node.connect(&mut node).await;
+        }
+
+        // Connect last node with the first if there are more than two
+        if idx + 1 == num_nodes && num_nodes > 2 {
+            if let Some(first_node) = nodes.first_mut() {
+                node.connect(first_node).await;
+            }
+        }
+
+        nodes.push(node);
+    }
+
+    Ok((nodes, tasks, Wallet::default().with_chain_id(chain_spec.chain().into())))
+}
+
+// Type aliases
+
+type TmpDB = Arc<TempDatabase<DatabaseEnv>>;
+type EvmType<N> = EvmProcessorFactory<<N as NodeTypes>::Evm>;
+type RethProvider<N> = BlockchainProvider<TmpDB, ShareableBlockchainTree<TmpDB, EvmType<N>>>;
+type TmpPool<N> = <<N as reth_node_builder::Node<TmpNodeAdapter<N>>>::PoolBuilder as PoolBuilder<
+    TmpNodeAdapter<N>,
+>>::Pool;
+type TmpNodeAdapter<N> = FullNodeTypesAdapter<N, TmpDB, RethProvider<N>>;
+
+/// Type alias for a type of NodeHelper
+pub type NodeHelperType<N> = NodeHelper<FullNodeComponentsAdapter<TmpNodeAdapter<N>, TmpPool<N>>>;

crates/e2e-test-utils/src/node.rs

@@ -4,21 +4,18 @@ use crate::{
 };
 use alloy_rpc_types::BlockNumberOrTag;
 use eyre::Ok;
+use futures_util::Future;
 use reth::{
     api::{BuiltPayload, EngineTypes, FullNodeComponents, PayloadBuilderAttributes},
     builder::FullNode,
-    providers::{BlockReaderIdExt, CanonStateSubscriptions},
+    providers::{BlockReader, BlockReaderIdExt, CanonStateSubscriptions, StageCheckpointReader},
     rpc::{
         eth::{error::EthResult, EthTransactions},
-        types::engine::PayloadAttributes,
+        types::engine::PayloadStatusEnum,
     },
 };
-use reth_payload_builder::EthPayloadBuilderAttributes;
-use reth_primitives::{Address, BlockNumber, Bytes, B256};
-use std::{
-    marker::PhantomData,
-    time::{SystemTime, UNIX_EPOCH},
-};
+use reth_primitives::{stage::StageId, BlockHash, BlockNumber, Bytes, B256};
+use std::{marker::PhantomData, pin::Pin};
 use tokio_stream::StreamExt;
 
 /// An helper struct to handle node actions
@@ -27,7 +24,7 @@ where
     Node: FullNodeComponents,
 {
     pub inner: FullNode<Node>,
-    payload: PayloadHelper<Node::Engine>,
+    pub payload: PayloadHelper<Node::Engine>,
     pub network: NetworkHelper,
     pub engine_api: EngineApiHelper<Node::Engine>,
 }
@@ -52,12 +49,53 @@ where
         })
     }
 
-    /// Advances the node forward
+    pub async fn connect(&mut self, node: &mut NodeHelper<Node>) {
+        self.network.add_peer(node.network.record()).await;
+        node.network.add_peer(self.network.record()).await;
+        node.network.expect_session().await;
+        self.network.expect_session().await;
+    }
+
+    /// Advances the chain `length` blocks.
+    ///
+    /// Returns the added chain as a Vec of block hashes.
     pub async fn advance(
+        &mut self,
+        length: u64,
+        tx_generator: impl Fn() -> Pin<Box<dyn Future<Output = Bytes>>>,
+        attributes_generator: impl Fn(u64) -> <Node::Engine as EngineTypes>::PayloadBuilderAttributes
+            + Copy,
+    ) -> eyre::Result<
+        Vec<(
+            <Node::Engine as EngineTypes>::BuiltPayload,
+            <Node::Engine as EngineTypes>::PayloadBuilderAttributes,
+        )>,
+    >
+    where
+        <Node::Engine as EngineTypes>::ExecutionPayloadV3:
+            From<<Node::Engine as EngineTypes>::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        let mut chain = Vec::with_capacity(length as usize);
+        for _ in 0..length {
+            let (payload, _) =
+                self.advance_block(tx_generator().await, attributes_generator).await?;
+            chain.push(payload);
+        }
+        Ok(chain)
+    }
+
+    /// Advances the node forward one block
+    pub async fn advance_block(
         &mut self,
         raw_tx: Bytes,
         attributes_generator: impl Fn(u64) -> <Node::Engine as EngineTypes>::PayloadBuilderAttributes,
-    ) -> eyre::Result<(B256, B256)>
+    ) -> eyre::Result<(
+        (
+            <Node::Engine as EngineTypes>::BuiltPayload,
+            <Node::Engine as EngineTypes>::PayloadBuilderAttributes,
+        ),
+        B256,
+    )>
     where
         <Node::Engine as EngineTypes>::ExecutionPayloadV3:
             From<<Node::Engine as EngineTypes>::BuiltPayload> + PayloadEnvelopeExt,
@@ -81,15 +119,54 @@ where
         let payload = self.payload.expect_built_payload().await?;
 
         // submit payload via engine api
-        let block_number = payload.block().number;
-        let block_hash = self.engine_api.submit_payload(payload, eth_attr.clone()).await?;
+        let block_hash = self
+            .engine_api
+            .submit_payload(payload.clone(), eth_attr.clone(), PayloadStatusEnum::Valid)
+            .await?;
 
         // trigger forkchoice update via engine api to commit the block to the blockchain
         self.engine_api.update_forkchoice(block_hash).await?;
 
         // assert the block has been committed to the blockchain
-        self.assert_new_block(tx_hash, block_hash, block_number).await?;
-        Ok((block_hash, tx_hash))
+        self.assert_new_block(tx_hash, block_hash, payload.block().number).await?;
+        Ok(((payload, eth_attr), tx_hash))
+    }
+
+    /// Waits for block to be available on node.
+    pub async fn wait_block(
+        &self,
+        number: BlockNumber,
+        expected_block_hash: BlockHash,
+        wait_finish_checkpoint: bool,
+    ) -> eyre::Result<()> {
+        let mut check = !wait_finish_checkpoint;
+        loop {
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+
+            if !check && wait_finish_checkpoint {
+                if let Some(checkpoint) =
+                    self.inner.provider.get_stage_checkpoint(StageId::Finish)?
+                {
+                    if checkpoint.block_number >= number {
+                        check = true
+                    }
+                }
+            }
+
+            if check {
+                if let Some(latest_block) = self.inner.provider.block_by_number(number)? {
+                    if latest_block.hash_slow() != expected_block_hash {
+                        // TODO: only if its awaiting a reorg
+                        continue
+                    }
+                    break
+                }
+                if wait_finish_checkpoint {
+                    panic!("Finish checkpoint matches, but could not fetch block.");
+                }
+            }
+        }
+        Ok(())
     }
 
     /// Injects a raw transaction into the node tx pool via RPC server
@@ -129,17 +206,3 @@ where
         Ok(())
     }
 }
-
-/// Helper function to create a new eth payload attributes
-pub fn eth_payload_attributes() -> EthPayloadBuilderAttributes {
-    let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
-
-    let attributes = PayloadAttributes {
-        timestamp,
-        prev_randao: B256::ZERO,
-        suggested_fee_recipient: Address::ZERO,
-        withdrawals: Some(vec![]),
-        parent_beacon_block_root: Some(B256::ZERO),
-    };
-    EthPayloadBuilderAttributes::new(B256::ZERO, attributes)
-}

crates/e2e-test-utils/src/payload.rs

@@ -7,7 +7,7 @@ use tokio_stream::wrappers::BroadcastStream;
 pub struct PayloadHelper<E: EngineTypes + 'static> {
     pub payload_event_stream: BroadcastStream<Events<E>>,
     payload_builder: PayloadBuilderHandle<E>,
-    timestamp: u64,
+    pub timestamp: u64,
 }
 
 impl<E: EngineTypes + 'static> PayloadHelper<E> {

crates/e2e-test-utils/src/wallet.rs

@@ -1,11 +1,11 @@
 use alloy_network::{eip2718::Encodable2718, EthereumSigner, TransactionBuilder};
 use alloy_rpc_types::{TransactionInput, TransactionRequest};
 use alloy_signer_wallet::{coins_bip39::English, LocalWallet, MnemonicBuilder};
-use reth_primitives::{Address, Bytes, U256};
+use reth_primitives::{hex, Address, Bytes, U256};
 /// One of the accounts of the genesis allocations.
 pub struct Wallet {
     inner: LocalWallet,
-    nonce: u64,
+    pub nonce: u64,
     chain_id: u64,
 }
 
@@ -27,6 +27,11 @@ impl Wallet {
         self.tx(None).await
     }
 
+    pub async fn optimism_l1_block_info_tx(&mut self) -> Bytes {
+        let l1_block_info = Bytes::from_static(&hex!("7ef9015aa044bae9d41b8380d781187b426c6fe43df5fb2fb57bd4466ef6a701e1f01e015694deaddeaddeaddeaddeaddeaddeaddeaddead000194420000000000000000000000000000000000001580808408f0d18001b90104015d8eb900000000000000000000000000000000000000000000000000000000008057650000000000000000000000000000000000000000000000000000000063d96d10000000000000000000000000000000000000000000000000000000000009f35273d89754a1e0387b89520d989d3be9c37c1f32495a88faf1ea05c61121ab0d1900000000000000000000000000000000000000000000000000000000000000010000000000000000000000002d679b567db6187c0c8323fa982cfb88b74dbcc7000000000000000000000000000000000000000000000000000000000000083400000000000000000000000000000000000000000000000000000000000f4240"));
+        self.tx(Some(l1_block_info)).await
+    }
+
     /// Creates a transaction with data and signs it
     pub async fn tx(&mut self, data: Option<Bytes>) -> Bytes {
         let tx = TransactionRequest {

crates/node-ethereum/tests/e2e/dev.rs

@@ -1,43 +1,27 @@
-use futures_util::StreamExt;
-use reth::{
-    api::FullNodeComponents,
-    builder::{FullNode, NodeBuilder, NodeHandle},
-    providers::CanonStateSubscriptions,
-    rpc::eth::EthTransactions,
-    tasks::TaskManager,
-};
-use reth_node_core::{args::RpcServerArgs, node_config::NodeConfig};
-use reth_node_ethereum::EthereumNode;
+use crate::utils::EthNode;
+use futures::StreamExt;
+use reth::rpc::eth::EthTransactions;
+use reth_e2e_test_utils::setup;
 use reth_primitives::{b256, hex, ChainSpec, Genesis};
+use reth_provider::CanonStateSubscriptions;
 use std::sync::Arc;
 
 #[tokio::test]
 async fn can_run_dev_node() -> eyre::Result<()> {
-    let tasks = TaskManager::current();
+    reth_tracing::init_test_tracing();
+    let (mut nodes, _tasks, _) = setup(1, custom_chain(), true).await?;
 
-    // create node config
-    let node_config = NodeConfig::test()
-        .dev()
-        .with_rpc(RpcServerArgs::default().with_http().with_unused_ports())
-        .with_chain(custom_chain());
-
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config)
-        .testing_node(tasks.executor())
-        .node(EthereumNode::default())
-        .launch()
-        .await?;
-
-    assert_chain_advances(node).await;
+    assert_chain_advances(nodes.pop().unwrap()).await;
     Ok(())
 }
 
-async fn assert_chain_advances<Node: FullNodeComponents>(mut node: FullNode<Node>) {
-    let mut notifications = node.provider.canonical_state_stream();
+async fn assert_chain_advances(mut node: EthNode) {
+    let mut notifications = node.inner.provider.canonical_state_stream();
 
     // submit tx through rpc
     let raw_tx = hex!("02f876820a28808477359400847735940082520894ab0840c0e43688012c1adb0f5e3fc665188f83d28a029d394a5d630544000080c080a0a044076b7e67b5deecc63f61a8d7913fab86ca365b344b5759d1fe3563b4c39ea019eab979dd000da04dfc72bb0377c092d30fd9e1cab5ae487de49586cc8b0090");
 
-    let eth_api = node.rpc_registry.eth_api();
+    let eth_api = node.inner.rpc_registry.eth_api();
 
     let hash = eth_api.send_raw_transaction(raw_tx.into()).await.unwrap();
 

crates/node-ethereum/tests/e2e/eth.rs

@@ -4,7 +4,7 @@ use reth::{
     builder::{NodeBuilder, NodeConfig, NodeHandle},
     tasks::TaskManager,
 };
-use reth_e2e_test_utils::{node::NodeHelper, wallet::Wallet};
+use reth_e2e_test_utils::{node::NodeHelper, setup, wallet::Wallet};
 use reth_node_ethereum::EthereumNode;
 use reth_primitives::{ChainSpecBuilder, Genesis, MAINNET};
 use std::sync::Arc;
@@ -13,38 +13,24 @@ use std::sync::Arc;
 async fn can_run_eth_node() -> eyre::Result<()> {
     reth_tracing::init_test_tracing();
 
-    let exec = TaskManager::current();
-    let exec = exec.executor();
+    let (mut nodes, _tasks, mut wallet) = setup::<EthereumNode>(
+        1,
+        Arc::new(
+            ChainSpecBuilder::default()
+                .chain(MAINNET.chain)
+                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
+                .cancun_activated()
+                .build(),
+        ),
+        false,
+    )
+    .await?;
 
-    // Chain spec with test allocs
-    let genesis: Genesis = serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
-    let chain_spec = Arc::new(
-        ChainSpecBuilder::default()
-            .chain(MAINNET.chain)
-            .genesis(genesis)
-            .cancun_activated()
-            .build(),
-    );
-
-    // Node setup
-    let node_config = NodeConfig::test()
-        .with_chain(chain_spec)
-        .with_unused_ports()
-        .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
-
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config)
-        .testing_node(exec)
-        .node(EthereumNode::default())
-        .launch()
-        .await?;
-    let mut node = NodeHelper::new(node).await?;
-
-    // Configure wallet from test mnemonic and create dummy transfer tx
-    let mut wallet = Wallet::default();
+    let mut node = nodes.pop().unwrap();
     let raw_tx = wallet.transfer_tx().await;
 
     // make the node advance
-    node.advance(raw_tx, eth_payload_attributes).await?;
+    node.advance_block(raw_tx, eth_payload_attributes).await?;
 
     Ok(())
 }
@@ -83,7 +69,7 @@ async fn can_run_eth_node_with_auth_engine_api_over_ipc() -> eyre::Result<()> {
     let raw_tx = wallet.transfer_tx().await;
 
     // make the node advance
-    node.advance(raw_tx, crate::utils::eth_payload_attributes).await?;
+    node.advance_block(raw_tx, crate::utils::eth_payload_attributes).await?;
 
     Ok(())
 }

crates/node-ethereum/tests/e2e/p2p.rs

@@ -1,71 +1,34 @@
-use std::sync::Arc;
-
 use crate::utils::eth_payload_attributes;
-use reth::{
-    args::{DiscoveryArgs, NetworkArgs, RpcServerArgs},
-    builder::{NodeBuilder, NodeConfig, NodeHandle},
-    tasks::TaskManager,
-};
-use reth_e2e_test_utils::{node::NodeHelper, wallet::Wallet};
+use reth_e2e_test_utils::setup;
 use reth_node_ethereum::EthereumNode;
-use reth_primitives::{ChainSpecBuilder, Genesis, MAINNET};
+use reth_primitives::{ChainSpecBuilder, MAINNET};
+use std::sync::Arc;
 
 #[tokio::test]
 async fn can_sync() -> eyre::Result<()> {
     reth_tracing::init_test_tracing();
 
-    let tasks = TaskManager::current();
-    let exec = tasks.executor();
+    let (mut nodes, _tasks, mut wallet) = setup::<EthereumNode>(
+        2,
+        Arc::new(
+            ChainSpecBuilder::default()
+                .chain(MAINNET.chain)
+                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
+                .cancun_activated()
+                .build(),
+        ),
+        false,
+    )
+    .await?;
 
-    let genesis: Genesis = serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
-    let chain_spec = Arc::new(
-        ChainSpecBuilder::default()
-            .chain(MAINNET.chain)
-            .genesis(genesis)
-            .cancun_activated()
-            .build(),
-    );
-
-    let network_config = NetworkArgs {
-        discovery: DiscoveryArgs { disable_discovery: true, ..DiscoveryArgs::default() },
-        ..NetworkArgs::default()
-    };
-
-    let node_config = NodeConfig::test()
-        .with_chain(chain_spec)
-        .with_network(network_config)
-        .with_unused_ports()
-        .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
-
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
-        .testing_node(exec.clone())
-        .node(EthereumNode::default())
-        .launch()
-        .await?;
-
-    let mut first_node = NodeHelper::new(node.clone()).await?;
-
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config)
-        .testing_node(exec)
-        .node(EthereumNode::default())
-        .launch()
-        .await?;
-
-    let mut second_node = NodeHelper::new(node).await?;
-
-    let mut wallet = Wallet::default();
     let raw_tx = wallet.transfer_tx().await;
-
-    // Make them peer
-    first_node.network.add_peer(second_node.network.record()).await;
-    second_node.network.add_peer(first_node.network.record()).await;
-
-    // Make sure they establish a new session
-    first_node.network.expect_session().await;
-    second_node.network.expect_session().await;
+    let mut second_node = nodes.pop().unwrap();
+    let mut first_node = nodes.pop().unwrap();
 
     // Make the first node advance
-    let (block_hash, tx_hash) = first_node.advance(raw_tx.clone(), eth_payload_attributes).await?;
+    let ((payload, _), tx_hash) =
+        first_node.advance_block(raw_tx.clone(), eth_payload_attributes).await?;
+    let block_hash = payload.block().hash();
 
     // only send forkchoice update to second node
     second_node.engine_api.update_forkchoice(block_hash).await?;

crates/node-ethereum/tests/e2e/utils.rs

@@ -1,7 +1,12 @@
 use reth::rpc::types::engine::PayloadAttributes;
+use reth_e2e_test_utils::NodeHelperType;
+use reth_node_ethereum::EthereumNode;
 use reth_payload_builder::EthPayloadBuilderAttributes;
 use reth_primitives::{Address, B256};
 
+/// Ethereum Node Helper type
+pub(crate) type EthNode = NodeHelperType<EthereumNode>;
+
 /// Helper function to create a new eth payload attributes
 pub(crate) fn eth_payload_attributes(timestamp: u64) -> EthPayloadBuilderAttributes {
     let attributes = PayloadAttributes {

crates/optimism/node/Cargo.toml

@@ -28,7 +28,7 @@ reth-network.workspace = true
 reth-interfaces.workspace = true
 reth-evm.workspace = true
 reth-revm.workspace = true
-
+reth-beacon-consensus.workspace = true
 revm.workspace = true
 revm-primitives.workspace = true
 
@@ -67,4 +67,5 @@ optimism = [
     "reth-rpc/optimism",
     "reth-revm/optimism",
     "reth-optimism-payload-builder/optimism",
+    "reth-beacon-consensus/optimism",
 ]

crates/optimism/node/tests/e2e/p2p.rs

@@ -1,80 +1,28 @@
+use crate::utils::{advance_chain, setup};
 use std::sync::Arc;
-
-use crate::utils::optimism_payload_attributes;
-use reth::{
-    args::{DiscoveryArgs, NetworkArgs, RpcServerArgs},
-    builder::{NodeBuilder, NodeConfig, NodeHandle},
-    tasks::TaskManager,
-};
-use reth_e2e_test_utils::{node::NodeHelper, wallet::Wallet};
-use reth_node_optimism::node::OptimismNode;
-use reth_primitives::{hex, Bytes, ChainSpecBuilder, Genesis, BASE_MAINNET};
+use tokio::sync::Mutex;
 
 #[tokio::test]
 async fn can_sync() -> eyre::Result<()> {
     reth_tracing::init_test_tracing();
 
-    let tasks = TaskManager::current();
-    let exec = tasks.executor();
+    let (mut nodes, _tasks, wallet) = setup(2).await?;
+    let wallet = Arc::new(Mutex::new(wallet));
 
-    let genesis: Genesis = serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
-    let chain_spec = Arc::new(
-        ChainSpecBuilder::default()
-            .chain(BASE_MAINNET.chain)
-            .genesis(genesis)
-            .ecotone_activated()
-            .build(),
-    );
-    let mut wallet = Wallet::default().with_chain_id(chain_spec.chain.into());
+    let second_node = nodes.pop().unwrap();
+    let mut first_node = nodes.pop().unwrap();
 
-    let network_config = NetworkArgs {
-        discovery: DiscoveryArgs { disable_discovery: true, ..DiscoveryArgs::default() },
-        ..NetworkArgs::default()
-    };
+    let tip: usize = 300;
+    let tip_index: usize = tip - 1;
 
-    let node_config = NodeConfig::test()
-        .with_chain(chain_spec)
-        .with_network(network_config)
-        .with_unused_ports()
-        .with_rpc(RpcServerArgs::default().with_unused_ports().with_http());
+    // On first node, create a chain up to block number 300a
+    let canonical_payload_chain = advance_chain(tip, &mut first_node, wallet.clone()).await?;
+    let canonical_chain =
+        canonical_payload_chain.iter().map(|p| p.0.block().hash()).collect::<Vec<_>>();
 
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
-        .testing_node(exec.clone())
-        .node(OptimismNode::default())
-        .launch()
-        .await?;
-
-    let mut first_node = NodeHelper::new(node.clone()).await?;
-
-    let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config)
-        .testing_node(exec)
-        .node(OptimismNode::default())
-        .launch()
-        .await?;
-
-    let mut second_node = NodeHelper::new(node).await?;
-
-    // Make them peer
-    first_node.network.add_peer(second_node.network.record()).await;
-    second_node.network.add_peer(first_node.network.record()).await;
-
-    // Make sure they establish a new session
-    first_node.network.expect_session().await;
-    second_node.network.expect_session().await;
-
-    // Taken from optimism tests
-    let l1_block_info = Bytes::from_static(&hex!("7ef9015aa044bae9d41b8380d781187b426c6fe43df5fb2fb57bd4466ef6a701e1f01e015694deaddeaddeaddeaddeaddeaddeaddeaddead000194420000000000000000000000000000000000001580808408f0d18001b90104015d8eb900000000000000000000000000000000000000000000000000000000008057650000000000000000000000000000000000000000000000000000000063d96d10000000000000000000000000000000000000000000000000000000000009f35273d89754a1e0387b89520d989d3be9c37c1f32495a88faf1ea05c61121ab0d1900000000000000000000000000000000000000000000000000000000000000010000000000000000000000002d679b567db6187c0c8323fa982cfb88b74dbcc7000000000000000000000000000000000000000000000000000000000000083400000000000000000000000000000000000000000000000000000000000f4240"));
-
-    // Make the first node advance
-    let raw_tx = wallet.tx(Some(l1_block_info)).await;
-    let (block_hash, tx_hash) =
-        first_node.advance(raw_tx.clone(), optimism_payload_attributes).await?;
-
-    // only send forkchoice update to second node
-    second_node.engine_api.update_forkchoice(block_hash).await?;
-
-    // expect second node advanced via p2p gossip
-    second_node.assert_new_block(tx_hash, block_hash, 1).await?;
+    // On second node, sync up to block number 300a
+    second_node.engine_api.update_forkchoice(canonical_chain[tip_index]).await?;
+    second_node.wait_block(tip as u64, canonical_chain[tip_index], true).await?;
 
     Ok(())
 }

crates/optimism/node/tests/e2e/utils.rs

@@ -1,7 +1,45 @@
-use reth::rpc::types::engine::PayloadAttributes;
-use reth_node_optimism::OptimismPayloadBuilderAttributes;
+use reth::{rpc::types::engine::PayloadAttributes, tasks::TaskManager};
+use reth_e2e_test_utils::{wallet::Wallet, NodeHelperType};
+use reth_node_optimism::{OptimismBuiltPayload, OptimismNode, OptimismPayloadBuilderAttributes};
 use reth_payload_builder::EthPayloadBuilderAttributes;
-use reth_primitives::{Address, B256};
+use reth_primitives::{Address, ChainSpecBuilder, Genesis, B256, BASE_MAINNET};
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+/// Optimism Node Helper type
+pub(crate) type OpNode = NodeHelperType<OptimismNode>;
+
+pub(crate) async fn setup(num_nodes: usize) -> eyre::Result<(Vec<OpNode>, TaskManager, Wallet)> {
+    let genesis: Genesis = serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
+    reth_e2e_test_utils::setup(
+        num_nodes,
+        Arc::new(
+            ChainSpecBuilder::default()
+                .chain(BASE_MAINNET.chain)
+                .genesis(genesis)
+                .ecotone_activated()
+                .build(),
+        ),
+        false,
+    )
+    .await
+}
+
+pub(crate) async fn advance_chain(
+    length: usize,
+    node: &mut OpNode,
+    wallet: Arc<Mutex<Wallet>>,
+) -> eyre::Result<Vec<(OptimismBuiltPayload, OptimismPayloadBuilderAttributes)>> {
+    node.advance(
+        length as u64,
+        || {
+            let wallet = wallet.clone();
+            Box::pin(async move { wallet.lock().await.optimism_l1_block_info_tx().await })
+        },
+        optimism_payload_attributes,
+    )
+    .await
+}
 
 /// Helper function to create a new eth payload attributes
 pub(crate) fn optimism_payload_attributes(timestamp: u64) -> OptimismPayloadBuilderAttributes {

crates/storage/provider/src/traits/full.rs

@@ -2,7 +2,7 @@
 
 use crate::{
     AccountReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
-    DatabaseProviderFactory, EvmEnvProvider, StateProviderFactory,
+    DatabaseProviderFactory, EvmEnvProvider, StageCheckpointReader, StateProviderFactory,
 };
 use reth_db::database::Database;
 
@@ -16,6 +16,7 @@ pub trait FullProvider<DB: Database>:
     + ChainSpecProvider
     + ChangeSetReader
     + CanonStateSubscriptions
+    + StageCheckpointReader
     + Clone
     + Unpin
     + 'static
@@ -31,6 +32,7 @@ impl<T, DB: Database> FullProvider<DB> for T where
         + ChainSpecProvider
         + ChangeSetReader
         + CanonStateSubscriptions
+        + StageCheckpointReader
         + Clone
         + Unpin
         + 'static