diff --git a/Cargo.lock b/Cargo.lock index 94036ec..bca639c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5205,6 +5205,7 @@ dependencies = [ "reth-errors", "reth-evm", "reth-execution-cache", + "reth-network", "reth-node-api", "reth-node-builder", "reth-node-core", diff --git a/Cargo.toml b/Cargo.toml index d39af98..b4311c5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -141,6 +141,7 @@ reth-evm = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-evm-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-execution-types = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-metrics = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } +reth-network = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-network-peers = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-node-api = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v2.5.2" } diff --git a/crates/node/Cargo.toml b/crates/node/Cargo.toml index afd7699..d06be88 100644 --- a/crates/node/Cargo.toml +++ b/crates/node/Cargo.toml @@ -33,6 +33,7 @@ reth-engine-tree.workspace = true reth-errors.workspace = true reth-evm.workspace = true reth-execution-cache.workspace = true +reth-network.workspace = true reth-node-api.workspace = true reth-node-builder.workspace = true reth-node-ethereum.workspace = true diff --git a/crates/node/src/components/mod.rs b/crates/node/src/components/mod.rs index 7ef3716..e30cbca 100644 --- a/crates/node/src/components/mod.rs +++ b/crates/node/src/components/mod.rs @@ -2,16 +2,19 @@ //! //! This module provides builders for the various components that make up a Morph node: //! - [`MorphPoolBuilder`]: Transaction pool with L1 fee validation +//! - [`MorphNetworkBuilder`]: P2P network that accepts transaction gossip from start-up //! - [`MorphExecutorBuilder`]: EVM executor with Morph-specific logic //! - [`MorphConsensusBuilder`]: Consensus validation for L2 blocks //! - [`MorphPayloadBuilderBuilder`]: Block building with L1 message handling mod consensus; mod executor; +mod network; mod payload; mod pool; pub use consensus::MorphConsensusBuilder; pub use executor::MorphExecutorBuilder; +pub use network::MorphNetworkBuilder; pub use payload::MorphPayloadBuilderBuilder; pub use pool::MorphPoolBuilder; diff --git a/crates/node/src/components/network.rs b/crates/node/src/components/network.rs new file mode 100644 index 0000000..ecfea8b --- /dev/null +++ b/crates/node/src/components/network.rs @@ -0,0 +1,46 @@ +//! Morph network builder. + +use reth_network::{NetworkSyncUpdater, SyncState}; +use reth_node_api::FullNodeTypes; +use reth_node_builder::{BuilderContext, components::NetworkBuilder}; +use reth_node_ethereum::EthereumNetworkBuilder; +use reth_transaction_pool::TransactionPool; + +/// Builder for the P2P network. +/// +/// Builds the standard reth network and accepts transaction gossip from start-up instead of +/// from the first block the consensus client imports. +#[derive(Debug, Default, Clone, Copy)] +#[non_exhaustive] +pub struct MorphNetworkBuilder; + +impl NetworkBuilder for MorphNetworkBuilder +where + Node: FullNodeTypes, + Pool: TransactionPool, + EthereumNetworkBuilder: NetworkBuilder, +{ + type Network = >::Network; + + async fn build_network( + self, + ctx: &BuilderContext, + pool: Pool, + ) -> eyre::Result { + let network = EthereumNetworkBuilder::default() + .build_network(ctx, pool) + .await?; + + // reth ignores peer transactions while the network is initially syncing: from the + // `Syncing` state the launcher sets on every start until the first switch to `Idle`, + // which otherwise waits for the first block the consensus client imports. That window + // drops the pool each peer announces only once, when its session opens, so a restarted + // sequencer would never see what RPC nodes held while it was down. Blocks arrive through + // the engine API only, so there is no p2p sync to wait for: switching once here, before + // any session can open, marks the initial sync done for the life of the process. + network.update_sync_state(SyncState::Syncing); + network.update_sync_state(SyncState::Idle); + + Ok(network) + } +} diff --git a/crates/node/src/node.rs b/crates/node/src/node.rs index 87fee57..ebdd7e6 100644 --- a/crates/node/src/node.rs +++ b/crates/node/src/node.rs @@ -7,6 +7,7 @@ //! //! The node is assembled from the following builders: //! - [`MorphPoolBuilder`]: Transaction pool with L1 fee validation +//! - [`MorphNetworkBuilder`]: P2P network that accepts transaction gossip from start-up //! - [`MorphExecutorBuilder`]: EVM executor with Morph-specific logic //! - [`MorphConsensusBuilder`]: Consensus validation for L2 blocks //! - [`MorphPayloadBuilderBuilder`]: Block building with L1 message handling @@ -16,7 +17,8 @@ use super::{ add_ons::MorphAddOns, args::MorphArgs, components::{ - MorphConsensusBuilder, MorphExecutorBuilder, MorphPayloadBuilderBuilder, MorphPoolBuilder, + MorphConsensusBuilder, MorphExecutorBuilder, MorphNetworkBuilder, + MorphPayloadBuilderBuilder, MorphPoolBuilder, }, }; use alloy_consensus::BlockHeader; @@ -32,7 +34,6 @@ use reth_node_builder::{ DebugNode, Node, NodeAdapter, components::{BasicPayloadServiceBuilder, ComponentsBuilder}, }; -use reth_node_ethereum::EthereumNetworkBuilder; use reth_payload_primitives::PayloadAttributesBuilder; use reth_primitives_traits::SealedHeader; use reth_provider::{ @@ -73,7 +74,7 @@ impl MorphNode { N, MorphPoolBuilder, BasicPayloadServiceBuilder, - EthereumNetworkBuilder, + MorphNetworkBuilder, MorphExecutorBuilder, MorphConsensusBuilder, > @@ -87,7 +88,7 @@ impl MorphNode { .payload(BasicPayloadServiceBuilder::new( MorphPayloadBuilderBuilder::new(payload_builder_config), )) - .network(EthereumNetworkBuilder::default()) + .network(MorphNetworkBuilder::default()) .consensus(MorphConsensusBuilder::default()) } } @@ -110,7 +111,7 @@ where N, MorphPoolBuilder, BasicPayloadServiceBuilder, - EthereumNetworkBuilder, + MorphNetworkBuilder, MorphExecutorBuilder, MorphConsensusBuilder, >; diff --git a/crates/node/src/test_utils.rs b/crates/node/src/test_utils.rs index 24d2256..623e4ea 100644 --- a/crates/node/src/test_utils.rs +++ b/crates/node/src/test_utils.rs @@ -204,6 +204,8 @@ pub struct TestNodeBuilder { num_nodes: usize, is_dev: bool, desired_gas_limit: Option, + debug_tip: Option, + trusted_peer: Option, morph_args: Option, } @@ -229,6 +231,8 @@ impl TestNodeBuilder { num_nodes: 1, is_dev: false, desired_gas_limit: None, + debug_tip: None, + trusted_peer: None, morph_args: None, } } @@ -285,6 +289,18 @@ impl TestNodeBuilder { self } + /// Start an initial pipeline backfill toward the given block hash. + pub fn with_debug_tip(mut self, tip: B256) -> Self { + self.debug_tip = Some(tip); + self + } + + /// Dial a trusted peer during node startup, before an initial backfill runs. + pub fn with_trusted_peer(mut self, enode: impl Into) -> Self { + self.trusted_peer = Some(enode.into()); + self + } + /// Override the maximum pool-transaction payload bytes included in a block. /// /// Restricted to single-node setups. @@ -323,6 +339,8 @@ impl TestNodeBuilder { // can carry `--builder.gaslimit`, which `setup_engine` gives no way to set. let is_dev = self.is_dev; let desired_gas_limit = self.desired_gas_limit; + let debug_tip = self.debug_tip; + let trusted_peer = self.trusted_peer; reth_e2e_test_utils::E2ETestSetupBuilder::::new( self.num_nodes, Arc::new(chain_spec), @@ -330,6 +348,10 @@ impl TestNodeBuilder { ) .with_node_config_modifier(move |mut config| { config.builder.gas_limit = desired_gas_limit; + config.debug.tip = debug_tip; + if let Some(ref enode) = trusted_peer { + config.network.trusted_peers = vec![enode.parse().expect("valid trusted enode")]; + } config.set_dev(is_dev) }) .build() diff --git a/crates/node/tests/it/main.rs b/crates/node/tests/it/main.rs index 6e532c6..8f1c33f 100644 --- a/crates/node/tests/it/main.rs +++ b/crates/node/tests/it/main.rs @@ -16,6 +16,7 @@ mod invalid_payload_recovery; mod l1_messages; mod mixed_block_pressure; mod morph_tx; +mod network; mod proof_history; mod reference_index; mod rpc; diff --git a/crates/node/tests/it/network.rs b/crates/node/tests/it/network.rs new file mode 100644 index 0000000..6766d63 --- /dev/null +++ b/crates/node/tests/it/network.rs @@ -0,0 +1,94 @@ +//! P2P network E2E tests. + +use morph_node::test_utils::{TestNodeBuilder, make_transfer_tx, wallet_at_index}; +use reth_network::NetworkInfo; +use reth_provider::BlockNumReader; +use reth_transaction_pool::TransactionPool; + +use super::helpers::{NETWORK_POLL_BUDGET, POLL_INTERVAL, assemble_l2_block, import_l2_block}; +use morph_payload_types::AssembleL2BlockParams; + +/// A transaction a peer already holds when the session opens reaches a node that has not +/// imported a block yet. +/// +/// Peers announce their pool only once, when the session opens. If the node dropped that +/// announcement until its first block, a sequencer restarting while RPC nodes hold pending +/// transactions would never receive them. +#[tokio::test(flavor = "multi_thread")] +async fn peer_pool_reaches_node_before_first_block() -> eyre::Result<()> { + reth_tracing::init_test_tracing(); + + // Built separately so they are not connected until the transaction is pending. + let (mut sequencers, wallet) = TestNodeBuilder::new().build().await?; + let (mut rpcs, _) = TestNodeBuilder::new().build().await?; + let mut sequencer = sequencers.pop().unwrap(); + let mut rpc = rpcs.pop().unwrap(); + + let tx = make_transfer_tx(wallet.chain_id, wallet_at_index(1, wallet.chain_id), 0).await; + let tx_hash = rpc.rpc.inject_tx(tx).await?; + + assert!( + sequencer.inner.network.is_syncing(), + "the node must still be in its start-up sync state" + ); + sequencer.connect(&mut rpc).await; + + let deadline = tokio::time::Instant::now() + NETWORK_POLL_BUDGET; + while !sequencer.inner.pool.contains(&tx_hash) { + assert!( + tokio::time::Instant::now() < deadline, + "the peer's pending transaction never reached the node" + ); + tokio::time::sleep(POLL_INTERVAL).await; + } + assert_eq!(sequencer.inner.provider.best_block_number()?, 0); + + Ok(()) +} + +/// A node starting pipeline backfill can still receive a peer's pending pool +/// announcement while its chain is catching up. +#[tokio::test(flavor = "multi_thread")] +async fn peer_pool_reaches_node_during_initial_backfill() -> eyre::Result<()> { + reth_tracing::init_test_tracing(); + + let (mut sources, wallet) = TestNodeBuilder::new().build().await?; + let source = sources.pop().unwrap(); + let mut tip = None; + for number in 1..=5 { + let mut params = AssembleL2BlockParams::empty(number); + params.timestamp = Some(number); + let block = assemble_l2_block(&source, params).await?; + import_l2_block(&source, block.clone()).await?; + tip = Some(block.hash); + } + let tip = tip.expect("five blocks were imported"); + + let tx = make_transfer_tx(wallet.chain_id, wallet_at_index(1, wallet.chain_id), 0).await; + let tx_hash = source.rpc.inject_tx(tx).await?; + + // Launch must be able to dial the source during backfill: the upstream + // launcher can wait for that initial backfill before returning the handle. + let enode = source.network.record().to_string(); + let (mut followers, _) = tokio::time::timeout( + NETWORK_POLL_BUDGET, + TestNodeBuilder::new() + .with_debug_tip(tip) + .with_trusted_peer(enode) + .build(), + ) + .await??; + let follower = followers.pop().unwrap(); + let deadline = tokio::time::Instant::now() + NETWORK_POLL_BUDGET; + while !follower.inner.pool.contains(&tx_hash) + || follower.inner.provider.best_block_number()? < 5 + { + assert!( + tokio::time::Instant::now() < deadline, + "backfill did not finish with the peer's pending transaction in the pool" + ); + tokio::time::sleep(POLL_INTERVAL).await; + } + + Ok(()) +}