chore(e2e): refactor e2e tests (#7773)

crates/e2e-test-utils/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "reth-e2e-test-utils"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+
+
+[dependencies]
+reth.workspace = true
+reth-node-core.workspace = true
+reth-primitives.workspace = true
+reth-node-ethereum.workspace = true
+reth-tracing.workspace = true
+reth-db.workspace = true
+reth-rpc.workspace = true
+reth-payload-builder = { workspace = true, features = ["test-utils"] }
+
+jsonrpsee.workspace = true
+
+futures-util.workspace = true
+eyre.workspace = true
+tokio.workspace = true
+tokio-stream.workspace = true
+serde_json.workspace = true
+rand.workspace = true
+secp256k1.workspace = true
+alloy-signer.workspace = true
+alloy-signer-wallet = { workspace = true, features = ["mnemonic"] }
+alloy-rpc-types.workspace = true
+alloy-network.workspace = true
+alloy-consensus.workspace = true

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

@@ -0,0 +1,67 @@
+use crate::traits::PayloadEnvelopeExt;
+use jsonrpsee::http_client::HttpClient;
+use reth::{
+    api::{EngineTypes, PayloadBuilderAttributes},
+    providers::CanonStateNotificationStream,
+    rpc::{api::EngineApiClient, types::engine::ForkchoiceState},
+};
+use reth_payload_builder::PayloadId;
+use reth_primitives::B256;
+use std::marker::PhantomData;
+
+/// Helper for engine api operations
+pub struct EngineApiHelper<E> {
+    pub canonical_stream: CanonStateNotificationStream,
+    pub engine_api_client: HttpClient,
+    pub _marker: PhantomData<E>,
+}
+
+impl<E: EngineTypes + 'static> EngineApiHelper<E> {
+    /// Retrieves a v3 payload from the engine api
+    pub async fn get_payload_v3(
+        &self,
+        payload_id: PayloadId,
+    ) -> eyre::Result<E::ExecutionPayloadV3> {
+        Ok(EngineApiClient::<E>::get_payload_v3(&self.engine_api_client, payload_id).await?)
+    }
+
+    /// Submits a payload to the engine api
+    pub async fn submit_payload(
+        &self,
+        payload: E::BuiltPayload,
+        payload_builder_attributes: E::PayloadBuilderAttributes,
+    ) -> eyre::Result<B256>
+    where
+        E::ExecutionPayloadV3: From<E::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        // setup payload for submission
+        let envelope_v3: <E as EngineTypes>::ExecutionPayloadV3 = payload.into();
+
+        // submit payload to engine api
+        let submission = EngineApiClient::<E>::new_payload_v3(
+            &self.engine_api_client,
+            envelope_v3.execution_payload(),
+            vec![],
+            payload_builder_attributes.parent_beacon_block_root().unwrap(),
+        )
+        .await?;
+        assert!(submission.is_valid(), "{}", submission);
+        Ok(submission.latest_valid_hash.unwrap())
+    }
+
+    /// Sends forkchoice update to the engine api
+    pub async fn update_forkchoice(&self, hash: B256) -> eyre::Result<()> {
+        EngineApiClient::<E>::fork_choice_updated_v2(
+            &self.engine_api_client,
+            ForkchoiceState {
+                head_block_hash: hash,
+                safe_block_hash: hash,
+                finalized_block_hash: hash,
+            },
+            None,
+        )
+        .await?;
+
+        Ok(())
+    }
+}

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

@@ -0,0 +1,17 @@
+/// Wrapper type to create test nodes
+pub mod node;
+
+/// Helper type to yield accounts from mnemonic
+pub mod wallet;
+
+/// Helper for payload operations
+mod payload;
+
+/// Helper for network operations
+mod network;
+
+/// Helper for engine api operations
+mod engine_api;
+
+/// Helper traits
+mod traits;

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

@@ -0,0 +1,44 @@
+use futures_util::StreamExt;
+use reth::network::{NetworkEvent, NetworkEvents, NetworkHandle, PeersInfo};
+use reth_primitives::NodeRecord;
+use reth_tracing::tracing::info;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+
+/// Helper for network operations
+pub struct NetworkHelper {
+    network_events: UnboundedReceiverStream<NetworkEvent>,
+    network: NetworkHandle,
+}
+
+impl NetworkHelper {
+    /// Creates a new network helper
+    pub fn new(network: NetworkHandle) -> Self {
+        let network_events = network.event_listener();
+        Self { network_events, network }
+    }
+
+    /// Adds a peer to the network node via network handle
+    pub async fn add_peer(&mut self, node_record: NodeRecord) {
+        self.network.peers_handle().add_peer(node_record.id, node_record.tcp_addr());
+
+        match self.network_events.next().await {
+            Some(NetworkEvent::PeerAdded(_)) => (),
+            _ => panic!("Expected a peer added event"),
+        }
+    }
+
+    /// Returns the network node record
+    pub fn record(&self) -> NodeRecord {
+        self.network.local_node_record()
+    }
+
+    /// Expects a session to be established
+    pub async fn expect_session(&mut self) {
+        match self.network_events.next().await {
+            Some(NetworkEvent::SessionEstablished { remote_addr, .. }) => {
+                info!(?remote_addr, "Session established")
+            }
+            _ => panic!("Expected session established event"),
+        }
+    }
+}

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

@@ -0,0 +1,145 @@
+use crate::{
+    engine_api::EngineApiHelper, network::NetworkHelper, payload::PayloadHelper,
+    traits::PayloadEnvelopeExt,
+};
+use alloy_rpc_types::BlockNumberOrTag;
+use eyre::Ok;
+use reth::{
+    api::{BuiltPayload, EngineTypes, FullNodeComponents, PayloadBuilderAttributes},
+    builder::FullNode,
+    providers::{BlockReaderIdExt, CanonStateSubscriptions},
+    rpc::{
+        eth::{error::EthResult, EthTransactions},
+        types::engine::PayloadAttributes,
+    },
+};
+use reth_payload_builder::EthPayloadBuilderAttributes;
+use reth_primitives::{Address, BlockNumber, Bytes, B256};
+use std::{
+    marker::PhantomData,
+    time::{SystemTime, UNIX_EPOCH},
+};
+use tokio_stream::StreamExt;
+
+/// An helper struct to handle node actions
+pub struct NodeHelper<Node>
+where
+    Node: FullNodeComponents,
+{
+    pub inner: FullNode<Node>,
+    payload: PayloadHelper<Node::Engine>,
+    pub network: NetworkHelper,
+    pub engine_api: EngineApiHelper<Node::Engine>,
+}
+
+impl<Node> NodeHelper<Node>
+where
+    Node: FullNodeComponents,
+{
+    /// Creates a new test node
+    pub async fn new(node: FullNode<Node>) -> eyre::Result<Self> {
+        let builder = node.payload_builder.clone();
+
+        Ok(Self {
+            inner: node.clone(),
+            network: NetworkHelper::new(node.network.clone()),
+            payload: PayloadHelper::new(builder).await?,
+            engine_api: EngineApiHelper {
+                engine_api_client: node.auth_server_handle().http_client(),
+                canonical_stream: node.provider.canonical_state_stream(),
+                _marker: PhantomData::<Node::Engine>,
+            },
+        })
+    }
+
+    /// Advances the node forward
+    pub async fn advance(
+        &mut self,
+        raw_tx: Bytes,
+        attributes_generator: impl Fn(u64) -> <Node::Engine as EngineTypes>::PayloadBuilderAttributes,
+    ) -> eyre::Result<(B256, B256)>
+    where
+        <Node::Engine as EngineTypes>::ExecutionPayloadV3:
+            From<<Node::Engine as EngineTypes>::BuiltPayload> + PayloadEnvelopeExt,
+    {
+        // push tx into pool via RPC server
+        let tx_hash = self.inject_tx(raw_tx).await?;
+
+        // trigger new payload building draining the pool
+        let eth_attr = self.payload.new_payload(attributes_generator).await.unwrap();
+
+        // first event is the payload attributes
+        self.payload.expect_attr_event(eth_attr.clone()).await?;
+
+        // wait for the payload builder to have finished building
+        self.payload.wait_for_built_payload(eth_attr.payload_id()).await;
+
+        // trigger resolve payload via engine api
+        self.engine_api.get_payload_v3(eth_attr.payload_id()).await?;
+
+        // ensure we're also receiving the built payload as event
+        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?;
+
+        // 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))
+    }
+
+    /// Injects a raw transaction into the node tx pool via RPC server
+    async fn inject_tx(&mut self, raw_tx: Bytes) -> EthResult<B256> {
+        let eth_api = self.inner.rpc_registry.eth_api();
+        eth_api.send_raw_transaction(raw_tx).await
+    }
+
+    /// Asserts that a new block has been added to the blockchain
+    /// and the tx has been included in the block
+    pub async fn assert_new_block(
+        &mut self,
+        tip_tx_hash: B256,
+        block_hash: B256,
+        block_number: BlockNumber,
+    ) -> eyre::Result<()> {
+        // get head block from notifications stream and verify the tx has been pushed to the
+        // pool is actually present in the canonical block
+        let head = self.engine_api.canonical_stream.next().await.unwrap();
+        let tx = head.tip().transactions().next();
+        assert_eq!(tx.unwrap().hash().as_slice(), tip_tx_hash.as_slice());
+
+        loop {
+            // wait for the block to commit
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+            if let Some(latest_block) =
+                self.inner.provider.block_by_number_or_tag(BlockNumberOrTag::Latest)?
+            {
+                if latest_block.number == block_number {
+                    // make sure the block hash we submitted via FCU engine api is the new latest
+                    // block using an RPC call
+                    assert_eq!(latest_block.hash_slow(), block_hash);
+                    break
+                }
+            }
+        }
+        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

@@ -0,0 +1,68 @@
+use futures_util::StreamExt;
+use reth::api::{BuiltPayload, EngineTypes, PayloadBuilderAttributes};
+use reth_payload_builder::{Events, PayloadBuilderHandle, PayloadId};
+use tokio_stream::wrappers::BroadcastStream;
+
+/// Helper for payload operations
+pub struct PayloadHelper<E: EngineTypes + 'static> {
+    pub payload_event_stream: BroadcastStream<Events<E>>,
+    payload_builder: PayloadBuilderHandle<E>,
+    timestamp: u64,
+}
+
+impl<E: EngineTypes + 'static> PayloadHelper<E> {
+    /// Creates a new payload helper
+    pub async fn new(payload_builder: PayloadBuilderHandle<E>) -> eyre::Result<Self> {
+        let payload_events = payload_builder.subscribe().await?;
+        let payload_event_stream = payload_events.into_stream();
+        // Cancun timestamp
+        Ok(Self { payload_event_stream, payload_builder, timestamp: 1710338135 })
+    }
+
+    /// Creates a new payload job from static attributes
+    pub async fn new_payload(
+        &mut self,
+        attributes_generator: impl Fn(u64) -> E::PayloadBuilderAttributes,
+    ) -> eyre::Result<E::PayloadBuilderAttributes> {
+        self.timestamp += 1;
+        let attributes: E::PayloadBuilderAttributes = attributes_generator(self.timestamp);
+        self.payload_builder.new_payload(attributes.clone()).await.unwrap();
+        Ok(attributes)
+    }
+
+    /// Asserts that the next event is a payload attributes event
+    pub async fn expect_attr_event(
+        &mut self,
+        attrs: E::PayloadBuilderAttributes,
+    ) -> eyre::Result<()> {
+        let first_event = self.payload_event_stream.next().await.unwrap()?;
+        if let reth::payload::Events::Attributes(attr) = first_event {
+            assert_eq!(attrs.timestamp(), attr.timestamp());
+        } else {
+            panic!("Expect first event as payload attributes.")
+        }
+        Ok(())
+    }
+
+    /// Wait until the best built payload is ready
+    pub async fn wait_for_built_payload(&self, payload_id: PayloadId) {
+        loop {
+            let payload = self.payload_builder.best_payload(payload_id).await.unwrap().unwrap();
+            if payload.block().body.is_empty() {
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                continue;
+            }
+            break;
+        }
+    }
+
+    /// Expects the next event to be a built payload event or panics
+    pub async fn expect_built_payload(&mut self) -> eyre::Result<E::BuiltPayload> {
+        let second_event = self.payload_event_stream.next().await.unwrap()?;
+        if let reth::payload::Events::BuiltPayload(payload) = second_event {
+            Ok(payload)
+        } else {
+            panic!("Expect a built payload event.");
+        }
+    }
+}

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

@@ -0,0 +1,22 @@
+use reth::rpc::types::{
+    engine::{ExecutionPayloadEnvelopeV3, OptimismExecutionPayloadEnvelopeV3},
+    ExecutionPayloadV3,
+};
+
+/// The execution payload envelope type.
+pub trait PayloadEnvelopeExt: Send + Sync + std::fmt::Debug {
+    /// Returns the execution payload V3 from the payload
+    fn execution_payload(&self) -> ExecutionPayloadV3;
+}
+
+impl PayloadEnvelopeExt for OptimismExecutionPayloadEnvelopeV3 {
+    fn execution_payload(&self) -> ExecutionPayloadV3 {
+        self.execution_payload.clone()
+    }
+}
+
+impl PayloadEnvelopeExt for ExecutionPayloadEnvelopeV3 {
+    fn execution_payload(&self) -> ExecutionPayloadV3 {
+        self.execution_payload.clone()
+    }
+}

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

@@ -0,0 +1,41 @@
+use alloy_network::{eip2718::Encodable2718, EthereumSigner, TransactionBuilder};
+use alloy_rpc_types::TransactionRequest;
+use alloy_signer_wallet::{coins_bip39::English, LocalWallet, MnemonicBuilder};
+use reth_primitives::{Address, Bytes, U256};
+/// One of the accounts of the genesis allocations.
+pub struct Wallet {
+    inner: LocalWallet,
+    nonce: u64,
+}
+
+impl Wallet {
+    /// Creates a new account from one of the secret/pubkeys of the genesis allocations (test.json)
+    pub(crate) fn new(phrase: &str) -> Self {
+        let inner = MnemonicBuilder::<English>::default().phrase(phrase).build().unwrap();
+        Self { inner, nonce: 0 }
+    }
+
+    /// Creates a static transfer and signs it
+    pub async fn transfer_tx(&mut self) -> Bytes {
+        let tx = TransactionRequest {
+            nonce: Some(self.nonce),
+            value: Some(U256::from(100)),
+            to: Some(Address::random()),
+            gas_price: Some(20e9 as u128),
+            gas: Some(21000),
+            chain_id: Some(1),
+            ..Default::default()
+        };
+        self.nonce += 1;
+        let signer = EthereumSigner::from(self.inner.clone());
+        tx.build(&signer).await.unwrap().encoded_2718().into()
+    }
+}
+
+const TEST_MNEMONIC: &str = "test test test test test test test test test test test junk";
+
+impl Default for Wallet {
+    fn default() -> Self {
+        Wallet::new(TEST_MNEMONIC)
+    }
+}