//! libp2p as a transport (`--transports libp2p`, in builds with the `libp2p` feature):
//! direct connections over TCP (Noise, Yamux) and QUIC, to addresses this node knows.
//!
//! A node's transport identity is its libp2p peer ID; where it is, its listen addresses
//! (multiaddrs ending in the peer ID, `locations`). Seeds are such multiaddrs, or ones
//! without the peer ID, which a connection finds out. Messages go one per stream of the
//! `/continuum/1` protocol: their length (4 bytes, big endian), then the message as
//! `wire` encodes it. A send is done once the stream is written and closed; a connection
//! that can't be made, or a stream that can't be written, refuses it. A stream's protocol
//! goes with its first data, without waiting for the other side to agree
//! (`swarm_config`): opening a stream costs no round trip.
//!
//! Connected peers tell each other where they listen (libp2p's identify protocol), so a
//! node learns the addresses of peers that dialled it. A node publishes where it can be
//! reached (`local_addresses`): the addresses it announces (`--libp2p-announce`, e.g. a
//! public address forwarded to it), or else those it listens on, loopback left out.
//!
//! A node nobody can dial (behind NAT, or a closed port) can still be reached through a
//! relay: a node of the swarm that relays (`--libp2p-relay`), or one configured
//! (`--libp2p-relays`), on which it reserves a slot; its address through the relay
//! (`<relay>/p2p-circuit/p2p/<it>`) is published with the others. Over a relayed connection, both sides then try to
//! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until
//! that works, or for good if it doesn't, within its limits (`RELAY_CIRCUIT_DURATION`,
//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send).
//!
//! Of several connections to one peer (both dialled at once, or AutoNAT dialled back),
//! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`),
//! by the side with the smaller peer ID.
//!
//! A relay says so where it publishes its addresses: next to each, the same address
//! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the
//! member list. A node that needs a relay picks up to `AUTO_RELAYS` relays
//! (`relay_check`): those configured first, then those listed in the document
//! (`learn_relays`), then members that relay. It needs one when
//! it has no address others could dial, or when AutoNAT says it can't be dialled:
//! AutoNAT asks connected peers (nodes of the swarm) to dial it back. Once AutoNAT says
//! it can be dialled, it lets the relays it picked go.
//!
//! With `--libp2p-mdns`, a node finds libp2p peers on its LAN (multicast DNS). Their
//! LAN addresses are noted, and those that turn out to speak continuum (identify says
//! so) are seeds while the node joins (`discovered_seeds`): nodes on one LAN find each
//! other without configured seeds.
//!
//! It uses public libp2p infrastructure only with `--libp2p-public`: then it joins the
//! IPFS DHT (Kademlia, through the IPFS bootstrap nodes), where it finds addresses of
//! peers the member list doesn't have, and public peers that relay; those are picked
//! after members that relay. Public relays keep circuits short (by default 2 minutes and
//! 128 KiB): enough to punch holes through NATs, not to carry a swarm's traffic. Without
//! it, relays are those configured, or members that say they relay.

use std::collections::HashMap;
use std::path::Path;
use std::str::FromStr;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use ::libp2p::multiaddr::Protocol;
use ::libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
use ::libp2p::swarm::{ConnectionId, SwarmEvent};
use ::libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm, identity::Keypair};
use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
use tokio::sync::{Semaphore, mpsc, oneshot, watch};
use tracing::{debug, info, trace, warn};

use crate::locks::MutexExt;
use crate::transport::stats::{
    Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats,
};
use crate::transport::{
    Addressed, Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport,
    hand_on_probe, is_probe, wire,
};

/// The scheme of this transport's addresses (`transport::paths`).
pub(crate) const SCHEME: &str = "libp2p";

const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1");
/// Sends taking this long are logged with where the time went (at debug).
const SLOW_SEND: Duration = Duration::from_millis(5);
/// The largest message read: anyone who can connect can send anything, so this bounds
/// what one stream costs. Sync messages above 16 chunks travel as blobs.
const MAX_MESSAGE: usize = 128 << 20;
/// How long a connection may take to come up.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// A connection nobody uses is closed after this; neighbours keep theirs busy.
const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
/// How long a stream may take to open and to carry a message, before what its size adds
/// (`stream_timeout`).
const STREAM_TIMEOUT: Duration = Duration::from_secs(10);
/// The slowest a message is expected to go, in bytes per second.
const MIN_RATE: u64 = 1 << 20;
/// Inbound streams read at once.
const READS: usize = 32;
/// Addresses kept per peer, and peers with addresses at most.
const ADDRESSES_PER_PEER: usize = 8;
const MAX_PEERS: usize = 4096;

/// A relayed connection (circuit) through this node lasts at most this long and carries
/// at most this much: generous, since the swarm's own relays may carry sync traffic for
/// good where hole punching fails (libp2p's defaults, 2 minutes and 128 KiB, are for
/// the punching only).
const RELAY_CIRCUIT_DURATION: Duration = Duration::from_secs(3600);
const RELAY_CIRCUIT_BYTES: u64 = 1 << 30;
/// Circuits through this node at once, and per peer.
const RELAY_CIRCUITS: usize = 64;
const RELAY_CIRCUITS_PER_PEER: usize = 8;
/// A configured relay that couldn't be reached to find out its peer ID is tried again
/// after this.
const RELAY_RETRY: Duration = Duration::from_secs(30);
/// Relays picked among the members, at most, and how often a node checks whether it
/// needs them.
const AUTO_RELAYS: usize = 2;
const RELAY_CHECK: Duration = Duration::from_secs(30);
/// A relay that lost the slot before granting it (it doesn't answer, or another peer
/// answers at its address: a relay restarted without its libp2p key) isn't picked again
/// for this long.
const RELAY_PAUSE: Duration = Duration::from_secs(600);
/// A peer nobody mentioned for this long (the member list, identify, mDNS, a
/// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted
/// without their libp2p key, whose new ones the member list has since. Members are
/// mentioned every few seconds (`swarm::topology`). Checked every `PRUNE_EVERY`.
const STALE_AFTER: Duration = Duration::from_secs(600);
const PRUNE_EVERY: Duration = Duration::from_secs(60);
/// Several connections to one peer (both dialled at once, or AutoNAT dialled back to
/// see whether it can) are cut to one (`Book::duplicates`) this long after the last one
/// opened: long enough for what it was dialled for. Looked for every `DEDUPE_EVERY`.
const DUPLICATE_GRACE: Duration = Duration::from_secs(10);
const DEDUPE_EVERY: Duration = Duration::from_secs(5);

/// How long a message of `bytes` may take on a stream.
fn stream_timeout(bytes: usize) -> Duration {
    STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE)
}

/// What a libp2p transport is made with.
pub(crate) struct Libp2pConfig {
    pub(crate) key: Keypair,
    /// Multiaddrs to listen on.
    pub(crate) listen: Vec<Multiaddr>,
    /// Multiaddrs to publish instead of the listen addresses, if any.
    pub(crate) announce: Vec<Multiaddr>,
    /// Relays to reserve a slot on (multiaddrs; without the relay's peer ID, a connection
    /// finds it out).
    pub(crate) relays: Vec<Multiaddr>,
    /// Relay for other nodes.
    pub(crate) relay: bool,
    /// Speak continuum (`PROTOCOL`): not on a node that only relays (`--relay-only`),
    /// which nobody should take for a node.
    pub(crate) continuum: bool,
    /// Find peers on the LAN (mDNS).
    pub(crate) mdns: bool,
    /// Use public infrastructure: the IPFS DHT, public relays.
    pub(crate) public: bool,
    /// libp2p's in-memory transport instead of TCP and QUIC: for tests.
    pub(crate) memory: bool,
}

impl Libp2pConfig {
    /// The key in `path` (32 raw bytes, an Ed25519 secret key), or a new one each start
    /// without a file: the peer ID only says where a node is, so a new one costs nothing
    /// but the WHO that finds it again (`transport::identity`).
    pub(crate) fn load_key(path: Option<&Path>) -> Result<Keypair> {
        let Some(path) = path else {
            return Ok(Keypair::generate_ed25519());
        };
        let mut bytes = std::fs::read(path)
            .with_context(|| format!("Reading the libp2p key from {}", path.display()))?;
        crate::swarm::auth::warn_if_readable_by_others(path, "libp2p key");
        if bytes.len() != 32 {
            bail!(
                "The libp2p key in {} has {} bytes; it has 32",
                path.display(),
                bytes.len()
            );
        }
        Keypair::ed25519_from_bytes(&mut bytes)
            .map_err(|error| anyhow!("The libp2p key in {}: {error}", path.display()))
    }
}

enum Command {
    /// Connect to `peer` (at `addresses`) unless connected.
    Connect {
        peer: PeerId,
        addresses: Vec<Multiaddr>,
        done: oneshot::Sender<Result<(), String>>,
    },
    /// Connect to `address` to find out which peer is there.
    Identify {
        address: Multiaddr,
        done: oneshot::Sender<Result<PeerId, String>>,
    },
    /// See whether relays are needed now (one was just learned of).
    CheckRelays,
    /// Find out which peer relays at `address`, named without the relay's peer ID
    /// (`Book::finding`).
    FindRelay { address: Multiaddr },
}

/// Who named a relay to pick, in the order they are picked.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Named {
    /// `Libp2pConfig::relays`.
    Configured,
    /// The relays listed in the document (`Transport::learn_relays`).
    Listed,
}

impl Named {
    fn as_str(self) -> &'static str {
        match self {
            Self::Configured => "configured",
            Self::Listed => "listed",
        }
    }
}

/// A relay named to pick.
#[derive(Clone, Debug, PartialEq, Eq)]
struct NamedRelay {
    /// Where, with the relay's peer ID.
    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).
    given: Multiaddr,
    named: Named,
}

/// Where peers are, and where we listen.
#[derive(Default)]
struct Book {
    addresses: HashMap<PeerId, Vec<Multiaddr>>,
    listening: Vec<Multiaddr>,
    /// `Libp2pConfig::announce`.
    announced: Vec<Multiaddr>,
    /// `Libp2pConfig::relay`: this node relays, and says so.
    serving: bool,
    /// Relays named to pick (configured, listed), by peer: picked first.
    named_relays: HashMap<PeerId, NamedRelay>,
    /// Relays named without their peer ID, being found out (`Command::FindRelay`).
    finding: HashMap<Multiaddr, Named>,
    /// The relays listed in the document, as last learned (`learn_listed`).
    listed: Vec<Multiaddr>,
    /// Members that say they relay, and where (with the relay's peer ID).
    relays: HashMap<PeerId, Multiaddr>,
    /// With `Libp2pConfig::public`: other peers that relay (identify says so), and
    /// where.
    public_relays: HashMap<PeerId, Multiaddr>,
    /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`).
    failed_relays: HashMap<PeerId, tokio::time::Instant>,
    /// Relays that granted us a slot we hold.
    slots: std::collections::HashSet<PeerId>,
    /// Peers holding a slot on this node, as a relay.
    slots_here: std::collections::HashSet<PeerId>,
    /// Open connections: to whom, whether through a relay, and the remote address.
    connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>,
    /// What AutoNAT says, and the address it confirmed while public.
    reachability: Reachability,
    public_address: Option<Multiaddr>,
    /// Since start (`stats`).
    counts: Counts,
    /// When each peer was last mentioned (`STALE_AFTER`).
    seen: HashMap<PeerId, tokio::time::Instant>,
    /// Peers found on the LAN (mDNS) not known yet to speak continuum.
    lan: std::collections::HashSet<PeerId>,
    /// Peers that speak continuum (identify says so).
    continuum: std::collections::HashSet<PeerId>,
    /// Peers found on the LAN that speak continuum, with their address: seeds not
    /// handed out yet (`discovered_seeds`).
    discovered: Vec<Multiaddr>,
}

/// What `Libp2pStats` counts since start.
#[derive(Default)]
struct Counts {
    slots_granted: u64,
    slots_lost: u64,
    slots_not_granted: u64,
    hole_punches: HolePunches,
    dial_failures: DialFailures,
}

/// Counts a dial that failed with `error` (not one that wasn't needed).
/// A dial error, shortly: for each address tried (without the peer ID the caller names
/// already), the innermost cause, once. libp2p's own text nests each transport's errors
/// in the next ("Multiple dial errors occurred: ..."), repeating the cause at each level.
fn describe_dial_error(error: &::libp2p::swarm::DialError) -> String {
    use ::libp2p::swarm::DialError;
    let DialError::Transport(errors) = error else {
        return error.to_string();
    };
    let described: Vec<String> = errors
        .iter()
        .map(|(address, error)| {
            let mut address = address.clone();
            if let Some(Protocol::P2p(_)) = address.iter().last() {
                address.pop();
            }
            format!("{address}: {}", innermost_causes(error))
        })
        .collect();
    if described.is_empty() {
        return error.to_string();
    }
    described.join("; ")
}

/// Why a slot on a relay ended, or was never granted, shortly: libp2p nests the
/// relay's answer several types deep (`Left(Left(Left(Reservation(...))))`), and the
/// common cases deserve a sentence.
fn relay_failure(reason: &std::io::Result<()>) -> String {
    let Err(error) = reason else {
        return "closed".to_owned();
    };
    relay_failure_text(&format!("{error:?}"), &innermost_causes(error))
}

/// `relay_failure` on the error's debug form, and its innermost cause as text.
fn relay_failure_text(debug: &str, innermost: &str) -> String {
    let known = [
        (
            "NoAddressesInReservation",
            "the relay gave no address to be reached through it",
        ),
        ("ResourceLimitExceeded", "the relay has no slot left"),
        ("PermissionDenied", "the relay doesn't let us reserve"),
        ("NoReservation", "the relay holds no slot for us"),
        (
            "connection gated",
            "the relay refused the connection (connection gated)",
        ),
        (
            "UnsupportedProtocol",
            "the relay doesn't relay (no circuit relay protocol)",
        ),
        ("Timeout", "the relay didn't answer in time"),
    ];
    if let Some((_, said)) = known.iter().find(|(pattern, _)| debug.contains(pattern)) {
        return (*said).to_owned();
    }
    // The protocol refused, but only after its first data (`swarm_config`).
    if innermost == "Failed" {
        return "the relay refused the circuit relay protocol (is it a relay?)".to_owned();
    }
    innermost.to_owned()
}

/// The innermost of `error`'s causes, each distinct one once: those libp2p lists in a
/// "Multiple dial errors occurred:" text (one per line, ` - <cause>: <its cause>...`),
/// or the last of the source chain.
fn innermost_causes(error: &(dyn std::error::Error + 'static)) -> String {
    let mut innermost = error;
    while let Some(source) = innermost.source() {
        innermost = source;
    }
    let text = innermost.to_string();
    let mut causes: Vec<&str> = Vec::new();
    for line in text.lines() {
        let Some(listed) = line.trim().strip_prefix("- ") else {
            continue;
        };
        // A line ends in the next level's heading when that level lists more.
        let cause = listed
            .split(": ")
            .filter(|part| !part.starts_with("Multiple dial errors"))
            .last()
            .unwrap_or(listed)
            .trim();
        if !cause.is_empty() && !causes.contains(&cause) {
            causes.push(cause);
        }
    }
    if causes.is_empty() {
        let cause = text.rsplit(": ").next().unwrap_or(&text).trim();
        return cause.to_owned();
    }
    causes.join(", ")
}

fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) {
    use ::libp2p::swarm::DialError;
    match error {
        DialError::DialPeerConditionFalse(_) => {}
        DialError::Transport(_) => failures.unreachable += 1,
        DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1,
        DialError::NoAddresses => failures.no_addresses += 1,
        _ => failures.other += 1,
    }
}

impl Book {
    /// How we are connected to `peer`: directly if any connection is direct.
    fn connection(&self, peer: &PeerId) -> Option<Connection> {
        let mut found = None;
        for (connected, relayed, _) in self.connections.values() {
            if connected == peer {
                if !relayed {
                    return Some(Connection::Direct);
                }
                found = Some(Connection::Relayed);
            }
        }
        found
    }

    /// 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`).
    fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> {
        if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) {
            return Vec::new();
        }
        let mut connections: Vec<(bool, ConnectionId)> = self
            .connections
            .iter()
            .filter(|(_, (connected, _, _))| connected == peer)
            .map(|(id, (_, relayed, _))| (*relayed, *id))
            .collect();
        // Direct (false) first, then the oldest.
        connections.sort();
        connections.into_iter().skip(1).map(|(_, id)| id).collect()
    }

    /// The remote address of a connection to `peer`: a direct one if there is one.
    fn location(&self, peer: &PeerId) -> Option<&Multiaddr> {
        let mut found = None;
        for (connected, relayed, address) in self.connections.values() {
            if connected == peer {
                if !relayed {
                    return Some(address);
                }
                found = Some(address);
            }
        }
        found
    }

    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(|at| at.elapsed() < RELAY_PAUSE)
                    .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.
    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.
    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.lan.remove(peer);
            self.failed_relays.remove(peer);
            self.seen.remove(peer);
        }
        stale.len()
    }

    /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first.
    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`), without the peer ID.
    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();
        published(&direct)
    }
}

/// Whether `address` only reaches this host (or this link): no use to other hosts, and
/// dialled by one, it reaches that host itself.
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).
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.
    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.
    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.
    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.
    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.
    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()
    }

    /// Whether all of `addresses` are known of `peer` (`remember`).
    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.
    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.
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`.
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
}

/// 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).
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.
fn split_peer(address: &Multiaddr) -> Option<PeerId> {
    match address.iter().last() {
        Some(Protocol::P2p(peer)) => Some(peer),
        _ => None,
    }
}

pub(crate) struct Libp2pTransport {
    local_id: String,
    peer_id: PeerId,
    control: libp2p_stream::Control,
    commands: mpsc::UnboundedSender<Command>,
    book: Arc<Mutex<Book>>,
    incoming: tokio::sync::Mutex<Option<libp2p_stream::IncomingStreams>>,
    stopped: watch::Receiver<bool>,
}

fn lock(book: &Mutex<Book>) -> MutexGuard<'_, Book> {
    book.lock_or_recover()
}

impl Libp2pTransport {
    /// Builds the swarm, listens, and runs it in a task of its own.
    pub(crate) fn start(config: Libp2pConfig) -> Result<Self> {
        let peer_id = config.key.public().to_peer_id();
        let features = Features::of(&config);
        let mut swarm = if config.memory {
            memory_swarm(config.key, features)?
        } else {
            network_swarm(config.key, features)?
        };
        for address in config.listen {
            swarm
                .listen_on(address.clone())
                .with_context(|| format!("Listening on {address}"))?;
        }
        for address in &config.announce {
            // Identify tells peers these.
            swarm.add_external_address(address.clone());
        }
        let mut control = swarm.behaviour().stream.new_control();
        let incoming = if config.continuum {
            Some(
                control
                    .accept(PROTOCOL)
                    .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?,
            )
        } else {
            None
        };
        let (commands, receiver) = mpsc::unbounded_channel();
        let mut book = Book {
            announced: config.announce,
            serving: config.relay,
            ..Book::default()
        };
        for relay in config.relays {
            if let Some(address) = book.name_relay(relay, Named::Configured, peer_id) {
                let _ = commands.send(Command::FindRelay { address });
            }
        }
        let book = Arc::new(Mutex::new(book));
        let (stopping, stopped) = watch::channel(false);
        tokio::spawn(run(
            swarm,
            receiver,
            commands.clone(),
            Arc::clone(&book),
            Relaying {
                serving: config.relay,
                public: config.public,
                local: peer_id,
            },
            stopping,
        ));
        info!(peer_id = %peer_id, "Transport started");
        Ok(Self {
            local_id: peer_id.to_string(),
            peer_id,
            control,
            commands,
            book,
            incoming: tokio::sync::Mutex::new(incoming),
            stopped,
        })
    }

    /// Connected to `peer`, if it can be had in time.
    async fn connect(&self, peer: PeerId) -> Result<()> {
        let addresses = lock(&self.book)
            .addresses
            .get(&peer)
            .map(|known| dialable(known, self.peer_id))
            .unwrap_or_default();
        let (done, connected) = oneshot::channel();
        self.commands
            .send(Command::Connect {
                peer,
                addresses,
                done,
            })
            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
        match tokio::time::timeout(CONNECT_TIMEOUT, connected).await {
            Ok(Ok(Ok(()))) => Ok(()),
            Ok(Ok(Err(error))) => bail!("couldn't connect to {peer}: {error}"),
            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
            Err(_) => bail!("connecting to {peer} took longer than {CONNECT_TIMEOUT:?}"),
        }
    }
}

/// The protocol version identify says (informational: nothing checks it).
const IDENTIFY_VERSION: &str = "/continuum/1";

#[derive(::libp2p::swarm::NetworkBehaviour)]
struct Behaviour {
    stream: libp2p_stream::Behaviour,
    identify: ::libp2p::identify::Behaviour,
    relay_client: ::libp2p::relay::client::Behaviour,
    /// Only with `Libp2pConfig::relay`.
    relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
    dcutr: ::libp2p::dcutr::Behaviour,
    autonat: ::libp2p::autonat::Behaviour,
    /// Only with `Libp2pConfig::mdns`.
    mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
    /// Only with `Libp2pConfig::public`: the IPFS DHT.
    kad: ::libp2p::swarm::behaviour::toggle::Toggle<
        ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>,
    >,
}

/// What the behaviour has besides what every node has.
#[derive(Clone, Copy)]
struct Features {
    relay: bool,
    mdns: bool,
    public: bool,
}

impl Features {
    fn of(config: &Libp2pConfig) -> Self {
        Self {
            relay: config.relay,
            mdns: config.mdns,
            public: config.public,
        }
    }
}

/// The IPFS DHT's protocol, and its bootstrap nodes (all at `IPFS_BOOTSTRAP_ADDRESS`).
const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");
const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io";
const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [
    "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
    "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
    "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
    "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
];
/// How often a node on the DHT refreshes its routing table.
const DHT_BOOTSTRAP: Duration = Duration::from_secs(300);

impl Behaviour {
    fn new(
        key: &Keypair,
        relay_client: ::libp2p::relay::client::Behaviour,
        features: Features,
    ) -> std::io::Result<Self> {
        let Features {
            relay,
            mdns,
            public,
        } = features;
        let peer_id = key.public().to_peer_id();
        let kad = public.then(|| {
            let mut dht = ::libp2p::kad::Behaviour::with_config(
                peer_id,
                ::libp2p::kad::store::MemoryStore::new(peer_id),
                ::libp2p::kad::Config::new(IPFS_DHT),
            );
            let bootstrap: Multiaddr = IPFS_BOOTSTRAP_ADDRESS.parse().expect("a multiaddr");
            for peer in IPFS_BOOTSTRAP_PEERS {
                let peer: PeerId = peer.parse().expect("a peer ID");
                let _ = dht.add_address(&peer, bootstrap.clone());
            }
            dht
        });
        let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
        let relay = relay.then(|| {
            ::libp2p::relay::Behaviour::new(
                peer_id,
                ::libp2p::relay::Config {
                    max_circuits: RELAY_CIRCUITS,
                    max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER,
                    max_circuit_duration: RELAY_CIRCUIT_DURATION,
                    max_circuit_bytes: RELAY_CIRCUIT_BYTES,
                    ..Default::default()
                },
            )
        });
        let mdns = match mdns {
            true => Some(::libp2p::mdns::tokio::Behaviour::new(
                ::libp2p::mdns::Config::default(),
                peer_id,
            )?),
            false => None,
        };
        Ok(Self {
            stream: libp2p_stream::Behaviour::new(),
            identify: ::libp2p::identify::Behaviour::new(identify),
            relay_client,
            relay: relay.into(),
            dcutr: ::libp2p::dcutr::Behaviour::new(peer_id),
            autonat: ::libp2p::autonat::Behaviour::new(
                peer_id,
                ::libp2p::autonat::Config {
                    // Swarms on private networks (a LAN, WireGuard) count too.
                    only_global_ips: false,
                    ..Default::default()
                },
            ),
            mdns: mdns.into(),
            kad: kad.into(),
        })
    }
}

fn network_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
    Ok(::libp2p::SwarmBuilder::with_existing_identity(key)
        .with_tokio()
        .with_tcp(
            ::libp2p::tcp::Config::default(),
            ::libp2p::noise::Config::new,
            ::libp2p::yamux::Config::default,
        )?
        .with_quic()
        .with_dns()?
        .with_relay_client(
            ::libp2p::noise::Config::new,
            ::libp2p::yamux::Config::default,
        )?
        .with_behaviour(|key, relay_client| {
            Behaviour::new(key, relay_client, features)
                .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>)
        })?
        .with_swarm_config(|config| swarm_config(config, features))
        .build())
}

/// The swarm's settings. Streams are opened without waiting for the other side to agree
/// on the protocol (`V1Lazy`): the protocol and the first data go together, saving a
/// round trip per message (one stream each, module docs). That is safe only towards
/// peers sure to speak the protocol, so not with public libp2p infrastructure
/// (`Libp2pConfig::public`), whose nodes speak only some: there a protocol they don't
/// would fail late, and unclearly.
fn swarm_config(config: ::libp2p::swarm::Config, features: Features) -> ::libp2p::swarm::Config {
    let config = config.with_idle_connection_timeout(IDLE_TIMEOUT);
    if features.public {
        config
    } else {
        config.with_substream_upgrade_protocol_override(::libp2p::core::upgrade::Version::V1Lazy)
    }
}

fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
    use ::libp2p::Transport as _;
    use ::libp2p::core::transport::{MemoryTransport, OrTransport};
    use ::libp2p::core::upgrade::Version;

    let peer_id = key.public().to_peer_id();
    let (relay_transport, relay_client) = ::libp2p::relay::client::new(peer_id);
    let transport = OrTransport::new(relay_transport, MemoryTransport::default())
        .boxed()
        .upgrade(Version::V1)
        .authenticate(::libp2p::noise::Config::new(&key)?)
        .multiplex(::libp2p::yamux::Config::default())
        .timeout(CONNECT_TIMEOUT)
        .boxed();
    Ok(Swarm::new(
        transport,
        Behaviour::new(&key, relay_client, features)?,
        peer_id,
        swarm_config(::libp2p::swarm::Config::with_tokio_executor(), features),
    ))
}

/// How a node relates to relays.
struct Relaying {
    /// It relays for others.
    serving: bool,
    /// Public relays may be picked, after members.
    public: bool,
    local: PeerId,
}

/// The relays to pick among `offers`, if this node needs relays, keeping `using`: a
/// node needs them when it has no address others could dial (`direct`), or AutoNAT says
/// it can't be dialled; up to `AUTO_RELAYS`, in an order of its own (the lowest hash of
/// both peer IDs), so that not every node picks the same.
fn relay_check(
    local: PeerId,
    direct: bool,
    nat: &::libp2p::autonat::NatStatus,
    offers: &HashMap<PeerId, Multiaddr>,
    using: &[PeerId],
) -> Vec<(PeerId, Multiaddr)> {
    use ::libp2p::autonat::NatStatus;
    let needed = match nat {
        NatStatus::Private => true,
        NatStatus::Public(_) => false,
        NatStatus::Unknown => !direct,
    };
    if !needed || using.len() >= AUTO_RELAYS {
        return Vec::new();
    }
    let score = |relay: &PeerId| {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        (local, relay).hash(&mut hasher);
        hasher.finish()
    };
    let mut candidates: Vec<(PeerId, Multiaddr)> = offers
        .iter()
        .filter(|(relay, _)| **relay != local && !using.contains(relay))
        .map(|(relay, address)| (*relay, address.clone()))
        .collect();
    candidates.sort_by_key(|(relay, _)| score(relay));
    candidates.truncate(AUTO_RELAYS - using.len());
    candidates
}

/// Reserves slots on the relays `relay_check` picks (configured ones first, then listed
/// ones, members, public ones), and lets those picked go once AutoNAT says this node can
/// be dialled.
fn pick_relays(
    swarm: &mut Swarm<Behaviour>,
    book: &Mutex<Book>,
    relaying: &Relaying,
    nat: &::libp2p::autonat::NatStatus,
    picked: &mut HashMap<::libp2p::core::transport::ListenerId, PeerId>,
) {
    if nat.is_public() && !picked.is_empty() {
        info!("Reachable directly (AutoNAT); letting the relays picked go");
        let mut book = lock(book);
        for (listener, relay) in std::mem::take(picked) {
            swarm.remove_listener(listener);
            book.slots.remove(&relay);
        }
        return;
    }
    let (direct, configured, listed, members, public) = {
        let mut book = lock(book);
        book.failed_relays
            .retain(|_, at| at.elapsed() < RELAY_PAUSE);
        let failed = book.failed_relays.clone();
        let usable = |offers: &HashMap<PeerId, Multiaddr>| -> HashMap<PeerId, Multiaddr> {
            offers
                .iter()
                .filter(|(relay, _)| !failed.contains_key(relay))
                .map(|(relay, address)| (*relay, address.clone()))
                .collect()
        };
        let public = if relaying.public {
            usable(&book.public_relays)
        } else {
            HashMap::new()
        };
        (
            !book.direct().is_empty(),
            usable(&book.named(Named::Configured)),
            usable(&book.named(Named::Listed)),
            usable(&book.relays),
            public,
        )
    };
    let mut using: Vec<PeerId> = picked.values().copied().collect();
    // Each kind for what the ones before left.
    for (kind, offers) in [
        (Named::Configured.as_str(), configured),
        (Named::Listed.as_str(), listed),
        ("member", members),
        ("public", public),
    ] {
        for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) {
            using.push(relay);
            match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
                Ok(listener) => {
                    info!(relay = %Addressed(SCHEME, &relay), address = %address, kind, "Picked a relay");
                    picked.insert(listener, relay);
                }
                Err(error) => {
                    debug!(relay = %Addressed(SCHEME, &relay), kind, error = %error, "Couldn't use a relay");
                }
            }
        }
    }
}

/// Asks the DHT (with `--libp2p-public`) where `peer` is, for the next attempt.
fn find_on_dht(swarm: &mut Swarm<Behaviour>, peer: PeerId) {
    if let Some(dht) = swarm.behaviour_mut().kad.as_mut() {
        dht.get_closest_peers(peer);
    }
}

/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
/// transport is dropped. `again`: to schedule commands.
async fn run(
    mut swarm: Swarm<Behaviour>,
    mut commands: mpsc::UnboundedReceiver<Command>,
    again: mpsc::UnboundedSender<Command>,
    book: Arc<Mutex<Book>>,
    relaying: Relaying,
    stopping: watch::Sender<bool>,
) {
    let serving = relaying.serving;
    let mut nat = ::libp2p::autonat::NatStatus::Unknown;
    // Relays picked, by listener.
    let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new();
    let mut check = tokio::time::interval(RELAY_CHECK);
    check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP);
    dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut connecting: HashMap<PeerId, Vec<oneshot::Sender<Result<(), String>>>> = HashMap::new();
    let mut identifying: HashMap<
        ConnectionId,
        (Multiaddr, oneshot::Sender<Result<PeerId, String>>),
    > = HashMap::new();
    let mut prune = tokio::time::interval(PRUNE_EVERY);
    prune.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    // Peers that may have several connections, and when to cut them to one. Only the
    // side with the smaller peer ID does that, so the two never close different ones.
    let mut dedupe_due: HashMap<PeerId, tokio::time::Instant> = HashMap::new();
    // Connections closed as duplicates, until they are: their "Connection closed" says so.
    let mut closing_duplicates: std::collections::HashSet<ConnectionId> =
        std::collections::HashSet::new();
    let mut dedupe = tokio::time::interval(DEDUPE_EVERY);
    dedupe.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = dedupe.tick(), if !dedupe_due.is_empty() => {
                let now = tokio::time::Instant::now();
                let due: Vec<PeerId> = dedupe_due
                    .iter()
                    .filter(|(_, at)| **at <= now)
                    .map(|(peer, _)| *peer)
                    .collect();
                for peer in due {
                    dedupe_due.remove(&peer);
                    let reserving = picked.values().any(|relay| *relay == peer);
                    let duplicates = lock(&book).duplicates(&peer, reserving);
                    for connection in duplicates {
                        if swarm.close_connection(connection) {
                            closing_duplicates.insert(connection);
                        }
                    }
                }
            }
            _ = prune.tick() => {
                let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer));
                if forgotten > 0 {
                    debug!(forgotten, "Forgot peers nobody mentions any more");
                }
            }
            _ = dht_bootstrap.tick(), if relaying.public => {
                if let Some(dht) = swarm.behaviour_mut().kad.as_mut()
                    && let Err(error) = dht.bootstrap()
                {
                    warn!(error = %error, "Couldn't join the IPFS DHT");
                }
            }
            _ = check.tick() => {
                pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
            }
            command = commands.recv() => match command {
                None => break,
                Some(Command::CheckRelays) => {
                    pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
                }
                Some(Command::Connect { peer, addresses, done }) => {
                    if swarm.is_connected(&peer) {
                        let _ = done.send(Ok(()));
                        continue;
                    }
                    let waiting = connecting.entry(peer).or_default();
                    waiting.push(done);
                    if waiting.len() > 1 {
                        continue; // Already dialing.
                    }
                    let mut opts = DialOpts::peer_id(peer)
                        .addresses(addresses)
                        .condition(PeerCondition::DisconnectedAndNotDialing);
                    if relaying.public {
                        // Addresses the DHT knows too.
                        opts = opts.extend_addresses_through_behaviour();
                    }
                    if let Err(error) = swarm.dial(opts.build()) {
                        count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                        for done in connecting.remove(&peer).into_iter().flatten() {
                            let _ = done.send(Err(error.to_string()));
                        }
                        find_on_dht(&mut swarm, peer);
                    }
                }
                Some(Command::FindRelay { address }) => {
                    let again = again.clone();
                    let local = relaying.local;
                    let book = Arc::clone(&book);
                    tokio::spawn(async move {
                        let (done, identified) = oneshot::channel();
                        if again
                            .send(Command::Identify { address: address.clone(), done })
                            .is_err()
                        {
                            return;
                        }
                        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
                            Ok(Ok(Ok(peer))) => {
                                if lock(&book).relay_found(address.clone(), peer, local) {
                                    debug!(relay = %Addressed(SCHEME, &address), peer = %Addressed(SCHEME, &peer), "Found a named relay");
                                    let _ = again.send(Command::CheckRelays);
                                }
                            }
                            _ => {
                                if !lock(&book).finding.contains_key(&address) {
                                    return; // No longer named.
                                }
                                warn!(relay = %Addressed(SCHEME, &address), "Couldn't reach a named relay; trying again later");
                                tokio::time::sleep(RELAY_RETRY).await;
                                let _ = again.send(Command::FindRelay { address });
                            }
                        }
                    });
                }
                Some(Command::Identify { address, done }) => {
                    let opts = DialOpts::unknown_peer_id().address(address.clone()).build();
                    let id = opts.connection_id();
                    match swarm.dial(opts) {
                        Ok(()) => {
                            identifying.insert(id, (address, done));
                        }
                        Err(error) => {
                            let _ = done.send(Err(error.to_string()));
                        }
                    }
                }
            },
            event = swarm.select_next_some() => match event {
                SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => {
                    {
                        let mut book = lock(&book);
                        book.touch(peer_id);
                        let relayed = endpoint.is_relayed();
                        book.connections.insert(
                            connection_id,
                            (peer_id, relayed, endpoint.get_remote_address().clone()),
                        );
                        if relaying.local < peer_id
                            && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1
                        {
                            dedupe_due.insert(peer_id, tokio::time::Instant::now() + DUPLICATE_GRACE);
                        }
                        debug!(
                            peer = %Addressed(SCHEME, &peer_id),
                            connection = %connection_id,
                            dialled = endpoint.is_dialer(),
                            address = %endpoint.get_remote_address(),
                            relayed,
                            "Connection opened"
                        );
                    }
                    for done in connecting.remove(&peer_id).into_iter().flatten() {
                        let _ = done.send(Ok(()));
                    }
                    if let Some((address, done)) = identifying.remove(&connection_id) {
                        lock(&book).remember(peer_id, address);
                        let _ = done.send(Ok(peer_id));
                    }
                }
                SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => {
                    lock(&book).connections.remove(&connection_id);
                    let cause = match cause {
                        _ if closing_duplicates.remove(&connection_id) => "a duplicate".to_owned(),
                        // By us, or idle: libp2p doesn't tell which.
                        None => "closed here".to_owned(),
                        Some(error) => error.to_string(),
                    };
                    debug!(
                        peer = %Addressed(SCHEME, &peer_id),
                        connection = %connection_id,
                        cause = ?cause,
                        "Connection closed"
                    );
                }
                SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
                    count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                    if let Some((_, done)) = identifying.remove(&connection_id) {
                        let _ = done.send(Err(describe_dial_error(&error)));
                    }
                    if let Some(peer) = peer_id
                        && !swarm.is_connected(&peer)
                    {
                        if let Some(waiting) = connecting.remove(&peer) {
                            for done in waiting {
                                let _ = done.send(Err(describe_dial_error(&error)));
                            }
                            find_on_dht(&mut swarm, peer);
                        }
                    }
                }
                SwarmEvent::NewListenAddr { address, .. } => {
                    info!(address = %address, "Listening");
                    // A relay tells those reserving where it is: its external addresses.
                    if serving && !is_local(&address) && !is_relayed(&address) {
                        swarm.add_external_address(address.clone());
                    }
                    let mut book = lock(&book);
                    if !book.listening.contains(&address) {
                        book.listening.push(address);
                    }
                }
                SwarmEvent::ExpiredListenAddr { address, .. } => {
                    info!(address = %address, "No longer listening");
                    if serving {
                        swarm.remove_external_address(&address);
                    }
                    lock(&book).listening.retain(|known| *known != address);
                }
                SwarmEvent::ListenerClosed { listener_id, addresses, reason } => {
                    lock(&book)
                        .listening
                        .retain(|known| !addresses.contains(known));
                    // A relay picked: the next check picks again, another one if this one
                    // never granted the slot.
                    if let Some(relay) = picked.remove(&listener_id) {
                        let lost = {
                            let mut book = lock(&book);
                            let lost = book.slots.remove(&relay);
                            if lost {
                                book.counts.slots_lost += 1;
                            } else {
                                book.counts.slots_not_granted += 1;
                                book.failed_relays.insert(relay, tokio::time::Instant::now());
                                // Named without its peer ID: maybe another one now.
                                if let Some(address) = book.find_again(relay) {
                                    let _ = again.send(Command::FindRelay { address });
                                }
                            }
                            lost
                        };
                        let reason = relay_failure(&reason);
                        if lost {
                            debug!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay");
                        } else {
                            info!(
                                relay = %Addressed(SCHEME, &relay),
                                pause = ?RELAY_PAUSE,
                                reason = %reason,
                                "A relay didn't grant a slot; picking others for a while"
                            );
                        }
                    }
                }
                SwarmEvent::ListenerError { error, .. } => {
                    warn!(error = %error, "Listener failed");
                }
                SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                    ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. },
                )) => {
                    {
                        let mut book = lock(&book);
                        book.slots.insert(relay_peer_id);
                        book.failed_relays.remove(&relay_peer_id);
                        if !renewal {
                            book.counts.slots_granted += 1;
                        }
                    }
                    if !renewal {
                        info!(relay = %Addressed(SCHEME, &relay_peer_id), "Reachable through a relay");
                    }
                }
                SwarmEvent::Behaviour(BehaviourEvent::Autonat(
                    ::libp2p::autonat::Event::StatusChanged { old, new },
                )) => {
                    info!(old = ?old, new = ?new, "Reachability (AutoNAT)");
                    {
                        use ::libp2p::autonat::NatStatus;
                        let mut book = lock(&book);
                        (book.reachability, book.public_address) = match &new {
                            NatStatus::Unknown => (Reachability::Unknown, None),
                            NatStatus::Private => (Reachability::Private, None),
                            NatStatus::Public(address) => {
                                (Reachability::Public, Some(address.clone()))
                            }
                        };
                    }
                    nat = new;
                }
                SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
                    {
                        use ::libp2p::relay::Event;
                        let mut book = lock(&book);
                        match &event {
                            Event::ReservationReqAccepted { src_peer_id, .. } => {
                                book.slots_here.insert(*src_peer_id);
                            }
                            Event::ReservationClosed { src_peer_id }
                            | Event::ReservationTimedOut { src_peer_id } => {
                                book.slots_here.remove(src_peer_id);
                            }
                            _ => {}
                        }
                    }
                    debug!(event = ?event, "Relay event");
                }
                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                    Ok(_) => {
                        lock(&book).counts.hole_punches.succeeded += 1;
                        info!(peer = %Addressed(SCHEME, &event.remote_peer_id), "Connected directly to a peer behind a relay");
                    }
                    Err(error) => {
                        lock(&book).counts.hole_punches.failed += 1;
                        debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Identify(
                    ::libp2p::identify::Event::Received { connection_id, peer_id, info },
                )) => {
                    let mut book = lock(&book);
                    let listen = published(&info.listen_addrs);
                    // Every new connection identifies again: logged when it says something new.
                    if book.knows(peer_id, &listen) {
                        trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again");
                    } else {
                        debug!(
                            peer = %Addressed(SCHEME, &peer_id),
                            connection = %connection_id,
                            listen = ?info.listen_addrs,
                            observed = %info.observed_addr,
                            agent = %info.agent_version,
                            "Peer identified"
                        );
                    }
                    // Last first: `remember` puts each in front.
                    for address in listen.into_iter().rev() {
                        book.remember(peer_id, address);
                    }
                    if relaying.public
                        && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME)
                        && (book.public_relays.len() < MAX_PEERS
                            || book.public_relays.contains_key(&peer_id))
                        && let Some(address) = book
                            .addresses
                            .get(&peer_id)
                            .and_then(|known| known.first())
                            .and_then(|address| address.clone().with_p2p(peer_id).ok())
                    {
                        book.public_relays.insert(peer_id, address);
                    }
                    if info.protocols.contains(&PROTOCOL) {
                        book.continuum.insert(peer_id);
                        if book.lan.remove(&peer_id) {
                            book.discover(peer_id);
                        }
                    }
                }
                SwarmEvent::Behaviour(BehaviourEvent::Mdns(
                    ::libp2p::mdns::Event::Discovered(found),
                )) => {
                    for (peer, address) in found {
                        if peer == relaying.local {
                            continue;
                        }
                        debug!(peer = %Addressed(SCHEME, &peer), address = %address, "Peer on the LAN");
                        let continuum = {
                            let mut book = lock(&book);
                            book.remember(peer, address.clone());
                            book.continuum.contains(&peer)
                        };
                        if continuum {
                            lock(&book).discover(peer);
                        } else if lock(&book).lan.len() < MAX_PEERS {
                            // Connected, identify says whether it speaks continuum.
                            lock(&book).lan.insert(peer);
                            let _ = swarm.dial(
                                DialOpts::peer_id(peer)
                                    .addresses(vec![address])
                                    .condition(PeerCondition::DisconnectedAndNotDialing)
                                    .build(),
                            );
                        }
                    }
                }
                _ => {}
            },
        }
    }
    let _ = stopping.send(true);
}

/// One message from an inbound stream.
async fn read_message(mut stream: ::libp2p::Stream) -> Result<Vec<u8>> {
    let mut length = [0u8; 4];
    tokio::time::timeout(STREAM_TIMEOUT, stream.read_exact(&mut length))
        .await
        .map_err(|_| anyhow!("no length within {STREAM_TIMEOUT:?}"))??;
    let length = u32::from_be_bytes(length) as usize;
    if length > MAX_MESSAGE {
        bail!("a message of {length} bytes; at most {MAX_MESSAGE} are read");
    }
    let mut body = vec![0u8; length];
    tokio::time::timeout(stream_timeout(length), stream.read_exact(&mut body))
        .await
        .map_err(|_| {
            anyhow!(
                "{length} bytes took longer than {:?}",
                stream_timeout(length)
            )
        })??;
    Ok(body)
}

#[async_trait]
impl Transport for Libp2pTransport {
    fn local_id(&self) -> &str {
        &self.local_id
    }

    fn is_peer_id(&self, id: &str) -> bool {
        PeerId::from_str(id).is_ok()
    }

    /// The longest a message may take on a stream.
    fn delivery_limit(&self) -> Duration {
        stream_timeout(MAX_MESSAGE)
    }

    /// Where we can be reached, each with our peer ID: the announced addresses, or else
    /// those we listen on (`published`), and those through relays. A relay adds each
    /// direct one again, ending in `/p2p-circuit`: where to reserve a slot.
    fn local_addresses(&self) -> Vec<String> {
        let book = lock(&self.book);
        let direct: Vec<Multiaddr> = book
            .direct()
            .into_iter()
            .filter_map(|address| address.with_p2p(self.peer_id).ok())
            .collect();
        let relayed = book
            .listening
            .iter()
            .filter(|address| is_relayed(address))
            .filter_map(|address| address.clone().with_p2p(self.peer_id).ok());
        let offers: Vec<Multiaddr> = if book.serving {
            direct
                .iter()
                .map(|address| address.clone().with(Protocol::P2pCircuit))
                .collect()
        } else {
            Vec::new()
        };
        direct
            .into_iter()
            .chain(relayed)
            .chain(offers)
            .map(|address| address.to_string())
            .collect()
    }

    /// A peer ID, or a multiaddr ending in one (noted as where that peer is).
    fn locate(&self, target: &str) -> Option<String> {
        if let Ok(peer) = PeerId::from_str(target) {
            return Some(peer.to_string());
        }
        let mut address = Multiaddr::from_str(target).ok()?;
        // A relay saying where to reserve a slot: `<address>/p2p/<relay>/p2p-circuit`.
        if let Some(Protocol::P2pCircuit) = address.iter().last() {
            address.pop();
            let relay = split_peer(&address)?;
            if relay != self.peer_id {
                let new = {
                    let mut book = lock(&self.book);
                    let new = book.relays.get(&relay) != Some(&address);
                    if new && (book.relays.len() < MAX_PEERS || book.relays.contains_key(&relay)) {
                        book.relays.insert(relay, address.clone());
                    }
                    book.remember(relay, address);
                    new
                };
                if new {
                    let _ = self.commands.send(Command::CheckRelays);
                }
            }
            return Some(relay.to_string());
        }
        let peer = split_peer(&address)?;
        if peer != self.peer_id {
            lock(&self.book).remember(peer, address);
        }
        Some(peer.to_string())
    }

    /// Multiaddrs, with or without the relay's peer ID; others are ignored.
    fn learn_relays(&self, targets: &[String]) {
        let listed = targets
            .iter()
            .filter_map(|target| Multiaddr::from_str(target).ok())
            .collect();
        let Some(found) = lock(&self.book).learn_listed(listed, self.peer_id) else {
            return;
        };
        for address in found {
            let _ = self.commands.send(Command::FindRelay { address });
        }
        let _ = self.commands.send(Command::CheckRelays);
    }

    fn connection(&self, target: &str) -> Option<Connection> {
        let peer = PeerId::from_str(target).ok()?;
        lock(&self.book).connection(&peer)
    }

    fn location(&self, target: &str) -> Option<String> {
        let peer = PeerId::from_str(target).ok()?;
        let address = lock(&self.book).location(&peer)?.clone();
        Some(
            address
                .with_p2p(peer)
                .unwrap_or_else(|address| address)
                .to_string(),
        )
    }

    fn stats(&self) -> TransportStats {
        TransportStats {
            libp2p: Some(lock(&self.book).stats()),
            ..TransportStats::default()
        }
    }

    /// Nodes found on the LAN (mDNS) that speak continuum.
    fn discovered_seeds(&self) -> Vec<String> {
        std::mem::take(&mut lock(&self.book).discovered)
            .iter()
            .map(Multiaddr::to_string)
            .collect()
    }

    /// Where `target` is, each address with its peer ID.
    fn locations(&self, target: &str) -> Vec<String> {
        let Ok(peer) = PeerId::from_str(target) else {
            return Vec::new();
        };
        lock(&self.book)
            .addresses
            .get(&peer)
            .into_iter()
            .flatten()
            .filter_map(|address| address.clone().with_p2p(peer).ok())
            .map(|address| address.to_string())
            .collect()
    }

    /// A multiaddr ending in a peer ID is that peer; one without is connected to, to
    /// find out. A bare peer ID resolves if this transport knows where it is.
    async fn resolve_seed(&self, seed: &str) -> Result<String> {
        if let Ok(peer) = PeerId::from_str(seed) {
            if lock(&self.book).addresses.contains_key(&peer) {
                return Ok(peer.to_string());
            }
            bail!("no address known for libp2p peer {peer}; give a multiaddr");
        }
        let address =
            Multiaddr::from_str(seed).with_context(|| format!("{seed} is no multiaddr"))?;
        if let Some(peer) = self.locate(seed) {
            return Ok(peer);
        }
        let (done, identified) = oneshot::channel();
        self.commands
            .send(Command::Identify { address, done })
            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
            Ok(Ok(Ok(peer))) => Ok(peer.to_string()),
            Ok(Ok(Err(error))) => bail!("couldn't connect to {seed}: {error}"),
            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
            Err(_) => bail!("connecting to {seed} took longer than {CONNECT_TIMEOUT:?}"),
        }
    }

    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
        let body = wire::encode(message)?;
        Ok(EncodedMessage {
            peer: peer.to_owned(),
            address: peer.to_owned(),
            payload_len: body.len(),
            body: body.into(),
            kind: message.kind(),
            message: None,
        })
    }

    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
        let peer = PeerId::from_str(&message.peer)
            .with_context(|| format!("{} is no libp2p peer ID", message.peer))?;
        let started = std::time::Instant::now();
        self.connect(peer).await?;
        let connected = started.elapsed();
        let mut control = self.control.clone();
        let body = message.body.clone();
        let write = async move {
            let mut stream = control.open_stream(peer, PROTOCOL).await?;
            let opened = started.elapsed();
            stream.write_all(&(body.len() as u32).to_be_bytes()).await?;
            stream.write_all(&body).await?;
            stream.close().await?;
            Ok::<_, anyhow::Error>(opened)
        };
        let timeout = stream_timeout(message.body.len());
        let opened = tokio::time::timeout(timeout, write)
            .await
            .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))?
            .with_context(|| format!("Sending to {peer}"))?;
        let took = started.elapsed();
        if took >= SLOW_SEND {
            // Where a send's time goes: the swarm task (connecting), the stream's protocol
            // negotiation (opening), or the connection (writing).
            debug!(
                peer = %Addressed(SCHEME, &peer),
                kind = ?message.kind,
                bytes = message.body.len(),
                connect_ms = connected.as_secs_f64() * 1000.0,
                open_ms = (opened - connected).as_secs_f64() * 1000.0,
                write_ms = (took - opened).as_secs_f64() * 1000.0,
                "Slow send"
            );
        }
        Ok(())
    }

    /// Reads inbound streams (`READS` at once) and handles their messages one at a
    /// time, in the order they were read; PINGs and PONGs at once (`is_probe`).
    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
        let Some(mut incoming) = self.incoming.lock().await.take() else {
            bail!("libp2p consumer already started, or this node only relays");
        };
        let handler: Arc<HandlerFn> = Arc::from(handler);
        let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS);
        let reads = Arc::new(Semaphore::new(READS));
        let probes = Arc::clone(&handler);
        tokio::spawn(async move {
            while let Some((peer, stream)) = incoming.next().await {
                let Ok(permit) = Arc::clone(&reads).acquire_owned().await else {
                    break;
                };
                let sender = sender.clone();
                let probes = Arc::clone(&probes);
                tokio::spawn(async move {
                    match read_message(stream).await {
                        Ok(body) => match wire::decode(&body) {
                            Ok(message) if is_probe(&message) => {
                                hand_on_probe(
                                    SCHEME,
                                    &probes,
                                    peer.to_string(),
                                    message,
                                    body.len(),
                                );
                            }
                            Ok(message) => {
                                let _ = sender.send((peer, message, body.len())).await;
                            }
                            Err(error) => {
                                debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Dropping a message that doesn't decode");
                            }
                        },
                        Err(error) => {
                            debug!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Dropping an unreadable stream");
                        }
                    }
                    drop(permit);
                });
            }
        });
        while let Some((peer, message, payload_len)) = received.recv().await {
            if let Err(error) = handler(peer.to_string(), message, payload_len).await {
                warn!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Failed to process inbound message");
            }
        }
        Ok(())
    }

    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
        let mut stopped = self.stopped.clone();
        let _ = stopped.wait_for(|stopped| *stopped).await;
        Err(std::io::Error::other("the libp2p swarm stopped"))
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicU64, Ordering};

    use super::*;
    use crate::document::replica::Epoch;
    use crate::transport::{HeadsSummary, Status};

    /// Memory ports are process-wide.
    static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000);

    /// A node on the in-memory transport, listening on a port of its own (or nowhere),
    /// and its address with its peer ID (or just its peer ID).
    fn memory_node_with(
        listen: bool,
        relays: Vec<Multiaddr>,
        relay: bool,
    ) -> (Libp2pTransport, Multiaddr) {
        let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
        let key = Keypair::generate_ed25519();
        let address: Multiaddr = format!("/memory/{port}").parse().unwrap();
        let peer = key.public().to_peer_id();
        let transport = Libp2pTransport::start(Libp2pConfig {
            key,
            listen: if listen {
                vec![address.clone()]
            } else {
                Vec::new()
            },
            announce: Vec::new(),
            relays,
            relay,
            continuum: true,
            mdns: false,
            public: false,
            memory: true,
        })
        .unwrap();
        let address = if listen {
            address.with_p2p(peer).unwrap()
        } else {
            Multiaddr::empty().with(Protocol::P2p(peer))
        };
        (transport, address)
    }

    fn memory_node() -> (Libp2pTransport, Multiaddr) {
        memory_node_with(true, Vec::new(), false)
    }

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

    /// Starts `transport`'s consumer, forwarding what it receives.
    fn received(
        transport: Arc<Libp2pTransport>,
    ) -> mpsc::UnboundedReceiver<(String, ParsedMessage)> {
        let (sender, receiver) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            transport
                .start_consumer(Box::new(move |from, message, _| {
                    let sender = sender.clone();
                    Box::pin(async move {
                        let _ = sender.send((from, message));
                        Ok(())
                    })
                }))
                .await
        });
        receiver
    }

    #[tokio::test]
    async fn messages_both_ways_over_one_connection() {
        let (a, a_address) = memory_node();
        let (b, _) = memory_node();
        let (a, b) = (Arc::new(a), Arc::new(b));
        let mut at_a = received(Arc::clone(&a));
        let mut at_b = received(Arc::clone(&b));

        assert_eq!(
            b.resolve_seed(&a_address.to_string()).await.unwrap(),
            a.local_id()
        );
        b.send_message(a.local_id(), &status()).await.unwrap();
        let (from, message) = at_a.recv().await.unwrap();
        assert_eq!(from, b.local_id());
        assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3));

        // a knows no address of b, but b's connection is there.
        a.send_message(b.local_id(), &status()).await.unwrap();
        let (from, _) = at_b.recv().await.unwrap();
        assert_eq!(from, a.local_id());

        assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
        assert_eq!(b.connection(a.local_id()), Some(Connection::Direct));
        let stats = a.stats().libp2p.unwrap();
        assert!(stats.connections.direct >= 1);
        assert_eq!(stats.connections.relayed, 0);
    }

    #[tokio::test]
    async fn a_seed_without_a_peer_id_is_connected_to_to_find_out() {
        let (a, a_address) = memory_node();
        let (b, _) = memory_node();
        let mut bare = a_address.clone();
        bare.pop();
        assert_eq!(
            b.resolve_seed(&bare.to_string()).await.unwrap(),
            a.local_id()
        );
        // And b now knows where a is.
        assert_eq!(b.locations(a.local_id()), [a_address.to_string()]);
    }

    #[tokio::test]
    async fn listen_addresses_carry_the_peer_id() {
        let (a, a_address) = memory_node();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while a.local_addresses().is_empty() && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(a.local_addresses(), [a_address.to_string()]);
        assert_eq!(
            a.locate(&a_address.to_string()).as_deref(),
            Some(a.local_id())
        );
    }

    /// Waits up to 5 s for `condition`.
    async fn eventually(mut condition: impl FnMut() -> bool) -> bool {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while !condition() {
            if tokio::time::Instant::now() >= deadline {
                return false;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        true
    }

    /// A node learns where a peer that dialled it listens (identify), and can then
    /// pass it on.
    #[tokio::test]
    async fn a_dialled_node_learns_where_the_dialler_listens() {
        let (a, a_address) = memory_node();
        let (b, b_address) = memory_node();
        let (a, b) = (Arc::new(a), Arc::new(b));
        let _at_a = received(Arc::clone(&a));
        b.resolve_seed(&a_address.to_string()).await.unwrap();
        b.send_message(a.local_id(), &status()).await.unwrap();
        let b_id = b.local_id().to_owned();
        assert!(
            eventually(|| a.locations(&b_id) == [b_address.to_string()]).await,
            "a knows {:?} of b",
            a.locations(&b_id)
        );
    }

    #[tokio::test]
    async fn announced_addresses_are_published_instead() {
        let key = Keypair::generate_ed25519();
        let peer = key.public().to_peer_id();
        let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
        let announced: Multiaddr = "/ip4/203.0.113.7/tcp/9661".parse().unwrap();
        let transport = Libp2pTransport::start(Libp2pConfig {
            key,
            listen: vec![format!("/memory/{port}").parse().unwrap()],
            announce: vec![announced.clone()],
            relays: Vec::new(),
            relay: false,
            continuum: true,
            mdns: false,
            public: false,
            memory: true,
        })
        .unwrap();
        assert_eq!(
            transport.local_addresses(),
            [announced.with_p2p(peer).unwrap().to_string()]
        );
    }

    #[test]
    fn local_addresses_are_left_out_unless_there_are_only_local_ones() {
        let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
        let loopback = parse("/ip4/127.0.0.1/tcp/1");
        let link = parse("/ip6/fe80::1/tcp/1");
        let lan = parse("/ip4/192.168.1.2/tcp/1");
        let wireguard = parse("/ip6/fd00::2/tcp/1");
        assert_eq!(
            published(&[
                loopback.clone(),
                link.clone(),
                lan.clone(),
                wireguard.clone()
            ]),
            [lan, wireguard]
        );
        assert_eq!(
            published(&[loopback.clone(), link.clone()]),
            [loopback, link]
        );
    }

    /// b listens nowhere but through the relay r; a reaches it there.
    #[tokio::test]
    async fn a_node_nobody_can_dial_is_reached_through_a_relay() {
        let (_r, r_address) = memory_node_with(true, Vec::new(), true);
        let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
        let b = Arc::new(b);
        let mut at_b = received(Arc::clone(&b));
        assert!(
            eventually(|| !b.local_addresses().is_empty()).await,
            "b got no slot on the relay"
        );
        let relayed = b.local_addresses()[0].clone();
        assert!(
            relayed.starts_with(&format!("{r_address}/p2p-circuit/")),
            "{relayed}"
        );

        assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await);
        let slots = b.stats().libp2p.unwrap().relay_slots;
        assert_eq!((slots.granted, slots.not_granted), (1, 0));

        let (a, _) = memory_node();
        assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
        a.send_message(b.local_id(), &status()).await.unwrap();
        let (from, _) = at_b.recv().await.unwrap();
        assert_eq!(from, a.local_id());
        assert!(a.connection(b.local_id()).is_some());
    }

    /// A relay that is no node (`--relay-only`) relays, and takes no messages itself.
    #[tokio::test]
    async fn a_relay_that_is_no_node() {
        let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
        let key = Keypair::generate_ed25519();
        let r_peer = key.public().to_peer_id();
        let r_listen: Multiaddr = format!("/memory/{port}").parse().unwrap();
        let r = Libp2pTransport::start(Libp2pConfig {
            key,
            listen: vec![r_listen.clone()],
            announce: Vec::new(),
            relays: Vec::new(),
            relay: true,
            continuum: false,
            mdns: false,
            public: false,
            memory: true,
        })
        .unwrap();
        let r_address = r_listen.with_p2p(r_peer).unwrap();
        let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
        let b = Arc::new(b);
        let mut at_b = received(Arc::clone(&b));
        assert!(
            eventually(|| !b.local_addresses().is_empty()).await,
            "b got no slot on the relay"
        );
        let (a, _) = memory_node();
        let relayed = b.local_addresses()[0].clone();
        assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
        a.send_message(b.local_id(), &status()).await.unwrap();
        let (from, _) = at_b.recv().await.unwrap();
        assert_eq!(from, a.local_id());

        // The relay itself doesn't speak continuum: it takes no messages (a send to it
        // isn't refused, though: the protocol goes with the first data, `swarm_config`,
        // so the relay's refusal comes too late; nodes never send to it, as it never
        // proves itself a node).
        assert_eq!(
            a.resolve_seed(&r_address.to_string()).await.unwrap(),
            r.local_id()
        );
        assert!(
            r.start_consumer(Box::new(|_, _, _| Box::pin(async { Ok(()) })))
                .await
                .is_err()
        );
    }

    /// A relay configured without its peer ID is found out, and used.
    #[tokio::test]
    async fn a_relay_configured_without_its_peer_id() {
        let (_r, r_address) = memory_node_with(true, Vec::new(), true);
        let mut bare = r_address.clone();
        bare.pop();
        let (b, _) = memory_node_with(false, vec![bare], false);
        assert!(
            eventually(|| !b.local_addresses().is_empty()).await,
            "b got no slot on the relay"
        );
        assert!(b.local_addresses()[0].starts_with(&format!("{r_address}/p2p-circuit/")));
    }

    fn peer() -> PeerId {
        Keypair::generate_ed25519().public().to_peer_id()
    }

    #[test]
    fn relays_are_picked_when_needed() {
        use ::libp2p::autonat::NatStatus;
        let local = peer();
        let offers: HashMap<PeerId, Multiaddr> = (0..4)
            .map(|port| (peer(), format!("/memory/{port}").parse().unwrap()))
            .collect();
        let check = |direct, nat: &NatStatus, using: &[PeerId]| {
            relay_check(local, direct, nat, &offers, using)
        };
        // Dialable, or not known yet with an address to dial: none.
        assert!(check(true, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
        assert!(check(false, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
        assert!(check(true, &NatStatus::Unknown, &[]).is_empty());
        // No address to dial, or AutoNAT says it can't be dialled: some.
        assert_eq!(check(false, &NatStatus::Unknown, &[]).len(), AUTO_RELAYS);
        let picked = check(true, &NatStatus::Private, &[]);
        assert_eq!(picked.len(), AUTO_RELAYS);
        // One in use: the next in the same order.
        let more = check(true, &NatStatus::Private, &[picked[0].0]);
        assert_eq!(more.len(), AUTO_RELAYS - 1);
        assert_eq!(more[0].0, picked[1].0);
        let all: Vec<PeerId> = picked.iter().map(|(relay, _)| *relay).collect();
        assert!(check(true, &NatStatus::Private, &all).is_empty());
        // Never itself.
        let only_us = HashMap::from([(local, Multiaddr::empty())]);
        assert!(relay_check(local, false, &NatStatus::Private, &only_us, &[]).is_empty());
    }

    /// A relay says so in its addresses; b, which nobody can dial, learns of it (as from
    /// the member list) and becomes reachable through it.
    #[tokio::test]
    async fn a_member_that_relays_is_picked() {
        let (r, _) = memory_node_with(true, Vec::new(), true);
        assert!(eventually(|| !r.local_addresses().is_empty()).await);
        let offer = r
            .local_addresses()
            .into_iter()
            .find(|address| address.ends_with("/p2p-circuit"))
            .expect("the relay offers itself");
        let (b, _) = memory_node_with(false, Vec::new(), false);
        assert_eq!(b.locate(&offer).as_deref(), Some(r.local_id()));
        assert!(
            eventually(|| b
                .local_addresses()
                .iter()
                .any(|address| address.contains("/p2p-circuit/")))
            .await,
            "b isn't reachable through r: {:?}",
            b.local_addresses()
        );
    }

    /// A peer on the LAN that speaks continuum becomes a seed once, at its address.
    #[test]
    fn lan_peers_become_seeds_once() {
        let mut book = Book::default();
        let lan = peer();
        let address: Multiaddr = "/ip4/192.168.1.5/tcp/9661".parse().unwrap();
        book.discover(lan); // No address known: nothing.
        assert!(book.discovered.is_empty());
        book.remember(lan, address.clone());
        book.discover(lan);
        book.discover(lan);
        assert_eq!(book.discovered, [address.with_p2p(lan).unwrap()]);
    }

    #[test]
    fn named_relays() {
        let (local, r1, r2) = (peer(), peer(), peer());
        let at = |port: u16| -> Multiaddr { format!("/ip6/fd00::1/tcp/{port}").parse().unwrap() };
        let mut book = Book::default();
        // With a peer ID: known at once; without: to be found out, once.
        assert_eq!(
            book.name_relay(at(1).with_p2p(r1).unwrap(), Named::Configured, local),
            None
        );
        assert_eq!(
            book.name_relay(at(2), Named::Configured, local),
            Some(at(2))
        );
        assert_eq!(book.name_relay(at(2), Named::Configured, local), None);
        assert_eq!(
            book.name_relay(at(3).with_p2p(local).unwrap(), Named::Configured, local),
            None
        );
        assert!(book.relay_found(at(2), r2, local));
        assert_eq!(book.named(Named::Configured).len(), 2);
        // A relay named with its peer ID isn't found out again; one without is.
        assert_eq!(book.find_again(r1), None);
        assert_eq!(book.find_again(r2), Some(at(2)));
        assert!(!book.named(Named::Configured).contains_key(&r2));

        // Listed: configured wins; unlisted ones go, unless also configured.
        let r3 = peer();
        let listed = vec![
            at(1).with_p2p(r1).unwrap(),
            at(4).with_p2p(r3).unwrap(),
            at(5),
        ];
        assert_eq!(book.learn_listed(listed.clone(), local), Some(vec![at(5)]));
        assert_eq!(book.learn_listed(listed, local), None);
        assert_eq!(book.named_relays[&r1].named, Named::Configured);
        assert_eq!(book.named(Named::Listed).keys().collect::<Vec<_>>(), [&r3]);
        assert_eq!(book.learn_listed(Vec::new(), local), Some(Vec::new()));
        assert!(book.named(Named::Listed).is_empty());
        assert!(!book.finding.contains_key(&at(5)));
        assert!(!book.relay_found(at(5), peer(), local)); // No longer listed.
        assert_eq!(book.named(Named::Configured).len(), 1);
    }

    #[test]
    fn known_addresses() {
        let mut book = Book::default();
        let (relay, us) = (peer(), peer());
        let direct: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap();
        let relayed: Multiaddr = format!("/ip6/fd00::2/tcp/9661/p2p/{relay}/p2p-circuit/p2p/{us}")
            .parse()
            .unwrap();
        assert!(!book.knows(us, &[direct.clone()]));
        book.remember(us, relayed.clone());
        book.remember(us, direct.clone());
        // In any order, with or without the peer ID.
        assert!(book.knows(us, &[relayed.clone(), direct.clone()]));
        assert!(book.knows(us, &[direct.clone().with_p2p(us).unwrap()]));
        assert!(!book.knows(us, &["/ip6/fd00::3/tcp/9661".parse().unwrap()]));
    }

    /// Peers nobody mentions any more are forgotten, but connected ones.
    #[test]
    fn stale_peers_are_forgotten() {
        let (old, current, connected) = (peer(), peer(), peer());
        let address: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap();
        let mut book = Book::default();
        for peer in [old, current, connected] {
            book.remember(peer, address.clone());
        }
        book.relays.insert(old, address.clone());
        book.continuum.insert(old);
        // From after the mentions above (taken at `Instant::now()`), however fast or
        // slow the clock moved meanwhile.
        let later = tokio::time::Instant::now() + STALE_AFTER;
        // `current` was mentioned again since.
        book.seen.insert(current, later);
        assert_eq!(book.prune(later, |peer| *peer == connected), 1);
        assert!(!book.addresses.contains_key(&old));
        assert!(!book.relays.contains_key(&old) && !book.continuum.contains(&old));
        assert!(book.addresses.contains_key(&current));
        assert!(book.addresses.contains_key(&connected));
        assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't.
    }

    #[test]
    fn relay_failures_are_short() {
        assert_eq!(
            relay_failure_text(
                "Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) }",
                "x"
            ),
            "the relay gave no address to be reached through it"
        );
        assert_eq!(
            relay_failure_text(
                "Codec(Error(Custom { kind: Other, error: Failed }))",
                "Failed"
            ),
            "the relay refused the circuit relay protocol (is it a relay?)"
        );
        assert_eq!(
            relay_failure_text("Io(…)", "connection reset"),
            "connection reset"
        );
        assert_eq!(relay_failure(&Ok(())), "closed");
    }

    /// libp2p's nested dial errors come out as each address's cause, once.
    #[test]
    fn dial_errors_are_short() {
        use ::libp2p::core::transport::TransportError;
        let peer = peer();
        let address: Multiaddr = format!("/ip6/fd00::1/tcp/9661/p2p/{peer}").parse().unwrap();
        let nested = std::io::Error::other(
            "Multiple dial errors occurred:\n - Network is unreachable (os error 101): Multiple dial errors occurred:\n - Network is unreachable (os error 101): Network is unreachable (os error 101)",
        );
        let error =
            ::libp2p::swarm::DialError::Transport(vec![(address, TransportError::Other(nested))]);
        assert_eq!(
            describe_dial_error(&error),
            "/ip6/fd00::1/tcp/9661: Network is unreachable (os error 101)"
        );
        let refused = ::libp2p::swarm::DialError::Transport(vec![(
            "/ip4/10.0.0.1/tcp/1".parse().unwrap(),
            TransportError::Other(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)),
        )]);
        assert_eq!(
            describe_dial_error(&refused),
            "/ip4/10.0.0.1/tcp/1: connection refused"
        );
    }

    /// A relay doesn't dial the nodes it relays for through itself.
    #[test]
    fn no_circuits_through_ourselves() {
        let (us, other_relay, target) = (peer(), peer(), peer());
        let through = |relay: PeerId| -> Multiaddr {
            format!("/ip6/fd00::1/tcp/9661/p2p/{relay}/p2p-circuit")
                .parse()
                .unwrap()
        };
        let direct: Multiaddr = "/ip6/fd00::2/tcp/9661".parse().unwrap();
        assert_eq!(relay_of(&through(us)), Some(us));
        assert_eq!(relay_of(&direct), None);
        assert_eq!(relay_of(&direct.clone().with_p2p(target).unwrap()), None);
        assert_eq!(
            dialable(&[through(us), through(other_relay), direct.clone()], us),
            [through(other_relay), direct]
        );
    }

    #[tokio::test]
    async fn unknown_peers_are_refused() {
        let (b, _) = memory_node();
        let stranger = Keypair::generate_ed25519()
            .public()
            .to_peer_id()
            .to_string();
        assert!(b.send_message(&stranger, &status()).await.is_err());
        assert!(b.resolve_seed(&stranger).await.is_err());
        assert_eq!(b.connection(&stranger), None);
    }

    #[test]
    fn a_direct_connection_wins_over_a_relayed_one() {
        let mut book = Book::default();
        let (relayed_only, both) = (peer(), peer());
        let mut id = 0;
        let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| {
            id += 1;
            let address: Multiaddr = format!("/ip4/10.0.0.{id}/tcp/1").parse().unwrap();
            book.connections
                .insert(ConnectionId::new_unchecked(id), (peer, relayed, address));
        };
        connect(&mut book, relayed_only, true);
        connect(&mut book, both, true);
        connect(&mut book, both, false);
        assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed));
        assert_eq!(book.connection(&both), Some(Connection::Direct));
        assert_eq!(book.connection(&peer()), None);
        assert_eq!(
            book.location(&both).map(|address| address.to_string()),
            Some("/ip4/10.0.0.3/tcp/1".to_owned())
        );
        let connections = book.stats().connections;
        assert_eq!((connections.direct, connections.relayed), (1, 2));
        // Of several, the oldest direct one stays; a relayed one only without a direct.
        connect(&mut book, both, false);
        assert_eq!(
            book.duplicates(&both, false),
            [
                ConnectionId::new_unchecked(4),
                ConnectionId::new_unchecked(2)
            ]
        );
        assert!(book.duplicates(&relayed_only, false).is_empty());
        // Not while one of them carries a slot, ours or the peer's, or we reserve one.
        assert!(book.duplicates(&both, true).is_empty());
        book.slots.insert(both);
        assert!(book.duplicates(&both, false).is_empty());
        book.slots.clear();
        book.slots_here.insert(both);
        assert!(book.duplicates(&both, false).is_empty());
    }
}