feat(node/p2p): extend network actor test with block propagation. reduce test flakiness

bin/node/src/flags/p2p.rs

@@ -402,6 +402,7 @@ impl P2PArgs {
             discovery_interval: Duration::from_secs(self.discovery_interval),
             discovery_address,
             discovery_randomize: self.discovery_randomize.map(Duration::from_secs),
+            enr_update: !static_ip,
             gossip_address,
             keypair,
             unsafe_block_signer: chain_unsafe_block_signer,

crates/node/gossip/src/driver.rs

@@ -185,7 +185,7 @@ where
     /// - Tells the swarm to listen on the given [`Multiaddr`].
     ///
     /// Waits for the swarm to start listen before returning and connecting to peers.
-    pub async fn start(&mut self) -> Result<(), TransportError<std::io::Error>> {
+    pub async fn start(&mut self) -> Result<Multiaddr, TransportError<std::io::Error>> {
         // Start the sync request/response protocol handler.
         self.sync_protocol_handler();
 
@@ -196,16 +196,18 @@ where
                 {
                     if id == listener_id {
                         info!(target: "gossip", "Swarm now listening on: {address}");
-                        break;
+
+                        self.addr = address.clone();
+
+                        return Ok(address);
                     }
                 }
             },
             Err(err) => {
                 error!(target: "gossip", "Fail to listen on {}: {err}", self.addr);
-                return Err(err);
+                Err(err)
             }
         }
-        Ok(())
     }
 
     /// Returns the local peer id.

crates/node/service/Cargo.toml

@@ -76,6 +76,8 @@ anyhow.workspace = true
 backon.workspace = true
 alloy-primitives = { workspace = true, features = ["k256"] }
 alloy-rpc-types-engine = { workspace = true, features = ["arbitrary"] }
+alloy-consensus = { workspace = true, features = ["arbitrary"] }
+op-alloy-consensus = { workspace = true, features = ["arbitrary", "k256"] }
 
 [features]
 default = []

crates/node/service/src/actors/network/builder.rs

@@ -24,6 +24,11 @@ pub struct NetworkBuilder {
     pub(super) gossip: GossipDriverBuilder,
     /// A signer for payloads.
     pub(super) signer: Option<BlockSigner>,
+    /// Whether to update the ENR socket after the libp2p Swarm is started.
+    /// This is set to true by default.
+    /// This may be set to false if the node is configured to use a static advertised address (when
+    /// used with a nat for example).
+    pub(super) enr_update: bool,
 }
 
 impl From<NetworkConfig> for NetworkBuilder {
@@ -35,7 +40,9 @@ impl From<NetworkConfig> for NetworkBuilder {
             config.keypair,
             config.discovery_address,
             config.discovery_config,
+            config.gossip_signer,
         )
+        .with_enr_update(config.enr_update)
         .with_discovery_randomize(config.discovery_randomize)
         .with_bootstore(config.bootstore)
         .with_bootnodes(config.bootnodes)
@@ -45,7 +52,6 @@ impl From<NetworkConfig> for NetworkBuilder {
         .with_peer_monitoring(config.monitor_peers)
         .with_topic_scoring(config.topic_scoring)
         .with_gater_config(config.gater_config)
-        .with_signer(config.gossip_signer)
     }
 }
 
@@ -58,6 +64,7 @@ impl NetworkBuilder {
         keypair: Keypair,
         discovery_address: LocalNode,
         discovery_config: discv5::Config,
+        signer: Option<BlockSigner>,
     ) -> Self {
         Self {
             discovery: Discv5Builder::new(
@@ -71,10 +78,16 @@ impl NetworkBuilder {
                 gossip_addr,
                 keypair,
             ),
-            signer: None,
+            signer,
+            enr_update: true,
         }
     }
 
+    /// Sets the ENR update flag for the [`NetworkBuilder`].
+    pub fn with_enr_update(self, enr_update: bool) -> Self {
+        Self { enr_update, ..self }
+    }
+
     /// Sets the configuration for the connection gater.
     pub fn with_gater_config(self, config: GaterConfig) -> Self {
         Self { gossip: self.gossip.with_gater_config(config), ..self }
@@ -150,7 +163,13 @@ impl NetworkBuilder {
         let (gossip, unsafe_block_signer_sender) = self.gossip.build()?;
         let discovery = self.discovery.build()?;
 
-        Ok(NetworkDriver { gossip, discovery, unsafe_block_signer_sender, signer: self.signer })
+        Ok(NetworkDriver {
+            gossip,
+            discovery,
+            unsafe_block_signer_sender,
+            signer: self.signer,
+            enr_update: self.enr_update,
+        })
     }
 }
 
@@ -199,6 +218,7 @@ mod tests {
             keypair,
             discovery_address,
             discovery_config,
+            None,
         )
     }
 

crates/node/service/src/actors/network/config.rs

@@ -22,6 +22,8 @@ pub struct NetworkConfig {
     pub discovery_interval: Duration,
     /// The interval to remove peers from the discovery service.
     pub discovery_randomize: Option<Duration>,
+    /// Whether to update the ENR socket when the gossip listen address changes.
+    pub enr_update: bool,
     /// The gossip address.
     pub gossip_address: libp2p::Multiaddr,
     /// The unsafe block signer.
@@ -84,6 +86,7 @@ impl NetworkConfig {
             discovery_randomize: Self::DEFAULT_DISCOVERY_RANDOMIZE,
             gossip_address,
             unsafe_block_signer,
+            enr_update: true,
             keypair: Keypair::generate_secp256k1(),
             bootnodes: Default::default(),
             bootstore: Default::default(),

crates/node/service/src/actors/network/driver.rs

@@ -1,9 +1,12 @@
+use std::net::{IpAddr, SocketAddr};
+
 use alloy_primitives::Address;
+use discv5::multiaddr::Protocol;
 use futures::future::OptionFuture;
 use kona_disc::Discv5Driver;
 use kona_gossip::{ConnectionGater, GossipDriver, PEER_SCORE_INSPECT_FREQUENCY};
 use kona_sources::{BlockSigner, BlockSignerStartError};
-use libp2p::TransportError;
+use libp2p::{Multiaddr, TransportError};
 use tokio::sync::watch;
 
 use crate::actors::network::handler::NetworkHandler;
@@ -15,6 +18,11 @@ pub struct NetworkDriver {
     pub gossip: GossipDriver<ConnectionGater>,
     /// The discovery driver.
     pub discovery: Discv5Driver,
+    /// Whether to update the ENR socket after the libp2p Swarm is started.
+    /// This is set to true by default.
+    /// This may be set to false if the node is configured to use a static advertised address (when
+    /// used with a nat for example).
+    pub enr_update: bool,
     /// The unsafe block signer sender.
     pub unsafe_block_signer_sender: watch::Sender<Address>,
     /// A block signer. This is optional and should be set if the node is configured to sign blocks
@@ -30,17 +38,46 @@ pub enum NetworkDriverError {
     /// An error occurred starting the block signer client.
     #[error("error starting block signer client: {0}")]
     BlockSignerStartError(#[from] BlockSignerStartError),
+    /// An error occurred parsing the gossip listen address.
+    #[error("error parsing gossip listen address: {0}")]
+    InvalidGossipListenAddr(Multiaddr),
 }
 
 impl NetworkDriver {
     /// Starts the network.
     pub async fn start(mut self) -> Result<NetworkHandler, NetworkDriverError> {
+        // Start the libp2p Swarm
+        let gossip_listen_addr = self.gossip.start().await?;
+
+        if self.enr_update {
+            // Update the local ENR socket to the gossip listen address.
+            // Parse the multiaddr to a socket address.
+            let ip_address = gossip_listen_addr
+                .iter()
+                .find_map(|p| match p {
+                    Protocol::Ip4(ip) => Some(IpAddr::V4(ip)),
+                    Protocol::Ip6(ip) => Some(IpAddr::V6(ip)),
+                    _ => None,
+                })
+                .ok_or_else(|| {
+                    NetworkDriverError::InvalidGossipListenAddr(gossip_listen_addr.clone())
+                })?;
+            let port = gossip_listen_addr
+                .iter()
+                .find_map(|p| match p {
+                    Protocol::Tcp(port) => Some(port),
+                    _ => None,
+                })
+                .ok_or_else(|| {
+                    NetworkDriverError::InvalidGossipListenAddr(gossip_listen_addr.clone())
+                })?;
+
+            self.discovery.disc.update_local_enr_socket(SocketAddr::new(ip_address, port), true);
+        }
+
         // Start the discovery service.
         let (handler, enr_receiver) = self.discovery.start();
 
-        // Start the libp2p Swarm
-        self.gossip.start().await?;
-
         // We are checking the peer scores every [`PEER_SCORE_INSPECT_FREQUENCY`] seconds.
         let peer_score_inspector = tokio::time::interval(*PEER_SCORE_INSPECT_FREQUENCY);
 

crates/node/service/tests/actors/generator/block_builder.rs

@@ -0,0 +1,135 @@
+use std::time::SystemTime;
+
+use alloy_consensus::{Block, EMPTY_OMMER_ROOT_HASH};
+use alloy_eips::Encodable2718;
+use alloy_primitives::Bytes;
+use arbitrary::{Arbitrary, Unstructured};
+use libp2p::bytes::BufMut;
+use op_alloy_consensus::OpTxEnvelope;
+use op_alloy_rpc_types_engine::{OpExecutionPayload, OpExecutionPayloadEnvelope};
+
+use crate::actors::generator::seed::SeedGenerator;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub(crate) enum PayloadVersion {
+    V1,
+    #[allow(dead_code)]
+    V2,
+    #[allow(dead_code)]
+    V3,
+    #[allow(dead_code)]
+    V4,
+}
+
+impl SeedGenerator {
+    /// Generate a random op execution payload.
+    pub(crate) fn random_valid_payload(
+        &mut self,
+        version: PayloadVersion,
+    ) -> anyhow::Result<OpExecutionPayloadEnvelope> {
+        let block: Block<OpTxEnvelope> = match version {
+            PayloadVersion::V1 => self.v1_valid_block(),
+            PayloadVersion::V2 => self.v2_valid_block(),
+            PayloadVersion::V3 => self.v3_valid_block(),
+            PayloadVersion::V4 => self.v4_valid_block(),
+        };
+
+        let (payload, _) = OpExecutionPayload::from_block_slow(&block);
+
+        let parent_beacon_block_root = block.header.parent_beacon_block_root;
+
+        let envelope =
+            OpExecutionPayloadEnvelope { parent_beacon_block_root, execution_payload: payload };
+
+        Ok(envelope)
+    }
+
+    fn valid_block(&mut self) -> Block<OpTxEnvelope> {
+        // Simulate some random data
+        let data = self.random_bytes(1024 * 1024);
+
+        // Create unstructured data with the random bytes
+        let u = Unstructured::new(&data);
+
+        // Generate a random instance of MyStruct
+        let mut block: Block<OpTxEnvelope> = Block::arbitrary_take_rest(u).unwrap();
+
+        let transactions: Vec<Bytes> =
+            block.body.transactions().map(|tx| tx.encoded_2718().into()).collect();
+
+        let transactions_root =
+            alloy_consensus::proofs::ordered_trie_root_with_encoder(&transactions, |item, buf| {
+                buf.put_slice(item)
+            });
+
+        block.header.transactions_root = transactions_root;
+
+        // We always need to set the base fee per gas to a positive value to ensure the block is
+        // valid.
+        block.header.base_fee_per_gas =
+            Some(block.header.base_fee_per_gas.unwrap_or_default().saturating_add(1));
+
+        let current_timestamp =
+            SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
+        block.header.timestamp = current_timestamp;
+
+        block
+    }
+
+    /// Make the block v1 compatible
+    fn v1_valid_block(&mut self) -> Block<OpTxEnvelope> {
+        let mut block = self.valid_block();
+        block.header.withdrawals_root = None;
+        block.header.blob_gas_used = None;
+        block.header.excess_blob_gas = None;
+        block.header.parent_beacon_block_root = None;
+        block.header.requests_hash = None;
+        block.header.ommers_hash = EMPTY_OMMER_ROOT_HASH;
+        block.header.difficulty = Default::default();
+        block.header.nonce = Default::default();
+
+        block
+    }
+
+    /// Make the block v2 compatible
+    pub(crate) fn v2_valid_block(&mut self) -> Block<OpTxEnvelope> {
+        let mut block = self.v1_valid_block();
+
+        block.body.withdrawals = Some(vec![].into());
+        let withdrawals_root = alloy_consensus::proofs::calculate_withdrawals_root(
+            &block.body.withdrawals.clone().unwrap_or_default(),
+        );
+
+        block.header.withdrawals_root = Some(withdrawals_root);
+
+        block
+    }
+
+    /// Make the block v3 compatible
+    pub(crate) fn v3_valid_block(&mut self) -> Block<OpTxEnvelope> {
+        let mut block = self.valid_block();
+
+        block.body.withdrawals = Some(vec![].into());
+        let withdrawals_root = alloy_consensus::proofs::calculate_withdrawals_root(
+            &block.body.withdrawals.clone().unwrap_or_default(),
+        );
+        block.header.withdrawals_root = Some(withdrawals_root);
+
+        block.header.blob_gas_used = Some(0);
+        block.header.excess_blob_gas = Some(0);
+        block.header.parent_beacon_block_root =
+            Some(block.header.parent_beacon_block_root.unwrap_or_default());
+
+        block.header.requests_hash = None;
+        block.header.ommers_hash = EMPTY_OMMER_ROOT_HASH;
+        block.header.difficulty = Default::default();
+        block.header.nonce = Default::default();
+
+        block
+    }
+
+    /// Make the block v4 compatible
+    pub(crate) fn v4_valid_block(&mut self) -> Block<OpTxEnvelope> {
+        self.v3_valid_block()
+    }
+}

crates/node/service/tests/actors/generator/mod.rs

@@ -0,0 +1,2 @@
+pub(crate) mod block_builder;
+pub(crate) mod seed;