diff --git a/CLAUDE.md b/CLAUDE.md index 05cf5048..957a5e68 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 dacda8d5..f2020929 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 712ba60f..3a032d24 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 51524ec6..61f0ca79 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 641c602f..60446ad0 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 ecaaf63d..3b57ac91 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 00000000..41e98be2 --- /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 685032f9..97ea702c 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 8513fe51..76dc05b8 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`