fix(node): accept peer transactions before the first block after startup

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" }

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

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;

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<Node, Pool> NetworkBuilder<Node, Pool> for MorphNetworkBuilder
+where
+    Node: FullNodeTypes,
+    Pool: TransactionPool,
+    EthereumNetworkBuilder: NetworkBuilder<Node, Pool>,
+{
+    type Network = <EthereumNetworkBuilder as NetworkBuilder<Node, Pool>>::Network;
+
+    async fn build_network(
+        self,
+        ctx: &BuilderContext<Node>,
+        pool: Pool,
+    ) -> eyre::Result<Self::Network> {
+        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)
+    }
+}

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<MorphPayloadBuilderBuilder>,
-        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<MorphPayloadBuilderBuilder>,
-        EthereumNetworkBuilder,
+        MorphNetworkBuilder,
         MorphExecutorBuilder,
         MorphConsensusBuilder,
     >;

crates/node/src/test_utils.rs

@@ -204,6 +204,8 @@ pub struct TestNodeBuilder {
     num_nodes: usize,
     is_dev: bool,
     desired_gas_limit: Option<u64>,
+    debug_tip: Option<B256>,
+    trusted_peer: Option<String>,
     morph_args: Option<MorphArgs>,
 }
 
@@ -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<String>) -> 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,13 +339,19 @@ 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::<MorphNode, _>::new(
             self.num_nodes,
             Arc::new(chain_spec),
             morph_payload_attributes,
         )
         .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()

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;

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(())
+}