fix(l1): back off failed dials, pace discovery by saturation and add --p2p.netrestrict

Cargo.toml

@@ -152,6 +152,7 @@ tempfile = "3.8"
 uuid = { version = "1.18.1", features = ["v4"] }
 tower-http = { version = "0.6.2", features = ["cors"] }
 indexmap = { version = "2.11.4" }
+ipnet = "2.12.0"
 k256 = "0.13.4"
 anyhow = "1.0.86"
 

cmd/ethrex/cli.rs

@@ -18,6 +18,7 @@ use ethrex_blockchain::{
 use ethrex_common::types::{Block, DEFAULT_BUILDER_GAS_CEIL, Genesis, validate_block_body};
 use ethrex_p2p::{
     discovery::INITIAL_LOOKUP_INTERVAL_MS,
+    netrestrict::IpNet,
     peer_table::TARGET_PEERS,
     sync::{HistoryChain, SyncMode},
     tx_broadcaster::BROADCAST_INTERVAL_MS,
@@ -484,6 +485,18 @@ pub struct Options {
         env = "ETHREX_P2P_LOOKUP_INTERVAL"
     )]
     pub lookup_interval: f64,
+    #[arg(
+        long = "p2p.netrestrict",
+        value_parser = clap::value_parser!(IpNet),
+        value_name = "CIDR_LIST",
+        value_delimiter = ',',
+        num_args = 1..,
+        help = "Restrict P2P traffic to the given IP networks (comma separated CIDRs).",
+        long_help = "Comma separated IP networks in CIDR notation, e.g. 10.0.0.0/8,172.16.0.0/12. Nodes discovered outside these networks are ignored, bootnodes outside are dropped, and inbound TCP and UDP from outside is discarded. Meant for private devnets and other closed networks. Unrestricted when not set.",
+        help_heading = "P2P options",
+        env = "ETHREX_P2P_NETRESTRICT"
+    )]
+    pub netrestrict: Vec<IpNet>,
     #[arg(
         long = "builder.extra-data",
         default_value = get_minimal_client_version(),
@@ -636,6 +649,7 @@ impl Default for Options {
             tx_broadcasting_time_interval: Default::default(),
             target_peers: Default::default(),
             lookup_interval: INITIAL_LOOKUP_INTERVAL_MS,
+            netrestrict: Default::default(),
             extra_data: get_minimal_client_version(),
             gas_limit: DEFAULT_BUILDER_GAS_CEIL,
             max_blobs_per_block: None,

cmd/ethrex/initializers.rs

@@ -17,6 +17,7 @@ use ethrex_metrics::rpc::initialize_rpc_metrics;
 use ethrex_p2p::rlpx::initiator::RLPxInitiator;
 use ethrex_p2p::{
     DiscoveryConfig,
+    netrestrict::NetRestrict,
     network::P2PContext,
     peer_handler::PeerHandler,
     peer_table::{PeerTable, PeerTableServer},
@@ -466,10 +467,16 @@ pub async fn init_network(
 
     let bootnodes = get_bootnodes(opts, network, datadir);
 
+    let netrestrict = NetRestrict::new(opts.netrestrict.clone());
+    if !netrestrict.is_unrestricted() {
+        info!(%netrestrict, "P2P traffic restricted to the configured networks");
+    }
+
     let discovery_config = DiscoveryConfig {
         discv4_enabled: opts.discv4_enabled,
         discv5_enabled: opts.discv5_enabled,
         nat_extip_set: opts.nat_extip.is_some(),
+        netrestrict,
     };
 
     ethrex_p2p::start_network(context, bootnodes, discovery_config, shared_local_node)
@@ -926,8 +933,12 @@ pub async fn init_l1(
         record: local_node_record,
     }));
 
-    let peer_table =
-        PeerTableServer::spawn(local_p2p_node.node_id(), opts.target_peers, store.clone());
+    let peer_table = PeerTableServer::spawn(
+        local_p2p_node.node_id(),
+        opts.target_peers,
+        store.clone(),
+        NetRestrict::new(opts.netrestrict.clone()),
+    );
 
     // TODO: Check every module starts properly.
     let tracker = TaskTracker::new();
@@ -946,6 +957,7 @@ pub async fn init_l1(
         None,
         opts.tx_broadcasting_time_interval,
         opts.lookup_interval,
+        NetRestrict::new(opts.netrestrict.clone()),
     )
     .expect("P2P context could not be created");
 

cmd/ethrex/l2/initializers.rs

@@ -16,6 +16,7 @@ use ethrex_common::types::fee_config::{FeeConfig, L1FeeConfig, OperatorFeeConfig
 use ethrex_l2::sequencer::block_producer::{self, block_producer_protocol};
 use ethrex_l2::sequencer::l1_committer::{self, l1_committer_protocol, regenerate_state};
 use ethrex_p2p::{
+    netrestrict::NetRestrict,
     network::P2PContext,
     peer_handler::PeerHandler,
     peer_table::PeerTableServer,
@@ -303,6 +304,7 @@ pub async fn init_l2(
             local_p2p_node.node_id(),
             opts.node_opts.target_peers,
             store.clone(),
+            NetRestrict::new(opts.node_opts.netrestrict.clone()),
         );
         let p2p_context = P2PContext::new(
             local_p2p_node.clone(),
@@ -332,6 +334,7 @@ pub async fn init_l2(
             }),
             opts.node_opts.tx_broadcasting_time_interval,
             opts.node_opts.lookup_interval,
+            NetRestrict::new(opts.node_opts.netrestrict.clone()),
         )
         .expect("P2P context could not be created");
         let initiator = RLPxInitiator::spawn(p2p_context.clone());

crates/networking/p2p/Cargo.toml

@@ -37,6 +37,7 @@ spawned-concurrency.workspace = true
 sha2.workspace = true
 futures.workspace = true
 indexmap.workspace = true
+ipnet.workspace = true
 rustc-hash.workspace = true
 rocksdb = { workspace = true, optional = true }
 prometheus = "0.14.0"

crates/networking/p2p/discovery/discv4_handlers.rs

@@ -129,11 +129,39 @@ impl DiscoveryServer {
         }
 
         // Remove finished lookups
+        let had_active = self
+            .discv4
+            .as_ref()
+            .is_some_and(|s| !s.active_lookups.is_empty());
         self.discv4
             .as_mut()
             .expect("discv4 state must exist")
             .active_lookups
             .retain(|(l, _)| !l.is_finished());
+        let just_finished = had_active
+            && self
+                .discv4
+                .as_ref()
+                .is_some_and(|s| s.active_lookups.is_empty());
+
+        // A lookup just ended. If it added nothing to the peer table, the
+        // network is saturated: rather than start the next one on this same
+        // tick, let the backoff in `get_lookup_interval` space them out.
+        if just_finished {
+            let count = self.peer_table.discovered_count().await?;
+            let discv4 = self.discv4.as_mut().expect("discv4 state must exist");
+            if count > discv4.lookup_started_at_count {
+                discv4.empty_lookups_in_a_row = 0;
+            } else {
+                discv4.empty_lookups_in_a_row = discv4.empty_lookups_in_a_row.saturating_add(1);
+                trace!(
+                    protocol = "discv4",
+                    empty_in_a_row = discv4.empty_lookups_in_a_row,
+                    "Lookup found nothing new, backing off"
+                );
+                return Ok(());
+            }
+        }
 
         // If a lookup is already active, advance it instead of starting a new
         // one. Lookups are timer-driven: each tick sends the next alpha queries.
@@ -180,7 +208,9 @@ impl DiscoveryServer {
         let mut buf = BytesMut::new();
         msg.encode_with_header(&mut buf, &self.signer);
 
+        let started_at_count = self.peer_table.discovered_count().await?;
         let discv4 = self.discv4.as_mut().expect("discv4 state must exist");
+        discv4.lookup_started_at_count = started_at_count;
         discv4.active_lookups.push((lookup, buf));
 
         // Fire the initial queries for the new lookup
@@ -338,18 +368,15 @@ impl DiscoveryServer {
         {
             self.discv4_send_ping(&node).await?;
         } else {
+            // A contact we hold no record for counts as seq 0, as on the discv5
+            // side. Otherwise a discv4 node's ENR, and with it the fork-id verdict
+            // that keeps it from being dialed on the wrong chain, would only ever
+            // be fetched by the slow random ENR lookup.
             let node_id = node_id(&sender_public_key);
-            let stored_enr_seq = self
-                .peer_table
-                .get_contact(node_id)
-                .await?
-                .and_then(|c| c.record)
-                .map(|r| r.seq);
-
-            let received_enr_seq = ping_message.enr_seq;
-
-            if let (Some(received), Some(stored)) = (received_enr_seq, stored_enr_seq)
-                && received > stored
+            if let Some(contact) = self.peer_table.get_contact(node_id).await?
+                && contact.was_validated()
+                && let Some(received) = ping_message.enr_seq
+                && received > contact.record.as_ref().map_or(0, |r| r.seq)
             {
                 self.discv4_send_enr_request(&node).await?;
             }
@@ -369,10 +396,10 @@ impl DiscoveryServer {
         let ping_id = Bytes::copy_from_slice(message.ping_hash.as_bytes());
         self.peer_table.record_pong_received(node_id, ping_id)?;
 
-        let stored_enr_seq = contact.record.map(|r| r.seq);
-        let received_enr_seq = message.enr_seq;
-        if let (Some(received), Some(stored)) = (received_enr_seq, stored_enr_seq)
-            && received > stored
+        // No record yet counts as seq 0 (see `discv4_handle_ping`).
+        let stored_enr_seq = contact.record.as_ref().map_or(0, |r| r.seq);
+        if let Some(received) = message.enr_seq
+            && received > stored_enr_seq
         {
             self.discv4_send_enr_request(&contact.node).await?;
         }

crates/networking/p2p/discovery/discv5_handlers.rs

@@ -296,11 +296,39 @@ impl DiscoveryServer {
         }
 
         // Remove finished lookups
+        let had_active = self
+            .discv5
+            .as_ref()
+            .is_some_and(|s| !s.active_lookups.is_empty());
         self.discv5
             .as_mut()
             .expect("discv5 state must exist")
             .active_lookups
             .retain(|l| !l.is_finished());
+        let just_finished = had_active
+            && self
+                .discv5
+                .as_ref()
+                .is_some_and(|s| s.active_lookups.is_empty());
+
+        // A lookup just ended. If it added nothing to the peer table, the
+        // network is saturated: rather than start the next one on this same
+        // tick, let the backoff in `get_lookup_interval` space them out.
+        if just_finished {
+            let count = self.peer_table.discovered_count().await?;
+            let discv5 = self.discv5.as_mut().expect("discv5 state must exist");
+            if count > discv5.lookup_started_at_count {
+                discv5.empty_lookups_in_a_row = 0;
+            } else {
+                discv5.empty_lookups_in_a_row = discv5.empty_lookups_in_a_row.saturating_add(1);
+                trace!(
+                    protocol = "discv5",
+                    empty_in_a_row = discv5.empty_lookups_in_a_row,
+                    "Lookup found nothing new, backing off"
+                );
+                return Ok(());
+            }
+        }
 
         // If a lookup is already active, advance it instead of starting a new
         // one. Lookups are timer-driven: each tick sends the next alpha queries.
@@ -338,7 +366,9 @@ impl DiscoveryServer {
             "Starting new iterative lookup"
         );
         let lookup = IterativeLookup::new(target_id, seed);
+        let started_at_count = self.peer_table.discovered_count().await?;
         let discv5 = self.discv5.as_mut().expect("discv5 state must exist");
+        discv5.lookup_started_at_count = started_at_count;
         discv5.active_lookups.push(lookup);
 
         // Fire the initial queries for the new lookup

crates/networking/p2p/discovery/mod.rs

@@ -20,6 +20,7 @@ pub mod server;
 pub use ip_predictor::IpPredictor;
 pub use server::{DiscoveryServer, DiscoveryServerError, is_discv4_packet};
 
+use crate::netrestrict::NetRestrict;
 use std::time::Duration;
 
 /// Configuration for which discovery protocols to enable.
@@ -30,6 +31,9 @@ pub struct DiscoveryConfig {
     /// Set to true when `--nat extip:<addr>` was supplied; locks the IP predictor
     /// from overwriting the user-specified external address.
     pub nat_extip_set: bool,
+    /// IP networks peers must fall in. Packets from outside are dropped before
+    /// they are decoded, and bootnodes outside are not contacted.
+    pub netrestrict: NetRestrict,
 }
 
 /// Lookup interval bounds for the RLPx initiator's connection attempts. The
@@ -53,3 +57,50 @@ pub fn lookup_interval_function(progress: f64, lower_limit: f64, upper_limit: f6
         (1000f64 * (ease_in_out_cubic * (upper_limit - lower_limit) + lower_limit)).round() as u64,
     )
 }
+
+/// Interval before the *next* lookup starts, once the previous one has finished.
+///
+/// `empty_lookups_in_a_row` counts finished lookups that added nothing to the
+/// peer table. Each one doubles the wait from `lower_limit`, up to
+/// `upper_limit`: a network whose every node is already known stops being asked
+/// the same question twice a second. The completion-based pacing still applies,
+/// so the result is never shorter than [`lookup_interval_function`] alone.
+pub fn next_lookup_interval(
+    progress: f64,
+    empty_lookups_in_a_row: u32,
+    lower_limit: f64,
+    upper_limit: f64,
+) -> Duration {
+    let paced = lookup_interval_function(progress, lower_limit, upper_limit);
+    if empty_lookups_in_a_row == 0 {
+        return paced;
+    }
+    // 2^16 already dwarfs any sane upper limit; the clamp only keeps `powi` finite.
+    let backoff_ms =
+        (lower_limit * 2f64.powi(empty_lookups_in_a_row.min(16) as i32)).min(upper_limit);
+    paced.max(Duration::from_micros((1000f64 * backoff_ms).round() as u64))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn saturation_backoff_doubles_from_the_lower_limit_and_caps() {
+        let at = |empty| next_lookup_interval(0.0, empty, 500.0, 10_000.0);
+        assert_eq!(at(0), Duration::from_millis(500));
+        assert_eq!(at(1), Duration::from_millis(1_000));
+        assert_eq!(at(2), Duration::from_millis(2_000));
+        assert_eq!(at(3), Duration::from_millis(4_000));
+        assert_eq!(at(4), Duration::from_millis(8_000));
+        assert_eq!(at(5), Duration::from_millis(10_000));
+        assert_eq!(at(1_000), Duration::from_millis(10_000));
+    }
+
+    #[test]
+    fn saturation_backoff_never_undercuts_completion_pacing() {
+        let paced = lookup_interval_function(0.9, 500.0, 10_000.0);
+        assert!(paced > Duration::from_millis(1_000));
+        assert_eq!(next_lookup_interval(0.9, 1, 500.0, 10_000.0), paced);
+    }
+}