path: root/src/transport/paths.rs

//! Paths: one address space over any number of transports, below the identity layer
//! (`transport::identity`). An address is `<scheme>:<target>`: the transport by its
//! scheme (`mycelium`, `libp2p`, …), and on it a transport identity (Mycelium's public
//! key, libp2p's peer ID) or a location (`Transport::locate`), e.g. a multiaddr.
//! Addresses as paths keeps them are `<scheme>:<transport identity>` (`canonical`).
//!
//! Every transport carries the same messages (`wire`); paths routes each send to the
//! transport of its address, hands every transport's messages to one handler under the
//! sender's address, and puts together what the transports say about themselves
//! (addresses, seeds they found, figures). The transports are in order of preference,
//! as configured (`--transports`).
//!
//! A message to a node with several addresses goes to the first of them that takes it
//! (`send_first`), in the order `rank` gives. A send that fails is tried at once at the
//! next address, the message encoded for its transport: when one transport fails, the
//! next carries on within the same send.
//!
//! Paths measures the addresses of such nodes itself (and every path the node's
//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of
//! within `ACTIVE_FOR` gets a PING with a random nonce, which the
//! other side's paths answers at once with a PONG, over the same path. Answers give each
//! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a
//! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up
//! first, the one in use first among them, and the rest by round-trip time; it changes
//! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least
//! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the
//! configured order decides.
//!
//! PINGs and PONGs then go up as other messages do: they tell the node the peer is
//! alive. The node's keepalive is a PING too (`swarm::peer::keepalive`), which paths
//! sends like any message, and notes as the probe of the path it went over.
//!
//! The debug firewall (`transport::firewall`) works here, below everything else, through
//! a `Gate`: messages from an address it blocks are dropped as they arrive, and sends to
//! one are dropped silently, as by a lossy network.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;

use anyhow::{Context, Result, anyhow};
use tokio::time::Instant;
use tracing::{debug, info};

use crate::locks::MutexExt;
use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats};
use crate::transport::traffic::Traffic;
use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};

/// Whether to drop what comes from or goes to an address (the debug firewall,
/// `transport::firewall`; the identity layer knows which node is where).
pub(crate) trait Gate: Send + Sync {
    fn blocks_inbound(&self, address: &str) -> bool;
    fn blocks_outbound(&self, address: &str) -> bool;
}

/// Called with the sender's address, the message and its `payload_len`.
pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>
    + Send
    + Sync
    + 'static;

/// One of the node's transports.
struct Link {
    scheme: &'static str,
    inner: Arc<dyn Transport>,
}

impl Link {
    fn address(&self, target: &str) -> String {
        format!("{}:{target}", self.scheme)
    }

    fn own_address(&self) -> String {
        self.address(self.inner.local_id())
    }
}

/// The scheme of `address`.
pub(crate) fn scheme_of(address: &str) -> &str {
    address.split_once(':').map_or("", |(scheme, _)| scheme)
}

/// A path whose send failed is down for this long (unless it answers a PING, or a
/// send there works again).
const DOWN_FOR: Duration = Duration::from_secs(60);
/// How often a path in use is pinged, how long an answer may take, and how many PINGs
/// in a row unanswered make it down.
const PROBE_EVERY: Duration = Duration::from_secs(30);
const PROBE_TIMEOUT: Duration = Duration::from_secs(15);
const UNANSWERED_DOWN: u32 = 3;
/// How often the probe task looks for paths due.
const PROBE_TICK: Duration = Duration::from_secs(5);
/// Paths this node had a choice of within this long are probed; ones it didn't use for
/// twice as long are forgotten.
const ACTIVE_FOR: Duration = Duration::from_secs(300);
/// The path in use is changed for one this much faster (a fraction of its round-trip
/// time), and at least `SWITCH_MIN`.
const SWITCH_GAIN: f64 = 0.25;
const SWITCH_MIN: Duration = Duration::from_millis(5);
/// Paths kept at most: they are of peers this node chose among, but bounded anyway.
const MAX_PATHS: usize = 4096;

/// What paths knows about one path (an address as paths keep it).
#[derive(Debug, Default)]
struct PathState {
    /// When it was last among several to choose from (`rank`).
    used: Option<Instant>,
    /// Smoothed round-trip time of its PINGs.
    rtt: Option<Duration>,
    /// The PING out, by nonce, and when it went.
    probe: Option<(u64, Instant)>,
    last_probe: Option<Instant>,
    /// PINGs in a row unanswered.
    unanswered: u32,
    /// When a send there last failed.
    failed_at: Option<Instant>,
}

impl PathState {
    fn is_down(&self, now: Instant) -> bool {
        self.unanswered >= UNANSWERED_DOWN
            || self
                .failed_at
                .is_some_and(|at| now.saturating_duration_since(at) < DOWN_FOR)
    }
}

/// Every path's state, and which one is in use for each set to choose from.
#[derive(Debug, Default)]
struct Table {
    paths: HashMap<String, PathState>,
    /// By the set's addresses (sorted, joined): the one in use.
    chosen: HashMap<String, String>,
    counts: PathCounts,
}

impl Table {
    /// `usable`, with each one's place in the configured order, best first (as
    /// `Paths::rank` says), noting the choice.
    fn rank(&mut self, mut usable: Vec<(String, usize)>, now: Instant) -> Vec<String> {
        usable.sort_by_key(|(_, preference)| *preference);
        usable.dedup_by(|a, b| a.0 == b.0);
        if usable.len() < 2 {
            return usable.into_iter().map(|(address, _)| address).collect();
        }
        if self.paths.len() >= MAX_PATHS {
            self.forget_unused(now);
        }
        for (address, _) in &usable {
            if self.paths.len() < MAX_PATHS || self.paths.contains_key(address) {
                self.paths.entry(address.clone()).or_default().used = Some(now);
            }
        }
        let rtt = |table: &Self, address: &str| table.paths.get(address).and_then(|path| path.rtt);
        let (up, down): (Vec<_>, Vec<_>) = usable.into_iter().partition(|(address, _)| {
            !self
                .paths
                .get(address)
                .is_some_and(|path| path.is_down(now))
        });
        let mut set: Vec<&str> = up
            .iter()
            .chain(down.iter())
            .map(|(address, _)| address.as_str())
            .collect();
        set.sort_unstable();
        let key = set.join(" ");
        let mut ranked: Vec<String> = Vec::new();
        if !up.is_empty() {
            let mut current = self
                .chosen
                .get(&key)
                .filter(|chosen| up.iter().any(|(address, _)| address == *chosen))
                .cloned()
                .unwrap_or_else(|| up[0].0.clone());
            let fastest = up
                .iter()
                .filter_map(|(address, _)| Some((address, rtt(self, address)?)))
                .min_by_key(|(_, rtt)| *rtt);
            if let (Some((fastest, fastest_rtt)), Some(current_rtt)) =
                (fastest, rtt(self, &current))
                && *fastest != current
                && current_rtt.saturating_sub(fastest_rtt)
                    >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN))
            {
                info!(
                    from = %current,
                    to = %fastest,
                    from_rtt = ?current_rtt,
                    to_rtt = ?fastest_rtt,
                    "Switching to a faster path"
                );
                self.counts.switches += 1;
                current = fastest.clone();
            }
            if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) {
                self.chosen.insert(key, current.clone());
            }
            // The rest: measured ones by round-trip time, then unmeasured ones in the
            // configured order.
            let mut rest: Vec<&(String, usize)> = up
                .iter()
                .filter(|(address, _)| *address != current)
                .collect();
            rest.sort_by_key(|(address, preference)| {
                (
                    rtt(self, address).is_none(),
                    rtt(self, address),
                    *preference,
                )
            });
            ranked.push(current);
            ranked.extend(rest.into_iter().map(|(address, _)| address.clone()));
        }
        ranked.extend(down.into_iter().map(|(address, _)| address));
        ranked
    }

    /// A send to `address` failed.
    fn failed(&mut self, address: &str, now: Instant) {
        self.counts.send_failures += 1;
        if let Some(path) = self.paths.get_mut(address) {
            path.failed_at = Some(now);
            // A PING that didn't go out gets no answer; it is no unanswered one.
            path.probe = None;
        }
    }

    /// A send to `address` worked.
    fn sent(&mut self, address: &str) {
        if let Some(path) = self.paths.get_mut(address) {
            path.failed_at = None;
        }
    }

    /// The PINGs due now, with their nonces (from `nonce`): to every path used within
    /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING
    /// out for `PROBE_TIMEOUT` counts as unanswered.
    fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> {
        self.forget_unused(now);
        let mut due = Vec::new();
        for (address, path) in &mut self.paths {
            if let Some((_, sent)) = path.probe
                && now.saturating_duration_since(sent) >= PROBE_TIMEOUT
            {
                path.probe = None;
                path.unanswered += 1;
                self.counts.probes_unanswered += 1;
                if path.unanswered == UNANSWERED_DOWN {
                    info!(address = %address, "A path stopped answering PINGs; using others");
                }
            }
            let active = path
                .used
                .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR);
            let rested = path
                .last_probe
                .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY);
            if active && rested && path.probe.is_none() {
                let nonce = nonce();
                path.probe = Some((nonce, now));
                path.last_probe = Some(now);
                self.counts.probes_sent += 1;
                due.push((address.clone(), nonce));
            }
        }
        due
    }

    /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that
    /// path, which its PONG measures, even if it is a peer's only path (then it isn't
    /// pinged otherwise: `due` pings only paths there was a choice of).
    fn pinged(&mut self, address: &str, nonce: u64, now: Instant) {
        if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) {
            self.forget_unused(now);
            if self.paths.len() >= MAX_PATHS {
                return;
            }
        }
        let path = self.paths.entry(address.to_owned()).or_default();
        path.probe = Some((nonce, now));
        path.last_probe = Some(now);
        self.counts.probes_sent += 1;
    }

    /// The PING with `nonce` to `address` (from `due`) goes now: its round trip counts
    /// from here, not from when it was found due.
    fn sending(&mut self, address: &str, nonce: u64, now: Instant) {
        if let Some(path) = self.paths.get_mut(address)
            && let Some((expected, sent)) = &mut path.probe
            && *expected == nonce
        {
            *sent = now;
        }
    }

    /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out.
    fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> {
        let path = self.paths.get_mut(address)?;
        let (expected, sent) = path.probe?;
        if expected != nonce {
            return None;
        }
        let sample = now.saturating_duration_since(sent);
        path.probe = None;
        if path.unanswered >= UNANSWERED_DOWN {
            info!(address = %address, "A path answers PINGs again");
        }
        path.unanswered = 0;
        path.failed_at = None;
        path.rtt = Some(match path.rtt {
            Some(rtt) => (rtt * 3 + sample) / 4,
            None => sample,
        });
        self.counts.probes_answered += 1;
        path.rtt
    }

    /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices
    /// among them.
    fn forget_unused(&mut self, now: Instant) {
        let recent = |at: Option<Instant>| {
            at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2)
        };
        self.paths
            .retain(|_, path| recent(path.used) || recent(path.last_probe));
        let paths = &self.paths;
        self.chosen
            .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address)));
    }

    fn info(&self, address: &str, now: Instant) -> PathInfo {
        let path = self.paths.get(address);
        PathInfo {
            address: address.to_owned(),
            location: None,
            rtt_ms: path
                .and_then(|path| path.rtt)
                .map(|rtt| rtt.as_secs_f64() * 1000.0),
            down: path.is_some_and(|path| path.is_down(now)),
        }
    }
}

pub(crate) struct Paths {
    /// In order of preference.
    links: Vec<Link>,
    /// Its own PINGs and PONGs are counted here, under the address (`traffic`).
    traffic: Option<Arc<Traffic>>,
    /// The debug firewall, if any.
    gate: Option<Weak<dyn Gate>>,
    /// A std mutex: never held across an await.
    table: Mutex<Table>,
}

impl Paths {
    /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs
    /// counted in `traffic`, the firewall through `gate`.
    pub(crate) fn new(
        transports: Vec<(&'static str, Arc<dyn Transport>)>,
        traffic: Option<Arc<Traffic>>,
        gate: Option<Weak<dyn Gate>>,
    ) -> Self {
        Self {
            links: transports
                .into_iter()
                .map(|(scheme, inner)| Link { scheme, inner })
                .collect(),
            traffic,
            gate,
            table: Mutex::default(),
        }
    }

    fn gate(&self) -> Option<Arc<dyn Gate>> {
        self.gate.as_ref()?.upgrade()
    }

    /// The transport `address` is on, and the transport identity at it
    /// (`Transport::locate`, which notes where that is if the address says); None if no
    /// transport of ours has its scheme, or it isn't one of that transport's.
    fn split(&self, address: &str) -> Option<(&Link, String)> {
        let (scheme, target) = address.split_once(':')?;
        let link = self.links.iter().find(|link| link.scheme == scheme)?;
        let target = link.inner.locate(target)?;
        Some((link, target))
    }

    /// Whether one of our transports can use `address`.
    pub(crate) fn usable(&self, address: &str) -> bool {
        self.split(address).is_some()
    }

    /// `address` as paths keep it: `<scheme>:<transport identity>`.
    pub(crate) fn canonical(&self, address: &str) -> Option<String> {
        self.split(address)
            .map(|(link, target)| link.address(&target))
    }

    /// Where the node at `address` (as paths keep it) is, as others can use it
    /// (`Transport::locations`).
    pub(crate) fn locations(&self, address: &str) -> Vec<String> {
        let Some((link, target)) = self.split(address) else {
            return Vec::new();
        };
        link.inner
            .locations(&target)
            .iter()
            .map(|location| link.address(location))
            .collect()
    }

    /// Where `address`'s transport is in the order of preference (none of ours: last).
    pub(crate) fn preference(&self, address: &str) -> usize {
        let scheme = scheme_of(address);
        self.links
            .iter()
            .position(|link| link.scheme == scheme)
            .unwrap_or(usize::MAX)
    }

    /// The usable ones of `addresses` (as paths keep them), best first: paths that are
    /// up before those down, the one in use first, then by round-trip time, then in
    /// the configured order (module docs). Several addresses are paths to measure.
    pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> {
        let usable: Vec<(String, usize)> = addresses
            .iter()
            .filter(|address| self.usable(address))
            .map(|address| (address.clone(), self.preference(address)))
            .collect();
        self.table.lock_or_recover().rank(usable, Instant::now())
    }

    /// The smoothed round-trip time of the path to `address`, if measured.
    pub(crate) fn rtt(&self, address: &str) -> Option<Duration> {
        self.table.lock_or_recover().paths.get(address)?.rtt
    }

    /// What paths knows about each of `addresses`.
    pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> {
        let now = Instant::now();
        let infos = {
            let table = self.table.lock_or_recover();
            addresses
                .iter()
                .map(|address| table.info(address, now))
                .collect()
        };
        self.located(infos)
    }

    /// `infos` with where each path goes (`Transport::location`); asked outside the
    /// table's lock.
    fn located(&self, mut infos: Vec<PathInfo>) -> Vec<PathInfo> {
        for info in &mut infos {
            info.location = self
                .split(&info.address)
                .and_then(|(link, target)| link.inner.location(&target));
        }
        infos
    }

    /// Whether `address` is one of ours.
    pub(crate) fn is_own(&self, address: &str) -> bool {
        self.links.iter().any(|link| link.own_address() == address)
    }

    /// Our address on the transport of `address`, as `address` names it there.
    pub(crate) fn own_address(&self, address: &str) -> Option<String> {
        let scheme = scheme_of(address);
        self.links
            .iter()
            .find(|link| link.scheme == scheme)
            .map(Link::own_address)
    }

    /// The longest of the transports': until then, something may still be on its way.
    pub(crate) fn delivery_limit(&self) -> Duration {
        self.links
            .iter()
            .map(|link| link.inner.delivery_limit())
            .max()
            .unwrap_or_default()
    }

    /// Each transport's (`Transport::local_addresses`), with its scheme.
    pub(crate) fn local_addresses(&self) -> Vec<String> {
        self.links
            .iter()
            .flat_map(|link| {
                link.inner
                    .local_addresses()
                    .into_iter()
                    .map(|target| link.address(&target))
            })
            .collect()
    }

    /// The schemes of the transports, in order of preference.
    pub(crate) fn transports(&self) -> Vec<String> {
        self.links
            .iter()
            .map(|link| link.scheme.to_owned())
            .collect()
    }

    /// Each transport's, with its scheme.
    pub(crate) fn discovered_seeds(&self) -> Vec<String> {
        self.links
            .iter()
            .flat_map(|link| {
                link.inner
                    .discovered_seeds()
                    .into_iter()
                    .map(|seed| link.address(&seed))
            })
            .collect()
    }

    /// The address of the node a seed leads to, as paths keep it: an address of ours
    /// (`<scheme>:...`) whose transport reads it at once; with a scheme of ours, what
    /// that transport resolves the rest to (an address the transport can't read as an
    /// identity, e.g. a libp2p multiaddr without its peer ID, still resolves); without
    /// one, what the first transport that can resolves it to.
    pub(crate) async fn resolve(&self, seed: &str) -> Result<String> {
        if let Some(address) = self.canonical(seed) {
            return Ok(address);
        }
        let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
            let link = self.links.iter().find(|link| link.scheme == scheme)?;
            Some((link, rest))
        });
        if let Some((link, rest)) = scheme_and_rest {
            let target = link
                .inner
                .resolve_seed(rest)
                .await
                .with_context(|| format!("{}: {seed}", link.scheme))?;
            return Ok(link.address(&target));
        }
        let mut failures = Vec::new();
        for link in &self.links {
            match link.inner.resolve_seed(seed).await {
                Ok(target) => return Ok(link.address(&target)),
                Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
            }
        }
        Err(anyhow!("{}", failures.join("; ")))
    }

    /// `message` encoded for `address`: `peer` and `address` of the result are
    /// `address`, and it keeps the message (to encode it for another address).
    pub(crate) fn encode(&self, address: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
        let (link, target) = self
            .split(address)
            .with_context(|| format!("no transport for {address}"))?;
        let mut encoded = link.inner.encode(&target, message)?;
        encoded.peer = address.to_owned();
        encoded.address = address.to_owned();
        encoded.message = Some(Arc::new(message.clone()));
        Ok(encoded)
    }

    /// Sends `message` to the first of `addresses` (`rank`ed) that takes it: to its own
    /// `address` as encoded, to others encoded again for their transport (if it keeps
    /// the message). Returns the address it went to; an error with every address's if
    /// none took it.
    pub(crate) async fn send_first(
        &self,
        message: &EncodedMessage,
        addresses: &[String],
    ) -> Result<String> {
        let mut failures: Vec<(String, anyhow::Error)> = Vec::new();
        for address in addresses {
            let encoded = if *address == message.address {
                message.clone()
            } else {
                let Some(parsed) = &message.message else {
                    continue;
                };
                match self.encode(address, parsed) {
                    Ok(mut encoded) => {
                        encoded.peer = message.peer.clone();
                        encoded
                    }
                    Err(error) => {
                        failures.push((address.clone(), error));
                        continue;
                    }
                }
            };
            // A PING is noted before it goes: on a fast link, its PONG may be back
            // before the send returns.
            if let Some(ParsedMessage::Ping(nonce)) = message.message.as_deref() {
                self.table
                    .lock_or_recover()
                    .pinged(address, *nonce, Instant::now());
            }
            match self.send(&encoded).await {
                Ok(()) => {
                    let mut table = self.table.lock_or_recover();
                    table.sent(address);
                    if !failures.is_empty() {
                        table.counts.failovers += 1;
                        drop(table);
                        debug!(
                            peer = %message.peer,
                            address = %address,
                            failed = %describe(&failures),
                            "Sent at another address after a send failed"
                        );
                    }
                    return Ok(address.clone());
                }
                Err(error) => {
                    self.table.lock_or_recover().failed(address, Instant::now());
                    failures.push((address.clone(), error));
                }
            }
        }
        match failures.len() {
            0 => Err(anyhow!("no address to send to")),
            // The transport's error as it is.
            1 => Err(failures.pop().expect("one").1),
            _ => Err(anyhow!("{}", describe(&failures))),
        }
    }

    /// Sends `message` to its `address` (dropped silently if the firewall blocks it).
    pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> {
        if self
            .gate()
            .is_some_and(|gate| gate.blocks_outbound(&message.address))
        {
            tracing::trace!(to = %message.address, "Firewall dropped outbound message");
            return Ok(());
        }
        let (link, target) = self
            .split(&message.address)
            .with_context(|| format!("no transport for {}", message.address))?;
        let mut inner = message.clone();
        inner.peer = target.clone();
        inner.address = target;
        link.inner.send_encoded(&inner).await
    }

    /// How the transport of `address` is connected there (`Transport::connection`).
    pub(crate) fn connection(&self, address: &str) -> Option<Connection> {
        let (link, target) = self.split(address)?;
        link.inner.connection(&target)
    }

    /// Paths' own figures, and every transport's.
    pub(crate) fn stats(&self) -> TransportStats {
        let mut stats = TransportStats::default();
        for link in &self.links {
            stats.merge(link.inner.stats());
        }
        let now = Instant::now();
        let (mut paths, counts) = {
            let table = self.table.lock_or_recover();
            let paths: Vec<PathInfo> = table
                .paths
                .keys()
                .map(|address| table.info(address, now))
                .collect();
            (paths, table.counts)
        };
        paths.sort_by(|a, b| a.address.cmp(&b.address));
        stats.paths = Some(PathsStats {
            paths: self.located(paths),
            counts,
        });
        stats
    }

    /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted
    /// (`traffic`).
    async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> {
        let message = if pong {
            ParsedMessage::Pong(nonce)
        } else {
            ParsedMessage::Ping(nonce)
        };
        let encoded = self.encode(address, &message)?;
        let sent = self.send(&encoded).await;
        if let Some(traffic) = &self.traffic {
            match &sent {
                Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len),
                Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len),
            }
        }
        sent
    }

    /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING
    /// goes on its own, timed from when it goes: a send that takes long (a dial) neither
    /// holds up the others nor counts in their round trips.
    async fn probe(self: Arc<Self>) {
        let mut tick = tokio::time::interval(PROBE_TICK);
        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tick.tick().await;
            let due = self
                .table
                .lock_or_recover()
                .due(Instant::now(), random_nonce);
            for (address, nonce) in due {
                let paths = Arc::clone(&self);
                tokio::spawn(async move {
                    paths
                        .table
                        .lock_or_recover()
                        .sending(&address, nonce, Instant::now());
                    if let Err(error) = paths.send_probe(&address, false, nonce).await {
                        debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
                        paths
                            .table
                            .lock_or_recover()
                            .failed(&address, Instant::now());
                    }
                });
            }
        }
    }

    /// A message from `address`, handed on to `handler`: a PING answered and a PONG
    /// noted first; dropped if the firewall blocks it.
    async fn receive(
        self: Arc<Self>,
        address: String,
        message: ParsedMessage,
        payload_len: usize,
        handler: &PathHandler,
    ) -> Result<()> {
        if self
            .gate()
            .is_some_and(|gate| gate.blocks_inbound(&address))
        {
            tracing::trace!(from = %address, "Firewall dropped inbound message");
            return Ok(());
        }
        match message {
            ParsedMessage::Ping(nonce) => {
                let paths = Arc::clone(&self);
                let to = address.clone();
                tokio::spawn(async move {
                    if let Err(error) = paths.send_probe(&to, true, nonce).await {
                        debug!(to = %to, error = %format!("{error:#}"), "Couldn't answer a PING");
                    }
                });
            }
            ParsedMessage::Pong(nonce) => {
                self.table
                    .lock_or_recover()
                    .pong(&address, nonce, Instant::now());
            }
            _ => {}
        }
        handler(address, message, payload_len).await
    }

    /// Every transport's consumer, each message handed to `handler` under the sender's
    /// address (PINGs and PONGs end here), and the probe task; returns when the first
    /// consumer does.
    pub(crate) async fn start_consumer(self: &Arc<Self>, handler: Arc<PathHandler>) -> Result<()> {
        let probing = tokio::spawn(Arc::clone(self).probe());
        let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
            Vec::new();
        for link in &self.links {
            let inner = Arc::clone(&link.inner);
            let scheme = link.scheme;
            let handler = Arc::clone(&handler);
            let paths = Arc::clone(self);
            consumers.push(Box::pin(async move {
                inner
                    .start_consumer(Box::new(move |sender, message, payload_len| {
                        let paths = Arc::clone(&paths);
                        let handler = Arc::clone(&handler);
                        Box::pin(async move {
                            paths
                                .receive(
                                    format!("{scheme}:{sender}"),
                                    message,
                                    payload_len,
                                    &*handler,
                                )
                                .await
                        })
                    }))
                    .await
            }));
        }
        let result = first(consumers)
            .await
            .unwrap_or_else(|| Err(anyhow!("no transports")));
        probing.abort();
        result
    }

    /// When the first transport exits.
    pub(crate) async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
        let waits = self
            .links
            .iter()
            .map(|link| {
                let inner = Arc::clone(&link.inner);
                Box::pin(async move { inner.wait_for_exit().await })
                    as std::pin::Pin<Box<dyn Future<Output = _> + Send>>
            })
            .collect();
        first(waits)
            .await
            .unwrap_or_else(|| Err(std::io::Error::other("no transports")))
    }
}

/// A PING's nonce: random, so a PONG can't be made up for a PING not seen.
pub(crate) fn random_nonce() -> u64 {
    let mut bytes = [0u8; 8];
    if getrandom::fill(&mut bytes).is_err() {
        // No randomness: a nonce still tells answers apart, if not unguessably.
        return std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |since| since.as_nanos() as u64);
    }
    u64::from_le_bytes(bytes)
}

/// `address: error; …` for each failed send.
fn describe(failures: &[(String, anyhow::Error)]) -> String {
    failures
        .iter()
        .map(|(address, error)| format!("{address}: {error:#}"))
        .collect::<Vec<_>>()
        .join("; ")
}

/// The first of the `results` to arrive, from tasks that run until they end.
async fn first<T: Send + 'static>(
    tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>,
) -> Option<T> {
    let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
    for task in tasks {
        let sender = sender.clone();
        tokio::spawn(async move {
            let _ = sender.send(task.await).await;
        });
    }
    drop(sender);
    receiver.recv().await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::replica::Epoch;
    use crate::transport::sim::SimNet;
    use crate::transport::{HeadsSummary, Status};

    /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too.
    fn two_nets() -> (Arc<Paths>, Arc<Paths>) {
        let (first, second) = (SimNet::new(1), SimNet::new(2));
        let on_both = |id: &str| {
            Arc::new(Paths::new(
                vec![
                    ("sim", first.transport(id) as Arc<dyn Transport>),
                    ("other", second.transport(id) as Arc<dyn Transport>),
                ],
                None,
                None,
            ))
        };
        (on_both("a"), on_both("b"))
    }

    #[test]
    fn addresses_by_scheme() {
        let (a, _b) = two_nets();
        assert_eq!(a.local_addresses(), ["sim:a", "other:a"]);
        assert_eq!(a.transports(), ["sim", "other"]);
        assert_eq!(a.canonical("other:b").as_deref(), Some("other:b"));
        assert_eq!(a.canonical("pigeon:b"), None);
        assert_eq!(a.canonical("b"), None); // No scheme.
        assert!(a.preference("sim:b") < a.preference("other:b"));
        assert_eq!(a.preference("pigeon:b"), usize::MAX);
        assert!(a.is_own("other:a") && !a.is_own("other:b"));
        assert_eq!(a.own_address("other:b").as_deref(), Some("other:a"));
        assert_eq!(a.own_address("pigeon:b"), None);
    }

    #[tokio::test]
    async fn seeds_resolve_to_addresses() {
        let (a, _b) = two_nets();
        assert_eq!(a.resolve("other:b").await.unwrap(), "other:b");
        // Without a scheme: the first transport that resolves it.
        assert_eq!(a.resolve("b").await.unwrap(), "sim:b");
        assert!(a.resolve("nobody").await.is_err());
    }

    fn paths(addresses: &[&str]) -> Vec<(String, usize)> {
        addresses
            .iter()
            .enumerate()
            .map(|(preference, address)| (address.to_string(), preference))
            .collect()
    }

    /// Answers every PING due `rtt` later, as from the path it went to (`rtts`).
    fn answer(table: &mut Table, now: Instant, rtts: &[(&str, u64)]) -> Instant {
        let mut nonces = 0;
        let due = table.due(now, || {
            nonces += 1;
            nonces
        });
        for (address, nonce) in due {
            let rtt = rtts
                .iter()
                .find(|(to, _)| *to == address)
                .map(|(_, ms)| *ms);
            if let Some(ms) = rtt {
                table.pong(&address, nonce, now + Duration::from_millis(ms));
            }
        }
        now + PROBE_EVERY
    }

    #[test]
    fn the_configured_order_until_measured_then_the_faster_path() {
        let mut table = Table::default();
        let mut now = Instant::now();
        let set = paths(&["libp2p:x", "mycelium:x"]);
        assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]);
        // libp2p a bit slower: not enough to switch.
        for _ in 0..8 {
            now = answer(&mut table, now, &[("libp2p:x", 110), ("mycelium:x", 100)]);
        }
        assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]);
        assert_eq!(table.counts.switches, 0);
        // Much slower: switched, and it stays switched while that holds.
        for _ in 0..8 {
            now = answer(&mut table, now, &[("libp2p:x", 300), ("mycelium:x", 100)]);
        }
        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
        assert_eq!(table.counts.switches, 1);
        // Small differences under SWITCH_MIN never switch.
        let lan = paths(&["libp2p:y", "mycelium:y"]);
        table.rank(lan.clone(), now);
        for _ in 0..8 {
            now = answer(&mut table, now, &[("libp2p:y", 4), ("mycelium:y", 1)]);
        }
        assert_eq!(table.rank(lan, now), ["libp2p:y", "mycelium:y"]);
    }

    #[test]
    fn unanswered_or_failing_paths_are_down() {
        let mut table = Table::default();
        let mut now = Instant::now();
        let set = paths(&["libp2p:x", "mycelium:x"]);
        table.rank(set.clone(), now);
        // libp2p stops answering: down after UNANSWERED_DOWN PINGs, then last.
        for _ in 0..=UNANSWERED_DOWN {
            now = answer(&mut table, now, &[("mycelium:x", 100)]);
            table.rank(set.clone(), now);
        }
        assert!(table.info("libp2p:x", now).down);
        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
        // It answers again: up, and measured.
        now = answer(&mut table, now, &[("libp2p:x", 50), ("mycelium:x", 100)]);
        assert!(!table.info("libp2p:x", now).down);

        // A failed send: down for DOWN_FOR, unless a send there works.
        table.failed("mycelium:x", now);
        assert!(table.info("mycelium:x", now).down);
        assert!(!table.info("mycelium:x", now + DOWN_FOR).down);
        table.sent("mycelium:x");
        assert!(!table.info("mycelium:x", now).down);
    }

    #[test]
    fn only_paths_in_use_are_pinged() {
        let mut table = Table::default();
        let now = Instant::now();
        let mut nonce = 0;
        let mut next = || {
            nonce += 1;
            nonce
        };
        // One address: nothing to choose, nothing to measure.
        table.rank(paths(&["libp2p:z"]), now);
        assert!(table.due(now, &mut next).is_empty());
        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
        assert_eq!(table.due(now, &mut next).len(), 2);
        // Not again within PROBE_EVERY; a stale answer is ignored.
        assert!(table.due(now + PROBE_TIMEOUT, &mut next).is_empty());
        assert_eq!(table.pong("libp2p:x", 99, now), None);
        // Unused for ACTIVE_FOR: no more PINGs; for twice as long: forgotten.
        assert!(table.due(now + ACTIVE_FOR, &mut next).is_empty());
        table.due(now + ACTIVE_FOR * 2, &mut next);
        assert!(table.paths.is_empty() && table.chosen.is_empty());
    }

    /// A PING's round trip counts from when it goes, not from when it was found due.
    #[test]
    fn round_trips_count_from_the_send() {
        let mut table = Table::default();
        let now = Instant::now();
        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
        let due = table.due(now, || 7);
        assert_eq!(due.len(), 2);
        // Sent a second late (after another path's slow send).
        let sent = now + Duration::from_secs(1);
        table.sending("libp2p:x", 7, sent);
        table.sending("libp2p:x", 8, sent + Duration::from_secs(1)); // not its PING
        table.pong("libp2p:x", 7, sent + Duration::from_millis(100));
        let rtt = table.info("libp2p:x", now).rtt_ms.unwrap();
        assert!((rtt - 100.0).abs() < 1e-6, "{rtt}");
    }

    /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and
    /// the PONG measures it.
    #[test]
    fn keepalive_pings_are_probes() {
        let mut table = Table::default();
        let now = Instant::now();
        let mut nonce = 100;
        let mut next = || {
            nonce += 1;
            nonce
        };
        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
        table.pinged("libp2p:x", 7, now);
        let due = table.due(now, &mut next);
        assert_eq!(due.len(), 1);
        assert_eq!(due[0].0, "mycelium:x");
        assert!(
            table
                .pong("libp2p:x", 7, now + Duration::from_millis(20))
                .is_some()
        );
        assert_eq!(table.counts.probes_sent, 2);
        // A peer's only path is measured by its keepalives, and not pinged otherwise.
        table.pinged("libp2p:z", 8, now);
        assert!(
            table
                .pong("libp2p:z", 8, now + Duration::from_millis(30))
                .is_some()
        );
        let rtt = table.info("libp2p:z", now).rtt_ms.unwrap();
        assert!((rtt - 30.0).abs() < 1e-6, "{rtt}");
        let later = now + PROBE_EVERY;
        assert!(
            table
                .due(later, &mut next)
                .iter()
                .all(|(address, _)| address != "libp2p:z")
        );
        // Forgotten once the keepalives stop.
        table.due(now + ACTIVE_FOR * 2, &mut next);
        assert!(!table.paths.contains_key("libp2p:z"));
    }

    /// Blocks one address, both ways.
    struct Blocking(&'static str);

    impl Gate for Blocking {
        fn blocks_inbound(&self, address: &str) -> bool {
            address == self.0
        }
        fn blocks_outbound(&self, address: &str) -> bool {
            address == self.0
        }
    }

    /// The firewall drops what comes from a blocked address before anything sees it
    /// (a PING isn't answered), and sends to one silently.
    #[tokio::test]
    async fn the_firewall_drops_both_ways() {
        let net = SimNet::new(4);
        let gate: Arc<dyn Gate> = Arc::new(Blocking("sim:a"));
        let b = Arc::new(Paths::new(
            vec![("sim", net.transport("b") as Arc<dyn Transport>)],
            None,
            Some(Arc::downgrade(&gate)),
        ));
        let a = Arc::new(Paths::new(
            vec![("sim", net.transport("a") as Arc<dyn Transport>)],
            None,
            None,
        ));
        let received = |paths: &Arc<Paths>| {
            let (sender, received) = tokio::sync::mpsc::unbounded_channel();
            let handler: Arc<PathHandler> = Arc::new(move |from, message, _| {
                let sender = sender.clone();
                Box::pin(async move {
                    let _ = sender.send((from, message.kind()));
                    Ok(())
                })
            });
            let paths = Arc::clone(paths);
            tokio::spawn(async move { paths.start_consumer(handler).await });
            received
        };
        let (mut at_a, mut at_b) = (received(&a), received(&b));

        // a's PING never reaches b, so nothing answers it.
        let ping = a.encode("sim:b", &ParsedMessage::Ping(5)).unwrap();
        a.send(&ping).await.unwrap();
        // b's send to a "works", and goes nowhere.
        let status = b.encode("sim:a", &some_status()).unwrap();
        b.send(&status).await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert!(at_b.try_recv().is_err());
        assert!(at_a.try_recv().is_err());

        // Without the gate: a PING is answered by b's paths, and goes on up there.
        drop(gate);
        a.send(&ping).await.unwrap();
        assert_eq!(
            at_b.recv().await.unwrap(),
            ("sim:a".to_owned(), crate::transport::MessageKind::Ping)
        );
        assert_eq!(
            at_a.recv().await.unwrap(),
            ("sim:b".to_owned(), crate::transport::MessageKind::Pong)
        );
    }

    /// A transport whose sends all fail.
    struct Broken;

    #[async_trait::async_trait]
    impl Transport for Broken {
        fn local_id(&self) -> &str {
            "us"
        }
        fn is_peer_id(&self, id: &str) -> bool {
            !id.is_empty()
        }
        fn delivery_limit(&self) -> Duration {
            Duration::from_secs(1)
        }
        async fn resolve_seed(&self, seed: &str) -> Result<String> {
            Ok(seed.to_owned())
        }
        fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
            crate::transport::mycelium::encode_message(peer, message)
        }
        async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> {
            Err(anyhow!("broken"))
        }
        async fn start_consumer(&self, _handler: crate::transport::MessageHandler) -> Result<()> {
            std::future::pending().await
        }
        async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
            std::future::pending().await
        }
    }

    fn some_status() -> ParsedMessage {
        ParsedMessage::Status(Status {
            heads: HeadsSummary::of(Vec::new()),
            epoch: Epoch::GENESIS,
            epoch_seq: 0,
            ready: false,
            syncs_with_you: false,
        })
    }

    /// When the preferred transport fails, the same send goes over the next one, and
    /// the failed address is ranked last for a while.
    #[tokio::test]
    async fn a_failed_send_goes_to_the_next_address() {
        let net = SimNet::new(3);
        let a = Paths::new(
            vec![
                ("broken", Arc::new(Broken) as Arc<dyn Transport>),
                ("sim", net.transport("a") as Arc<dyn Transport>),
            ],
            None,
            None,
        );
        let b = Arc::new(Paths::new(
            vec![("sim", net.transport("b") as Arc<dyn Transport>)],
            None,
            None,
        ));
        let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
        let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {
            let sender = sender.clone();
            Box::pin(async move {
                let _ = sender.send(from);
                Ok(())
            })
        });
        tokio::spawn(async move { b.start_consumer(handler).await });

        let addresses = vec!["sim:b".to_owned(), "broken:b".to_owned()];
        let ranked = a.rank(&addresses);
        assert_eq!(ranked, ["broken:b", "sim:b"]); // The configured order.
        let mut encoded = a.encode(&ranked[0], &some_status()).unwrap();
        encoded.peer = "node-b".to_owned();
        assert_eq!(a.send_first(&encoded, &ranked).await.unwrap(), "sim:b");
        assert_eq!(received.recv().await.unwrap(), "sim:a");
        assert_eq!(a.rank(&addresses), ["sim:b", "broken:b"]);

        // Without the message kept, nothing else can be tried: the transport's error.
        encoded.message = None;
        let error = a
            .send_first(&encoded, &["broken:b".to_owned()])
            .await
            .unwrap_err();
        assert_eq!(error.to_string(), "broken");
        assert!(a.send_first(&encoded, &[]).await.is_err());
    }

    /// A message sent to an address arrives under the sender's address on that
    /// transport.
    #[tokio::test]
    async fn messages_arrive_under_the_senders_address() {
        let (a, b) = two_nets();
        let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
        let handler: Arc<PathHandler> = Arc::new(move |from, message, _| {
            let sender = sender.clone();
            Box::pin(async move {
                let _ = sender.send((from, message.kind()));
                Ok(())
            })
        });
        tokio::spawn(async move { b.start_consumer(handler).await });
        let status = some_status();
        for to in ["other:b", "sim:b"] {
            let encoded = a.encode(to, &status).unwrap();
            assert_eq!(encoded.address, to);
            a.send(&encoded).await.unwrap();
        }
        let mut from: Vec<String> = Vec::new();
        for _ in 0..2 {
            from.push(received.recv().await.unwrap().0);
        }
        from.sort();
        assert_eq!(from, ["other:a", "sim:a"]);
        assert!(a.encode("pigeon:b", &status).is_err());
    }
}