feat: Add more complex E2E test (#12005)

Cargo.lock

@@ -7020,6 +7020,7 @@ dependencies = [
  "reth",
  "reth-chainspec",
  "reth-db",
+ "reth-engine-local",
  "reth-network-peers",
  "reth-node-builder",
  "reth-payload-builder",
@@ -7033,6 +7034,7 @@ dependencies = [
  "tokio",
  "tokio-stream",
  "tracing",
+ "url",
 ]
 
 [[package]]
@@ -7966,8 +7968,11 @@ dependencies = [
  "alloy-consensus",
  "alloy-genesis",
  "alloy-primitives",
+ "alloy-provider",
+ "alloy-signer",
  "eyre",
  "futures",
+ "rand 0.8.5",
  "reth",
  "reth-auto-seal-consensus",
  "reth-basic-payload-builder",

crates/e2e-test-utils/Cargo.toml

@@ -23,9 +23,11 @@ reth-node-builder = { workspace = true, features = ["test-utils"] }
 reth-tokio-util.workspace = true
 reth-stages-types.workspace = true
 reth-network-peers.workspace = true
+reth-engine-local.workspace = true
 
 # rpc
 jsonrpsee.workspace = true
+url.workspace = true
 
 # ethereum
 alloy-primitives.workspace = true

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

@@ -12,19 +12,22 @@ use reth::{
         types::engine::{ForkchoiceState, PayloadStatusEnum},
     },
 };
+use reth_chainspec::EthereumHardforks;
+use reth_node_builder::BuiltPayload;
 use reth_payload_builder::PayloadId;
 use reth_rpc_layer::AuthClientService;
-use std::marker::PhantomData;
+use std::{marker::PhantomData, sync::Arc};
 
 /// Helper for engine api operations
 #[derive(Debug)]
-pub struct EngineApiTestContext<E> {
+pub struct EngineApiTestContext<E, ChainSpec> {
+    pub chain_spec: Arc<ChainSpec>,
     pub canonical_stream: CanonStateNotificationStream,
     pub engine_api_client: HttpClient<AuthClientService<HttpBackend>>,
     pub _marker: PhantomData<E>,
 }
 
-impl<E: EngineTypes> EngineApiTestContext<E> {
+impl<E: EngineTypes, ChainSpec: EthereumHardforks> EngineApiTestContext<E, ChainSpec> {
     /// Retrieves a v3 payload from the engine api
     pub async fn get_payload_v3(
         &self,
@@ -51,18 +54,40 @@ impl<E: EngineTypes> EngineApiTestContext<E> {
     ) -> eyre::Result<B256>
     where
         E::ExecutionPayloadEnvelopeV3: From<E::BuiltPayload> + PayloadEnvelopeExt,
+        E::ExecutionPayloadEnvelopeV4: From<E::BuiltPayload> + PayloadEnvelopeExt,
     {
-        // setup payload for submission
-        let envelope_v3: <E as EngineTypes>::ExecutionPayloadEnvelopeV3 = payload.into();
-
         // submit payload to engine api
-        let submission = EngineApiClient::<E>::new_payload_v3(
-            &self.engine_api_client,
-            envelope_v3.execution_payload(),
-            versioned_hashes,
-            payload_builder_attributes.parent_beacon_block_root().unwrap(),
-        )
-        .await?;
+        let submission = if self
+            .chain_spec
+            .is_prague_active_at_timestamp(payload_builder_attributes.timestamp())
+        {
+            let requests = payload
+                .executed_block()
+                .unwrap()
+                .execution_outcome()
+                .requests
+                .first()
+                .unwrap()
+                .clone();
+            let envelope: <E as EngineTypes>::ExecutionPayloadEnvelopeV4 = payload.into();
+            EngineApiClient::<E>::new_payload_v4(
+                &self.engine_api_client,
+                envelope.execution_payload(),
+                versioned_hashes,
+                payload_builder_attributes.parent_beacon_block_root().unwrap(),
+                requests,
+            )
+            .await?
+        } else {
+            let envelope: <E as EngineTypes>::ExecutionPayloadEnvelopeV3 = payload.into();
+            EngineApiClient::<E>::new_payload_v3(
+                &self.engine_api_client,
+                envelope.execution_payload(),
+                versioned_hashes,
+                payload_builder_attributes.parent_beacon_block_root().unwrap(),
+            )
+            .await?
+        };
 
         assert_eq!(submission.status, expected_status);
 

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

@@ -11,11 +11,13 @@ use reth::{
 };
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_db::{test_utils::TempDatabase, DatabaseEnv};
+use reth_engine_local::LocalPayloadAttributesBuilder;
 use reth_node_builder::{
-    components::NodeComponentsBuilder, rpc::RethRpcAddOns, FullNodeTypesAdapter, Node, NodeAdapter,
-    NodeComponents, NodeTypesWithDBAdapter, NodeTypesWithEngine, RethFullAdapter,
+    components::NodeComponentsBuilder, rpc::RethRpcAddOns, EngineNodeLauncher,
+    FullNodeTypesAdapter, Node, NodeAdapter, NodeComponents, NodeTypesWithDBAdapter,
+    NodeTypesWithEngine, PayloadAttributesBuilder, PayloadTypes,
 };
-use reth_provider::providers::BlockchainProvider;
+use reth_provider::providers::{BlockchainProvider, BlockchainProvider2};
 use tracing::{span, Level};
 use wallet::Wallet;
 
@@ -102,21 +104,102 @@ where
     Ok((nodes, tasks, Wallet::default().with_chain_id(chain_spec.chain().into())))
 }
 
+/// Creates the initial setup with `num_nodes` started and interconnected.
+pub async fn setup_engine<N>(
+    num_nodes: usize,
+    chain_spec: Arc<N::ChainSpec>,
+    is_dev: bool,
+) -> eyre::Result<(
+    Vec<NodeHelperType<N, N::AddOns, BlockchainProvider2<NodeTypesWithDBAdapter<N, TmpDB>>>>,
+    TaskManager,
+    Wallet,
+)>
+where
+    N: Default
+        + Node<TmpNodeAdapter<N, BlockchainProvider2<NodeTypesWithDBAdapter<N, TmpDB>>>>
+        + NodeTypesWithEngine<ChainSpec: EthereumHardforks>,
+    N::ComponentsBuilder: NodeComponentsBuilder<
+        TmpNodeAdapter<N, BlockchainProvider2<NodeTypesWithDBAdapter<N, TmpDB>>>,
+        Components: NodeComponents<
+            TmpNodeAdapter<N, BlockchainProvider2<NodeTypesWithDBAdapter<N, TmpDB>>>,
+            Network: PeersHandleProvider,
+        >,
+    >,
+    N::AddOns: RethRpcAddOns<Adapter<N, BlockchainProvider2<NodeTypesWithDBAdapter<N, TmpDB>>>>,
+    LocalPayloadAttributesBuilder<N::ChainSpec>: PayloadAttributesBuilder<
+        <<N as NodeTypesWithEngine>::Engine as PayloadTypes>::PayloadAttributes,
+    >,
+{
+    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<NodeTestContext<_, _>> = Vec::with_capacity(num_nodes);
+
+    for idx in 0..num_nodes {
+        let node_config = NodeConfig::new(chain_spec.clone())
+            .with_network(network_config.clone())
+            .with_unused_ports()
+            .with_rpc(RpcServerArgs::default().with_unused_ports().with_http())
+            .set_dev(is_dev);
+
+        let span = span!(Level::INFO, "node", idx);
+        let _enter = span.enter();
+        let node = N::default();
+        let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config.clone())
+            .testing_node(exec.clone())
+            .with_types_and_provider::<N, BlockchainProvider2<_>>()
+            .with_components(node.components_builder())
+            .with_add_ons(node.add_ons())
+            .launch_with_fn(|builder| {
+                let launcher = EngineNodeLauncher::new(
+                    builder.task_executor().clone(),
+                    builder.config().datadir(),
+                    Default::default(),
+                );
+                builder.launch_with(launcher)
+            })
+            .await?;
+
+        let mut node = NodeTestContext::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 TmpNodeAdapter<N> = FullNodeTypesAdapter<
-    NodeTypesWithDBAdapter<N, TmpDB>,
-    BlockchainProvider<NodeTypesWithDBAdapter<N, TmpDB>>,
->;
+type TmpNodeAdapter<N, Provider = BlockchainProvider<NodeTypesWithDBAdapter<N, TmpDB>>> =
+    FullNodeTypesAdapter<NodeTypesWithDBAdapter<N, TmpDB>, Provider>;
 
 /// Type alias for a `NodeAdapter`
-pub type Adapter<N> = NodeAdapter<
-    RethFullAdapter<TmpDB, N>,
-    <<N as Node<TmpNodeAdapter<N>>>::ComponentsBuilder as NodeComponentsBuilder<
-        RethFullAdapter<TmpDB, N>,
+pub type Adapter<N, Provider = BlockchainProvider<NodeTypesWithDBAdapter<N, TmpDB>>> = NodeAdapter<
+    TmpNodeAdapter<N, Provider>,
+    <<N as Node<TmpNodeAdapter<N, Provider>>>::ComponentsBuilder as NodeComponentsBuilder<
+        TmpNodeAdapter<N, Provider>,
     >>::Components,
 >;
 
 /// Type alias for a type of `NodeHelper`
-pub type NodeHelperType<N, AO> = NodeTestContext<Adapter<N>, AO>;
+pub type NodeHelperType<N, AO, Provider = BlockchainProvider<NodeTypesWithDBAdapter<N, TmpDB>>> =
+    NodeTestContext<Adapter<N, Provider>, AO>;

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

@@ -18,9 +18,10 @@ use reth::{
     },
 };
 use reth_chainspec::EthereumHardforks;
-use reth_node_builder::{rpc::RethRpcAddOns, NodeTypesWithEngine};
+use reth_node_builder::{rpc::RethRpcAddOns, NodeTypes, NodeTypesWithEngine};
 use reth_stages_types::StageId;
 use tokio_stream::StreamExt;
+use url::Url;
 
 use crate::{
     engine_api::EngineApiTestContext, network::NetworkTestContext, payload::PayloadTestContext,
@@ -41,7 +42,10 @@ where
     /// Context for testing network functionalities.
     pub network: NetworkTestContext<Node::Network>,
     /// Context for testing the Engine API.
-    pub engine_api: EngineApiTestContext<<Node::Types as NodeTypesWithEngine>::Engine>,
+    pub engine_api: EngineApiTestContext<
+        <Node::Types as NodeTypesWithEngine>::Engine,
+        <Node::Types as NodeTypes>::ChainSpec,
+    >,
     /// Context for testing RPC features.
     pub rpc: RpcTestContext<Node, AddOns::EthApi>,
 }
@@ -63,6 +67,7 @@ where
             payload: PayloadTestContext::new(builder).await?,
             network: NetworkTestContext::new(node.network.clone()),
             engine_api: EngineApiTestContext {
+                chain_spec: node.chain_spec(),
                 engine_api_client: node.auth_server_handle().http_client(),
                 canonical_stream: node.provider.canonical_state_stream(),
                 _marker: PhantomData::<Engine>,
@@ -89,6 +94,7 @@ where
     ) -> eyre::Result<Vec<(Engine::BuiltPayload, Engine::PayloadBuilderAttributes)>>
     where
         Engine::ExecutionPayloadEnvelopeV3: From<Engine::BuiltPayload> + PayloadEnvelopeExt,
+        Engine::ExecutionPayloadEnvelopeV4: From<Engine::BuiltPayload> + PayloadEnvelopeExt,
         AddOns::EthApi: EthApiSpec + EthTransactions + TraceExt + FullEthApiTypes,
     {
         let mut chain = Vec::with_capacity(length as usize);
@@ -137,6 +143,8 @@ where
     where
         <Engine as EngineTypes>::ExecutionPayloadEnvelopeV3:
             From<Engine::BuiltPayload> + PayloadEnvelopeExt,
+        <Engine as EngineTypes>::ExecutionPayloadEnvelopeV4:
+            From<Engine::BuiltPayload> + PayloadEnvelopeExt,
     {
         let (payload, eth_attr) = self.new_payload(attributes_generator).await?;
 
@@ -236,4 +244,10 @@ where
         }
         Ok(())
     }
+
+    /// Returns the RPC URL.
+    pub fn rpc_url(&self) -> Url {
+        let addr = self.inner.rpc_server_handle().http_local_addr().unwrap();
+        format!("http://{}", addr).parse().unwrap()
+    }
 }

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

@@ -1,4 +1,5 @@
-use op_alloy_rpc_types_engine::OpExecutionPayloadEnvelopeV3;
+use alloy_rpc_types::engine::ExecutionPayloadEnvelopeV4;
+use op_alloy_rpc_types_engine::{OpExecutionPayloadEnvelopeV3, OpExecutionPayloadEnvelopeV4};
 use reth::rpc::types::engine::{ExecutionPayloadEnvelopeV3, ExecutionPayloadV3};
 
 /// The execution payload envelope type.
@@ -13,8 +14,20 @@ impl PayloadEnvelopeExt for OpExecutionPayloadEnvelopeV3 {
     }
 }
 
+impl PayloadEnvelopeExt for OpExecutionPayloadEnvelopeV4 {
+    fn execution_payload(&self) -> ExecutionPayloadV3 {
+        self.execution_payload.clone()
+    }
+}
+
 impl PayloadEnvelopeExt for ExecutionPayloadEnvelopeV3 {
     fn execution_payload(&self) -> ExecutionPayloadV3 {
         self.execution_payload.clone()
     }
 }
+
+impl PayloadEnvelopeExt for ExecutionPayloadEnvelopeV4 {
+    fn execution_payload(&self) -> ExecutionPayloadV3 {
+        self.execution_payload.clone()
+    }
+}

crates/ethereum/node/Cargo.toml

@@ -52,6 +52,9 @@ alloy-genesis.workspace = true
 tokio.workspace = true
 serde_json.workspace = true
 alloy-consensus.workspace = true
+alloy-provider.workspace = true
+rand.workspace = true
+alloy-signer.workspace = true
 
 [features]
 default = []

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

@@ -1,7 +1,19 @@
 use crate::utils::eth_payload_attributes;
+use alloy_consensus::TxType;
+use alloy_primitives::bytes;
+use alloy_provider::{
+    network::{
+        Ethereum, EthereumWallet, NetworkWallet, TransactionBuilder, TransactionBuilder7702,
+    },
+    Provider, ProviderBuilder, SendableTx,
+};
+use alloy_signer::SignerSync;
+use rand::{rngs::StdRng, seq::SliceRandom, Rng, SeedableRng};
+use reth::rpc::types::TransactionRequest;
 use reth_chainspec::{ChainSpecBuilder, MAINNET};
-use reth_e2e_test_utils::{setup, transaction::TransactionTestContext};
+use reth_e2e_test_utils::{setup, setup_engine, transaction::TransactionTestContext};
 use reth_node_ethereum::EthereumNode;
+use revm::primitives::{AccessListItem, Authorization};
 use std::sync::Arc;
 
 #[tokio::test]
@@ -45,3 +57,115 @@ async fn can_sync() -> eyre::Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn e2e_test_send_transactions() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let seed: [u8; 32] = rand::thread_rng().gen();
+    let mut rng = StdRng::from_seed(seed);
+    println!("Seed: {:?}", seed);
+
+    let chain_spec = Arc::new(
+        ChainSpecBuilder::default()
+            .chain(MAINNET.chain)
+            .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
+            .cancun_activated()
+            .prague_activated()
+            .build(),
+    );
+
+    let (mut nodes, _tasks, wallet) =
+        setup_engine::<EthereumNode>(2, chain_spec.clone(), false).await?;
+    let mut node = nodes.pop().unwrap();
+    let signers = wallet.gen();
+    let provider = ProviderBuilder::new().with_recommended_fillers().on_http(node.rpc_url());
+
+    // simple contract which writes to storage on any call
+    let dummy_bytecode = bytes!("6080604052348015600f57600080fd5b50602880601d6000396000f3fe4360a09081523360c0526040608081905260e08152902080805500fea164736f6c6343000810000a");
+    let mut call_destinations = signers.iter().map(|s| s.address()).collect::<Vec<_>>();
+
+    // Produce 100 random blocks with random transactions
+    for _ in 0..100 {
+        let tx_count = rng.gen_range(1..20);
+
+        let mut pending = vec![];
+        for _ in 0..tx_count {
+            let signer = signers.choose(&mut rng).unwrap();
+            let tx_type = TxType::try_from(rng.gen_range(0..=4)).unwrap();
+
+            let mut tx = TransactionRequest::default().with_from(signer.address());
+
+            let should_create =
+                rng.gen::<bool>() && tx_type != TxType::Eip4844 && tx_type != TxType::Eip7702;
+            if should_create {
+                tx = tx.into_create().with_input(dummy_bytecode.clone());
+            } else {
+                tx = tx.with_to(*call_destinations.choose(&mut rng).unwrap()).with_input(
+                    (0..rng.gen_range(0..10000)).map(|_| rng.gen()).collect::<Vec<u8>>(),
+                );
+            }
+
+            if matches!(tx_type, TxType::Legacy | TxType::Eip2930) {
+                tx = tx.with_gas_price(provider.get_gas_price().await?);
+            }
+
+            if rng.gen::<bool>() || tx_type == TxType::Eip2930 {
+                tx = tx.with_access_list(
+                    vec![AccessListItem {
+                        address: *call_destinations.choose(&mut rng).unwrap(),
+                        storage_keys: (0..rng.gen_range(0..100)).map(|_| rng.gen()).collect(),
+                    }]
+                    .into(),
+                );
+            }
+
+            if tx_type == TxType::Eip7702 {
+                let signer = signers.choose(&mut rng).unwrap();
+                let auth = Authorization {
+                    chain_id: provider.get_chain_id().await?,
+                    address: *call_destinations.choose(&mut rng).unwrap(),
+                    nonce: provider.get_transaction_count(signer.address()).await?,
+                };
+                let sig = signer.sign_hash_sync(&auth.signature_hash())?;
+                tx = tx.with_authorization_list(vec![auth.into_signed(sig)])
+            }
+
+            let SendableTx::Builder(tx) = provider.fill(tx).await? else { unreachable!() };
+            let tx =
+                NetworkWallet::<Ethereum>::sign_request(&EthereumWallet::new(signer.clone()), tx)
+                    .await?;
+
+            pending.push(provider.send_tx_envelope(tx).await?);
+        }
+
+        let (payload, _) = node.advance_block(vec![], eth_payload_attributes).await?;
+        assert!(payload.block().raw_transactions().len() == tx_count);
+
+        for pending in pending {
+            let receipt = pending.get_receipt().await?;
+            if let Some(address) = receipt.contract_address {
+                call_destinations.push(address);
+            }
+        }
+    }
+
+    let second_node = nodes.pop().unwrap();
+    let second_provider =
+        ProviderBuilder::new().with_recommended_fillers().on_http(second_node.rpc_url());
+
+    assert_eq!(second_provider.get_block_number().await?, 0);
+
+    let head = provider.get_block_by_number(Default::default(), false).await?.unwrap().header.hash;
+    second_node.engine_api.update_forkchoice(head, head).await?;
+
+    let start = std::time::Instant::now();
+
+    while provider.get_block_number().await? != second_provider.get_block_number().await? {
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        assert!(start.elapsed() <= std::time::Duration::from_secs(10), "timed out");
+    }
+
+    Ok(())
+}