From 228eb88d89efb1edf875abbc3c8a8449a916061c Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 04:20:41 +0100 Subject: libp2p: explode! --- src/transport/libp2p/book.rs | 524 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 524 insertions(+) create mode 100644 src/transport/libp2p/book.rs (limited to 'src/transport/libp2p/book.rs') diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs new file mode 100644 index 0000000..5658c56 --- /dev/null +++ b/src/transport/libp2p/book.rs @@ -0,0 +1,524 @@ +//! The book: who is where (peers' addresses, connections and their round trips), the +//! relays and their slots, hole punching and upgrades, and the counts the figures show. + +use super::*; + +/// Who named a relay to pick, in the order they are picked. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Named { + /// `Libp2pConfig::relays`. + Configured, + /// The relays listed in the document (`Transport::learn_relays`). + Listed, +} + +impl Named { + pub(super) fn as_str(self) -> &'static str { + match self { + Self::Configured => "configured", + Self::Listed => "listed", + } + } +} + +/// A relay named to pick. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct NamedRelay { + /// Where, with the relay's peer ID. + pub(super) at: Multiaddr, + /// As named: without the peer ID, it is found out again when the relay doesn't + /// grant a slot (it may have a new one). + pub(super) given: Multiaddr, + pub(super) named: Named, +} + +/// Where peers are, and where we listen. +#[derive(Default)] +pub(super) struct Book { + pub(super) addresses: HashMap>, + pub(super) listening: Vec, + /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`). + pub(super) mapped: Vec, + /// `Libp2pConfig::announce`. + pub(super) announced: Vec, + /// `Libp2pConfig::ignore`: not published (but announced), not dialled. + pub(super) ignored: Vec, + /// `Libp2pConfig::relay`: this node relays, and says so. + pub(super) serving: bool, + /// Relays named to pick (configured, listed), by peer: picked first. + pub(super) named_relays: HashMap, + /// Relays named without their peer ID, being found out (`Command::FindRelay`). + pub(super) finding: HashMap, + /// The relays listed in the document, as last learned (`learn_listed`). + pub(super) listed: Vec, + /// Members that say they relay, and where (with the relay's peer ID). + pub(super) relays: HashMap, + /// With `Libp2pConfig::public`: other peers that relay (identify says so), and + /// where. + pub(super) public_relays: HashMap, + /// Relays not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`). + pub(super) failed_relays: HashMap, + /// Relays that granted us a slot we hold. + pub(super) slots: std::collections::HashSet, + /// Peers holding a slot on this node, as a relay. + pub(super) slots_here: std::collections::HashSet, + /// Peers connected only through a relay, dialled directly (`upgrade`): when to again + /// at the soonest, and how long to wait after that. + pub(super) upgrades: HashMap, + /// The shortest round trip measured on each open connection (libp2p's ping). + pub(super) rtts: HashMap, + /// Peers whose hole punching failed: how often in a row, and when to try again + /// (`PUNCH_RETRIES`). + pub(super) punching: HashMap, + /// Open connections: to whom, whether through a relay, and the remote address. + pub(super) connections: HashMap, + /// What AutoNAT says, and the address it confirmed while public. + pub(super) reachability: Reachability, + pub(super) public_address: Option, + /// Since start (`stats`). + pub(super) counts: Counts, + /// When each peer was last mentioned (`STALE_AFTER`). + pub(super) seen: HashMap, + /// Peers found on the LAN (mDNS) not known yet to speak continuum. + pub(super) lan: std::collections::HashSet, + /// Peers that speak continuum (identify says so). + pub(super) continuum: std::collections::HashSet, + /// Peers identify says don't speak continuum, and whether they relay: not sent to + /// (`send_encoded`), so that a seed that's no node says so. + pub(super) not_nodes: HashMap, + /// Peers found on the LAN that speak continuum, with their address: seeds not + /// handed out yet (`discovered_seeds`). + pub(super) discovered: Vec, +} + +/// What `Libp2pStats` counts since start. +#[derive(Default)] +pub(super) struct Counts { + pub(super) slots_granted: u64, + pub(super) slots_lost: u64, + pub(super) slots_not_granted: u64, + pub(super) hole_punches: HolePunches, + pub(super) dial_failures: DialFailures, +} + +impl Book { + /// How we are connected to `peer`: directly if any connection is direct. + pub(super) fn connection(&self, peer: &PeerId) -> Option { + book_connection(&self.connections, peer) + } + + /// The connections to `peer` to close, of several: all but one, the oldest direct + /// one (the oldest relayed one if none is direct). + /// + /// None between a relay and a node with a slot on it, or reserving one (`reserving`): + /// a slot lives on one connection, which libp2p doesn't tell, and closing it would + /// end the slot, and with it every circuit to the node (the relay then answers + /// `NoReservation`). + pub(super) fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec { + if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) { + return Vec::new(); + } + self.ranked(peer).into_iter().skip(1).collect() + } + + /// The connections to `peer`, the one to keep first: direct ones before relayed + /// ones, measured ones by their round trip (`rtts`: over the LAN before a VPN's hub, + /// whichever opened first), then the oldest. + pub(super) fn ranked(&self, peer: &PeerId) -> Vec { + let mut connections: Vec<(bool, Duration, ConnectionId)> = self + .connections + .iter() + .filter(|(_, (connected, _, _))| connected == peer) + .map(|(id, (_, relayed, _))| { + let rtt = self.rtts.get(id).copied().unwrap_or(Duration::MAX); + (*relayed, rtt, *id) + }) + .collect(); + connections.sort(); + connections.into_iter().map(|(_, _, id)| id).collect() + } + + /// The remote address of the connection to `peer` that is kept (`ranked`). + pub(super) fn location(&self, peer: &PeerId) -> Option<&Multiaddr> { + let kept = *self.ranked(peer).first()?; + self.connections.get(&kept).map(|(_, _, address)| address) + } + + pub(super) fn stats(&self) -> Libp2pStats { + let mut connections = Connections::default(); + for (_, relayed, _) in self.connections.values() { + if *relayed { + connections.relayed += 1; + } else { + connections.direct += 1; + } + } + Libp2pStats { + reachability: self.reachability, + public_address: self.public_address.as_ref().map(Multiaddr::to_string), + connections, + relay_slots: RelaySlots { + held: self.slots.len(), + paused: self + .failed_relays + .values() + .filter(|until| tokio::time::Instant::now() < **until) + .count(), + granted: self.counts.slots_granted, + lost: self.counts.slots_lost, + not_granted: self.counts.slots_not_granted, + }, + hole_punches: self.counts.hole_punches, + dial_failures: self.counts.dial_failures, + } + } + + /// `peer` was mentioned just now. + pub(super) fn touch(&mut self, peer: PeerId) { + self.seen.insert(peer, tokio::time::Instant::now()); + } + + /// Forgets the peers not mentioned within `STALE_AFTER` before `now`, but those + /// `connected`. Returns how many. + pub(super) fn prune( + &mut self, + now: tokio::time::Instant, + connected: impl Fn(&PeerId) -> bool, + ) -> usize { + let stale: Vec = self + .addresses + .keys() + .chain(self.relays.keys()) + .chain(self.public_relays.keys()) + .chain(self.continuum.iter()) + .chain(self.lan.iter()) + .filter(|peer| { + !connected(peer) + && self + .seen + .get(peer) + .is_none_or(|at| now.saturating_duration_since(*at) >= STALE_AFTER) + }) + .copied() + .collect::>() + .into_iter() + .collect(); + for peer in &stale { + self.addresses.remove(peer); + self.relays.remove(peer); + self.public_relays.remove(peer); + self.continuum.remove(peer); + self.not_nodes.remove(peer); + self.lan.remove(peer); + self.failed_relays.remove(peer); + self.upgrades.remove(peer); + self.punching.remove(peer); + self.seen.remove(peer); + } + stale.len() + } + + /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first. + pub(super) fn discover(&mut self, peer: PeerId) { + let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else { + return; + }; + if let Ok(address) = address.clone().with_p2p(peer) + && !self.discovered.contains(&address) + && self.discovered.len() < MAX_PEERS + { + self.discovered.push(address); + } + } + + /// Where others could dial us directly: the announced addresses, or else those we + /// listen on (`published`), where the router forwards to them (`mapped`), and where + /// AutoNAT found others can dial us (`public_address`: a port forwarded by hand, say), + /// without the peer ID. + pub(super) fn direct(&self) -> Vec { + if !self.announced.is_empty() { + return self.announced.clone(); + } + let direct: Vec = self + .listening + .iter() + .filter(|address| !is_relayed(address)) + .cloned() + .collect(); + let mut direct = published(&direct); + let confirmed = self.public_address.iter().filter_map(|address| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + (!address.is_empty() && !is_relayed(&address)).then_some(address) + }); + // First: the member list keeps only so many (`members::MAX_ADDRESSES`), and a + // node listening everywhere has many. + let mut first: Vec = Vec::new(); + for address in self.mapped.iter().cloned().chain(confirmed) { + if !first.contains(&address) { + first.push(address); + } + } + direct.retain(|address| !first.contains(address)); + first.extend(direct); + first.retain(|address| !ignored(&self.ignored, address)); + first + } +} + +/// Whether `address` only reaches this host (or this link): no use to other hosts, and +/// dialled by one, it reaches that host itself. +pub(super) fn is_local(address: &Multiaddr) -> bool { + match address.iter().next() { + Some(Protocol::Ip4(ip)) => ip.is_loopback() || ip.is_link_local() || ip.is_unspecified(), + Some(Protocol::Ip6(ip)) => { + ip.is_loopback() || ip.is_unicast_link_local() || ip.is_unspecified() + } + _ => false, + } +} + +/// The addresses of `all` worth publishing: all but local ones, unless they're all +/// local (a test on one host). +pub(super) fn published(all: &[Multiaddr]) -> Vec { + let reachable: Vec = all + .iter() + .filter(|address| !is_local(address)) + .cloned() + .collect(); + if reachable.is_empty() { + all.to_vec() + } else { + reachable + } +} + +impl Book { + /// `named` names the relay at `given`; returns it if its peer is to be found out. + /// `local`: this node, never its own relay. + pub(super) fn name_relay( + &mut self, + given: Multiaddr, + named: Named, + local: PeerId, + ) -> Option { + match split_peer(&given) { + Some(peer) if peer == local => None, + Some(peer) => { + // Configured wins over listed. + if self + .named_relays + .get(&peer) + .is_none_or(|known| known.named == Named::Listed || named == Named::Configured) + { + self.named_relays.insert( + peer, + NamedRelay { + at: given.clone(), + given, + named, + }, + ); + } + None + } + None => { + let known = self.finding.contains_key(&given) + || self.named_relays.values().any(|relay| relay.given == given); + if known { + return None; + } + self.finding.insert(given.clone(), named); + Some(given) + } + } + } + + /// The relay named at `given` (without a peer ID) is `peer`: unless it is no longer + /// named (or is this node). Returns whether that's news. + pub(super) fn relay_found(&mut self, given: Multiaddr, peer: PeerId, local: PeerId) -> bool { + let Some(named) = self.finding.remove(&given) else { + return false; + }; + let Ok(at) = given.clone().with_p2p(peer) else { + return false; + }; + if peer == local + || (named == Named::Listed + && self + .named_relays + .get(&peer) + .is_some_and(|known| known.named == Named::Configured)) + { + return false; + } + self.named_relays + .insert(peer, NamedRelay { at, given, named }); + true + } + + /// `relay` didn't grant a slot: if it was named without its peer ID, its address is + /// to be found out again, returned. + pub(super) fn find_again(&mut self, relay: PeerId) -> Option { + if self + .named_relays + .get(&relay) + .is_none_or(|known| split_peer(&known.given).is_some()) + { + return None; + } + let known = self.named_relays.remove(&relay)?; + self.finding.insert(known.given.clone(), known.named); + Some(known.given) + } + + /// The relays listed in the document are `listed` now: returns those to find out, + /// None if nothing changed. + pub(super) fn learn_listed( + &mut self, + listed: Vec, + local: PeerId, + ) -> Option> { + if listed == self.listed { + return None; + } + self.named_relays + .retain(|_, relay| relay.named != Named::Listed || listed.contains(&relay.given)); + self.finding + .retain(|given, named| *named != Named::Listed || listed.contains(given)); + self.listed = listed.clone(); + Some( + listed + .into_iter() + .filter_map(|given| self.name_relay(given, Named::Listed, local)) + .collect(), + ) + } + + /// The named relays `named` names, by peer, where. + pub(super) fn named(&self, named: Named) -> HashMap { + self.named_relays + .iter() + .filter(|(_, relay)| relay.named == named) + .map(|(peer, relay)| (*peer, relay.at.clone())) + .collect() + } + + /// Where `peer` may be reached through the relays we know (named, then members that + /// relay): for a peer known by its ID only (a seed `/p2p/`), which may hold + /// a slot on one of them. `local`: this node, never its own relay. + pub(super) fn through_relays(&self, peer: PeerId, local: PeerId) -> Vec { + let mut relays: Vec<(PeerId, Multiaddr)> = self + .named_relays + .iter() + .map(|(relay, named)| (*relay, named.at.clone())) + .collect(); + relays.sort_by_key(|(_, address)| address.to_string()); + let mut members: Vec<(PeerId, Multiaddr)> = self + .relays + .iter() + .filter(|(relay, _)| !self.named_relays.contains_key(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect(); + members.sort_by_key(|(_, address)| address.to_string()); + relays + .into_iter() + .chain(members) + .filter(|(relay, _)| *relay != local && *relay != peer) + .map(|(_, address)| address.with(Protocol::P2pCircuit).with(Protocol::P2p(peer))) + .take(ADDRESSES_PER_PEER) + .collect() + } + + /// Whether all of `addresses` are known of `peer` (`remember`). + pub(super) fn knows(&self, peer: PeerId, addresses: &[Multiaddr]) -> bool { + let Some(known) = self.addresses.get(&peer) else { + return false; + }; + addresses.iter().all(|address| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + known.contains(&address) + }) + } + + /// `peer` is at `address` (without its `/p2p` part): kept first. + pub(super) fn remember(&mut self, peer: PeerId, mut address: Multiaddr) { + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + if !self.addresses.contains_key(&peer) && self.addresses.len() >= MAX_PEERS { + return; + } + self.touch(peer); + let addresses = self.addresses.entry(peer).or_default(); + addresses.retain(|known| *known != address); + addresses.insert(0, address); + addresses.truncate(ADDRESSES_PER_PEER); + } +} + +/// Whether `address` goes through a relay. +pub(super) fn is_relayed(address: &Multiaddr) -> bool { + address + .iter() + .any(|protocol| matches!(protocol, Protocol::P2pCircuit)) +} + +/// The relay `address` goes through, if it goes through one: the peer before +/// `/p2p-circuit`. +pub(super) fn relay_of(address: &Multiaddr) -> Option { + let mut relay = None; + for protocol in address.iter() { + match protocol { + Protocol::P2p(peer) => relay = Some(peer), + Protocol::P2pCircuit => return relay, + _ => {} + } + } + None +} + +/// How `peer` is connected, of `connections` (`Book::connections`): directly if one is +/// direct. +pub(super) fn book_connection( + connections: &HashMap, + peer: &PeerId, +) -> Option { + let mut found = None; + for (connected, relayed, _) in connections.values() { + if connected == peer { + if !relayed { + return Some(Connection::Direct); + } + found = Some(Connection::Relayed); + } + } + found +} + +/// The addresses of `known` this node can dial: not those through itself as the relay +/// (a relay reaches the nodes it relays for only while they are connected to it). +pub(super) fn dialable(known: &[Multiaddr], local: PeerId) -> Vec { + known + .iter() + .filter(|address| relay_of(address) != Some(local)) + .cloned() + .collect() +} + +/// `address` with the peer ID it ends in, if it ends in one. +pub(super) fn split_peer(address: &Multiaddr) -> Option { + match address.iter().last() { + Some(Protocol::P2p(peer)) => Some(peer), + _ => None, + } +} + +pub(super) fn lock(book: &Mutex) -> MutexGuard<'_, Book> { + book.lock_or_recover() +} -- cgit v1.3.1