libp2p: explode!

1 files changed, 524 insertions, 0 deletions

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<PeerId, Vec<Multiaddr>>,
+    pub(super) listening: Vec<Multiaddr>,
+    /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`).
+    pub(super) mapped: Vec<Multiaddr>,
+    /// `Libp2pConfig::announce`.
+    pub(super) announced: Vec<Multiaddr>,
+    /// `Libp2pConfig::ignore`: not published (but announced), not dialled.
+    pub(super) ignored: Vec<IpRange>,
+    /// `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<PeerId, NamedRelay>,
+    /// Relays named without their peer ID, being found out (`Command::FindRelay`).
+    pub(super) finding: HashMap<Multiaddr, Named>,
+    /// The relays listed in the document, as last learned (`learn_listed`).
+    pub(super) listed: Vec<Multiaddr>,
+    /// Members that say they relay, and where (with the relay's peer ID).
+    pub(super) relays: HashMap<PeerId, Multiaddr>,
+    /// With `Libp2pConfig::public`: other peers that relay (identify says so), and
+    /// where.
+    pub(super) public_relays: HashMap<PeerId, Multiaddr>,
+    /// Relays not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`).
+    pub(super) failed_relays: HashMap<PeerId, tokio::time::Instant>,
+    /// Relays that granted us a slot we hold.
+    pub(super) slots: std::collections::HashSet<PeerId>,
+    /// Peers holding a slot on this node, as a relay.
+    pub(super) slots_here: std::collections::HashSet<PeerId>,
+    /// 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<PeerId, (tokio::time::Instant, Duration)>,
+    /// The shortest round trip measured on each open connection (libp2p's ping).
+    pub(super) rtts: HashMap<ConnectionId, Duration>,
+    /// Peers whose hole punching failed: how often in a row, and when to try again
+    /// (`PUNCH_RETRIES`).
+    pub(super) punching: HashMap<PeerId, (usize, tokio::time::Instant)>,
+    /// Open connections: to whom, whether through a relay, and the remote address.
+    pub(super) connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>,
+    /// What AutoNAT says, and the address it confirmed while public.
+    pub(super) reachability: Reachability,
+    pub(super) public_address: Option<Multiaddr>,
+    /// Since start (`stats`).
+    pub(super) counts: Counts,
+    /// When each peer was last mentioned (`STALE_AFTER`).
+    pub(super) seen: HashMap<PeerId, tokio::time::Instant>,
+    /// Peers found on the LAN (mDNS) not known yet to speak continuum.
+    pub(super) lan: std::collections::HashSet<PeerId>,
+    /// Peers that speak continuum (identify says so).
+    pub(super) continuum: std::collections::HashSet<PeerId>,
+    /// 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<PeerId, bool>,
+    /// Peers found on the LAN that speak continuum, with their address: seeds not
+    /// handed out yet (`discovered_seeds`).
+    pub(super) discovered: Vec<Multiaddr>,
+}
+
+/// 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<Connection> {
+        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<ConnectionId> {
+        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<ConnectionId> {
+        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<PeerId> = 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::<std::collections::HashSet<_>>()
+            .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<Multiaddr> {
+        if !self.announced.is_empty() {
+            return self.announced.clone();
+        }
+        let direct: Vec<Multiaddr> = 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<Multiaddr> = 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<Multiaddr> {
+    let reachable: Vec<Multiaddr> = 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<Multiaddr> {
+        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<Multiaddr> {
+        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<Multiaddr>,
+        local: PeerId,
+    ) -> Option<Vec<Multiaddr>> {
+        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<PeerId, Multiaddr> {
+        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/<peer ID>`), 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<Multiaddr> {
+        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<PeerId> {
+    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<ConnectionId, (PeerId, bool, Multiaddr)>,
+    peer: &PeerId,
+) -> Option<Connection> {
+    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<Multiaddr> {
+    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<PeerId> {
+    match address.iter().last() {
+        Some(Protocol::P2p(peer)) => Some(peer),
+        _ => None,
+    }
+}
+
+pub(super) fn lock(book: &Mutex<Book>) -> MutexGuard<'_, Book> {
+    book.lock_or_recover()
+}