From f9b7eb861ce4f361b6d2b7e8d884bef467deca0b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:28:25 -0300 Subject: [PATCH 1/4] feat(discovery): report libp2p connections so discv5 paces its lookups ethrex used to pace its discv5 lookups off its peer table's connected count, which only RLPx could fill. A libp2p client registered nothing, so the count was permanently 0, completion was permanently 0, and lookups never eased off the 500ms startup rate: roughly 20x the intended steady-state FindNode traffic, for the life of the process. ethrex PR #7217 moves the contact table into the discovery server and gives a consumer `DiscoveryHandle::record_peer_event`, so any consumer can report its own connections whatever transport they run over. Pin the three ethrex crates to that PR's head and report every libp2p connection opened and closed. `--discovery.target-peers` now feeds the pacing it was deliberately kept out of, and `PEER_TABLE_TARGET_PEERS`, the placeholder that only existed to keep the old division off `NaN`, is gone. Reporting needs the discv5 node id behind a libp2p `PeerId`. Both names derive from the same secp256k1 key, and libp2p inlines a key that short into the peer id, so it is recoverable: `node_id` does that round trip, and pins it against `Node::from_enr` so the two halves cannot drift. The peer table is gone from this crate with it. Two consequences: - The filter is what discovery now takes, and `next_dial_candidate` hands back a `Node` rather than a `Contact`, so the ENR the dial loop used to read is no longer in reach. Judging a record is the one moment we hold it, so `LeanFilter` files each admitted peer's multiaddr and `attnets` in `DialTargets` as it goes, and the dial loop looks the node id back up there. Discovery still decides *whether* to dial; we only supply *how*. - `--discovery.advertise-ip` maps onto ethrex's new `nat_extip_set`, so an operator who names an external address keeps it: discv5's PONG voting no longer overwrites what was explicitly asked for. The pin is a commit rather than a branch, for the reason the release tag it replaces was chosen over `main`. It goes back to a tag once #7217 is in a release. --- CLAUDE.md | 6 +- Cargo.lock | 42 ++-- crates/net/p2p/Cargo.toml | 18 +- crates/net/p2p/src/discovery/admission.rs | 229 +++++++++++++++++----- crates/net/p2p/src/discovery/dial.rs | 110 +++++++---- crates/net/p2p/src/discovery/mod.rs | 143 ++++++++------ crates/net/p2p/src/discovery/node_id.rs | 83 ++++++++ crates/net/p2p/src/lib.rs | 7 +- docs/discovery.md | 54 ++--- 9 files changed, 498 insertions(+), 194 deletions(-) create mode 100644 crates/net/p2p/src/discovery/node_id.rs diff --git a/CLAUDE.md b/CLAUDE.md index 05cf50485..957a5e685 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -282,9 +282,11 @@ actual_slot = finalized_slot + 1 + relative_index ### Peer Discovery (discv5, opt-in) - Off by default; `--discovery.enable` plus `--discovery.port` (own UDP socket, must differ from `--gossipsub-port`) -- Reuses ethrex's `DiscoveryServer` + `PeerTable` with discv4 disabled; `spawn` takes the prepared lean ENR, so the record ethrex serves is the one we report +- Reuses ethrex's `DiscoveryServer` with discv4 disabled; `spawn` takes the prepared lean ENR, so the record ethrex serves is the one we report - ENR follows the beacon phase0 spec: `ip`/`udp`/`quic`/`secp256k1`/`eth2`/`attnets` -- Admission mirrors lighthouse: `eth2.fork_digest` must match, `next_fork_*` may differ, `quic` entry required. Handed to the peer table as `LeanFilter: PeerFilter`, so records are judged on arrival, not at dial time; a reject is re-judged on a higher-`seq` ENR +- Admission mirrors lighthouse: `eth2.fork_digest` must match, `next_fork_*` may differ, `quic` entry required. Handed to discovery as `LeanFilter: PeerFilter`, so records are judged on arrival, not at dial time; a reject is re-judged on a higher-`seq` ENR +- The filter files each admitted peer in `DialTargets`, since discovery hands back a `Node` that carries no `quic`/`attnets`; the dial loop looks the node id back up there +- Every libp2p connection is reported with `DiscoveryHandle::record_peer_event`, which is what lets discovery pace its lookups against `--discovery.target-peers` - Candidates ranked by uncovered attestation subnets. See [`docs/discovery.md`](docs/discovery.md) ### Retry Strategy on Block Requests diff --git a/Cargo.lock b/Cargo.lock index dacda8d5a..f2020929c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -172,7 +172,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -183,7 +183,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1848,7 +1848,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2128,7 +2128,7 @@ dependencies = [ [[package]] name = "ethrex-blockchain" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "crossbeam", @@ -2153,7 +2153,7 @@ dependencies = [ [[package]] name = "ethrex-common" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "crc32fast", @@ -2187,7 +2187,7 @@ dependencies = [ [[package]] name = "ethrex-crypto" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "ark-bn254", "ark-ec", @@ -2209,7 +2209,7 @@ dependencies = [ [[package]] name = "ethrex-guest-program" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "ethereum-types", @@ -2232,7 +2232,7 @@ dependencies = [ [[package]] name = "ethrex-l2-common" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "ethereum-types", @@ -2250,7 +2250,7 @@ dependencies = [ [[package]] name = "ethrex-levm" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "derive_more 1.0.0", @@ -2268,7 +2268,7 @@ dependencies = [ [[package]] name = "ethrex-metrics" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "ethrex-common", "serde", @@ -2280,7 +2280,7 @@ dependencies = [ [[package]] name = "ethrex-p2p" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "aes", "aes-gcm", @@ -2322,7 +2322,7 @@ dependencies = [ [[package]] name = "ethrex-rlp" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "ethereum-types", @@ -2332,7 +2332,7 @@ dependencies = [ [[package]] name = "ethrex-storage" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "anyhow", "bytes", @@ -2354,7 +2354,7 @@ dependencies = [ [[package]] name = "ethrex-trie" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "anyhow", "bytes", @@ -2374,7 +2374,7 @@ dependencies = [ [[package]] name = "ethrex-vm" version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?tag=v25.0.0#f3b90bc6dc57704940d1ff4f2bc74a8dd5788df2" +source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" dependencies = [ "bytes", "derive_more 1.0.0", @@ -3134,7 +3134,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.62.2", + "windows-core 0.57.0", ] [[package]] @@ -5057,7 +5057,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6519,7 +6519,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7003,7 +7003,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7274,7 +7274,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8100,7 +8100,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/crates/net/p2p/Cargo.toml b/crates/net/p2p/Cargo.toml index 712ba60f7..3a032d24c 100644 --- a/crates/net/p2p/Cargo.toml +++ b/crates/net/p2p/Cargo.toml @@ -35,13 +35,17 @@ rand = "0.8" # Required for NodeEnr parsing. # -# Pinned to an ethrex release tag, for the unified discovery module, the peer -# table, and the `PeerFilter` seam this crate builds on. A tag rather than a -# branch: a `main` that moves under us would change what this crate builds -# against with nothing in the manifest saying so. -ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", tag = "v25.0.0" } -ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", tag = "v25.0.0" } -ethrex-common = { git = "https://github.com/lambdaclass/ethrex", tag = "v25.0.0" } +# Pinned to the head of ethrex PR #7217, which moves the contact table into the +# discovery server and gives a consumer `DiscoveryHandle::record_peer_event` to +# report its own connections with. That is what lets a libp2p client feed +# ethrex's lookup pacing; see `docs/discovery.md`. +# +# A commit rather than a branch, for the reason the release tag it replaces was +# chosen over `main`: what this crate builds against must not move without the +# manifest saying so. Goes back to a tag once #7217 is in a release. +ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } +ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } +ethrex-common = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } # Version pinned to ethrex's workspace: `SecretKey` crosses the API boundary. secp256k1 = { version = "0.30.0", default-features = false, features = ["global-context"] } diff --git a/crates/net/p2p/src/discovery/admission.rs b/crates/net/p2p/src/discovery/admission.rs index 51524ec65..61f0ca79e 100644 --- a/crates/net/p2p/src/discovery/admission.rs +++ b/crates/net/p2p/src/discovery/admission.rs @@ -7,18 +7,22 @@ //! //! Lighthouse applies these inside the discovery query itself, via //! `discv5.find_node_predicate`. ethlambda hands them to ethrex as a -//! [`LeanFilter`], which the peer table consults the moment each ENR arrives. A -//! peer that does not belong is judged where the record lands, not at dial time, -//! and is not offered for dialing again until it publishes a higher-`seq` -//! record, which the peer table runs through the filter afresh. +//! [`LeanFilter`], which the discovery server consults the moment each ENR +//! arrives. A peer that does not belong is judged where the record lands, not at +//! dial time, and is not offered for dialing again until it publishes a +//! higher-`seq` record, which discovery runs through the filter afresh. //! -//! So the dial loop filters nothing: every contact it draws has already passed, -//! and all it does is turn the record into something dialable -//! ([`LeanFilter::dial_target`]) and rank what it got +//! So the dial loop filters nothing: every node it draws has already passed. It +//! does not even read ENRs, because discovery hands out an +//! [`ethrex_p2p::types::Node`], which knows nothing of `quic` or `attnets`. +//! Instead the filter files what it admits in [`DialTargets`] as it judges it, +//! and the dial loop looks the node id back up there and ranks what it got //! ([`rank_by_uncovered_subnets`]). -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex}; +use ethrex_common::H256; use ethrex_p2p::peer_filter::PeerFilter; use ethrex_p2p::types::NodeRecord; use libp2p::{Multiaddr, PeerId}; @@ -29,20 +33,87 @@ use super::enr::{ ATTNETS_ENR_KEY, ETH2_ENR_KEY, EnrForkId, read_ip, read_public_key, read_quic_port, subnets_from_attnets, }; +use super::node_id::node_id_from_public_key; use crate::quic_multiaddr; +/// Dial targets held at once, past which a new one is dropped. +/// +/// Only records that pass admission land here, so on any real network the map +/// is the size of the lean population and never approaches this. The cap is for +/// the case where that population is manufactured: minting ENRs costs a +/// signature, and an unbounded map would grow with however many an adversary +/// cares to gossip. Matching ethrex's own connection-pool bound keeps our +/// memory in the same order as the table that feeds us. +const MAX_DIAL_TARGETS: usize = 10_000; + /// A peer that passed admission and is ready to dial. #[derive(Debug, Clone, PartialEq)] pub(crate) struct DiscoveredPeer { + /// The discv5 node id, which is how ethrex names this same peer. + pub(crate) node_id: H256, pub(crate) peer_id: PeerId, pub(crate) addr: Multiaddr, /// Attestation subnets the peer advertises in `attnets`. pub(crate) subnets: Vec, } +/// What the filter admitted, keyed by the node id discovery will name it with. +/// +/// Written inside the discovery actor as ENRs arrive, read by the dial loop in +/// the P2P actor, hence the lock. Every critical section is one map operation on +/// data already in hand, so nothing is held across an `.await` and the discovery +/// message loop is never parked on it. +/// +/// This exists because the two halves know different things about the same peer. +/// Discovery decides *whether* to dial, from facts it tracks itself: whether the +/// peer knows us, whether we already tried it, whether we are connected. It +/// hands back a `Node`, which carries an address and a key and nothing else. +/// *How* to dial a lean peer lives in ENR entries discovery has no opinion +/// about, so we keep what we read out of the record the one time we saw it. +#[derive(Clone, Default)] +pub(crate) struct DialTargets(Arc>>); + +impl DialTargets { + /// File a peer the filter just admitted, replacing what an earlier record + /// said about it. + fn record(&self, peer: DiscoveredPeer) { + let mut targets = self.lock(); + if targets.len() >= MAX_DIAL_TARGETS && !targets.contains_key(&peer.node_id) { + debug!( + node_id = %peer.node_id, + "Dropping dial target: already holding MAX_DIAL_TARGETS" + ); + return; + } + targets.insert(peer.node_id, peer); + } + + /// Forget a peer whose latest record no longer passes admission. + /// + /// Discovery stops offering it either way, so this is about not keeping a + /// stale multiaddr for a peer that moved off our network. + fn forget(&self, node_id: &H256) { + self.lock().remove(node_id); + } + + /// How to dial `node_id`, or `None` for a node we never saw an admissible + /// record for. That covers every bootnode discovery knows only as a bare + /// endpoint, which is why it is an ordinary answer rather than a surprise. + pub(crate) fn get(&self, node_id: &H256) -> Option { + self.lock().get(node_id).cloned() + } + + /// The only way the map is reached, so the reason a poisoned lock is + /// impossible is stated once: every critical section is a `HashMap` call + /// with no user code inside it, so no panic can leave the map torn. + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.0.lock().expect("dial targets lock is never poisoned") + } +} + /// Why a discovered peer was turned away. /// -/// No reason is final: the peer table re-runs the filter on every higher-`seq` +/// No reason is final: discovery re-runs the filter on every higher-`seq` /// record, so a peer that adds a `quic` entry or gains an address through /// discv5's IP voting is reconsidered without restarting the process. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -60,18 +131,21 @@ pub(crate) enum RejectReason { MissingAddress, } -/// The spec's admission checks, in the shape ethrex's peer table wants them. +/// The spec's admission checks, in the shape ethrex's discovery server wants +/// them. /// /// Holds what [`admit`] needs to judge a record, so the dial loop no longer -/// carries the local fork id and committee count around: the value handed to -/// [`PeerTableServer::spawn_with_filter`](ethrex_p2p::peer_table::PeerTableServer::spawn_with_filter) -/// judges by the same rules the dial loop later asks for dial targets. It is -/// [`Clone`] because the peer table takes ownership of the filter it runs, and -/// both fields are plain data. -#[derive(Clone)] +/// carries the local fork id and committee count around. Judging and publishing +/// are the same act here: a record that passes is filed in [`DialTargets`] on +/// the way out, which is the only moment the ENR behind a peer is in hand. +/// +/// Not [`Clone`]: discovery takes the filter by value and there is nothing left +/// for a second copy to do. What the dial loop needs is [`Self::dial_targets`], +/// which shares the map rather than the policy. pub struct LeanFilter { fork_id: EnrForkId, attestation_committee_count: u64, + dial_targets: DialTargets, } impl LeanFilter { @@ -79,28 +153,26 @@ impl LeanFilter { Self { fork_id, attestation_committee_count, + dial_targets: DialTargets::default(), } } - /// What to dial for a record that has already been admitted, or `None` if it - /// would not be. - /// - /// The `None` arm is unreachable for a contact drawn from the peer table, - /// since the same policy already judged the same record. It is not an - /// `expect` because the two are only guaranteed to agree while the record is - /// unchanged, and the peer table hands out clones: a caller that reaches - /// this with an arbitrary record should get nothing to dial, not a panic. - pub(crate) fn dial_target(&self, record: &NodeRecord) -> Option { - admit(record, &self.fork_id, self.attestation_committee_count).ok() + /// A handle on what this filter has admitted so far, for the dial loop. + pub(crate) fn dial_targets(&self) -> DialTargets { + self.dial_targets.clone() } } impl PeerFilter for LeanFilter { fn accepts(&self, record: &NodeRecord) -> bool { - admit(record, &self.fork_id, self.attestation_committee_count) - // The only place a rejection is visible: the peer table records - // that the record failed the filter but says nothing about why. - .inspect_err(|reason| { + match admit(record, &self.fork_id, self.attestation_committee_count) { + Ok(peer) => { + self.dial_targets.record(peer); + true + } + // The only place a rejection is visible: discovery records that the + // record failed the filter but says nothing about why. + Err(reason) => { debug!( ip = ?record.pairs().ip, udp_port = ?record.pairs().udp_port, @@ -108,8 +180,16 @@ impl PeerFilter for LeanFilter { ?reason, "Rejecting discovered peer" ); - }) - .is_ok() + // A peer that used to pass and no longer does must not leave a + // dialable address behind: this is the same record being judged + // afresh at a higher `seq`, which is how a node announces it + // moved. Nothing to remove for a peer that never passed. + if let Some(key) = read_public_key(record.pairs()) { + self.dial_targets.forget(&node_id_from_public_key(&key)); + } + false + } + } } } @@ -147,6 +227,7 @@ fn admit( let quic_port = read_quic_port(record).ok_or(RejectReason::NoQuicPort)?; let public_key = read_public_key(pairs).ok_or(RejectReason::BadPublicKey)?; + let node_id = node_id_from_public_key(&public_key); let peer_id = PeerId::from_public_key(&libp2p::identity::PublicKey::from(public_key)); let ip = read_ip(pairs).ok_or(RejectReason::MissingAddress)?; @@ -157,6 +238,7 @@ fn admit( .unwrap_or_default(); Ok(DiscoveredPeer { + node_id, peer_id, addr: quic_multiaddr(ip, quic_port, peer_id), subnets, @@ -260,9 +342,11 @@ mod tests { } impl DiscoveredPeer { - /// A candidate carrying only what ranking looks at. + /// A candidate carrying only what ranking looks at. Ranking never reads + /// the identity, so both ids are placeholders. fn for_test(subnets: Vec) -> Self { Self { + node_id: H256::zero(), peer_id: PeerId::random(), addr: Multiaddr::empty(), subnets, @@ -389,12 +473,36 @@ mod tests { assert_eq!(peer.subnets, vec![2]); } - // --- what the peer table sees, and what it hands back --- + // --- what discovery sees, and what the dial loop finds afterwards --- fn filter() -> LeanFilter { LeanFilter::new(EnrForkId::local(), TEST_COMMITTEE_COUNT) } + /// What the dial loop would find for `record`, going the way it really + /// goes: discovery hands it a node id, and the target must have been filed + /// under that id when the filter judged the record. + fn filed_target(policy: &LeanFilter, record: &NodeRecord) -> Option { + let key = read_public_key(record.pairs())?; + policy.dial_targets().get(&node_id_from_public_key(&key)) + } + + #[test] + fn a_target_is_keyed_by_the_node_id_discovery_will_name_it_with() { + // The dial loop looks a target up under the id carried by the `Node` + // discovery hands back, so our derivation has to be ethrex's. Were it + // not, every lookup would miss and nothing would surface it: the loop + // would keep drawing candidates and dial none of them. + let record = record_with(set_admissible_entries); + + let peer = admit_record(&record).expect("accepted"); + + assert_eq!( + peer.node_id, + Node::from_enr(&record).expect("a node").node_id() + ); + } + #[test] fn a_well_formed_record_is_accepted_and_dialable() { let record = record_with(|pairs| { @@ -402,8 +510,9 @@ mod tests { set_admissible_entries(pairs); }); - assert!(filter().accepts(&record)); - let peer = filter().dial_target(&record).expect("dialable"); + let policy = filter(); + assert!(policy.accepts(&record)); + let peer = filed_target(&policy, &record).expect("dialable"); assert_eq!(peer.subnets, vec![2, 5]); } @@ -416,21 +525,48 @@ mod tests { set_quic(pairs, 9001); }); - assert!(!filter().accepts(&record)); - assert!(filter().dial_target(&record).is_none()); + let policy = filter(); + assert!(!policy.accepts(&record)); + assert!(filed_target(&policy, &record).is_none()); } #[test] fn a_missing_quic_port_is_rejected() { // Discoverable, but not over the only transport we speak. The peer can - // add a `quic` entry and republish: the peer table runs the filter - // again on a higher-`seq` record. This is what the dial-time - // `set_unwanted` this replaced could not express, since ethrex never - // clears that flag. + // add a `quic` entry and republish: discovery runs the filter again on a + // higher-`seq` record. This is what the dial-time `set_unwanted` this + // replaced could not express, since ethrex never clears that flag. let record = record_with(|pairs| set_eth2(pairs, EnrForkId::local())); - assert!(!filter().accepts(&record)); - assert!(filter().dial_target(&record).is_none()); + let policy = filter(); + assert!(!policy.accepts(&record)); + assert!(filed_target(&policy, &record).is_none()); + } + + #[test] + fn a_peer_that_leaves_our_network_stops_being_dialable() { + // The same node, republishing at a higher `seq` with a foreign fork + // digest. Discovery re-judges it and stops offering it, but a target + // left filed under its node id would still be dialed by the one draw + // already in flight, and would sit there for the life of the process. + let signer = secp256k1::SecretKey::new(&mut rand::rngs::OsRng); + let public_key = public_key_from_signing_key(&signer); + let node = Node::new(IpAddr::from(Ipv4Addr::LOCALHOST), 9010, 0, public_key); + let mut record = NodeRecord::from_node(&node, 1, &signer).unwrap(); + record.edit(&signer, set_admissible_entries).unwrap(); + + let policy = filter(); + assert!(policy.accepts(&record)); + assert!(filed_target(&policy, &record).is_some()); + + let mut foreign = EnrForkId::local(); + foreign.fork_digest = [0xde, 0xad, 0xbe, 0xef]; + record + .edit(&signer, |pairs| set_eth2(pairs, foreign)) + .unwrap(); + + assert!(!policy.accepts(&record)); + assert_eq!(filed_target(&policy, &record), None); } #[test] @@ -455,7 +591,10 @@ mod tests { let policy = filter(); let mut admitted: Vec<_> = [honest, hostile] .iter() - .map(|record| policy.dial_target(record).expect("both are admitted")) + .map(|record| { + assert!(policy.accepts(record)); + filed_target(&policy, record).expect("both are admitted") + }) .collect(); assert!( admitted diff --git a/crates/net/p2p/src/discovery/dial.rs b/crates/net/p2p/src/discovery/dial.rs index 641c602f4..60446ad08 100644 --- a/crates/net/p2p/src/discovery/dial.rs +++ b/crates/net/p2p/src/discovery/dial.rs @@ -1,27 +1,31 @@ //! The dial loop: turn what discv5 found into libp2p QUIC connections. //! -//! Runs as a `P2PServer` tick every [`DISCOVERY_DIAL_INTERVAL`], drawing -//! candidates from the ethrex peer table, ranking them by subnet coverage, and -//! dialing one per tick until [`DiscoveryState::target_peers`] are connected. +//! Runs as a `P2PServer` tick every [`DISCOVERY_DIAL_INTERVAL`], asking +//! discovery for candidates, ranking them by subnet coverage, and dialing one +//! per tick until [`DiscoveryState::target_peers`] are connected. +//! +//! Also the other direction: [`report_peer_connected`] and +//! [`report_peer_disconnected`] tell discovery what the swarm did, which is the +//! only way a libp2p connection becomes visible to it. use std::collections::{HashMap, HashSet, VecDeque}; -use ethrex_p2p::peer_table::{PeerTable, PeerTableServerProtocol as _}; +use ethrex_p2p::discovery::{DiscoveryHandle, PeerEvent}; use libp2p::PeerId; use tracing::info; -use super::admission::{DiscoveredPeer, LeanFilter, rank_by_uncovered_subnets}; -use super::{DISCOVERY_CANDIDATE_BATCH, DiscoveryHandle}; +use super::admission::{DialTargets, DiscoveredPeer, rank_by_uncovered_subnets}; +use super::node_id::node_id_from_peer_id; +use super::{DISCOVERY_CANDIDATE_BATCH, SpawnedDiscovery}; use crate::{P2PServer, metrics}; /// Everything the dial loop needs from a running discovery server. pub(crate) struct DiscoveryState { - peer_table: PeerTable, - /// The same policy the peer table judges records with, asked here for the - /// dial target behind an already-admitted record. - filter: LeanFilter, - /// Admitted candidates, best first, drained one per tick. Refilled from the - /// peer table when empty. + handle: DiscoveryHandle, + /// How to dial the peers the filter admitted, filed as their ENRs arrived. + dial_targets: DialTargets, + /// Admitted candidates, best first, drained one per tick. Refilled from + /// discovery when empty. candidates: VecDeque, /// Subnets advertised by peers we dialed from discovery. peer_attnets: HashMap>, @@ -32,18 +36,48 @@ pub(crate) struct DiscoveryState { } impl DiscoveryState { - pub(crate) fn new(handle: DiscoveryHandle, local_peer_id: PeerId) -> Self { + pub(crate) fn new(spawned: SpawnedDiscovery, local_peer_id: PeerId) -> Self { Self { - peer_table: handle.peer_table, - filter: handle.filter, + handle: spawned.handle, + dial_targets: spawned.dial_targets, candidates: VecDeque::new(), peer_attnets: HashMap::new(), local_peer_id, - target_peers: handle.target_peers, + target_peers: spawned.target_peers, } } } +/// Tell discovery we are connected to `peer_id`. +/// +/// Two things follow upstream: the peer stops being offered as a dial +/// candidate, and it counts towards the completion figure that paces discv5 +/// lookups, which eases off the startup rate as we approach `target_peers`. +/// Without this a libp2p node looks permanently peerless to discovery and keeps +/// looking at the startup rate for the life of the process. +/// +/// A no-op when discovery is off, or for a peer whose id carries no secp256k1 +/// key to name it by; see [`node_id_from_peer_id`]. +pub(crate) fn report_peer_connected(server: &P2PServer, peer_id: &PeerId) { + report(server, peer_id, PeerEvent::Connected); +} + +/// Tell discovery our connection to `peer_id` is gone, so it is dialable again +/// and no longer counted. +pub(crate) fn report_peer_disconnected(server: &P2PServer, peer_id: &PeerId) { + report(server, peer_id, PeerEvent::Disconnected); +} + +fn report(server: &P2PServer, peer_id: &PeerId, event: PeerEvent) { + let Some(discovery) = server.discovery.as_ref() else { + return; + }; + let Some(node_id) = node_id_from_peer_id(peer_id) else { + return; + }; + discovery.handle.record_peer_event(node_id, event); +} + /// Drop a peer's discovery bookkeeping. /// /// Called from both teardown paths — a connection that closed and a dial that @@ -60,8 +94,8 @@ pub(crate) fn forget_discovered_peer(server: &mut P2PServer, peer_id: &PeerId) { pub(crate) async fn dial_tick(server: &mut P2PServer) { // Snapshot what the refill needs before any `.await`, so no borrow of // `server.discovery` has to live across the async boundary. Both are handle - // clones: an actor ref and two `Copy` fields, taken only when a refill is - // actually due rather than on every tick that just drains the queue. + // clones: an actor ref and an `Arc`, taken only when a refill is actually + // due rather than on every tick that just drains the queue. let Some(discovery) = server.discovery.as_ref() else { return; }; @@ -71,9 +105,9 @@ pub(crate) async fn dial_tick(server: &mut P2PServer) { let refill = discovery .candidates .is_empty() - .then(|| (discovery.peer_table.clone(), discovery.filter.clone())); + .then(|| (discovery.handle.clone(), discovery.dial_targets.clone())); let mut admitted = match refill { - Some((peer_table, filter)) => draw_candidates(&peer_table, &filter).await, + Some((handle, dial_targets)) => draw_candidates(&handle, &dial_targets).await, None => Vec::new(), }; @@ -120,27 +154,31 @@ pub(crate) async fn dial_tick(server: &mut P2PServer) { } } -/// Draw up to [`DISCOVERY_CANDIDATE_BATCH`] dialable peers from the peer table. +/// Draw up to [`DISCOVERY_CANDIDATE_BATCH`] dialable peers from discovery. +/// +/// ethrex serves one node per call, skipping anything it has written off, is +/// already connected to, or whose ENR our [`LeanFilter`](super::admission::LeanFilter) +/// turned down, and records each as tried before returning it. So successive +/// calls never repeat, an early `None` means the pool is exhausted, and +/// everything that arrives here has already passed admission. /// -/// ethrex serves one contact per call, skipping anything its `PeerFilter` (ours: -/// [`LeanFilter`]) already rejected, and records each as tried before returning -/// it. So successive calls never repeat, an early `None` means the pool is -/// exhausted, and everything that arrives here has already passed admission. -async fn draw_candidates(peer_table: &PeerTable, filter: &LeanFilter) -> Vec { +/// What arrives is a `Node`, which knows an address and a key and nothing about +/// `quic` or `attnets`, so the dial target is looked back up in [`DialTargets`] +/// under the node's id. A miss is ordinary: bootnodes are dialable to ethrex +/// before they have published an ENR, and one we never read a record for is one +/// we cannot build a libp2p multiaddr for. Skipping costs nothing, since the +/// node was marked tried on the way out either way and that set is cleared once +/// a full scan finds nothing eligible. +async fn draw_candidates( + handle: &DiscoveryHandle, + dial_targets: &DialTargets, +) -> Vec { let mut admitted = Vec::with_capacity(DISCOVERY_CANDIDATE_BATCH); for _ in 0..DISCOVERY_CANDIDATE_BATCH { - let Ok(Some(contact)) = peer_table.get_contact_to_initiate().await else { + let Some(node) = handle.next_dial_candidate().await else { break; }; - // A contact whose ENR has not arrived is unjudged, so the peer table - // still offers it, but it carries no address or peer id to dial. - // Skipping it costs nothing: it was marked tried on the way out either - // way, and that set is cleared once a full scan finds nothing eligible. - let Some(peer) = contact - .record - .as_ref() - .and_then(|record| filter.dial_target(record)) - else { + let Some(peer) = dial_targets.get(&node.node_id()) else { continue; }; admitted.push(peer); diff --git a/crates/net/p2p/src/discovery/mod.rs b/crates/net/p2p/src/discovery/mod.rs index ecaaf63d4..3b57ac91d 100644 --- a/crates/net/p2p/src/discovery/mod.rs +++ b/crates/net/p2p/src/discovery/mod.rs @@ -1,29 +1,35 @@ //! discv5 peer discovery, built on ethrex's discovery stack. //! -//! ethrex's `DiscoveryServer` runs discv5-only on its own UDP socket and writes -//! what it finds into an ethrex `PeerTable`. ethlambda's `P2PServer` polls that -//! table, applies the spec checks in [`admission`], and dials the survivors over -//! libp2p QUIC. Static bootnode dialing is untouched. +//! ethrex's `DiscoveryServer` runs discv5-only on its own UDP socket and keeps +//! what it finds to itself. ethlambda asks it for dial candidates, applies the +//! spec checks in [`admission`] as each ENR arrives, and dials the survivors +//! over libp2p QUIC. Static bootnode dialing is untouched. +//! +//! The traffic across that boundary runs one way, into discovery: a request for +//! the next candidate, and a cast per connection opened or closed. Those casts +//! are what let discovery pace its own lookups, since a libp2p connection is +//! otherwise invisible to it. //! //! See `docs/discovery.md` for the operator-facing description. pub mod admission; pub(crate) mod dial; pub mod enr; +pub(crate) mod node_id; use std::collections::HashSet; use std::net::{IpAddr, SocketAddr}; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use std::time::Duration; -use ethrex_p2p::discovery::{DiscoveryConfig, DiscoveryServer}; -use ethrex_p2p::peer_table::{PeerTable, PeerTableServer}; -use ethrex_p2p::types::Node; +use ethrex_p2p::discovery::{DiscoveryConfig, DiscoveryHandle, DiscoveryServer}; +use ethrex_p2p::types::{LocalNode, Node}; use tokio::net::UdpSocket; +use tokio::sync::watch; use tracing::{info, warn}; use crate::Bootnode; -use admission::LeanFilter; +use admission::{DialTargets, LeanFilter}; use enr::{EnrForkId, LocalEnrParams, build_local_enr}; /// How often the dial loop looks for a new peer. @@ -33,20 +39,7 @@ pub const DISCOVERY_DIAL_INTERVAL: Duration = Duration::from_secs(5); /// Overridable per node via [`DiscoverySpawnConfig::target_peers`]. pub const DEFAULT_DISCOVERY_TARGET_PEERS: usize = 200; -/// The target we hand ethrex's peer table, which is not -/// [`DiscoverySpawnConfig::target_peers`] and deliberately so. -/// -/// ethrex's table counts only peers registered through `NewConnectedPeer`, which -/// carries an RLPx `PeerConnection`. ethlambda connects over libp2p and never -/// registers anything, so `peers.len()` is permanently 0 and the table's target -/// cannot mean "how many peers we have". Its one live consumer here is the discv5 -/// lookup pacing, which divides by it: passing 0 would yield `0/0 = NaN`, and -/// `NaN as u64` saturates to zero, turning the lookup timer into an unthrottled -/// re-fire loop. Any non-zero value gives the same pacing, so this is 1 with the -/// reason attached rather than a number pretending to be a peer budget. -const PEER_TABLE_TARGET_PEERS: usize = 1; - -/// Candidates drawn from the peer table per refill. +/// Candidates drawn from discovery per refill. pub const DISCOVERY_CANDIDATE_BATCH: usize = 8; /// Why discovery could not be started. Every variant is fatal at startup. @@ -86,27 +79,28 @@ pub struct DiscoverySpawnConfig { /// [`DEFAULT_DISCOVERY_TARGET_PEERS`]; a target of 0 leaves the dial loop /// ticking without ever dialing. /// - /// Governs the dial loop only. ethrex's peer table is handed a fixed value - /// instead, because it counts only peers registered over RLPx and so can - /// never see ours; see `PEER_TABLE_TARGET_PEERS`. + /// The same number governs how hard discovery looks: it eases its lookups + /// off the startup rate as our connected count approaches this, which it can + /// only do because we report those connections to it. pub target_peers: usize, } /// What the P2P actor needs from a running discovery server. -pub struct DiscoveryHandle { - pub peer_table: PeerTable, +pub struct SpawnedDiscovery { + /// The running server. Three things travel over it: a request for the next + /// dial candidate, and a report for each connection opened and closed. + pub handle: DiscoveryHandle, + /// How to dial the peers the filter admitted; see [`DialTargets`]. + pub(crate) dial_targets: DialTargets, /// This node's ENR as an `enr:`-prefixed string. `spawn_discovery` already /// logs it; this copy is what the tests assert the published record against, /// and what a future RPC identity endpoint would read. Reflects startup /// state; discv5 may bump the sequence number later if PONG voting changes /// our external IP. pub local_enr: String, - /// The admission policy the peer table judges records with, kept so the dial - /// loop can apply the same rules when it turns a contact into a dial target. - /// See [`LeanFilter`]. - pub filter: LeanFilter, /// The configured [`DiscoverySpawnConfig::target_peers`], carried through to - /// the dial loop, which is the only thing it governs. + /// the dial loop's cutoff. Discovery is handed the same number separately, + /// for its lookup pacing. pub target_peers: usize, } @@ -121,7 +115,7 @@ pub struct DiscoveryHandle { /// are still dialed statically by `build_swarm`. pub async fn spawn_discovery( config: DiscoverySpawnConfig, -) -> Result { +) -> Result { let signer = secp256k1::SecretKey::from_slice(&config.node_key).map_err(DiscoveryError::NodeKey)?; @@ -147,20 +141,16 @@ pub async fn spawn_discovery( let local_record = build_local_enr(¶ms)?; let local_enr = local_record.enr_url().map_err(DiscoveryError::EncodeEnr)?; - // `spawn` rather than `spawn_with_filter` would install ethrex's own filter, - // which wants an EIP-2124 `eth` entry compatible with an execution chain lean - // does not have and rejects any record without one, so every lean contact - // would be stamped rejected and never dialed. + // Our own filter rather than ethrex's `EthForkIdFilter`, which wants an + // EIP-2124 `eth` entry compatible with an execution chain lean does not have + // and rejects any record without one, so every lean contact would be stamped + // rejected and never dialed. `AcceptAllFilter` would go the other way and + // hand us every node on the DHT. // - // The peer table owns the filter it runs, so the dial loop keeps a clone - // rather than sharing one: the two carry the same fork id and committee - // count, which is what makes their judgments agree. + // Discovery takes the filter by value, so what the dial loop keeps is the + // map the filter writes into, not a second copy of the policy. let filter = LeanFilter::new(EnrForkId::local(), config.attestation_committee_count); - let peer_table = PeerTableServer::spawn_with_filter( - local_node.node_id(), - PEER_TABLE_TARGET_PEERS, - filter.clone(), - ); + let dial_targets = filter.dial_targets(); let seeds: Vec = config .bootnodes @@ -179,21 +169,57 @@ pub async fn spawn_discovery( // `attnets`, `quic`) and a lean peer applying our own admission rules to it // admits us. ethrex re-signs it under `params.signer` whenever IP voting // bumps the sequence number, keeping the extra entries. - DiscoveryServer::spawn( + // + // `shared_local_node` is ethrex's live mirror of that identity, which its RPC + // layer serves the current record from. Nothing on this side reads it yet, so + // it is written and dropped; it costs one `Arc`. + // + // `fork_id` publishes an EIP-2124 fork id for discovery to stamp on the ENR + // when the chain crosses a fork. Lean has no fork schedule and its digest + // lives in `eth2`, not `eth`, so there is nothing to publish: the sender is + // dropped straight away, which discovery reads as "the publisher is gone" and + // stops checking. + let (fork_id_tx, fork_id_rx) = watch::channel(None); + drop(fork_id_tx); + let shared_local_node = Arc::new(RwLock::new(LocalNode { + node: local_node.clone(), + record: local_record.clone(), + })); + + let server = DiscoveryServer::spawn( local_node, local_record, params.signer, Arc::new(socket), - peer_table.clone(), + Box::new(filter), seeds, DiscoveryConfig { discv4_enabled: false, discv5_enabled: true, + // The real target, unlike the placeholder this replaces: discovery + // divides our reported connection count by it to decide how hard to + // keep looking. A target of 0 is answered as "complete" upstream, so + // it no longer has to be worked around here. + target_peers: config.target_peers, + // `--discovery.advertise-ip` is this node's `--nat extip:`: an + // operator naming the address peers should reach it on. Locking the + // predictor keeps discv5's PONG voting from overwriting it, which is + // the whole point of having said it. + nat_extip_set: config.advertise_ip.is_some(), }, + shared_local_node, + fork_id_rx, ) .await .map_err(|err| DiscoveryError::Server(err.to_string()))?; + // The handle exists to be published once and cloned everywhere, which is how + // ethrex's own consumer starts discovery after the context that reaches it. + // We have the running server in hand, so the publish is unconditional and its + // "already set" answer cannot be anything but true. + let discovery = DiscoveryHandle::new(); + discovery.set(server); + info!(enr = %local_enr, "Local ENR"); if advertise_ip.is_unspecified() { warn!( @@ -204,10 +230,10 @@ pub async fn spawn_discovery( ); } - Ok(DiscoveryHandle { - peer_table, + Ok(SpawnedDiscovery { + handle: discovery, + dial_targets, local_enr, - filter, target_peers: config.target_peers, }) } @@ -220,6 +246,10 @@ mod tests { use ethrex_rlp::decode::RLPDecode; use std::net::Ipv4Addr; + /// The committee count `config` below spawns with, named so the filter the + /// self-admission test rebuilds cannot silently drift from it. + const TEST_COMMITTEE_COUNT: u64 = 4; + fn config(discovery_port: u16, advertise_ip: Option) -> DiscoverySpawnConfig { DiscoverySpawnConfig { node_key: secp256k1::SecretKey::new(&mut rand::rngs::OsRng) @@ -229,7 +259,7 @@ mod tests { discovery_port, quic_port: 9001, subscription_subnets: HashSet::from([0u64]), - attestation_committee_count: 4, + attestation_committee_count: TEST_COMMITTEE_COUNT, bootnodes: Vec::new(), advertise_ip, target_peers: DEFAULT_DISCOVERY_TARGET_PEERS, @@ -263,10 +293,13 @@ mod tests { let advertised_port = record.pairs().udp_port.expect("udp entry"); assert_ne!(advertised_port, 0); - // The policy handed to the peer table must admit our own record. A peer + // The policy handed to discovery must admit our own record. A peer // running this code applies exactly these rules to what we publish, so a - // record we would reject ourselves is one nobody dials. - assert!(handle.filter.accepts(&record)); + // record we would reject ourselves is one nobody dials. Rebuilt here + // from the same two values `spawn_discovery` constructs it with, since + // the filter itself was moved into the discovery server. + let filter = LeanFilter::new(EnrForkId::local(), TEST_COMMITTEE_COUNT); + assert!(filter.accepts(&record)); } #[tokio::test] diff --git a/crates/net/p2p/src/discovery/node_id.rs b/crates/net/p2p/src/discovery/node_id.rs new file mode 100644 index 000000000..41e98be23 --- /dev/null +++ b/crates/net/p2p/src/discovery/node_id.rs @@ -0,0 +1,83 @@ +//! The same peer, named two ways. +//! +//! discv5 identifies a node by its Kademlia node id, the keccak256 of its +//! uncompressed secp256k1 public key. libp2p identifies the same node by a +//! [`PeerId`], a multihash of the protobuf encoding of that same key. Both are +//! derived from one key, so either name can be computed from it, and ethlambda +//! needs the round trip in both directions: +//! +//! - reading an ENR, to key a dial target by the node id ethrex will later name +//! it with ([`node_id_from_public_key`]); +//! - reporting a live connection, where all the swarm gives us is a `PeerId` +//! ([`node_id_from_peer_id`]). + +use ethrex_common::{H256, H512}; +use ethrex_p2p::utils::node_id; +use libp2p::PeerId; +use libp2p::identity::secp256k1::PublicKey; + +/// Multihash code under which libp2p inlines a public key rather than hashing +/// it. Keys short enough to fit are stored verbatim, which is what makes the +/// key recoverable from a `PeerId` at all. +const IDENTITY_MULTIHASH_CODE: u64 = 0x00; + +/// The discv5 node id for a libp2p secp256k1 key. +/// +/// ethrex hashes the 64-byte uncompressed point, so the `0x04` prefix libp2p +/// emits is dropped first. +pub(crate) fn node_id_from_public_key(key: &PublicKey) -> H256 { + node_id(&H512::from_slice(&key.to_bytes_uncompressed()[1..])) +} + +/// The discv5 node id behind a `PeerId`, or `None` when the peer id does not +/// carry a secp256k1 key we can read back. +/// +/// A `PeerId` is a hash of the key in general, but libp2p inlines any encoding +/// short enough, and a secp256k1 key always is. Lean's spec mandates secp256k1 +/// for the libp2p identity, so the `None` arm covers a peer speaking something +/// else: nothing to report about it, rather than an error. +pub(crate) fn node_id_from_peer_id(peer_id: &PeerId) -> Option { + let multihash = peer_id.as_ref(); + if multihash.code() != IDENTITY_MULTIHASH_CODE { + return None; + } + let key = libp2p::identity::PublicKey::try_decode_protobuf(multihash.digest()).ok()?; + Some(node_id_from_public_key(&key.try_into_secp256k1().ok()?)) +} + +#[cfg(test)] +mod tests { + use super::*; + use libp2p::identity::Keypair; + + #[test] + fn a_peer_id_resolves_to_the_node_id_of_the_key_behind_it() { + let keypair = Keypair::generate_secp256k1(); + let key = keypair.public().try_into_secp256k1().expect("secp256k1"); + let peer_id = PeerId::from_public_key(&keypair.public()); + + // The two directions must agree, or a peer admitted from its ENR and the + // same peer seen connecting would be reported under different ids: the + // connected set would count it twice and never drop it on disconnect. + assert_eq!( + node_id_from_peer_id(&peer_id), + Some(node_id_from_public_key(&key)) + ); + } + + #[test] + fn a_non_secp256k1_peer_id_has_no_node_id() { + // ed25519 keys are also short enough to be inlined, so this reaches the + // key decode rather than being turned away by the multihash code. + let peer_id = PeerId::from_public_key(&Keypair::generate_ed25519().public()); + + assert_eq!(node_id_from_peer_id(&peer_id), None); + } + + #[test] + fn a_hashed_peer_id_has_no_node_id() { + // `PeerId::random` produces a sha256 multihash, the form a long key + // takes. There is no key inside it to recover. + assert_eq!(node_id_from_peer_id(&PeerId::random()), None); + } +} diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 685032f91..97ea702ce 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -37,7 +37,10 @@ use tracing::{debug, info, trace, warn}; use crate::{ discovery::{ DISCOVERY_DIAL_INTERVAL, DiscoveryError, DiscoverySpawnConfig, - dial::{DiscoveryState, dial_tick, forget_discovered_peer}, + dial::{ + DiscoveryState, dial_tick, forget_discovered_peer, report_peer_connected, + report_peer_disconnected, + }, enr::{read_ip, read_public_key, read_quic_port}, spawn_discovery, }, @@ -645,6 +648,7 @@ async fn handle_swarm_event( let direction = connection_direction(&endpoint); if num_established.get() == 1 { server.connected_peers.insert(peer_id); + report_peer_connected(server, &peer_id); let peer_count = server.connected_peers.len(); metrics::notify_peer_connected( server.resolve_node_name(Some(&peer_id)), @@ -702,6 +706,7 @@ async fn handle_swarm_event( }; if num_established == 0 { server.connected_peers.remove(&peer_id); + report_peer_disconnected(server, &peer_id); forget_discovered_peer(server, &peer_id); let peer_count = server.connected_peers.len(); metrics::notify_peer_disconnected( diff --git a/docs/discovery.md b/docs/discovery.md index 8513fe51c..76dc05b80 100644 --- a/docs/discovery.md +++ b/docs/discovery.md @@ -31,9 +31,11 @@ The discv5 socket always binds the wildcard `0.0.0.0`, since that is where we listen, not where peers should dial us. Without `--discovery.advertise-ip` the published ENR inherits that same `0.0.0.0`, which is not a dialable address: set the flag to `127.0.0.1` for a local devnet or to the host's public address -so the ENR is usable as soon as it is published. discv5's PONG-based IP voting -may still replace the advertised address later, once a peer's response tells -the node what its external address looks like. +so the ENR is usable as soon as it is published. Left unset, discv5's PONG-based +IP voting replaces the advertised address once peers' responses agree on what +the node's external address looks like. Set, the flag wins and voting cannot +overwrite it: naming an address is the operator saying they know better than the +vote. ## The ENR @@ -70,20 +72,28 @@ A differing `next_fork_version` or `next_fork_epoch` is *not* grounds for rejection: the spec permits connecting to a peer that is incompatible with an upcoming fork but compatible now. -These checks are handed to ethrex's peer table as a `PeerFilter`, so each record -is judged the moment it arrives and a peer that fails is not offered for dialing. -No rejection is final: the peer table runs the filter again as soon as the peer -publishes a higher-`seq` ENR, so a node that adds a `quic` entry, or gains an -address through discv5's IP voting, is reconsidered without a restart. +These checks are handed to ethrex's discovery server as a `PeerFilter`, so each +record is judged the moment it arrives and a peer that fails is not offered for +dialing. No rejection is final: discovery runs the filter again as soon as the +peer publishes a higher-`seq` ENR, so a node that adds a `quic` entry, or gains +an address through discv5's IP voting, is reconsidered without a restart. + +Judging is also the only moment ethlambda holds a peer's ENR, so the filter files +what it admits as it goes. Discovery hands back an `enode`-shaped `Node`, which +knows an address and a public key and nothing about `quic` or `attnets`; the dial +loop looks the node id back up in what the filter filed to recover the multiaddr +and the subnets. A node discovery offers that we never read an admissible record +for is skipped, which is what happens to a bootnode still known only as a bare +endpoint. Admitted peers are ranked by how many attestation subnets they advertise that no currently connected peer covers, so discovery preferentially fills gaps in subnet coverage. A peer advertising no `attnets` is ranked last but never dropped. -Dialing stops once `--discovery.target-peers` peers are connected, and resumes -if that count drops. That is all the flag does: it is the dial loop's cutoff, and -nothing in ethrex's peer table or discv5's own pacing enforces it (see -[below](#discv5-lookups-run-at-the-startup-rate)). +Dialing stops once `--discovery.target-peers` peers are connected, and resumes if +that count drops. The same number also paces discovery itself: every connection +the swarm opens or closes is reported to it, so it knows how close to the target +we are and eases its lookups off the startup rate accordingly. ## Bootnodes @@ -122,21 +132,11 @@ another**: two devnets running this code will peer with each other. Closing that gap requires lean adopting a genesis-derived fork digest, which is a cross-client change to gossip topic names. -### discv5 lookups run at the startup rate - -ethrex paces its discv5 iterative lookups by how full its own peer table is, -easing from one lookup every 500ms at startup to one every 10s once the table -reaches its target. That table only counts peers registered through -`NewConnectedPeer`, which carries an RLPx connection; ethlambda connects over -libp2p and registers nothing, so the count is permanently zero and the pacing -never eases off the startup rate. A lean node therefore keeps looking up every -500ms rather than settling at 10s, roughly 20x the intended steady-state -`FindNode` traffic, for the life of the process. - -`--discovery.target-peers` deliberately does *not* feed that computation, since a -target of `0` would make it divide by zero and re-fire the lookup timer with no -delay at all. Closing the gap properly means ethrex learning about non-RLPx -connections, which is an upstream change. +### Only peers dialed from discovery contribute to subnet ranking + +An inbound peer never tells us its `attnets`, so a subnet it covers is not +credited when the dial loop ranks candidates. Treating an unknown peer as +covering nothing makes the ranking more eager than it needs to be, never wrong. ### `attnets` is not a fixed-width SSZ `Bitvector` From 6e3c8f605374414b2484eb6f5b50c620cf4e978d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:21:49 -0300 Subject: [PATCH 2/4] chore(deps): pin ethrex to feat/discv5-predicate-queries The branch adds DiscoveryHandle::find_nodes and records_matching, discv5 predicate queries: discovery runs a consumer's test on every record a walk sees and answers with the matches whose nodes replied. That is what lets the dial loop look for peers on a given attestation subnet instead of waiting for the crawl to stumble on one. It is built on ethrex #7217, which the previous pin was, so the API this crate uses is unchanged and it builds as is. The lockfile moves only the ethrex crates (v25.0.0 -> v30.0.1 at 8f847524); make cooldown-check passes. A commit rather than the branch, for the same reason as before: what this crate builds against must not move without the manifest saying so. --- Cargo.lock | 172 +++++++++++--------------------------- crates/net/p2p/Cargo.toml | 19 +++-- 2 files changed, 61 insertions(+), 130 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 87d5e4b99..c1f6c1267 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -172,7 +172,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -183,7 +183,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -720,7 +720,7 @@ dependencies = [ "bitflags 2.11.1", "cexpr", "clang-sys", - "itertools 0.13.0", + "itertools 0.10.5", "log", "prettyplease", "proc-macro2", @@ -1870,7 +1870,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] @@ -2253,8 +2253,8 @@ dependencies = [ [[package]] name = "ethrex-blockchain" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "crossbeam", @@ -2278,8 +2278,8 @@ dependencies = [ [[package]] name = "ethrex-common" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "crc32fast", @@ -2287,6 +2287,7 @@ dependencies = [ "ethrex-crypto", "ethrex-rlp", "ethrex-trie", + "foldhash 0.2.0", "hex", "hex-literal", "hex-simd", @@ -2312,8 +2313,8 @@ dependencies = [ [[package]] name = "ethrex-crypto" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "ark-bn254", "ark-ec", @@ -2334,8 +2335,8 @@ dependencies = [ [[package]] name = "ethrex-guest-program" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "ethereum-types", @@ -2357,8 +2358,8 @@ dependencies = [ [[package]] name = "ethrex-l2-common" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "ethereum-types", @@ -2375,8 +2376,8 @@ dependencies = [ [[package]] name = "ethrex-levm" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "derive_more 1.0.0", @@ -2393,8 +2394,8 @@ dependencies = [ [[package]] name = "ethrex-metrics" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "ethrex-common", "serde", @@ -2405,8 +2406,8 @@ dependencies = [ [[package]] name = "ethrex-p2p" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "aes", "aes-gcm", @@ -2447,8 +2448,8 @@ dependencies = [ [[package]] name = "ethrex-rlp" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "ethereum-types", @@ -2457,8 +2458,8 @@ dependencies = [ [[package]] name = "ethrex-storage" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "anyhow", "bytes", @@ -2467,6 +2468,7 @@ dependencies = [ "ethrex-rlp", "ethrex-trie", "fastbloom", + "indexmap 2.14.0", "lru", "rayon", "rustc-hash", @@ -2479,8 +2481,8 @@ dependencies = [ [[package]] name = "ethrex-trie" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "anyhow", "bytes", @@ -2499,8 +2501,8 @@ dependencies = [ [[package]] name = "ethrex-vm" -version = "25.0.0" -source = "git+https://github.com/lambdaclass/ethrex?rev=4462253ea2785b20d8ee8822034f2a98b49ae839#4462253ea2785b20d8ee8822034f2a98b49ae839" +version = "30.0.1" +source = "git+https://github.com/lambdaclass/ethrex?rev=8f84752408a126cfa8a9c21c3ebaa76658023cb7#8f84752408a126cfa8a9c21c3ebaa76658023cb7" dependencies = [ "bytes", "derive_more 1.0.0", @@ -3282,7 +3284,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.62.2", ] [[package]] @@ -5094,7 +5096,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -5702,7 +5704,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.117", @@ -5838,7 +5840,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] @@ -6396,7 +6398,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] @@ -6902,7 +6904,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7150,7 +7152,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] @@ -7962,7 +7964,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.52.0", ] [[package]] @@ -7978,7 +7980,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12342cb4d8e3b046f3d80effd474a7a02447231330ef77d71daa6fbc40681143" dependencies = [ "windows-core 0.57.0", - "windows-targets 0.52.6", + "windows-targets", ] [[package]] @@ -8011,7 +8013,7 @@ dependencies = [ "windows-implement 0.57.0", "windows-interface 0.57.0", "windows-result 0.1.2", - "windows-targets 0.52.6", + "windows-targets", ] [[package]] @@ -8115,7 +8117,7 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e383302e8ec8515204254685643de10811af0ed97ea37210dc26fb0032647f8" dependencies = [ - "windows-targets 0.52.6", + "windows-targets", ] [[package]] @@ -8142,16 +8144,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets 0.52.6", -] - -[[package]] -name = "windows-sys" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" -dependencies = [ - "windows-targets 0.53.5", + "windows-targets", ] [[package]] @@ -8169,31 +8162,14 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm 0.52.6", - "windows_aarch64_msvc 0.52.6", - "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm 0.52.6", - "windows_i686_msvc 0.52.6", - "windows_x86_64_gnu 0.52.6", - "windows_x86_64_gnullvm 0.52.6", - "windows_x86_64_msvc 0.52.6", -] - -[[package]] -name = "windows-targets" -version = "0.53.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" -dependencies = [ - "windows-link", - "windows_aarch64_gnullvm 0.53.1", - "windows_aarch64_msvc 0.53.1", - "windows_i686_gnu 0.53.1", - "windows_i686_gnullvm 0.53.1", - "windows_i686_msvc 0.53.1", - "windows_x86_64_gnu 0.53.1", - "windows_x86_64_gnullvm 0.53.1", - "windows_x86_64_msvc 0.53.1", + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", ] [[package]] @@ -8211,96 +8187,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" - [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" -[[package]] -name = "windows_aarch64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" - [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" -[[package]] -name = "windows_i686_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" - [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" -[[package]] -name = "windows_i686_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" - [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" -[[package]] -name = "windows_i686_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" - [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" -[[package]] -name = "windows_x86_64_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" - [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" - [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" -[[package]] -name = "windows_x86_64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" - [[package]] name = "winnow" version = "1.0.3" diff --git a/crates/net/p2p/Cargo.toml b/crates/net/p2p/Cargo.toml index d6ddb7abb..cecda325d 100644 --- a/crates/net/p2p/Cargo.toml +++ b/crates/net/p2p/Cargo.toml @@ -51,17 +51,20 @@ rand = "0.8" # Required for NodeEnr parsing. # -# Pinned to the head of ethrex PR #7217, which moves the contact table into the -# discovery server and gives a consumer `DiscoveryHandle::record_peer_event` to -# report its own connections with. That is what lets a libp2p client feed -# ethrex's lookup pacing; see `docs/discovery.md`. +# Pinned to the head of ethrex's `feat/discv5-predicate-queries`, which builds +# on PR #7217. #7217 moves the contact table into the discovery server and gives +# a consumer `DiscoveryHandle::record_peer_event` to report its own connections +# with, which is what lets a libp2p client feed ethrex's lookup pacing. The +# branch adds `DiscoveryHandle::find_nodes`/`records_matching`, the predicate +# queries the subnet hunt in `discovery/subnets.rs` runs on. See +# `docs/discovery.md`. # # A commit rather than a branch, for the reason the release tag it replaces was # chosen over `main`: what this crate builds against must not move without the -# manifest saying so. Goes back to a tag once #7217 is in a release. -ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } -ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } -ethrex-common = { git = "https://github.com/lambdaclass/ethrex", rev = "4462253ea2785b20d8ee8822034f2a98b49ae839" } +# manifest saying so. Goes back to a tag once both are in a release. +ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", rev = "8f84752408a126cfa8a9c21c3ebaa76658023cb7" } +ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", rev = "8f84752408a126cfa8a9c21c3ebaa76658023cb7" } +ethrex-common = { git = "https://github.com/lambdaclass/ethrex", rev = "8f84752408a126cfa8a9c21c3ebaa76658023cb7" } # `rand` seeds the throwaway signing keys the ENR tests build records with. secp256k1.workspace = true From b005e04c5a2bdb13c20755d43e68933b5a9be5be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:21:54 -0300 Subject: [PATCH 3/4] feat(discovery): hunt subnet peers with predicate queries, serve the live ENR The dial loop ranks what the crawl happened to find, which only reorders random draws. About one mainnet record in 32 advertises a given subnet, so a subscribed subnet nobody connected covers could stay uncovered for as long as the crawl took to stumble on a peer for it. The subnet hunt (discovery/subnets.rs) searches for them instead. Each dial tick publishes the subscribed subnets with fewer than 3 connected peers advertising them (Lighthouse's TARGET_SUBNET_PEERS). A task off the P2P actor takes the matches discovery already holds, then walks discv5 with find_nodes for the rest, one hunt at a time, at most 6 peers each, 30 s apart. The predicate is admission plus one wanted attnets bit, as in Lighthouse's subnet predicate. Finds join the dial loop's candidate queue and rank after custody gaps, ahead of other subnets. Nothing is hunted while the loop has no budget to dial, and a peer handed over is left out of hunts for 10 minutes, since discovery keeps offering a mainnet peer that connected and said goodbye. /eth/v1/node/identity now reports the ENR discv5 serves rather than the one built at startup. ethrex mirrors every record it re-signs (on discv5 IP voting) into the SharedLocalNode it was spawned with; LocalEnr reads that mirror, and the RPC crate gets it as an EnrSource closure, since it does not depend on ethlambda-p2p. The other direction stays unsupported, now documented against the new API: ethrex has no message that edits its local record. Its one consumer-driven edit writes the EIP-2124 `eth` entry, not `eth2`, and a write into the mirror is overwritten by discovery's next edit. So the served eth2 still keeps the startup digest across a fork boundary. --- CLAUDE.md | 12 +- bin/ethlambda/src/main.rs | 22 +- crates/net/p2p/src/beacon/transition.rs | 2 +- crates/net/p2p/src/discovery/admission.rs | 98 ++++- crates/net/p2p/src/discovery/dial.rs | 60 ++- crates/net/p2p/src/discovery/enr.rs | 12 +- crates/net/p2p/src/discovery/mod.rs | 127 +++--- crates/net/p2p/src/discovery/subnets.rs | 467 ++++++++++++++++++++++ crates/net/p2p/src/lib.rs | 15 +- crates/net/p2p/src/metrics.rs | 14 + crates/net/rpc/src/beacon/node.rs | 4 +- crates/net/rpc/src/lib.rs | 38 +- docs/discovery.md | 69 +++- docs/metrics.md | 1 + 14 files changed, 830 insertions(+), 111 deletions(-) create mode 100644 crates/net/p2p/src/discovery/subnets.rs diff --git a/CLAUDE.md b/CLAUDE.md index cc7ea964a..39dde0b86 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -366,9 +366,10 @@ actual_slot = finalized_slot + 1 + relative_index - Fork-digest rollover happens at runtime (`ForkSchedule`, `beacon::transition`): the next digest's topics are joined `SUBSCRIBE_LEAD_EPOCHS` before its boundary and the previous one's left `UNSUBSCRIBE_LAG_EPOCHS` after, blob-parameter (BPO) boundaries included. - **Known gap**: ethrex's `DiscoveryServer` cannot replace the served ENR at runtime, so its - `eth2` entry keeps the startup digest until a restart; peers discovering the node after a - boundary reject the record + **Known gap**: ethrex's `DiscoveryServer` has no message that edits the served ENR (its one + consumer-driven edit, the `fork_id` watch, writes `eth`, not `eth2`), so its `eth2` entry + keeps the startup digest until a restart; peers discovering the node after a boundary + reject the record. See `discovery::update_served_enr` - `voluntary_exit`, `proposer_slashing`, `attester_slashing` and `bls_to_execution_change` validate in p2p like aggregates (`gossip::operations`, against the unadvanced head state, which must be electra or fulu: under a gloas head every operation is IGNOREd) and feed @@ -394,12 +395,13 @@ actual_slot = finalized_slot + 1 + relative_index ### Peer Discovery (discv5) - Always on for `beacon` (mainnet bootnode ENRs are not statically dialable, so a crawl is its only way to find a peer); opt-in for `node` behind the lean-only `--discovery.enable`, off by default, so a lean node peers from `--bootnodes` alone and binds no discovery socket. `Network::discovery_enabled` is the one answer; `P2P::spawn` takes `Option` and `P2PServer.discovery` is an `Option`, `None` leaving the dial loop unscheduled - Where it runs: `DEFAULT_DISCOVERY_PORT` (9000) unless `--discovery.port` says otherwise (own UDP socket, must differ from `--gossipsub-port`; checked once by `Options::validate_ports`, which skips the discovery rules when it is off). Co-located nodes that run discovery must each pass `--discovery.port` -- Reuses ethrex's `DiscoveryServer` with discv4 disabled; `spawn` takes the prepared lean ENR, so the record ethrex serves is the one we report +- Reuses ethrex's `DiscoveryServer` with discv4 disabled, pinned to a commit of ethrex's `feat/discv5-predicate-queries` (on top of ethrex PR #7217) until both are in a release; `spawn` takes the prepared lean ENR. ethrex re-signs it on IP voting and mirrors it into the `SharedLocalNode` it was spawned with; `discovery::LocalEnr` reads that mirror, so `/eth/v1/node/identity` reports the record discv5 currently serves. Read-only: nothing written there reaches discv5 - ENR follows the beacon phase0 spec: `ip`/`udp`/`quic`/`tcp`/`secp256k1`/`eth2`/`attnets` - Admission mirrors lighthouse: `eth2.fork_digest` must match, `next_fork_*` may differ, a `quic` or `tcp` entry required. Handed to discovery as `LeanFilter: PeerFilter`, so records are judged on arrival, not at dial time; a reject is re-judged on a higher-`seq` ENR - The filter files each admitted peer in `DialTargets`, since discovery hands back a `Node` that carries no `quic`/`tcp`/`attnets`; the dial loop looks the node id back up there - Every libp2p connection is reported with `DiscoveryHandle::record_peer_event`, which is what lets discovery pace its lookups against `--discovery.target-peers` -- Candidates ranked by uncovered attestation subnets. See [`docs/discovery.md`](docs/discovery.md) +- Candidates ranked by custody gaps, then the subnet hunt's wanted subnets, then uncovered attestation subnets +- Subnet hunt (`discovery/subnets.rs`): subscribed subnets with fewer than `TARGET_SUBNET_PEERS` (3) connected peers advertising them are searched with `DiscoveryHandle::records_matching` then `find_nodes` (ethrex's predicate queries), one hunt at a time on its own task; finds join the dial loop's candidate queue. Duty subnets from `beacon_committee_subscriptions` are not fed to it. See [`docs/discovery.md`](docs/discovery.md) ### Retry Strategy on Block Requests - Exponential backoff: doubling from `INITIAL_BACKOFF_MS` (5ms → 2560ms) diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 6ea55328f..4a14da807 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -119,12 +119,12 @@ fn main() -> eyre::Result<()> { } /// What `/eth/v1/node/identity` reports beyond the peer id: the ENR discv5 -/// publishes, and the multiaddrs peers reach this node on. The addresses need -/// the node's externally reachable IP, which only `--discovery.advertise-ip` -/// supplies (a node bound to `0.0.0.0` does not know it), so without the flag -/// they are left empty rather than guessed. +/// serves, read live, and the multiaddrs peers reach this node on. The +/// addresses need the node's externally reachable IP, which only +/// `--discovery.advertise-ip` supplies (a node bound to `0.0.0.0` does not know +/// it), so without the flag they are left empty rather than guessed. fn beacon_identity( - enr: &str, + enr: ethlambda_rpc::EnrSource, advertise_ip: Option, gossipsub_port: u16, discovery_port: u16, @@ -132,13 +132,13 @@ fn beacon_identity( ) -> ethlambda_rpc::BeaconIdentity { let Some(ip) = advertise_ip else { return ethlambda_rpc::BeaconIdentity { - enr: enr.to_string(), + enr, ..Default::default() }; }; let family = if ip.is_ipv4() { "ip4" } else { "ip6" }; ethlambda_rpc::BeaconIdentity { - enr: enr.to_string(), + enr, // QUIC first, matching the order this node dials in. p2p_addresses: vec![ format!("/{family}/{ip}/udp/{gossipsub_port}/quic-v1/p2p/{peer_id}"), @@ -768,8 +768,14 @@ async fn run_node(options: Options) -> eyre::Result<()> { } => ethlambda_rpc::CustodyColumns(custody_columns.clone()), ChainActor::Lean(..) => ethlambda_rpc::CustodyColumns::default(), }; + // Read through discovery's own copy of the record, so the endpoint follows + // it when discv5's IP voting re-signs it. No ENR without discovery. + let rpc_enr = p2p + .local_enr() + .map(|local_enr| ethlambda_rpc::EnrSource::new(move || local_enr.current())) + .unwrap_or_default(); let rpc_identity = beacon_identity( - p2p.local_enr().unwrap_or_default(), + rpc_enr, common.discovery.advertise_ip, common.gossipsub_port, common.discovery.port, diff --git a/crates/net/p2p/src/beacon/transition.rs b/crates/net/p2p/src/beacon/transition.rs index 9bc6f378f..67cdf18cb 100644 --- a/crates/net/p2p/src/beacon/transition.rs +++ b/crates/net/p2p/src/beacon/transition.rs @@ -244,7 +244,7 @@ pub(crate) fn apply(server: &mut P2PServer, epoch: Epoch) -> Changes { .iter() .map(|topics| topics.fork_digest) .collect(); - // Without discovery there is no peer table whose admission to update. + // Without discovery there is no admission policy to update. if let Some(state) = &server.discovery { state.filter().set_fork_id(fork_id, also_admitted); } diff --git a/crates/net/p2p/src/discovery/admission.rs b/crates/net/p2p/src/discovery/admission.rs index 94e0fcf71..759fb2fd5 100644 --- a/crates/net/p2p/src/discovery/admission.rs +++ b/crates/net/p2p/src/discovery/admission.rs @@ -206,6 +206,15 @@ impl LeanFilter { self.dial_targets.clone() } + /// What to dial for `record`, or `None` if it would not be admitted. + /// + /// Judges without filing: the subnet hunt reads records out of + /// discovery's answers, which hand back the record itself rather than a + /// node id to look up in [`DialTargets`]. + pub(crate) fn dial_target(&self, record: &NodeRecord) -> Option { + self.judge(record).ok() + } + fn judge(&self, record: &NodeRecord) -> Result { let policy = self.policy.read().expect("admission policy lock poisoned"); admit( @@ -345,8 +354,14 @@ impl DiscoveredPeer { /// Order candidates so the ones filling this node's gaps are dialed first: /// custody columns it samples that are undersupplied (fewer connected -/// custodians than `dial::CUSTODY_REDUNDANCY_TARGET`), then attestation -/// subnets no connected peer covers. +/// custodians than `dial::CUSTODY_REDUNDANCY_TARGET`), then the subscribed +/// attestation subnets the subnet hunt is looking for (`wanted_subnets`, see +/// [`super::subnets`]), then attestation subnets no connected peer covers. +/// +/// A wanted subnet outranks any other uncovered one because this node needs +/// it: it subscribes to it, and publishes its own attestations there. The +/// subnet hunt's finds land in the same queue and rank here like any other +/// candidate, which is what puts them ahead of the crawl's random draws. /// /// Custody outranks subnets because the two shortfalls do not cost the same. /// At the extreme, a column no connected peer custodies cannot be fetched by @@ -370,6 +385,7 @@ pub(crate) fn rank_candidates( candidates: &mut [DiscoveredPeer], covered_subnets: &HashSet, wanted_columns: &HashSet, + wanted_subnets: &HashSet, ) { candidates.sort_by_key(|candidate| { // Skipped rather than computed and discarded when nothing is wanted, @@ -381,12 +397,21 @@ pub(crate) fn rank_candidates( } else { candidate.custody_coverage(wanted_columns) }; + let wanted = candidate + .subnets + .iter() + .filter(|subnet| wanted_subnets.contains(subnet)) + .count(); let subnets = candidate .subnets .iter() .filter(|subnet| !covered_subnets.contains(subnet)) .count(); - (std::cmp::Reverse(columns), std::cmp::Reverse(subnets)) + ( + std::cmp::Reverse(columns), + std::cmp::Reverse(wanted), + std::cmp::Reverse(subnets), + ) }); } @@ -727,7 +752,7 @@ mod tests { #[test] fn a_fork_boundary_moves_what_every_clone_admits() { - // The peer table owns a clone of the filter, so a boundary has to reach + // Discovery owns a clone of the filter, so a boundary has to reach // it through the shared policy rather than through the dial loop's copy. let table_copy = filter(); let dial_copy = table_copy.clone(); @@ -828,7 +853,12 @@ mod tests { "no admitted peer may advertise a subnet outside the local committee" ); - rank_candidates(&mut admitted, &HashSet::new(), &HashSet::new()); + rank_candidates( + &mut admitted, + &HashSet::new(), + &HashSet::new(), + &HashSet::new(), + ); assert_eq!( admitted[0].subnets, vec![3], @@ -845,11 +875,46 @@ mod tests { DiscoveredPeer::for_test(vec![2]), DiscoveredPeer::for_test(vec![2, 3]), ]; - rank_candidates(&mut candidates, &HashSet::from([0u64, 1]), &HashSet::new()); + rank_candidates( + &mut candidates, + &HashSet::from([0u64, 1]), + &HashSet::new(), + &HashSet::new(), + ); let order: Vec<_> = candidates.iter().map(|c| c.subnets.clone()).collect(); assert_eq!(order, vec![vec![2, 3], vec![2], vec![0]]); } + #[test] + fn a_wanted_subnet_outranks_more_uncovered_ones() { + // Nothing is covered, so `[1, 2, 3]` fills three gaps and `[0]` one. + // But subnet 0 is one this node subscribes to and is hunting peers + // for, and the others are not, so the one peer on it goes first. + let mut candidates = vec![ + DiscoveredPeer::for_test(vec![1, 2, 3]), + DiscoveredPeer::for_test(vec![0]), + ]; + rank_candidates( + &mut candidates, + &HashSet::new(), + &HashSet::new(), + &HashSet::from([0u64]), + ); + let order: Vec<_> = candidates.iter().map(|c| c.subnets.clone()).collect(); + assert_eq!(order, vec![vec![0], vec![1, 2, 3]]); + + // A column gap still comes first: the hunt sits below custody. + let supernode = custodian(constants::NUMBER_OF_CUSTODY_GROUPS); + let mut candidates = vec![DiscoveredPeer::for_test(vec![0]), supernode.clone()]; + rank_candidates( + &mut candidates, + &HashSet::new(), + &HashSet::from([97u64]), + &HashSet::from([0u64]), + ); + assert_eq!(candidates[0].peer_id, supernode.peer_id); + } + /// A candidate whose peer id is a real secp256k1 key, so the node id /// behind it can be recovered and its custody set computed. `PeerId::random` /// is not that: it is an arbitrary multihash, which is exactly the case @@ -880,7 +945,12 @@ mod tests { let well_subnetted = DiscoveredPeer::for_test(vec![0, 1, 2, 3]); let mut candidates = vec![well_subnetted.clone(), supernode.clone()]; - rank_candidates(&mut candidates, &HashSet::new(), &HashSet::from([97u64])); + rank_candidates( + &mut candidates, + &HashSet::new(), + &HashSet::from([97u64]), + &HashSet::new(), + ); assert_eq!( candidates[0].peer_id, supernode.peer_id, "a column no peer holds stops the chain; an uncovered subnet only narrows the view" @@ -888,7 +958,12 @@ mod tests { // With every column covered, subnet coverage decides again. let mut candidates = vec![supernode.clone(), well_subnetted.clone()]; - rank_candidates(&mut candidates, &HashSet::new(), &HashSet::new()); + rank_candidates( + &mut candidates, + &HashSet::new(), + &HashSet::new(), + &HashSet::new(), + ); assert_eq!(candidates[0].peer_id, well_subnetted.peer_id); } @@ -908,7 +983,12 @@ mod tests { DiscoveredPeer::for_test(vec![]), DiscoveredPeer::for_test(vec![7]), ]; - rank_candidates(&mut candidates, &HashSet::new(), &HashSet::new()); + rank_candidates( + &mut candidates, + &HashSet::new(), + &HashSet::new(), + &HashSet::new(), + ); let order: Vec<_> = candidates.iter().map(|c| c.subnets.clone()).collect(); assert_eq!( order, diff --git a/crates/net/p2p/src/discovery/dial.rs b/crates/net/p2p/src/discovery/dial.rs index 1b92da1a8..93c3131eb 100644 --- a/crates/net/p2p/src/discovery/dial.rs +++ b/crates/net/p2p/src/discovery/dial.rs @@ -7,6 +7,10 @@ //! Also the other direction: [`report_peer_connected`] and //! [`report_peer_disconnected`] tell discovery what the swarm did, which is the //! only way a libp2p connection becomes visible to it. +//! +//! Every tick also tells the subnet hunt ([`SubnetHunt`]) which subscribed +//! attestation subnets are short of peers, and ranks what it found in with the +//! rest. use std::collections::{HashMap, HashSet, VecDeque}; use std::time::Duration; @@ -19,6 +23,7 @@ use tracing::trace; use super::admission::{DialTargets, DiscoveredPeer, LeanFilter, rank_candidates}; use super::node_id::node_id_from_peer_id; +use super::subnets::SubnetHunt; use super::{ DIAL_INTERVAL_AT_TARGET, DIAL_INTERVAL_AT_ZERO_PEERS, DISCOVERY_CANDIDATE_BATCH, SpawnedDiscovery, @@ -39,8 +44,11 @@ pub(crate) struct DiscoveryState { /// holds `&mut P2PServer` for the length of a tick. contacts: mpsc::Receiver, /// Admitted candidates, best first, drained one per tick. Refilled from - /// `contacts` when empty. + /// `contacts` when empty, and joined by the subnet hunt's finds whenever + /// some arrive. candidates: VecDeque, + /// The search for peers on subscribed subnets this node is short of. + hunt: SubnetHunt, /// Subnets advertised by peers we dialed from discovery. peer_attnets: HashMap>, /// Our own peer ID, so the loop never dials itself. @@ -53,11 +61,17 @@ pub(crate) struct DiscoveryState { } impl DiscoveryState { - /// Starts the contact poll task, so it must be called from inside a tokio - /// runtime. Every call site builds a `P2PServer` from an async context. + /// Starts the contact poll and subnet hunt tasks, so it must be called from + /// inside a tokio runtime. Every call site builds a `P2PServer` from an + /// async context. pub(crate) fn new(spawned: SpawnedDiscovery, local_peer_id: PeerId) -> Self { Self { contacts: spawn_contact_poll(spawned.handle.clone(), spawned.filter.dial_targets()), + hunt: SubnetHunt::spawn( + spawned.handle.clone(), + spawned.filter.clone(), + spawned.subscription_subnets, + ), handle: spawned.handle, candidates: VecDeque::new(), peer_attnets: HashMap::new(), @@ -293,6 +307,13 @@ pub(crate) async fn dial_tick(server: &mut P2PServer, target_peers: usize) -> bo // borrow held across it keeps any swarm event from touching // `connected_peers` for the length of the tick. let budget = dial_budget(server, target_peers); + // Before the early return, so a loop with no budget tells the hunt to + // stand down rather than leaving it walking for peers nobody will dial. + if let Some(discovery) = server.discovery.as_ref() { + discovery + .hunt + .update(&discovery.peer_attnets, &server.connected_peers, budget > 0); + } if budget == 0 { return false; } @@ -372,31 +393,34 @@ pub(crate) async fn dial_tick(server: &mut P2PServer, target_peers: usize) -> bo } /// Rank whatever [`spawn_contact_poll`] has admitted since the last refill -/// into the candidate queue, once that queue has run out. +/// into the candidate queue, once that queue has run out, and whatever the +/// subnet hunt found into it at once. /// -/// Refilled only when empty, which is what leaves `spawn_contact_poll`'s -/// bounded channel able to do its job: draining on every tick would move -/// contacts into this unbounded queue as fast as the table could serve them, -/// and the backpressure that stops it being drawn down for nobody would be -/// gone. Ranking is per refill either way, since it scores a batch against -/// coverage this node has right now. +/// The poll's contacts are taken only when the queue is empty, which is what +/// leaves `spawn_contact_poll`'s bounded channel able to do its job: draining +/// on every tick would move contacts into this unbounded queue as fast as the +/// table could serve them, and the backpressure that stops it being drawn down +/// for nobody would be gone. The hunt's finds need no such wait, since a hunt +/// hands over a handful at most, and they are worth more than what is queued: +/// the queue is ranked again with them in it, so they go ahead of it. fn refill_candidates(server: &mut P2PServer) { let Some(discovery) = server.discovery.as_mut() else { return; }; - if !discovery.candidates.is_empty() { - return; - } - let mut admitted = Vec::with_capacity(DISCOVERY_CANDIDATE_BATCH); - while let Ok(peer) = discovery.contacts.try_recv() { - admitted.push(peer); + let mut admitted = discovery.hunt.drain_found(); + if discovery.candidates.is_empty() { + while let Ok(peer) = discovery.contacts.try_recv() { + admitted.push(peer); + } } if admitted.is_empty() { return; } + admitted.extend(discovery.candidates.drain(..)); let covered = covered_subnets(&discovery.peer_attnets, &server.connected_peers); - let wanted = undersupplied_custody_columns(server); - rank_candidates(&mut admitted, &covered, &wanted); + let wanted_subnets = discovery.hunt.wanted(); + let wanted_columns = undersupplied_custody_columns(server); + rank_candidates(&mut admitted, &covered, &wanted_columns, &wanted_subnets); if let Some(discovery) = server.discovery.as_mut() { discovery.candidates.extend(admitted); } diff --git a/crates/net/p2p/src/discovery/enr.rs b/crates/net/p2p/src/discovery/enr.rs index 3c894d97b..3160d9bd0 100644 --- a/crates/net/p2p/src/discovery/enr.rs +++ b/crates/net/p2p/src/discovery/enr.rs @@ -70,13 +70,17 @@ pub(crate) fn encode_attnets(subnets: &HashSet, committee_count: u64) -> Ve /// forever. A subnet we have no committee for cannot be useful to us regardless. pub(crate) fn subnets_from_attnets(bits: &[u8], committee_count: u64) -> Vec { (0..committee_count) - .filter(|subnet| { - bits.get((subnet / 8) as usize) - .is_some_and(|byte| byte & (1 << (subnet % 8)) != 0) - }) + .filter(|subnet| attnets_has(bits, *subnet)) .collect() } +/// Whether the `attnets` bitfield `bits` sets `subnet`'s bit. A bitfield too +/// short to reach it reads as unsubscribed, as in [`subnets_from_attnets`]. +pub(crate) fn attnets_has(bits: &[u8], subnet: u64) -> bool { + bits.get((subnet / 8) as usize) + .is_some_and(|byte| byte & (1 << (subnet % 8)) != 0) +} + /// The discv5 node id `node_key` derives, independent of any address or port. /// /// `keccak256` of the uncompressed public key: exactly what diff --git a/crates/net/p2p/src/discovery/mod.rs b/crates/net/p2p/src/discovery/mod.rs index e59e8a4bb..9ca962b9a 100644 --- a/crates/net/p2p/src/discovery/mod.rs +++ b/crates/net/p2p/src/discovery/mod.rs @@ -8,9 +8,13 @@ //! and let libp2p race them. //! //! The traffic across that boundary runs one way, into discovery: a request for -//! the next candidate, and a cast per connection opened or closed. Those casts -//! are what let discovery pace its own lookups, since a libp2p connection is -//! otherwise invisible to it. +//! the next candidate, a cast per connection opened or closed, and the +//! [`subnets`] hunt's predicate queries. The casts are what let discovery pace +//! its own lookups, since a libp2p connection is otherwise invisible to it. +//! +//! Discovery owns this node's ENR once it starts: it re-signs the record when +//! discv5's IP voting moves the address. [`LocalEnr`] reads the record it +//! currently serves, which is what the Beacon API reports. //! //! See `docs/discovery.md` for the operator-facing description. @@ -18,6 +22,7 @@ pub mod admission; pub(crate) mod dial; pub mod enr; pub(crate) mod node_id; +pub(crate) mod subnets; use std::collections::HashSet; use std::net::{IpAddr, SocketAddr}; @@ -25,7 +30,7 @@ use std::sync::{Arc, RwLock}; use std::time::Duration; use ethrex_p2p::discovery::{DiscoveryConfig, DiscoveryHandle, DiscoveryServer}; -use ethrex_p2p::types::{LocalNode, Node}; +use ethrex_p2p::types::{LocalNode, Node, SharedLocalNode}; use tokio::net::UdpSocket; use tokio::sync::watch; use tracing::{info, warn}; @@ -129,20 +134,20 @@ pub struct DiscoverySpawnConfig { /// What the P2P actor needs from a running discovery server. pub struct SpawnedDiscovery { - /// The running server. Three things travel over it: a request for the next - /// dial candidate, and a report for each connection opened and closed. + /// The running server. Over it travel a request for the next dial + /// candidate, a report for each connection opened and closed, and the + /// subnet hunt's queries. pub handle: DiscoveryHandle, /// A clone of the admission policy discovery runs. Clones share their /// policy and their [`DialTargets`](admission::DialTargets), so this is /// both how a fork boundary moves what discovery admits and how the dial /// loop reads what discovery filed. pub(crate) filter: LeanFilter, - /// This node's ENR as an `enr:`-prefixed string. `spawn_discovery` already - /// logs it; this copy is what the tests assert the published record against, - /// and what a future RPC identity endpoint would read. Reflects startup - /// state; discv5 may bump the sequence number later if PONG voting changes - /// our external IP. - pub local_enr: String, + /// The ENR discovery serves, read live; see [`LocalEnr`]. + pub local_enr: LocalEnr, + /// The attestation subnets this node subscribes to, which its ENR's + /// `attnets` advertises. What the subnet hunt keeps supplied with peers. + pub(crate) subscription_subnets: HashSet, /// The configured [`DiscoverySpawnConfig::target_peers`], carried through to /// the dial loop's cutoff. Discovery is handed the same number separately, /// for its lookup pacing. @@ -176,6 +181,7 @@ pub async fn spawn_discovery( let bound = socket.local_addr().map_err(DiscoveryError::SocketAddr)?; let advertise_ip = config.advertise_ip.unwrap_or(config.bind_ip); + let subscription_subnets = config.subscription_subnets.clone(); let params = LocalEnrParams { signer, ip: advertise_ip, @@ -188,12 +194,10 @@ pub async fn spawn_discovery( }; let local_node = params.local_node(); let local_record = build_local_enr(¶ms)?; - // EIP-778's text form is URL-safe base64 *without* padding - let local_enr = local_record - .enr_url() - .map_err(DiscoveryError::EncodeEnr)? - .trim_end_matches('=') - .to_string(); + // Encoded once up front so a record that cannot be is a startup error, not + // an empty answer from the identity endpoint later. ethrex writes EIP-778's + // unpadded base64, which lighthouse requires. + let startup_enr = local_record.enr_url().map_err(DiscoveryError::EncodeEnr)?; // Our own filter rather than ethrex's `EthForkIdFilter`, which wants an // EIP-2124 `eth` entry compatible with an execution chain we do not judge @@ -224,9 +228,9 @@ pub async fn spawn_discovery( // re-signs it under `params.signer` whenever IP voting bumps the sequence // number, keeping the extra entries. // - // `shared_local_node` is ethrex's live mirror of that identity, which its RPC - // layer serves the current record from. Nothing on this side reads it yet, so - // it is written and dropped; it costs one `Arc`. + // `shared_local_node` is ethrex's live mirror of that identity: every + // re-signed record is written back into it. [`LocalEnr`] keeps a clone, so + // the identity endpoint reports the record discv5 actually serves. // // `fork_id` publishes an EIP-2124 fork id for discovery to stamp on the ENR // when the chain crosses a fork. Neither chain here keys its peers on an @@ -239,6 +243,7 @@ pub async fn spawn_discovery( node: local_node.clone(), record: local_record.clone(), })); + let local_enr = LocalEnr(shared_local_node.clone()); let server = DiscoveryServer::spawn( local_node, @@ -274,7 +279,7 @@ pub async fn spawn_discovery( let discovery = DiscoveryHandle::new(); discovery.set(server); - info!(enr = %local_enr, "Local ENR"); + info!(enr = %startup_enr, "Local ENR"); if advertise_ip.is_unspecified() { warn!( "Local ENR advertises an unspecified IP; peers can still reach us once \ @@ -288,24 +293,61 @@ pub async fn spawn_discovery( handle: discovery, filter, local_enr, + subscription_subnets, target_peers: config.target_peers, }) } +/// This node's ENR as discovery serves it right now. +/// +/// discv5's IP voting re-signs the record at runtime with a new address and a +/// higher `seq`, and ethrex writes every record it re-signs into the +/// [`SharedLocalNode`] it was started with. Reading through that, instead of +/// keeping the string built at startup, is what keeps the Beacon API's +/// `/eth/v1/node/identity` reporting the record peers get over discv5. +/// +/// One direction only: nothing written here would reach discv5, which serves +/// its own copy and overwrites this one on its next edit. See +/// [`update_served_enr`] for the other direction. +#[derive(Clone)] +pub struct LocalEnr(SharedLocalNode); + +impl LocalEnr { + /// The served record as an `enr:` string, or `None` if it cannot be read. + /// + /// Encoding cannot fail for a record `spawn_discovery` already encoded, + /// since IP voting edits only the address. A poisoned lock means ethrex + /// panicked mid-edit; this answers `None` rather than spread the panic + /// into whichever caller asked. + pub fn current(&self) -> Option { + // Cloned out so the guard, which is `!Send`, is dropped right here. + let record = self.0.read().ok()?.record.clone(); + record + .enr_url() + .inspect_err(|err| warn!(%err, "Could not encode the local ENR")) + .ok() + } +} + /// Publish a new `eth2` entry in the ENR this node serves over discv5. /// /// Not possible with the discovery stack this node is built on, so this only -/// says so. [`DiscoveryServer::spawn`] takes the local record by value, moves it -/// into an actor it starts itself, and returns `()`: no actor handle comes back, -/// and `DiscoveryServerProtocol` (`raw_packet`, `lookup_*`, `revalidate_*`, -/// `enr_lookup`, `prune`, `shutdown`) has no message that replaces the record. -/// The record is re-signed only internally, by discv5 IP voting -/// (`update_local_ip`), which edits the address and nothing else. What would -/// close the gap is a protocol message on `DiscoveryServer` that edits the -/// local record and bumps its `seq`, or `spawn` returning the actor's handle. +/// says so. [`DiscoveryServer::spawn`] now returns the actor's handle, but +/// `DiscoveryServerProtocol` still has no message that edits the local record: +/// its consumer-facing messages report peer events, prune, ask for dial +/// candidates, run predicate queries and refresh a *remote* ENR. ethrex edits +/// its own record in two places only, both internal: discv5 IP voting +/// (`update_local_ip`, the address) and `refresh_fork_id`, fed by the `fork_id` +/// watch channel `spawn` takes. That one is consumer-driven, but it writes the +/// EIP-2124 `eth` entry, an execution-chain fork id, not the consensus `eth2` +/// entry this node needs moved. Writing into the [`SharedLocalNode`] mirror +/// does not help either: discovery serves its own copy and overwrites the +/// mirror on its next edit. What would close the gap is a protocol message +/// that applies an edit through `NodeRecord::edit` (one `seq` bump, re-signed) +/// and mirrors the result, as `refresh_fork_id` already does for `eth`. /// /// Until then the record served to discv5 queries keeps the `eth2` entry it -/// started with. Peers already connected are unaffected (they handshake on +/// started with, and so does [`LocalEnr`], which reads what discovery serves. Peers already connected are unaffected (they handshake on /// `Status`, which does follow the switch) and the admission filter does move, /// but a peer that discovers this node after a boundary reads the stale digest /// and rejects the record, so a long-running node becomes harder to find until @@ -327,10 +369,6 @@ mod tests { use ethrex_rlp::decode::RLPDecode; use std::net::Ipv4Addr; - /// The committee count `config` below spawns with, named so the filter the - /// self-admission test rebuilds cannot silently drift from it. - const TEST_COMMITTEE_COUNT: u64 = 4; - fn config(discovery_port: u16, advertise_ip: Option) -> DiscoverySpawnConfig { DiscoverySpawnConfig { node_key: secp256k1::SecretKey::new(&mut rand::rngs::OsRng) @@ -340,7 +378,7 @@ mod tests { discovery_port, p2p_port: 9001, subscription_subnets: HashSet::from([0u64]), - attestation_committee_count: TEST_COMMITTEE_COUNT, + attestation_committee_count: 4, bootnodes: Vec::new(), advertise_ip, target_peers: DEFAULT_DISCOVERY_TARGET_PEERS, @@ -367,11 +405,12 @@ mod tests { .await .expect("discovery spawns"); - assert!(handle.local_enr.starts_with("enr:")); + let local_enr = handle.local_enr.current().expect("the served ENR encodes"); + assert!(local_enr.starts_with("enr:")); // EIP-778's text form carries no base64 padding, and lighthouse // refuses a record that does. - assert!(!handle.local_enr.contains('='), "{}", handle.local_enr); - let record = decode_enr(&handle.local_enr); + assert!(!local_enr.contains('='), "{local_enr}"); + let record = decode_enr(&local_enr); // With discovery_port: 0 the OS picks the real port, and the published // ENR must advertise that one: were it built from the requested 0 @@ -381,11 +420,8 @@ mod tests { // The policy handed to discovery must admit our own record. A peer // running this code applies exactly these rules to what we publish, so a - // record we would reject ourselves is one nobody dials. Rebuilt here - // from the same two values `spawn_discovery` constructs it with, since - // the filter itself was moved into the discovery server. - let filter = LeanFilter::new(EnrForkId::local(), TEST_COMMITTEE_COUNT); - assert!(filter.accepts(&record)); + // record we would reject ourselves is one nobody dials. + assert!(handle.filter.accepts(&record)); } #[tokio::test] @@ -400,7 +436,8 @@ mod tests { .await .expect("discovery spawns"); - assert_eq!(decode_enr(&handle.local_enr).pairs().ip, Some(advertised)); + let local_enr = handle.local_enr.current().expect("the served ENR encodes"); + assert_eq!(decode_enr(&local_enr).pairs().ip, Some(advertised)); } #[tokio::test] diff --git a/crates/net/p2p/src/discovery/subnets.rs b/crates/net/p2p/src/discovery/subnets.rs new file mode 100644 index 000000000..fa72fb2b7 --- /dev/null +++ b/crates/net/p2p/src/discovery/subnets.rs @@ -0,0 +1,467 @@ +//! The subnet hunt: aim discv5 at the attestation subnets this node subscribes +//! to but has too few connected peers advertising. +//! +//! The dial loop draws whatever the crawl happened to find and ranks it, which +//! only reorders random draws. About one mainnet record in 32 advertises a given +//! subnet, so a subscribed subnet nobody connected covers can stay that way for +//! as long as the crawl takes to stumble on a peer for it. ethrex's predicate +//! queries let the search be aimed instead: discovery runs our test on every +//! record it sees and hands back the ones that pass. +//! +//! ```text +//! dial tick ── wanted subnets ──► watch ──► hunt task (one hunt at a time) +//! ▲ │ 1. records_matching: what discovery holds +//! │ │ 2. find_nodes: walk the DHT for the rest +//! └──── admitted peers ◄──── mpsc ◄───────┘ +//! ``` +//! +//! The dial tick says what is short, the task does the asking, and the finds +//! come back as ordinary candidates: [`rank_candidates`] puts a peer on a +//! wanted subnet ahead of the crawl's random draws, and the dial loop dials it +//! under the same budget as anything else. The task owns every `.await`, since +//! the dial loop holds `&mut P2PServer` for its whole tick and must never wait +//! on discovery. +//! +//! Lighthouse does the same with sigp/discv5's `find_node_predicate`, whose +//! semantics ethrex's `find_nodes` copies: a match counts only once its node has +//! answered the query, so dead records stay out. The predicate is +//! Lighthouse's too: everything admission checks, plus at least one wanted +//! `attnets` bit. +//! +//! [`rank_candidates`]: super::admission::rank_candidates + +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::time::{Duration, Instant}; + +use ethrex_common::H256; +use ethrex_p2p::discovery::DiscoveryHandle; +use ethrex_p2p::types::NodeRecord; +use libp2p::PeerId; +use tokio::sync::{mpsc, watch}; +use tracing::debug; + +use super::admission::{DiscoveredPeer, LeanFilter}; +use super::enr::{ATTNETS_ENR_KEY, attnets_has}; +use crate::ConnectionDirection; + +/// Connected peers advertising a subscribed subnet, below which the hunt looks +/// for more. Lighthouse's `TARGET_SUBNET_PEERS`. +/// +/// More than one because a subnet's only peer is one disconnect away from +/// leaving it uncovered, and the crawl is in no hurry to replace it. +pub(crate) const TARGET_SUBNET_PEERS: usize = 3; + +/// Most peers one hunt hands the dial loop. Lighthouse asks the same number of +/// each grouped subnet query (`TARGET_PEERS_FOR_GROUPED_QUERY`). +/// +/// Not the shortfall itself: a match needs only one wanted bit, so the peers a +/// hunt finds can all land on the same subnet. The next hunt recounts. +const MAX_HUNT_RESULTS: usize = 6; + +/// How long one network walk may run before it answers with what it found. +/// +/// Short against ethrex's own cap (`MAX_QUERY_TIMEOUT`, 60 s): a find is only +/// worth something once it is dialed, and the next hunt can always walk again. +const HUNT_QUERY_TIMEOUT: Duration = Duration::from_secs(10); + +/// Pause between the end of one hunt and the start of the next. +/// +/// Long enough for the last hunt's dials to become connections, so the next +/// one counts them instead of hunting for peers already on their way; a dial +/// that times out takes ten seconds by itself. On a small network that has no +/// more peers on a subnet to give, this is also the rate a hunt that finds +/// nothing is repeated at. +const HUNT_INTERVAL: Duration = Duration::from_secs(30); + +/// How long a peer the hunt handed over stays out of later hunts. +/// +/// Discovery keeps offering a peer that turned us away: a mainnet node at its +/// inbound cap completes the handshake and then says goodbye, which reads as a +/// connection that closed, not one refused. Without this every hunt would +/// return the same few full nodes and the dial loop would spend its budget +/// on them every time. +const REOFFER_AFTER: Duration = Duration::from_secs(10 * 60); + +/// The dial loop's end of the hunt. +pub(crate) struct SubnetHunt { + /// Subnets this node subscribes to, which are what its ENR's `attnets` + /// advertises. Fixed for the life of the process: neither chain rotates + /// them (see `beacon::subnets`). + subscribed: HashSet, + /// What is short right now, read by the hunt task. Empty means nothing to + /// hunt for, which is also what is published while the dial loop has no + /// budget to dial a find with. + wanted: watch::Sender>, + /// Peers the hunt found, already through admission. + found: mpsc::UnboundedReceiver, +} + +impl SubnetHunt { + /// Start the hunt task, idle until [`Self::update`] publishes a subnet. + /// Must be called from inside a tokio runtime. + /// + /// The task ends once this value is dropped, at the end of the hunt it is + /// in, which is when the `P2PServer` holding it goes away. + pub(crate) fn spawn( + handle: DiscoveryHandle, + filter: LeanFilter, + subscribed: HashSet, + ) -> Self { + let (wanted, wanted_rx) = watch::channel(BTreeSet::new()); + let (found_tx, found) = mpsc::unbounded_channel(); + tokio::spawn(hunt(handle, filter, wanted_rx, found_tx)); + Self { + subscribed, + wanted, + found, + } + } + + /// Recount and publish which subscribed subnets are short of + /// [`TARGET_SUBNET_PEERS`]. + /// + /// `may_dial` is whether the dial loop has budget for a find this tick. + /// Without it nothing is wanted, so the task does not walk the network for + /// peers nobody would dial. + pub(crate) fn update( + &self, + peer_attnets: &HashMap>, + connected_peers: &HashMap, + may_dial: bool, + ) { + let short = if may_dial { + short_subnets(&self.subscribed, peer_attnets, connected_peers) + } else { + BTreeSet::new() + }; + // Only a change wakes the task, and the dial tick runs up to fifty + // times a second. + self.wanted.send_if_modified(|wanted| { + let changed = *wanted != short; + *wanted = short; + changed + }); + } + + /// The subnets the hunt is looking for right now, for ranking. + pub(crate) fn wanted(&self) -> HashSet { + self.wanted.borrow().iter().copied().collect() + } + + /// Whatever the hunt has found since the last call. + pub(crate) fn drain_found(&mut self) -> Vec { + let mut found = Vec::new(); + while let Ok(peer) = self.found.try_recv() { + found.push(peer); + } + found + } +} + +/// The subscribed subnets fewer than [`TARGET_SUBNET_PEERS`] connected peers +/// advertise. +/// +/// Counted from `peer_attnets`, which holds the `attnets` of peers this node +/// dialed from discovery: an inbound peer never shows us its ENR, so it covers +/// nothing here. That undercounts, which makes the hunt more eager than it has +/// to be, never blind to a real gap. +fn short_subnets( + subscribed: &HashSet, + peer_attnets: &HashMap>, + connected_peers: &HashMap, +) -> BTreeSet { + let mut peers_on: HashMap = HashMap::new(); + for subnets in peer_attnets + .iter() + .filter(|(peer, _)| connected_peers.contains_key(peer)) + .map(|(_, subnets)| subnets) + { + for subnet in subnets.iter().filter(|subnet| subscribed.contains(subnet)) { + *peers_on.entry(*subnet).or_insert(0) += 1; + } + } + subscribed + .iter() + .copied() + .filter(|subnet| peers_on.get(subnet).copied().unwrap_or(0) < TARGET_SUBNET_PEERS) + .collect() +} + +/// The hunt task: one hunt per [`HUNT_INTERVAL`] while anything is wanted. +/// +/// A hunt first offers the matches discovery already holds, which costs no +/// network round trip, and walks the network only for what that left short. +/// Both answers go through the same predicate, [`hunt_predicate`]. +async fn hunt( + handle: DiscoveryHandle, + filter: LeanFilter, + mut wanted: watch::Receiver>, + found: mpsc::UnboundedSender, +) { + let mut offered: HashMap = HashMap::new(); + loop { + // An error is the dial loop's sender going away, which is shutdown. + let Ok(subnets) = wanted + .wait_for(|wanted| !wanted.is_empty()) + .await + .map(|wanted| wanted.clone()) + else { + return; + }; + offered.retain(|_, at| at.elapsed() < REOFFER_AFTER); + + let predicate = hunt_predicate(&filter, &subnets, &offered); + let cached = handle.records_matching(predicate).await; + let mut handed = offer(&filter, cached, MAX_HUNT_RESULTS, &mut offered, &found); + + if handed < MAX_HUNT_RESULTS { + // Rebuilt, so the walk does not count the peers just offered. + let predicate = hunt_predicate(&filter, &subnets, &offered); + match handle + .find_nodes(predicate, MAX_HUNT_RESULTS - handed, HUNT_QUERY_TIMEOUT) + .await + { + Ok(result) => { + let walked = offer( + &filter, + result.records, + MAX_HUNT_RESULTS - handed, + &mut offered, + &found, + ); + debug!( + ?subnets, + cached = handed, + walked, + outcome = ?result.outcome, + nodes_seen = result.nodes_seen, + findnodes_sent = result.findnodes_sent, + "Subnet hunt finished" + ); + handed += walked; + } + // Not running is shutdown or a discovery that failed to start, + // and a refusal is a full query slot: either way, the next hunt + // asks again. + Err(err) => debug!(?subnets, cached = handed, %err, "Subnet hunt could not walk"), + } + } else { + debug!( + ?subnets, + cached = handed, + "Subnet hunt satisfied from known records" + ); + } + if found.is_closed() { + return; + } + crate::metrics::inc_subnet_hunt_peers(handed as u64); + tokio::time::sleep(HUNT_INTERVAL).await; + } +} + +/// Which records a hunt for `subnets` wants: one advertising at least one of +/// them, that admission would dial, and that the hunt has not offered within +/// [`REOFFER_AFTER`]. +/// +/// Runs inside discovery's message loop on every record a walk sees, so the +/// one-bit test goes first and turns most records away before admission +/// decodes anything. The rest costs what [`PeerFilter`] does on every record +/// discovery takes in, which is the bound ethrex's documentation asks for. +/// +/// [`PeerFilter`]: ethrex_p2p::peer_filter::PeerFilter +fn hunt_predicate( + filter: &LeanFilter, + subnets: &BTreeSet, + offered: &HashMap, +) -> impl Fn(&NodeRecord) -> bool + Send + 'static { + let filter = filter.clone(); + let subnets = subnets.clone(); + let offered: HashSet = offered.keys().copied().collect(); + move |record| { + let advertises_one = record + .pairs() + .extra(ATTNETS_ENR_KEY) + .is_some_and(|bits| subnets.iter().any(|subnet| attnets_has(&bits, *subnet))); + advertises_one + && filter + .dial_target(record) + .is_some_and(|peer| !offered.contains(&peer.node_id)) + } +} + +/// Send up to `limit` of `records` to the dial loop, noting each in `offered`. +/// Returns how many went. +/// +/// Admission runs again here rather than trusting the predicate: a fork +/// boundary can move the policy between discovery matching a record and the +/// answer arriving, and a record it no longer admits has nothing to dial. +fn offer( + filter: &LeanFilter, + records: Vec, + limit: usize, + offered: &mut HashMap, + found: &mpsc::UnboundedSender, +) -> usize { + let mut handed = 0; + for peer in records + .iter() + .filter_map(|record| filter.dial_target(record)) + { + if handed == limit { + break; + } + if offered.contains_key(&peer.node_id) { + continue; + } + offered.insert(peer.node_id, Instant::now()); + if found.send(peer).is_err() { + break; + } + handed += 1; + } + handed +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::discovery::enr::{EnrForkId, LocalEnrParams, build_local_enr}; + use libp2p::identity::Keypair; + use std::net::Ipv4Addr; + + /// The committee count every record and filter below is built with. + const COMMITTEE_COUNT: u64 = 8; + + fn random_peer() -> PeerId { + PeerId::from_public_key(&Keypair::generate_ed25519().public()) + } + + /// A peer's ENR as this node would publish it, on `fork_id`'s network and + /// subscribed to `subnets`. + fn peer_record(fork_id: EnrForkId, subnets: &[u64]) -> NodeRecord { + build_local_enr(&LocalEnrParams { + signer: secp256k1::SecretKey::new(&mut rand::rngs::OsRng), + ip: Ipv4Addr::LOCALHOST.into(), + discovery_port: 9000, + p2p_port: 9001, + subscription_subnets: subnets.iter().copied().collect(), + attestation_committee_count: COMMITTEE_COUNT, + fork_id, + custody_group_count: None, + }) + .expect("a record builds") + } + + fn filter() -> LeanFilter { + LeanFilter::new(EnrForkId::local(), COMMITTEE_COUNT) + } + + #[test] + fn the_hunt_wants_an_admissible_peer_on_a_wanted_subnet() { + let wanted = BTreeSet::from([2]); + let predicate = hunt_predicate(&filter(), &wanted, &HashMap::new()); + + assert!(predicate(&peer_record(EnrForkId::local(), &[2, 5]))); + assert!( + !predicate(&peer_record(EnrForkId::local(), &[5])), + "a peer on none of the wanted subnets is not what the hunt is for" + ); + let mut elsewhere = EnrForkId::local(); + elsewhere.fork_digest = [0xde, 0xad, 0xbe, 0xef]; + assert!( + !predicate(&peer_record(elsewhere, &[2])), + "the wanted bit does not get a peer on another network past admission" + ); + } + + #[test] + fn a_peer_already_offered_is_left_out_of_the_next_hunt() { + let filter = filter(); + let record = peer_record(EnrForkId::local(), &[2]); + let node_id = filter.dial_target(&record).expect("admitted").node_id; + let offered = HashMap::from([(node_id, Instant::now())]); + + let predicate = hunt_predicate(&filter, &BTreeSet::from([2]), &offered); + + assert!(!predicate(&record)); + } + + #[test] + fn offer_stops_at_its_limit_and_never_hands_over_a_peer_twice() { + let filter = filter(); + let records: Vec = (0..3) + .map(|_| peer_record(EnrForkId::local(), &[1])) + .collect(); + let (found, mut received) = mpsc::unbounded_channel(); + let mut offered = HashMap::new(); + + assert_eq!(offer(&filter, records.clone(), 2, &mut offered, &found), 2); + // The same three again: two were handed over already, one is left. + assert_eq!(offer(&filter, records, 2, &mut offered, &found), 1); + + let mut handed = HashSet::new(); + while let Ok(peer) = received.try_recv() { + assert!(handed.insert(peer.node_id), "handed over twice"); + } + assert_eq!(handed.len(), 3); + } + + #[tokio::test] + async fn nothing_is_wanted_while_the_dial_loop_cannot_dial() { + // An unset handle: the task it starts finds discovery not running, + // which is all this test needs, since it reads only what is published. + let hunt = SubnetHunt::spawn(DiscoveryHandle::new(), filter(), HashSet::from([1, 4])); + + hunt.update(&HashMap::new(), &HashMap::new(), true); + assert_eq!(hunt.wanted(), HashSet::from([1, 4])); + + hunt.update(&HashMap::new(), &HashMap::new(), false); + assert!(hunt.wanted().is_empty()); + } + + fn connected(peers: &[PeerId]) -> HashMap { + peers + .iter() + .map(|peer| (*peer, ConnectionDirection::Outbound)) + .collect() + } + + #[test] + fn a_subscribed_subnet_below_target_is_short() { + let peers: Vec = (0..TARGET_SUBNET_PEERS).map(|_| random_peer()).collect(); + // Subnet 1 has a full complement, subnet 4 is one peer short of it. + let mut peer_attnets: HashMap> = + peers.iter().map(|peer| (*peer, vec![1u64])).collect(); + for peer in peers.iter().skip(1) { + peer_attnets.get_mut(peer).unwrap().push(4); + } + + let short = short_subnets(&HashSet::from([1, 4]), &peer_attnets, &connected(&peers)); + + assert_eq!(short, BTreeSet::from([4])); + } + + #[test] + fn only_connected_peers_and_subscribed_subnets_count() { + let here = random_peer(); + let gone = random_peer(); + // `gone` advertised subnet 2, but is no longer connected; `here` + // covers subnet 9, which this node does not subscribe to. + let peer_attnets = HashMap::from([(here, vec![9u64]), (gone, vec![2u64])]); + + let short = short_subnets(&HashSet::from([2]), &peer_attnets, &connected(&[here])); + + assert_eq!(short, BTreeSet::from([2])); + } + + #[test] + fn a_node_subscribed_to_nothing_hunts_for_nothing() { + let peer = random_peer(); + let peer_attnets = HashMap::from([(peer, vec![0u64])]); + + let short = short_subnets(&HashSet::new(), &peer_attnets, &connected(&[peer])); + + assert!(short.is_empty()); + } +} diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index d656307d7..aaca1847f 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -1132,9 +1132,9 @@ pub fn build_swarm(config: SwarmConfig) -> Result { /// Public handle to the P2P actor. pub struct P2P { handle: ActorRef, - /// This node's ENR as published at startup, for the Beacon API's + /// This node's ENR as discovery currently serves it, for the Beacon API's /// `/eth/v1/node/identity`. `None` when discovery is disabled. - local_enr: Option, + local_enr: Option, } impl P2P { @@ -1286,12 +1286,11 @@ impl P2P { &self.handle } - /// This node's ENR, `enr:`-prefixed, as published at startup. discv5 may - /// re-sign it later with a higher sequence number if IP voting changes the - /// external address; this is the startup record. `None` when discovery is - /// disabled, since then no ENR is published. - pub fn local_enr(&self) -> Option<&str> { - self.local_enr.as_deref() + /// A reader for this node's ENR as discovery serves it, which follows the + /// re-signed record when discv5's IP voting changes the external address. + /// `None` when discovery is disabled, since then no ENR is published. + pub fn local_enr(&self) -> Option { + self.local_enr.clone() } } diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index 643300967..4dc55b588 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/crates/net/p2p/src/metrics.rs @@ -313,6 +313,20 @@ pub fn inc_discovered_peers_dialed() { LEAN_DISCOVERED_PEERS_DIALED.inc(); } +/// Counts peers the subnet hunt handed the dial loop: discovery records on a +/// subscribed attestation subnet too few connected peers advertise. Whether +/// each one connected is in the peer connect and disconnect metrics. +pub fn inc_subnet_hunt_peers(count: u64) { + static LEAN_DISCOVERY_SUBNET_HUNT_PEERS: LazyLock = LazyLock::new(|| { + register_int_counter!( + "lean_discovery_subnet_hunt_peers_total", + "Peers found by discv5 subnet hunts, for subscribed attestation subnets short of peers" + ) + .unwrap() + }); + LEAN_DISCOVERY_SUBNET_HUNT_PEERS.inc_by(count); +} + /// Refresh the gossipsub mesh peers gauge from the current mesh peer set. pub fn update_gossip_mesh_peers<'a>( peers: impl Iterator, diff --git a/crates/net/rpc/src/beacon/node.rs b/crates/net/rpc/src/beacon/node.rs index aa2c0cabf..a895ffd92 100644 --- a/crates/net/rpc/src/beacon/node.rs +++ b/crates/net/rpc/src/beacon/node.rs @@ -108,7 +108,7 @@ async fn get_identity(peer_id: String, identity: Option) crate::json_response(serde_json::json!({ "data": { "peer_id": peer_id, - "enr": identity.enr, + "enr": identity.enr.current(), "p2p_addresses": identity.p2p_addresses, "discovery_addresses": identity.discovery_addresses, "metadata": { @@ -239,7 +239,7 @@ mod tests { async fn identity_reports_the_enr_and_listen_addresses() { let fixture = beacon_fixture(ANCHOR_SLOT); let identity = crate::BeaconIdentity { - enr: "enr:-abc".to_string(), + enr: "enr:-abc".to_string().into(), p2p_addresses: vec!["/ip4/10.0.0.1/tcp/9001/p2p/test-peer".to_string()], discovery_addresses: vec!["/ip4/10.0.0.1/udp/9000/p2p/test-peer".to_string()], }; diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 9173f0ac4..ee94c139d 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -196,14 +196,48 @@ pub struct CustodyColumns(pub Vec); /// How peers reach this node, as `/eth/v1/node/identity` reports it. #[derive(Debug, Clone, Default)] pub struct BeaconIdentity { - /// The node's ENR, `enr:`-prefixed. - pub enr: String, + /// The node's ENR, `enr:`-prefixed, read when the endpoint is asked. + pub enr: EnrSource, /// libp2p multiaddrs the node listens on, each ending in `/p2p/`. pub p2p_addresses: Vec, /// discv5 multiaddrs, likewise. pub discovery_addresses: Vec, } +/// Where `/eth/v1/node/identity` reads the node's ENR from. +/// +/// A reader rather than a string because the record changes after startup: +/// discv5's IP voting re-signs it with a new address and a higher `seq`, and a +/// copy taken at startup would go on reporting a record peers no longer get. +/// The default reads nothing, which the endpoint reports as an empty ENR, the +/// answer for a node that publishes none. +#[derive(Clone, Default)] +pub struct EnrSource(Option Option + Send + Sync>>); + +impl EnrSource { + pub fn new(read: impl Fn() -> Option + Send + Sync + 'static) -> Self { + Self(Some(Arc::new(read))) + } + + /// The ENR right now, or an empty string when there is none to read. + pub fn current(&self) -> String { + self.0.as_ref().and_then(|read| read()).unwrap_or_default() + } +} + +/// A fixed ENR, for a record that never changes. +impl From for EnrSource { + fn from(enr: String) -> Self { + Self::new(move || Some(enr.clone())) + } +} + +impl std::fmt::Debug for EnrSource { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_tuple("EnrSource").field(&self.current()).finish() + } +} + /// What the Beacon API's endpoints reach beyond the store. pub struct BeaconApiHandles { /// Through which the pool, aggregate and block endpoints gossip what a diff --git a/docs/discovery.md b/docs/discovery.md index e2545aee1..d17544ed1 100644 --- a/docs/discovery.md +++ b/docs/discovery.md @@ -81,10 +81,23 @@ a `0` either way, matching every reader's rule that `0` means absent. The local ENR is logged once at startup. This same record is handed to ethrex's `DiscoveryServer`, so it is what answers -discv5 queries: what we report and what peers see are the same bytes. If IP -voting later changes our external address, ethrex edits and re-signs that record -rather than rebuilding one, so the consensus entries survive the bump; only the -sequence number and `ip` move, which the reported ENR then lags. +discv5 queries. If IP voting later changes our external address, ethrex edits and +re-signs that record rather than rebuilding one, so the consensus entries survive +the bump; only the sequence number and `ip` move. + +`/eth/v1/node/identity` reports the record discovery serves *now*, not the one +logged at startup. ethrex writes every record it re-signs into a shared mirror +(`SharedLocalNode`) it was started with, and the endpoint reads through that +(`discovery::LocalEnr`), so what the Beacon API reports and what peers get over +discv5 are the same bytes even after IP voting. + +The other direction is not possible yet: nothing on this side can change the +record discovery serves. ethrex's discovery protocol has no message that edits +the local record. Its only consumer-driven edit is the EIP-2124 `eth` entry +(an execution-chain fork id), not the `eth2` entry a fork-digest boundary moves, +and writing into the mirror does nothing, since discovery serves its own copy and +overwrites the mirror on its next edit. See the next section for what that costs +at a boundary. ## Which peers get dialed @@ -98,8 +111,9 @@ The digest compared against is the caller's, not a constant: lean passes its hardcoded dummy and `beacon` passes the digest it derived at startup and moves it at each fork-digest boundary, so one admission policy serves both chains. The record this node serves is not refreshed at a boundary (ethrex's -`DiscoveryServer` cannot replace it at runtime), so its `eth2` entry keeps the -startup digest until a restart; see [beacon_wire.md](./beacon_wire.md#the-fork-digest). +`DiscoveryServer` offers no way to edit it at runtime; see [The ENR](#the-enr)), +so its `eth2` entry keeps the startup digest until a restart; see +[beacon_wire.md](./beacon_wire.md#the-fork-digest). A differing `next_fork_version` or `next_fork_epoch` is *not* grounds for rejection: the spec permits connecting to a peer that is incompatible with an @@ -179,6 +193,42 @@ network with fewer peers to offer than `--discovery.target-peers` — every lean devnet, against a default target of 200 — from holding the loop at its floor for the life of the process, re-drawing candidates it is already connected to. +## Hunting for subnet peers + +Ranking only reorders what the crawl happened to find, and about one mainnet +record in 32 advertises a given subnet. So the subnets this node subscribes to, +the ones its own `attnets` advertises, are also searched for directly, with +ethrex's predicate queries (Lighthouse does the same with sigp/discv5's +`find_node_predicate`): + +```text + dial tick ── subscribed subnets with < 3 connected peers ──► hunt task + ▲ │ 1. records discovery already holds + │ │ 2. a discv5 walk for the rest + └───────────── admitted peers, ranked with the rest ◄──────┘ +``` + +- **What is short.** A subscribed subnet with fewer than `TARGET_SUBNET_PEERS` + (3, Lighthouse's number) connected peers advertising it. Counted from the + `attnets` of peers dialed from discovery, so an inbound peer covers nothing + here, as in ranking. +- **What matches.** A record that passes admission and sets at least one short + subnet's `attnets` bit. A match counts only once its node has answered the + walk, so dead records stay out. +- **How it runs.** One hunt at a time, off the P2P actor: first the matches + discovery already holds, then a walk of up to 10 s for the rest, at most 6 + peers per hunt, then a 30 s pause. Nothing is hunted while the dial loop has + no budget to dial a find. A peer a hunt handed over is left out of hunts for + 10 minutes, since discovery keeps offering a mainnet peer that connected and + said goodbye. +- **Where finds go.** Into the same candidate queue as everything else, ranked + after custody gaps and ahead of other subnets, and dialed under the same + budget. `lean_discovery_subnet_hunt_peers_total` counts them. + +The hunt covers the subnets this node subscribes to, not the ones a validator +client's duties point at (`beacon_committee_subscriptions`): those are not yet +fed to discovery. + ## Bootnodes The three entries a bootnode ENR can carry are read independently, because they @@ -262,11 +312,12 @@ mainnet's fork schedule and `genesis_validators_root`, so its `eth2` check separates mainnet from every other beacon network; see [`beacon_wire.md`](./beacon_wire.md). -### Only peers dialed from discovery contribute to subnet ranking +### Only peers dialed from discovery contribute to subnet ranking and the hunt An inbound peer never tells us its `attnets`, so a subnet it covers is not -credited when the dial loop ranks candidates. Treating an unknown peer as -covering nothing makes the ranking more eager than it needs to be, never wrong. +credited when the dial loop ranks candidates or when the subnet hunt counts what +is short. Treating an unknown peer as covering nothing makes both more eager +than they need to be, never wrong. ### `attnets` is not a fixed-width SSZ `Bitvector` diff --git a/docs/metrics.md b/docs/metrics.md index 1ea07953f..5a78dd2dd 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -143,6 +143,7 @@ here: a discovery dial that succeeds or fails shows up in | Name | Type | Usage | Sample collection event | Labels | |------|------|-------|-------------------------|--------| | `lean_discovered_peers_dialed_total` | Counter | Peers dialed as a result of discv5 discovery | On dialing a discovered peer | | +| `lean_discovery_subnet_hunt_peers_total` | Counter | Peers found by discv5 subnet hunts, for subscribed attestation subnets short of peers | On each subnet hunt, by the peers it handed the dial loop | | ### Transport Mix From 424a337c8aca4f99ecdda1f62edeab4529f112ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:25:20 -0300 Subject: [PATCH 4/4] docs(rpc): say /eth/v1/node/identity reports the ENR discv5 serves now The endpoint reads discovery's own copy of the record since the previous commit, so after IP voting it reports the re-signed record, IP included, rather than the startup one the paragraph described. --- docs/rpc.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/docs/rpc.md b/docs/rpc.md index 04d725892..aee1cf6fa 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -839,11 +839,14 @@ the block under it really sits at that slot. This matters because it is the slot a checkpoint-syncing peer asks for right after reading the finalized state. A slot the store holds nothing at is still a `404`. -`/eth/v1/node/identity` reports the peer id, the ENR discv5 publishes, and the -node's QUIC, TCP and discv5 multiaddrs. The multiaddrs need -`--discovery.advertise-ip`, since a node bound to `0.0.0.0` does not know the -address peers reach it on; without the flag they are empty and the ENR carries -no IP. `metadata` is still a placeholder. +`/eth/v1/node/identity` reports the peer id, the ENR discv5 serves at the moment +of the request, and the node's QUIC, TCP and discv5 multiaddrs. The ENR is read +from discovery's own copy, so once discv5's IP voting re-signs it with the +external address, the endpoint reports that record rather than the startup one. +The multiaddrs need `--discovery.advertise-ip`, since a node bound to `0.0.0.0` +does not know the address peers reach it on; without the flag they are empty, +and the ENR carries no usable IP until IP voting supplies one. `metadata` is +still a placeholder. ### `GET /eth/v1/events`