diff --git a/CLAUDE.md b/CLAUDE.md index d08ee404f..138c6f3c6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -394,9 +394,11 @@ 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` + `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`/`tcp`/`secp256k1`/`eth2`/`attnets` -- Admission mirrors lighthouse: `eth2.fork_digest` must match, `next_fork_*` may differ, a `quic` or `tcp` 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, 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) ### Retry Strategy on Block Requests diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..67cd20979 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2300,7 +2300,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", @@ -2325,7 +2325,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", @@ -2359,7 +2359,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", @@ -2381,7 +2381,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", @@ -2404,7 +2404,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", @@ -2422,7 +2422,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", @@ -2440,7 +2440,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", @@ -2452,7 +2452,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", @@ -2494,7 +2494,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", @@ -2504,7 +2504,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", @@ -2526,7 +2526,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", @@ -2546,7 +2546,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", diff --git a/crates/net/p2p/Cargo.toml b/crates/net/p2p/Cargo.toml index 22154e141..e6768b82f 100644 --- a/crates/net/p2p/Cargo.toml +++ b/crates/net/p2p/Cargo.toml @@ -64,13 +64,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" } # `rand` seeds the throwaway signing keys the ENR tests build records with. secp256k1.workspace = true diff --git a/crates/net/p2p/src/discovery/admission.rs b/crates/net/p2p/src/discovery/admission.rs index 6a1c6adf5..94e0fcf71 100644 --- a/crates/net/p2p/src/discovery/admission.rs +++ b/crates/net/p2p/src/discovery/admission.rs @@ -7,19 +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_candidates`]). -use std::collections::HashSet; -use std::sync::{Arc, RwLock}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, RwLock}; +use ethrex_common::H256; use ethrex_p2p::peer_filter::PeerFilter; use ethrex_p2p::types::NodeRecord; use libp2p::{Multiaddr, PeerId}; @@ -34,11 +37,24 @@ use super::enr::{ ATTNETS_ENR_KEY, CGC_ENR_KEY, ETH2_ENR_KEY, EnrForkId, node_id_from_peer_id, read_ip, read_public_key, read_quic_port, read_tcp_port, subnets_from_attnets, }; +use super::node_id::node_id_from_public_key; use crate::dial_addrs; +/// 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, /// Every dial target for this peer, built from whichever of the two ports /// the record actually advertises. Never empty: [`admit`] rejects a record @@ -56,11 +72,65 @@ pub(crate) struct DiscoveredPeer { pub(crate) custody_group_count: Option, } +/// 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` -/// record, so a peer that adds a `quic` entry or gains an address through -/// discv5's IP voting is reconsidered without restarting the process. +/// No reason is final: discovery re-runs the filter on every higher-`seq` +/// record, so a peer that adds a `quic` or `tcp` entry or gains an address +/// through discv5's IP voting is reconsidered without restarting the process. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RejectReason { /// No `eth2` entry, or one that does not decode. Cannot be our network. @@ -77,20 +147,23 @@ 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. +/// 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. /// -/// [`Clone`] because the peer table takes ownership of the filter it runs, and -/// every clone shares one [`Policy`], so [`Self::set_fork_id`] at a fork -/// boundary changes what the table's copy admits as well as the dial loop's. +/// [`Clone`] because discovery takes ownership of the filter it runs, and every +/// clone shares one [`Policy`] and one [`DialTargets`], so [`Self::set_fork_id`] +/// at a fork boundary changes what discovery's copy admits as well as the dial +/// loop's, and what discovery's copy files is what the dial loop reads. #[derive(Clone)] pub struct LeanFilter { policy: Arc>, attestation_committee_count: u64, + dial_targets: DialTargets, } /// The part of admission that moves at a fork boundary. @@ -111,13 +184,14 @@ impl LeanFilter { also_admitted: Vec::new(), })), attestation_committee_count, + dial_targets: DialTargets::default(), } } /// Replace the local `eth2` entry and the further digests admitted beside /// it. /// - /// The peer table re-judges a record only when its `seq` rises, so a peer + /// Discovery re-judges a record only when its `seq` rises, so a peer /// rejected under the old policy stays rejected until it republishes; /// peers cross a boundary by doing exactly that. pub(crate) fn set_fork_id(&self, fork_id: EnrForkId, also_admitted: Vec) { @@ -127,16 +201,9 @@ impl LeanFilter { }; } - /// 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 { - self.judge(record).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() } fn judge(&self, record: &NodeRecord) -> Result { @@ -152,10 +219,14 @@ impl LeanFilter { impl PeerFilter for LeanFilter { fn accepts(&self, record: &NodeRecord) -> bool { - self.judge(record) - // 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 self.judge(record) { + 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, @@ -163,8 +234,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 + } + } } } @@ -211,6 +290,7 @@ fn admit( } 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)?; @@ -231,6 +311,7 @@ fn admit( }); Ok(DiscoveredPeer { + node_id, peer_id, addrs: dial_addrs(ip, quic_port, tcp_port, peer_id), subnets, @@ -389,9 +470,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(), addrs: vec![Multiaddr::empty()], subnets, @@ -585,12 +668,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| { @@ -598,8 +705,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]); } @@ -612,8 +720,9 @@ 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] @@ -649,14 +758,40 @@ mod tests { #[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] @@ -681,7 +816,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 @@ -723,10 +861,12 @@ mod tests { ) .map(libp2p::identity::secp256k1::Keypair::from) .expect("a valid key"); + let node_id = node_id_from_public_key(keypair.public()); DiscoveredPeer { peer_id: libp2p::identity::Keypair::from(keypair) .public() .to_peer_id(), + node_id, addrs: vec![Multiaddr::empty()], subnets: vec![], custody_group_count: Some(custody_group_count), diff --git a/crates/net/p2p/src/discovery/dial.rs b/crates/net/p2p/src/discovery/dial.rs index 229573c26..1b92da1a8 100644 --- a/crates/net/p2p/src/discovery/dial.rs +++ b/crates/net/p2p/src/discovery/dial.rs @@ -1,35 +1,42 @@ //! The dial loop: turn what discv5 found into libp2p connections. //! -//! Runs as a `P2PServer` tick paced by [`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 paced by [`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 std::time::Duration; -use ethrex_p2p::discovery::lookup_interval_function; -use ethrex_p2p::peer_table::{PeerTable, PeerTableServerProtocol as _}; +use ethrex_p2p::discovery::{DiscoveryHandle, PeerEvent, lookup_interval_function}; use libp2p::PeerId; use libp2p::swarm::dial_opts::DialOpts; use tokio::sync::mpsc; use tracing::trace; -use super::admission::{DiscoveredPeer, LeanFilter, rank_candidates}; +use super::admission::{DialTargets, DiscoveredPeer, LeanFilter, rank_candidates}; +use super::node_id::node_id_from_peer_id; use super::{ DIAL_INTERVAL_AT_TARGET, DIAL_INTERVAL_AT_ZERO_PEERS, DISCOVERY_CANDIDATE_BATCH, - DiscoveryHandle, + SpawnedDiscovery, }; use crate::swarm_adapter::DialOutcome; use crate::{ConnectionDirection, P2PServer, metrics}; /// Everything the dial loop needs from a running discovery server. pub(crate) struct DiscoveryState { - /// Contacts [`spawn_contact_poll`]'s task has drawn from the peer table and - /// passed through admission, waiting to be ranked into `candidates`. + /// The running discovery server, kept for the connection reports; see + /// [`report_peer_connected`]. + handle: DiscoveryHandle, + /// Nodes [`spawn_contact_poll`]'s task has drawn from discovery and found a + /// dial target for, waiting to be ranked into `candidates`. /// - /// A receiver rather than the `PeerTable` itself, because drawing a contact - /// is an actor round trip and this side of it must never await one: the dial - /// loop holds `&mut P2PServer` for the length of a tick. + /// A receiver rather than the handle itself, because drawing a node is an + /// actor round trip and this side of it must never await one: the dial loop + /// 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. @@ -40,7 +47,7 @@ pub(crate) struct DiscoveryState { local_peer_id: PeerId, /// Connected-peer count above which the loop stops dialing. target_peers: usize, - /// A handle on the admission policy the peer table also runs, so a fork + /// A handle on the admission policy discovery also runs, so a fork /// boundary can move both with one call. See [`LeanFilter::set_fork_id`]. filter: LeanFilter, } @@ -48,18 +55,19 @@ 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. - pub(crate) fn new(handle: DiscoveryHandle, local_peer_id: PeerId) -> Self { + pub(crate) fn new(spawned: SpawnedDiscovery, local_peer_id: PeerId) -> Self { Self { - contacts: spawn_contact_poll(handle.peer_table, handle.filter.clone()), + contacts: spawn_contact_poll(spawned.handle.clone(), spawned.filter.dial_targets()), + handle: spawned.handle, candidates: VecDeque::new(), peer_attnets: HashMap::new(), local_peer_id, - target_peers: handle.target_peers, - filter: handle.filter, + target_peers: spawned.target_peers, + filter: spawned.filter, } } - /// The admission policy shared with the peer table. + /// The admission policy shared with discovery. pub(crate) fn filter(&self) -> &LeanFilter { &self.filter } @@ -70,7 +78,37 @@ impl DiscoveryState { } } -/// How long the poll task waits after the peer table offers nothing. +/// 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); +} + +/// How long the poll task waits after discovery offers nothing. /// /// Not an end: ethrex clears its tried set once a full scan finds nothing /// eligible, and discv5 keeps crawling underneath, so contacts reappear on @@ -245,7 +283,7 @@ fn dial_budget_from( /// shortfall permanently open. Pacing on the shortfall alone would then hold /// the loop at the floor forever, emptying the contact buffer tens of times a /// second for peers it is already connected to and keeping -/// [`spawn_contact_poll`] drawing the peer table down to refill it. +/// [`spawn_contact_poll`] drawing discovery's eligible set down to refill it. /// /// `target_peers` is the running loop's own [`DiscoveryState::target_peers`], /// read by the caller that already had to check discovery is on. @@ -364,20 +402,27 @@ fn refill_candidates(server: &mut P2PServer) { } } -/// Draw dialable peers from the peer table, off the p2p actor's thread. +/// Draw dialable peers from discovery, off the p2p actor's thread. /// -/// `get_contact_to_initiate` is an actor round trip, and the dial loop used to +/// `next_dial_candidate` is an actor round trip, and the dial loop used to /// await up to [`DISCOVERY_CANDIDATE_BATCH`] of them inline, holding /// `&mut P2PServer` across every one. That parked the entire actor — gossip -/// forwarding, req/resp, swarm events, every tick — behind the peer table for -/// the length of a refill. The round trips happen in this task now, and what +/// forwarding, req/resp, swarm events, every tick — behind discovery for the +/// length of a refill. The round trips happen in this task now, and what /// reaches the loop is a channel of already-admitted [`DiscoveredPeer`]s it can /// drain without awaiting anything. /// -/// 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, and everything that arrives -/// here has already passed admission. +/// 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, and +/// everything that arrives here has already passed admission. +/// +/// What arrives is a `Node`, which knows an address and a key and nothing about +/// `quic`, `tcp` 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. /// /// That "marked tried on the way out" is why the channel is bounded and why the /// task reserves its slot before asking. A contact drawn with nowhere to put it @@ -388,25 +433,24 @@ fn refill_candidates(server: &mut P2PServer) { /// /// The task ends when the receiver does, which is when the `P2PServer` holding /// [`DiscoveryState`] is dropped. -fn spawn_contact_poll(peer_table: PeerTable, filter: LeanFilter) -> mpsc::Receiver { +fn spawn_contact_poll( + handle: DiscoveryHandle, + dial_targets: DialTargets, +) -> mpsc::Receiver { let (contacts, receiver) = mpsc::channel(DISCOVERY_CANDIDATE_BATCH); tokio::spawn(async move { while let Ok(permit) = contacts.reserve().await { - let Ok(Some(contact)) = peer_table.get_contact_to_initiate().await else { + let Some(node) = handle.next_dial_candidate().await else { drop(permit); tokio::time::sleep(CONTACT_POLL_IDLE).await; continue; }; - // 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. The permit is released with it, so the slot goes to - // the next contact rather than to this one's absence. - if let Some(peer) = contact - .record - .as_ref() - .and_then(|record| filter.dial_target(record)) - { + // A node with no dial target filed is one whose ENR we never read, + // so it carries no address or peer id to dial. Skipping it costs + // nothing: it was marked tried on the way out either way. The + // permit is released with it, so the slot goes to the next node + // rather than to this one's absence. + if let Some(peer) = dial_targets.get(&node.node_id()) { permit.send(peer); } } diff --git a/crates/net/p2p/src/discovery/mod.rs b/crates/net/p2p/src/discovery/mod.rs index 4b161f68b..63ba86aa3 100644 --- a/crates/net/p2p/src/discovery/mod.rs +++ b/crates/net/p2p/src/discovery/mod.rs @@ -1,27 +1,33 @@ //! 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. Both dial paths, discovered peers and static bootnodes alike, hand +//! 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. Both dial paths, discovered peers and static bootnodes alike, hand //! the swarm every address a peer advertises (`quic` and `tcp`) in one attempt //! 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. +//! //! 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; @@ -63,24 +69,11 @@ pub const DIAL_INTERVAL_AT_TARGET: Duration = /// 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; - /// Admitted contacts [`dial::spawn_contact_poll`] keeps buffered ahead of the /// dial loop, and so the most one refill can draw. /// -/// A bound rather than a batch size, because the peer table marks every contact -/// tried on the way out: this is how far ahead of the dialing the table may be +/// A bound rather than a batch size, because discovery marks every node tried +/// on the way out: this is how far ahead of the dialing its eligible set may be /// drawn down. See [`dial::spawn_contact_poll`]. pub const DISCOVERY_CANDIDATE_BATCH: usize = 8; @@ -124,9 +117,9 @@ 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, /// The `eth2` entry to publish and to compare discovered peers against. pub fork_id: EnrForkId, @@ -135,20 +128,24 @@ pub struct DiscoverySpawnConfig { } /// 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, + /// A clone of the admission policy discovery runs. It shares discovery's + /// [`DialTargets`](admission::DialTargets), which is how the dial loop knows + /// how to dial what was admitted, and its policy, which is how a fork + /// boundary moves what discovery admits. See [`LeanFilter`]. + 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 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, } @@ -165,7 +162,7 @@ pub struct DiscoveryHandle { /// neither. pub async fn spawn_discovery( config: DiscoverySpawnConfig, -) -> Result { +) -> Result { let signer = secp256k1::SecretKey::from_slice(&config.node_key).map_err(DiscoveryError::NodeKey)?; @@ -198,20 +195,17 @@ pub async fn spawn_discovery( .trim_end_matches('=') .to_string(); - // `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 we do not judge + // peers by, 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 the dial loop keeps a clone. The + // two share one policy and one dial-target map, so what discovery's copy + // files is what the dial loop reads, and a fork boundary set through the + // dial loop's copy moves what discovery's admits. let filter = LeanFilter::new(config.fork_id, config.attestation_committee_count); - let peer_table = PeerTableServer::spawn_with_filter( - local_node.node_id(), - PEER_TABLE_TARGET_PEERS, - filter.clone(), - ); let seeds: Vec = config .bootnodes @@ -231,21 +225,57 @@ pub async fn spawn_discovery( // 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. Neither chain here keys its peers on an + // `eth` entry (both carry their digest in `eth2`), 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.clone()), 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!( @@ -256,10 +286,10 @@ pub async fn spawn_discovery( ); } - Ok(DiscoveryHandle { - peer_table, - local_enr, + Ok(SpawnedDiscovery { + handle: discovery, filter, + local_enr, target_peers: config.target_peers, }) } @@ -267,14 +297,15 @@ pub async fn spawn_discovery( /// 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`] takes the local record by value and moves +/// it into the actor it starts. The handle it returns speaks +/// `DiscoveryServerProtocol` (`raw_packet`, `record_peer_event`, `enr_lookup`, +/// `prune`, `next_dial_candidate`, `shutdown`), none of which replaces the +/// record. The record is re-signed only internally: by the `fork_id` watch +/// `spawn` takes, which stamps an EIP-2124 `eth` entry rather than `eth2`, and by +/// discv5 IP voting (`apply_predicted_ip`), which edits the address. What would +/// close the gap is a protocol message on `DiscoveryServer` that edits the local +/// record's `eth2` entry and bumps its `seq`. /// /// Until then the record served to discv5 queries keeps the `eth2` entry it /// started with. Peers already connected are unaffected (they handshake on @@ -347,7 +378,7 @@ 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)); 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..6d7eaf74e --- /dev/null +++ b/crates/net/p2p/src/discovery/node_id.rs @@ -0,0 +1,77 @@ +//! 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; + +/// 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. +/// +/// The decode itself is [`super::enr::node_id_from_peer_id`]'s, which custody +/// reads peers' node ids through; this only names its answer the way discovery +/// does. +pub(crate) fn node_id_from_peer_id(peer_id: &PeerId) -> Option { + super::enr::node_id_from_peer_id(peer_id).map(H256) +} + +#[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 cc9ca2bd1..6a29273bb 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -92,7 +92,10 @@ use tracing::{debug, info, trace, warn}; use crate::{ discovery::{ DIAL_INTERVAL_AT_TARGET, DIAL_INTERVAL_AT_ZERO_PEERS, DiscoveryError, DiscoverySpawnConfig, - dial::{DiscoveryState, dial_interval, dial_progress, dial_tick, forget_discovered_peer}, + dial::{ + DiscoveryState, dial_interval, dial_progress, dial_tick, forget_discovered_peer, + report_peer_connected, report_peer_disconnected, + }, enr::{dialable_port, read_ip, read_public_key, read_quic_port, read_tcp_port}, spawn_discovery, }, @@ -1972,6 +1975,7 @@ async fn handle_swarm_event( if num_established.get() == 1 { server.connected_peers.insert(peer_id, direction); server.refresh_peer_metrics(); + report_peer_connected(server, &peer_id); let peer_count = server.connected_peers.len(); metrics::notify_peer_connected( server.resolve_node_name(Some(&peer_id)), @@ -2071,6 +2075,7 @@ async fn handle_swarm_event( .connected_peers .remove(&peer_id) .unwrap_or(closed_direction); + report_peer_disconnected(server, &peer_id); forget_discovered_peer(server, &peer_id); server.refresh_peer_metrics(); let peer_count = server.connected_peers.len(); diff --git a/docs/discovery.md b/docs/discovery.md index 3040ea64a..e2545aee1 100644 --- a/docs/discovery.md +++ b/docs/discovery.md @@ -42,9 +42,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 @@ -108,11 +110,19 @@ libp2p races them within one attempt and a `quic` entry that does not answer falls back to TCP rather than ending the dial. Most mainnet beacon nodes are in exactly that state, which is why a QUIC-only dialer peered so poorly with them. -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 transport 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 transport 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`, `tcp` or `attnets`; +the dial loop looks the node id back up in what the filter filed to recover the +multiaddrs 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. A peer's dial list carries every address it advertises, `quic` and `tcp` both, in one dial attempt, `quic` first. libp2p walks that list `dial_concurrency_factor` @@ -133,9 +143,10 @@ 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. 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. On `beacon` the flag is more than the dial loop's cutoff: the swarm's connection limits are derived from it too, so what this node refuses and what it goes @@ -251,21 +262,11 @@ 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). -### 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`