//! 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 (punched again now and then, `PUNCH_RETRIES`),
//! within its limits (`RELAY_CIRCUIT_DURATION`,
//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). Hole
//! punching needs both sides to listen (`HolePunching`: at their LAN addresses too);
//! where only one does, the other dials it directly once it knows where (`upgrade`).
//!
//! 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: it keeps the direct one with the shortest round
//! trip (libp2p's ping, on each connection), so a path over the LAN wins over one
//! through a VPN, whichever opened first.
//!
//! 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-ignore`, address ranges are left out (VPNs): not published, not
//! offered for hole punching, not dialled, and connections there refused (`Ignoring`).
//!
//! With `--libp2p-upnp`, a node asks its router to forward its listen ports (UPnP), and
//! publishes the router's public address for them with its own (`Book::direct`): then
//! it can be dialled from elsewhere, whatever the other side's NAT.
//!
//! 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);
/// Addresses of one dial tried at once (libp2p's default: 8). Hole punching (`HolePunching`)
/// dials every address both sides offer, where they listen (LAN, VPNs, public IPv6) and
/// where others saw them, often more than 8, while the other side punches towards all of
/// them at the same moment: an address dialled only after others failed (each after
/// QUIC's 5 s handshake timeout) misses its punch, and the LAN's is often among the
/// last. Dials of the member list's addresses (`ADDRESSES_PER_PEER`) are unchanged.
const DIAL_CONCURRENCY: std::num::NonZeroU8 = std::num::NonZeroU8::new(32).unwrap();
/// 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 (either end: a node behind the
/// relay is an end of every circuit to it). libp2p's rate limits on circuits and slots
/// (a burst, then one every minute or two, per peer and per IP address: every node
/// behind one NAT shares one) are left out: they're for public relays serving
/// strangers, and refused circuits used them up too.
const RELAY_CIRCUITS: usize = 1024;
const RELAY_CIRCUITS_PER_PEER: usize = 64;
/// 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 refused a slot (`relay_refused`: none left, not allowed, not a relay,
/// another peer at its address: a relay restarted without its libp2p key) isn't picked
/// again for this long. One that didn't grant it for another reason (it doesn't answer,
/// the connection closed) isn't for `RELAY_RETRY`; a slot lost is reserved again at
/// once.
const RELAY_PAUSE: Duration = Duration::from_secs(600);
/// A peer connected only through a relay is dialled directly where it can be reached
/// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to
/// `RELAY_PAUSE`.
const UPGRADE_RETRY: Duration = Duration::from_secs(30);
/// 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, the fastest (`Book::duplicates`), this long after
/// the last one opened: long enough for what it was dialled for, and for each to be
/// pinged (at once, as it opens). Looked for every `DEDUPE_EVERY`.
const DUPLICATE_GRACE: Duration = Duration::from_secs(10);
/// The same while a peer is connected only through relays: a new relayed connection is
/// a hole punch under way (`PUNCH_RETRIES`), which takes its tries (up to a QUIC
/// handshake's 5 s each, three times) on that connection.
const PUNCH_GRACE: Duration = Duration::from_secs(60);
/// After libp2p's hole punching gave up on a peer (three tries, as the relayed
/// connection opened), it tries again after these, the last one over and over: what
/// failed may have been passing (a NAT's mapping, a busy link). Each time through a new
/// connection through the relay, on which the peer starts the punch (`retry_punches`).
const PUNCH_RETRIES: [Duration; 3] = [
    Duration::from_secs(60),
    Duration::from_secs(300),
    Duration::from_secs(1800),
];
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,
    /// Ask the router to forward our listen ports (UPnP), and publish where.
    pub(crate) upnp: bool,
    /// Address ranges neither published nor dialled, nor connected over (`Ignoring`).
    pub(crate) ignore: Vec<IpRange>,
    /// 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,
}

/// The multiaddrs of option `option`.
fn parse_addresses(option: &str, addresses: &[String]) -> Result<Vec<Multiaddr>> {
    addresses
        .iter()
        .map(|address| {
            address
                .parse::<Multiaddr>()
                .with_context(|| format!("{option} {address:?}"))
        })
        .collect()
}

/// The ranges of `libp2p-ignore`.
fn parse_ranges(ranges: &[String]) -> Result<Vec<IpRange>> {
    ranges
        .iter()
        .map(|range| range.parse().context("libp2p-ignore"))
        .collect()
}

/// This transport's options, checked when the configuration is loaded
/// (`config::Settings::check`), so `--print-config` tells a wrong one too: the
/// multiaddrs and ranges parse, and a relay that is no node (`--relay-only`) can be
/// dialled and keeps its peer ID.
pub(crate) fn check(settings: &crate::config::Settings) -> Result<()> {
    parse_addresses("libp2p-listen", &settings.libp2p_listen)?;
    parse_addresses("libp2p-announce", &settings.libp2p_announce)?;
    parse_addresses("libp2p-relays", &settings.libp2p_relays)?;
    parse_ranges(&settings.libp2p_ignore)?;
    if settings.relay_only {
        if settings.libp2p_listen.is_empty() {
            bail!("relay-only needs libp2p-listen: where nodes reach the relay");
        }
        if settings.libp2p_key_file.is_none() {
            bail!(
                "relay-only needs libp2p-key-file: without it, the relay's peer ID changes with every start and nodes lose it"
            );
        }
    }
    Ok(())
}

impl Libp2pConfig {
    /// The configuration `settings` give; `relay_only`: a relay that is no node
    /// (`--relay-only`).
    pub(crate) fn from_settings(
        settings: &crate::config::Settings,
        relay_only: bool,
    ) -> Result<Self> {
        let parse = parse_addresses;
        Ok(Self {
            key: Self::load_key(settings.libp2p_key_file.as_deref())?,
            listen: parse("libp2p-listen", &settings.libp2p_listen)?,
            announce: parse("libp2p-announce", &settings.libp2p_announce)?,
            relays: parse("libp2p-relays", &settings.libp2p_relays)?,
            relay: settings.libp2p_relay || relay_only,
            continuum: !relay_only,
            mdns: settings.libp2p_mdns,
            upnp: settings.libp2p_upnp,
            ignore: parse_ranges(&settings.libp2p_ignore)?,
            public: settings.libp2p_public,
            memory: false,
        })
    }

    /// 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>,
    /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`).
    mapped: Vec<Multiaddr>,
    /// `Libp2pConfig::announce`.
    announced: Vec<Multiaddr>,
    /// `Libp2pConfig::ignore`: not published (but announced), not dialled.
    ignored: Vec<IpRange>,
    /// `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 not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`).
    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>,
    /// Peers connected only through a relay, dialled directly (`upgrade`): when to again
    /// at the soonest, and how long to wait after that.
    upgrades: HashMap<PeerId, (tokio::time::Instant, Duration)>,
    /// The shortest round trip measured on each open connection (libp2p's ping).
    rtts: HashMap<ConnectionId, Duration>,
    /// Peers whose hole punching failed: how often in a row, and when to try again
    /// (`PUNCH_RETRIES`).
    punching: HashMap<PeerId, (usize, tokio::time::Instant)>,
    /// 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 identify says don't speak continuum, and whether they relay: not sent to
    /// (`send_encoded`), so that a seed that's no node says so.
    not_nodes: HashMap<PeerId, bool>,
    /// 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()
}

/// Whether a slot ended, or wasn't granted, because the relay refused it (its error's
/// debug form and innermost cause, as for `relay_failure_text`), rather than because it
/// didn't answer or the connection closed: only then is it worth not asking again for a
/// while.
fn relay_refused(debug: &str, innermost: &str) -> bool {
    let refusals = [
        "NoAddressesInReservation",
        "ResourceLimitExceeded",
        "PermissionDenied",
        "connection gated",
        "UnsupportedProtocol",
        "WrongPeerId",
    ];
    refusals.iter().any(|pattern| debug.contains(pattern)) || innermost == "Failed"
}

/// 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> {
        book_connection(&self.connections, peer)
    }

    /// The connections to `peer` to close, of several: all but one, the oldest direct
    /// one (the oldest relayed one if none is direct).
    ///
    /// None between a relay and a node with a slot on it, or reserving one (`reserving`):
    /// a slot lives on one connection, which libp2p doesn't tell, and closing it would
    /// end the slot, and with it every circuit to the node (the relay then answers
    /// `NoReservation`).
    fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> {
        if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) {
            return Vec::new();
        }
        self.ranked(peer).into_iter().skip(1).collect()
    }

    /// The connections to `peer`, the one to keep first: direct ones before relayed
    /// ones, measured ones by their round trip (`rtts`: over the LAN before a VPN's hub,
    /// whichever opened first), then the oldest.
    fn ranked(&self, peer: &PeerId) -> Vec<ConnectionId> {
        let mut connections: Vec<(bool, Duration, ConnectionId)> = self
            .connections
            .iter()
            .filter(|(_, (connected, _, _))| connected == peer)
            .map(|(id, (_, relayed, _))| {
                let rtt = self.rtts.get(id).copied().unwrap_or(Duration::MAX);
                (*relayed, rtt, *id)
            })
            .collect();
        connections.sort();
        connections.into_iter().map(|(_, _, id)| id).collect()
    }

    /// The remote address of the connection to `peer` that is kept (`ranked`).
    fn location(&self, peer: &PeerId) -> Option<&Multiaddr> {
        let kept = *self.ranked(peer).first()?;
        self.connections.get(&kept).map(|(_, _, address)| address)
    }

    fn stats(&self) -> Libp2pStats {
        let mut connections = Connections::default();
        for (_, relayed, _) in self.connections.values() {
            if *relayed {
                connections.relayed += 1;
            } else {
                connections.direct += 1;
            }
        }
        Libp2pStats {
            reachability: self.reachability,
            public_address: self.public_address.as_ref().map(Multiaddr::to_string),
            connections,
            relay_slots: RelaySlots {
                held: self.slots.len(),
                paused: self
                    .failed_relays
                    .values()
                    .filter(|until| tokio::time::Instant::now() < **until)
                    .count(),
                granted: self.counts.slots_granted,
                lost: self.counts.slots_lost,
                not_granted: self.counts.slots_not_granted,
            },
            hole_punches: self.counts.hole_punches,
            dial_failures: self.counts.dial_failures,
        }
    }

    /// `peer` was mentioned just now.
    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.not_nodes.remove(peer);
            self.lan.remove(peer);
            self.failed_relays.remove(peer);
            self.upgrades.remove(peer);
            self.punching.remove(peer);
            self.seen.remove(peer);
        }
        stale.len()
    }

    /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first.
    fn discover(&mut self, peer: PeerId) {
        let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else {
            return;
        };
        if let Ok(address) = address.clone().with_p2p(peer)
            && !self.discovered.contains(&address)
            && self.discovered.len() < MAX_PEERS
        {
            self.discovered.push(address);
        }
    }

    /// Where others could dial us directly: the announced addresses, or else those we
    /// listen on (`published`), where the router forwards to them (`mapped`), and where
    /// AutoNAT found others can dial us (`public_address`: a port forwarded by hand, say),
    /// without the peer ID.
    fn direct(&self) -> Vec<Multiaddr> {
        if !self.announced.is_empty() {
            return self.announced.clone();
        }
        let direct: Vec<Multiaddr> = self
            .listening
            .iter()
            .filter(|address| !is_relayed(address))
            .cloned()
            .collect();
        let mut direct = published(&direct);
        let confirmed = self.public_address.iter().filter_map(|address| {
            let mut address = address.clone();
            if let Some(Protocol::P2p(_)) = address.iter().last() {
                address.pop();
            }
            (!address.is_empty() && !is_relayed(&address)).then_some(address)
        });
        // First: the member list keeps only so many (`members::MAX_ADDRESSES`), and a
        // node listening everywhere has many.
        let mut first: Vec<Multiaddr> = Vec::new();
        for address in self.mapped.iter().cloned().chain(confirmed) {
            if !first.contains(&address) {
                first.push(address);
            }
        }
        direct.retain(|address| !first.contains(address));
        first.extend(direct);
        first.retain(|address| !ignored(&self.ignored, address));
        first
    }
}

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

    /// Where `peer` may be reached through the relays we know (named, then members that
    /// relay): for a peer known by its ID only (a seed `/p2p/<peer ID>`), which may hold
    /// a slot on one of them. `local`: this node, never its own relay.
    fn through_relays(&self, peer: PeerId, local: PeerId) -> Vec<Multiaddr> {
        let mut relays: Vec<(PeerId, Multiaddr)> = self
            .named_relays
            .iter()
            .map(|(relay, named)| (*relay, named.at.clone()))
            .collect();
        relays.sort_by_key(|(_, address)| address.to_string());
        let mut members: Vec<(PeerId, Multiaddr)> = self
            .relays
            .iter()
            .filter(|(relay, _)| !self.named_relays.contains_key(relay))
            .map(|(relay, address)| (*relay, address.clone()))
            .collect();
        members.sort_by_key(|(_, address)| address.to_string());
        relays
            .into_iter()
            .chain(members)
            .filter(|(relay, _)| *relay != local && *relay != peer)
            .map(|(_, address)| address.with(Protocol::P2pCircuit).with(Protocol::P2p(peer)))
            .take(ADDRESSES_PER_PEER)
            .collect()
    }

    /// Whether all of `addresses` are known of `peer` (`remember`).
    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).
/// An address range (`10.244.0.0/16`, `42::/16`; a bare address is a range of one), for
/// `Libp2pConfig::ignore`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct IpRange {
    network: std::net::IpAddr,
    prefix: u8,
}

impl FromStr for IpRange {
    type Err = anyhow::Error;

    fn from_str(text: &str) -> Result<Self> {
        let (address, prefix) = match text.split_once('/') {
            Some((address, prefix)) => (address, Some(prefix)),
            None => (text, None),
        };
        let network: std::net::IpAddr = address
            .parse()
            .with_context(|| format!("{text:?} is no address range (e.g. 10.244.0.0/16)"))?;
        let bits = if network.is_ipv4() { 32 } else { 128 };
        let prefix = match prefix {
            Some(prefix) => prefix
                .parse::<u8>()
                .ok()
                .filter(|prefix| *prefix <= bits)
                .with_context(|| format!("{text:?}: the prefix is 0 to {bits}"))?,
            None => bits,
        };
        Ok(Self { network, prefix })
    }
}

impl IpRange {
    fn contains(&self, ip: std::net::IpAddr) -> bool {
        use std::net::IpAddr;
        match (self.network, ip) {
            (IpAddr::V4(network), IpAddr::V4(ip)) => {
                let mask = u32::MAX
                    .checked_shl(32 - u32::from(self.prefix))
                    .unwrap_or(0);
                u32::from(network) & mask == u32::from(ip) & mask
            }
            (IpAddr::V6(network), IpAddr::V6(ip)) => {
                let mask = u128::MAX
                    .checked_shl(128 - u32::from(self.prefix))
                    .unwrap_or(0);
                u128::from(network) & mask == u128::from(ip) & mask
            }
            _ => false,
        }
    }
}

/// Whether `address` is in one of `ranges`: a direct address there (one through a relay
/// is judged by the connection to the relay).
fn ignored(ranges: &[IpRange], address: &Multiaddr) -> bool {
    if ranges.is_empty() || is_relayed(address) {
        return false;
    }
    let ip = match address.iter().next() {
        Some(Protocol::Ip4(ip)) => std::net::IpAddr::V4(ip),
        Some(Protocol::Ip6(ip)) => std::net::IpAddr::V6(ip),
        _ => return false,
    };
    ranges.iter().any(|range| range.contains(ip))
}

/// Keeps connections off the address ranges `libp2p-ignore` names, whoever dialled:
/// hole punching and identify bring addresses there too, and a peer may dial us there.
struct Ignoring {
    ranges: Vec<IpRange>,
}

#[derive(Debug, thiserror::Error)]
#[error("{0} is in a range left out (libp2p-ignore)")]
struct IgnoredAddress(Multiaddr);

impl ::libp2p::swarm::NetworkBehaviour for Ignoring {
    type ConnectionHandler = ::libp2p::swarm::dummy::ConnectionHandler;
    type ToSwarm = std::convert::Infallible;

    fn handle_pending_inbound_connection(
        &mut self,
        _connection_id: ConnectionId,
        _local_addr: &Multiaddr,
        remote_addr: &Multiaddr,
    ) -> Result<(), ::libp2p::swarm::ConnectionDenied> {
        if ignored(&self.ranges, remote_addr) {
            return Err(::libp2p::swarm::ConnectionDenied::new(IgnoredAddress(
                remote_addr.clone(),
            )));
        }
        Ok(())
    }

    fn handle_established_inbound_connection(
        &mut self,
        _connection_id: ConnectionId,
        _peer: PeerId,
        _local_addr: &Multiaddr,
        _remote_addr: &Multiaddr,
    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
        Ok(::libp2p::swarm::dummy::ConnectionHandler)
    }

    fn handle_established_outbound_connection(
        &mut self,
        _connection_id: ConnectionId,
        _peer: PeerId,
        addr: &Multiaddr,
        _role_override: ::libp2p::core::Endpoint,
        _port_use: ::libp2p::core::transport::PortUse,
    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
        if ignored(&self.ranges, addr) {
            return Err(::libp2p::swarm::ConnectionDenied::new(IgnoredAddress(
                addr.clone(),
            )));
        }
        Ok(::libp2p::swarm::dummy::ConnectionHandler)
    }

    fn on_swarm_event(&mut self, _event: ::libp2p::swarm::FromSwarm) {}

    fn on_connection_handler_event(
        &mut self,
        _peer_id: PeerId,
        _connection_id: ConnectionId,
        event: ::libp2p::swarm::THandlerOutEvent<Self>,
    ) {
        match event {}
    }

    fn poll(
        &mut self,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<
        ::libp2p::swarm::ToSwarm<Self::ToSwarm, ::libp2p::swarm::THandlerInEvent<Self>>,
    > {
        std::task::Poll::Pending
    }
}

/// How `peer` is connected, of `connections` (`Book::connections`): directly if one is
/// direct.
fn book_connection(
    connections: &HashMap<ConnectionId, (PeerId, bool, Multiaddr)>,
    peer: &PeerId,
) -> Option<Connection> {
    let mut found = None;
    for (connected, relayed, _) in connections.values() {
        if connected == peer {
            if !relayed {
                return Some(Connection::Direct);
            }
            found = Some(Connection::Relayed);
        }
    }
    found
}

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, config.ignore.clone())?
        } else {
            network_swarm(config.key, features, config.ignore.clone())?
        };
        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,
            ignored: config.ignore.clone(),
            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 = {
            let book = lock(&self.book);
            let mut known = book
                .addresses
                .get(&peer)
                .map(|known| dialable(known, self.peer_id))
                .unwrap_or_default();
            known.retain(|address| !ignored(&book.ignored, address));
            if known.is_empty() {
                book.through_relays(peer, self.peer_id)
            } else {
                known
            }
        };
        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 {
    /// First: connections it denies need nothing from the others.
    ignoring: Ignoring,
    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: HolePunching,
    /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest
    /// is kept.
    ping: ::libp2p::ping::Behaviour,
    autonat: ::libp2p::autonat::Behaviour,
    /// Only with `Libp2pConfig::mdns`.
    mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
    /// Only with `Libp2pConfig::upnp`: maps our listen ports on the router, and confirms
    /// the mapped addresses as ours (identify then tells them).
    upnp: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::upnp::tokio::Behaviour>,
    /// Only with `Libp2pConfig::public`: the IPFS DHT.
    kad: ::libp2p::swarm::behaviour::toggle::Toggle<
        ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>,
    >,
}

/// libp2p's hole punching (DCUtR), offering our listen addresses too. DCUtR offers only
/// where others saw us (external address candidates, from identify): seen by a relay
/// out on the internet, two nodes on one LAN are both at their router's public address,
/// and punching through it needs the router to hairpin, which many don't. Their LAN
/// addresses work, the firewalls on the way being stateful. Only DCUtR is told:
/// AutoNAT would test every candidate, and fail on all of these.
struct HolePunching {
    inner: ::libp2p::dcutr::Behaviour,
    /// Where we listen (`offered`).
    listening: Vec<Multiaddr>,
    /// `Libp2pConfig::ignore`: not offered.
    ignored: Vec<IpRange>,
}

/// Whether DCUtR offers our listen address `address` too: neither loopback nor
/// link-local (as for publishing), nor through a relay.
fn offered(address: &Multiaddr) -> bool {
    !is_local(address) && !is_relayed(address)
}

impl HolePunching {
    fn new(local: PeerId, ignored: Vec<IpRange>) -> Self {
        Self {
            inner: ::libp2p::dcutr::Behaviour::new(local),
            listening: Vec::new(),
            ignored,
        }
    }

    /// Tells DCUtR where we listen, as if others had seen us there. Again after every
    /// other candidate, as DCUtR keeps the latest 20.
    fn offer_listening(&mut self) {
        use ::libp2p::swarm::{FromSwarm, NetworkBehaviour, NewExternalAddrCandidate};
        for addr in &self.listening {
            self.inner
                .on_swarm_event(FromSwarm::NewExternalAddrCandidate(
                    NewExternalAddrCandidate { addr },
                ));
        }
    }
}

impl ::libp2p::swarm::NetworkBehaviour for HolePunching {
    type ConnectionHandler =
        <::libp2p::dcutr::Behaviour as ::libp2p::swarm::NetworkBehaviour>::ConnectionHandler;
    type ToSwarm = ::libp2p::dcutr::Event;

    fn handle_pending_inbound_connection(
        &mut self,
        connection_id: ConnectionId,
        local_addr: &Multiaddr,
        remote_addr: &Multiaddr,
    ) -> Result<(), ::libp2p::swarm::ConnectionDenied> {
        self.inner
            .handle_pending_inbound_connection(connection_id, local_addr, remote_addr)
    }

    fn handle_established_inbound_connection(
        &mut self,
        connection_id: ConnectionId,
        peer: PeerId,
        local_addr: &Multiaddr,
        remote_addr: &Multiaddr,
    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
        if is_relayed(local_addr) {
            // DCUtR offers its candidates on this connection: where others saw us, and
            // these.
            debug!(peer = %Addressed(SCHEME, &peer), listening = ?self.listening, "Reached through a relay; hole punching offers where we listen too");
        }
        self.inner.handle_established_inbound_connection(
            connection_id,
            peer,
            local_addr,
            remote_addr,
        )
    }

    fn handle_pending_outbound_connection(
        &mut self,
        connection_id: ConnectionId,
        maybe_peer: Option<PeerId>,
        addresses: &[Multiaddr],
        effective_role: ::libp2p::core::Endpoint,
    ) -> Result<Vec<Multiaddr>, ::libp2p::swarm::ConnectionDenied> {
        self.inner.handle_pending_outbound_connection(
            connection_id,
            maybe_peer,
            addresses,
            effective_role,
        )
    }

    fn handle_established_outbound_connection(
        &mut self,
        connection_id: ConnectionId,
        peer: PeerId,
        addr: &Multiaddr,
        role_override: ::libp2p::core::Endpoint,
        port_use: ::libp2p::core::transport::PortUse,
    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
        self.inner.handle_established_outbound_connection(
            connection_id,
            peer,
            addr,
            role_override,
            port_use,
        )
    }

    fn on_swarm_event(&mut self, event: ::libp2p::swarm::FromSwarm) {
        use ::libp2p::swarm::FromSwarm;
        self.inner.on_swarm_event(event);
        match event {
            FromSwarm::NewListenAddr(new)
                if offered(new.addr) && !ignored(&self.ignored, new.addr) =>
            {
                if !self.listening.contains(new.addr) {
                    debug!(address = %new.addr, "Offering where we listen for hole punching");
                    self.listening.push(new.addr.clone());
                }
                self.offer_listening();
            }
            FromSwarm::ExpiredListenAddr(expired) => {
                self.listening.retain(|address| address != expired.addr);
            }
            FromSwarm::NewExternalAddrCandidate(_) => self.offer_listening(),
            _ => {}
        }
    }

    fn on_connection_handler_event(
        &mut self,
        peer_id: PeerId,
        connection_id: ConnectionId,
        event: ::libp2p::swarm::THandlerOutEvent<Self>,
    ) {
        self.inner
            .on_connection_handler_event(peer_id, connection_id, event)
    }

    fn poll(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<
        ::libp2p::swarm::ToSwarm<Self::ToSwarm, ::libp2p::swarm::THandlerInEvent<Self>>,
    > {
        self.inner.poll(cx)
    }
}

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

impl Features {
    fn of(config: &Libp2pConfig) -> Self {
        Self {
            relay: config.relay,
            mdns: config.mdns,
            upnp: config.upnp,
            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,
        ignored: Vec<IpRange>,
    ) -> std::io::Result<Self> {
        let Features {
            relay,
            mdns,
            upnp,
            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,
                    reservation_rate_limiters: Vec::new(),
                    circuit_src_rate_limiters: Vec::new(),
                    ..Default::default()
                },
            )
        });
        let mdns = match mdns {
            true => Some(::libp2p::mdns::tokio::Behaviour::new(
                ::libp2p::mdns::Config::default(),
                peer_id,
            )?),
            false => None,
        };
        Ok(Self {
            ignoring: Ignoring {
                ranges: ignored.clone(),
            },
            stream: libp2p_stream::Behaviour::new(),
            identify: ::libp2p::identify::Behaviour::new(identify),
            relay_client,
            relay: relay.into(),
            dcutr: HolePunching::new(peer_id, ignored),
            ping: ::libp2p::ping::Behaviour::new(::libp2p::ping::Config::new()),
            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(),
            upnp: upnp.then(::libp2p::upnp::tokio::Behaviour::default).into(),
            kad: kad.into(),
        })
    }
}

fn network_swarm(
    key: Keypair,
    features: Features,
    ignored: Vec<IpRange>,
) -> 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, ignored.clone())
                .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)
        .with_dial_concurrency_factor(DIAL_CONCURRENCY);
    if features.public {
        config
    } else {
        config.with_substream_upgrade_protocol_override(::libp2p::core::upgrade::Version::V1Lazy)
    }
}

fn memory_swarm(
    key: Keypair,
    features: Features,
    ignored: Vec<IpRange>,
) -> 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, ignored)?,
        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);
        let now = tokio::time::Instant::now();
        book.failed_relays.retain(|_, until| now < *until);
        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");
                }
            }
        }
    }
}

/// Dials `peer` directly if it's connected only through a relay and is known elsewhere
/// too (the member list, identify, mDNS), since the first connection, made through a
/// relay (a relayed seed), would otherwise be kept for good: a peer connected isn't
/// dialled again, and hole punching needs both sides to listen. The direct connection,
/// once open, wins over the relayed one (`Book::duplicates`).
fn upgrade(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, peer: PeerId, local: PeerId) {
    let direct: Vec<Multiaddr> = {
        let mut book = lock(book);
        if book.connection(&peer) != Some(Connection::Relayed) {
            return;
        }
        let direct: Vec<Multiaddr> = book
            .addresses
            .get(&peer)
            .map(|known| dialable(known, local))
            .unwrap_or_default()
            .into_iter()
            .filter(|address| !is_relayed(address) && !ignored(&book.ignored, address))
            .collect();
        if direct.is_empty() {
            return;
        }
        let now = tokio::time::Instant::now();
        let (due, wait) = book
            .upgrades
            .get(&peer)
            .copied()
            .unwrap_or((now, UPGRADE_RETRY));
        if now < due {
            return;
        }
        book.upgrades
            .insert(peer, (now + wait, (wait * 2).min(RELAY_PAUSE)));
        direct
    };
    debug!(peer = %Addressed(SCHEME, &peer), addresses = ?direct, "Connected only through a relay; dialling directly");
    let _ = swarm.dial(
        DialOpts::peer_id(peer)
            .addresses(direct)
            .condition(PeerCondition::NotDialing)
            .build(),
    );
}

/// Tries hole punching again with the peers it failed with that are due (`punching`).
/// libp2p punches as a relayed connection opens, started by the side that was dialled
/// through the relay; that side is where a punch's failure is told, so it is this one.
/// A new relayed connection starts a new punch: we dial the peer through a relay if it
/// has a slot on one; else we close our relayed connections to it, and its next message
/// (a probe, within a minute) dials us through ours again. Peers connected directly by
/// now are dropped; those not connected now are kept, failures and all.
fn retry_punches(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, local: PeerId) {
    let now = tokio::time::Instant::now();
    let mut close: Vec<ConnectionId> = Vec::new();
    let due: Vec<(PeerId, usize, Vec<Multiaddr>)> = {
        let mut book = lock(book);
        let book = &mut *book;
        let mut due = Vec::new();
        book.punching.retain(|peer, (failed, next)| {
            match book_connection(&book.connections, peer) {
                Some(Connection::Direct) => return false,
                // Kept, failures and all: one we closed to have it dial anew comes back.
                None => return true,
                Some(Connection::Relayed) if now < *next => return true,
                Some(Connection::Relayed) => {}
            }
            let through: Vec<Multiaddr> = book
                .addresses
                .get(peer)
                .map(|known| dialable(known, local))
                .unwrap_or_default()
                .into_iter()
                .filter(is_relayed)
                .collect();
            // Until its punch fails again (or the peer tries from its side).
            *next = now + PUNCH_RETRIES[(*failed).min(PUNCH_RETRIES.len() - 1)];
            if through.is_empty() {
                close.extend(
                    book.connections
                        .iter()
                        .filter(|(_, (connected, relayed, _))| connected == peer && *relayed)
                        .map(|(id, _)| *id),
                );
            } else {
                due.push((*peer, *failed, through));
            }
            true
        });
        due
    };
    for connection in close {
        debug!(connection = %connection, "Hole punching again: closing a relayed connection, for the peer to dial anew");
        swarm.close_connection(connection);
    }
    for (peer, failed, through) in due {
        debug!(peer = %Addressed(SCHEME, &peer), failed, "Hole punching again, through a new relayed connection");
        let _ = swarm.dial(
            DialOpts::peer_id(peer)
                .addresses(through)
                .condition(PeerCondition::Always)
                .build(),
        );
    }
}

/// 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);
                retry_punches(&mut swarm, &book, relaying.local);
            }
            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(()));
                        upgrade(&mut swarm, &book, peer, relaying.local);
                        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);
                                }
                            }
                            failed => {
                                if !lock(&book).finding.contains_key(&address) {
                                    return; // No longer named.
                                }
                                let error = match failed {
                                    Ok(Ok(Err(error))) => error,
                                    Ok(_) => "the libp2p swarm stopped".to_owned(),
                                    Err(_) => format!("no answer within {CONNECT_TIMEOUT:?}"),
                                };
                                warn!(relay = %Addressed(SCHEME, &address), error = %error, retry = ?RELAY_RETRY, "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 !relayed {
                            book.upgrades.remove(&peer_id);
                            book.punching.remove(&peer_id);
                        }
                        if relaying.local < peer_id
                            && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1
                        {
                            // Only through relays: maybe a punch under way.
                            let grace = match book.connection(&peer_id) {
                                Some(Connection::Relayed) => PUNCH_GRACE,
                                _ => DUPLICATE_GRACE,
                            };
                            dedupe_due.insert(peer_id, tokio::time::Instant::now() + 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, .. } => {
                    {
                        let mut book = lock(&book);
                        book.connections.remove(&connection_id);
                        book.rtts.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 refused = match &reason {
                            Ok(()) => false,
                            Err(error) => relay_refused(&format!("{error:?}"), &innermost_causes(error)),
                        };
                        let (lost, pause) = {
                            let mut book = lock(&book);
                            let lost = book.slots.remove(&relay);
                            let pause = match (refused, lost) {
                                (true, _) => Some(RELAY_PAUSE),
                                (false, true) => None,
                                (false, false) => Some(RELAY_RETRY),
                            };
                            if lost {
                                book.counts.slots_lost += 1;
                            } else {
                                book.counts.slots_not_granted += 1;
                                // Named without its peer ID: maybe another one now.
                                if let Some(address) = book.find_again(relay) {
                                    let _ = again.send(Command::FindRelay { address });
                                }
                            }
                            if let Some(pause) = pause {
                                book.failed_relays.insert(relay, tokio::time::Instant::now() + pause);
                            }
                            (lost, pause)
                        };
                        // Again at once, or others meanwhile.
                        let _ = again.send(Command::CheckRelays);
                        let reason = relay_failure(&reason);
                        match (lost, pause) {
                            (true, None) => {
                                info!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay; reserving again");
                            }
                            (true, Some(pause)) => {
                                info!(relay = %Addressed(SCHEME, &relay), pause = ?pause, reason = %reason, "Lost the slot on a relay; picking others for a while");
                            }
                            (false, pause) => {
                                info!(
                                    relay = %Addressed(SCHEME, &relay),
                                    pause = ?pause.unwrap_or_default(),
                                    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::Ping(::libp2p::ping::Event {
                    peer,
                    connection,
                    result,
                })) => match result {
                    Ok(rtt) => {
                        let mut book = lock(&book);
                        if book.connections.contains_key(&connection) {
                            let shortest = book.rtts.entry(connection).or_insert(rtt);
                            *shortest = (*shortest).min(rtt);
                        }
                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, rtt = ?rtt, "Connection's round trip");
                    }
                    Err(error) => {
                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, error = %error, "Connection's ping failed");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Upnp(event)) => match event {
                    ::libp2p::upnp::Event::NewExternalAddr {
                        local_addr,
                        external_addr: address,
                    } => {
                        info!(address = %address, to = %local_addr, "The router forwards to us (UPnP)");
                        let mut book = lock(&book);
                        if !book.mapped.contains(&address) {
                            book.mapped.push(address);
                        }
                    }
                    ::libp2p::upnp::Event::ExpiredExternalAddr {
                        local_addr,
                        external_addr: address,
                    } => {
                        info!(address = %address, to = %local_addr, "The router no longer forwards to us (UPnP)");
                        lock(&book).mapped.retain(|mapped| *mapped != address);
                    }
                    ::libp2p::upnp::Event::GatewayNotFound => {
                        info!("No router that maps ports (UPnP) answered");
                    }
                    ::libp2p::upnp::Event::NonRoutableGateway => {
                        warn!("The router that maps ports (UPnP) is behind another NAT itself; its mappings don't reach us from outside");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                    Ok(_) => {
                        let mut book = lock(&book);
                        book.counts.hole_punches.succeeded += 1;
                        book.punching.remove(&event.remote_peer_id);
                        drop(book);
                        info!(peer = %Addressed(SCHEME, &event.remote_peer_id), "Connected directly to a peer behind a relay");
                    }
                    Err(error) => {
                        let retry = {
                            let mut book = lock(&book);
                            book.counts.hole_punches.failed += 1;
                            let failed = book
                                .punching
                                .get(&event.remote_peer_id)
                                .map_or(0, |(failed, _)| *failed);
                            let retry = PUNCH_RETRIES[failed.min(PUNCH_RETRIES.len() - 1)];
                            book.punching.insert(
                                event.remote_peer_id,
                                (failed + 1, tokio::time::Instant::now() + retry),
                            );
                            retry
                        };
                        debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, retry = ?retry, "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.not_nodes.len() < MAX_PEERS
                                || book.not_nodes.contains_key(&peer_id))
                        {
                            let relays = info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME);
                            book.not_nodes.insert(peer_id, relays);
                        }
                        if info.protocols.contains(&PROTOCOL) {
                            book.not_nodes.remove(&peer_id);
                            book.continuum.insert(peer_id);
                            if book.lan.remove(&peer_id) {
                                book.discover(peer_id);
                            }
                        }
                    }
                    upgrade(&mut swarm, &book, peer_id, relaying.local);
                }
                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)
                        };
                        upgrade(&mut swarm, &book, peer, relaying.local);
                        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)?;
        address.pop();
        // A peer ID only says nothing about where.
        if peer != self.peer_id && !address.is_empty() {
            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 punching(&self, target: &str) -> Option<crate::transport::Punching> {
        let peer = PeerId::from_str(target).ok()?;
        let (failed, next) = *lock(&self.book).punching.get(&peer)?;
        Some(crate::transport::Punching {
            failed,
            next_in_secs: next
                .saturating_duration_since(tokio::time::Instant::now())
                .as_secs(),
        })
    }

    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> {
        let address = match PeerId::from_str(seed) {
            Ok(peer) => Multiaddr::empty().with(Protocol::P2p(peer)),
            Err(_) => {
                Multiaddr::from_str(seed).with_context(|| format!("{seed} is no multiaddr"))?
            }
        };
        // A peer ID only (`/p2p/<peer ID>`): reached where it's known, or through the
        // relays we know.
        if let [Protocol::P2p(peer)] = address.iter().collect::<Vec<_>>()[..] {
            let book = lock(&self.book);
            if book.addresses.contains_key(&peer)
                || !book.through_relays(peer, self.peer_id).is_empty()
            {
                return Ok(peer.to_string());
            }
            bail!(
                "no address known for libp2p peer {peer}, and no relay to reach it through; \
                 give a multiaddr, or a relay (libp2p-relays)"
            );
        }
        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?;
        // The protocol goes with the first data (`swarm_config`), so a peer that doesn't
        // speak it seems to take the message: once identify has told, say so.
        if let Some(relays) = lock(&self.book).not_nodes.get(&peer).copied() {
            if relays {
                bail!(
                    "{peer} is a relay, not a node: give a node behind it as the seed \
                     (<relay>/p2p-circuit/p2p/<peer ID>, or /p2p/<peer ID> with libp2p-relays)"
                );
            }
            bail!("{peer} doesn't speak continuum");
        }
        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?;
            // Kept until the peer has closed its side too. Over QUIC, dropping a stream
            // stops its reading half (STOP_SENDING), and the peer's answer to the
            // protocol (sent with the first data, `swarm_config`) then fails, and with it
            // the stream, before the peer has read the message: identify, which reads
            // its answers, works, while every message is lost. Yamux (TCP) doesn't mind.
            // Off the send, which is done.
            tokio::spawn(async move {
                let mut rest = [0u8; 64];
                let _ = tokio::time::timeout(STREAM_TIMEOUT, async {
                    while let Ok(read) = stream.read(&mut rest).await {
                        if read == 0 {
                            break;
                        }
                    }
                })
                .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,
            upnp: false,
            ignore: Vec::new(),
            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,
            upnp: false,
            ignore: Vec::new(),
            public: false,
            memory: true,
        })
        .unwrap();
        assert_eq!(
            transport.local_addresses(),
            [announced.with_p2p(peer).unwrap().to_string()]
        );
    }

    #[test]
    fn address_ranges() {
        let range = |text: &str| text.parse::<IpRange>().unwrap();
        let address = |text: &str| text.parse::<Multiaddr>().unwrap();
        let ranges = [range("10.244.0.0/16"), range("42::/16"), range("192.0.2.7")];
        for left_out in [
            "/ip4/10.244.3.2/udp/9661/quic-v1",
            "/ip6/42:1:a46e:be4a:cb58:e460:bf0:142c/tcp/9661",
            "/ip4/192.0.2.7/tcp/1",
        ] {
            assert!(ignored(&ranges, &address(left_out)), "{left_out}");
        }
        for kept in [
            "/ip4/10.23.1.144/udp/9661/quic-v1",
            "/ip4/10.245.0.1/tcp/1",
            "/ip6/2a0a:4cc0:c1:5eb0::1/udp/9661/quic-v1",
            "/ip4/192.0.2.8/tcp/1",
            "/dns4/example.org/tcp/1",
        ] {
            assert!(!ignored(&ranges, &address(kept)), "{kept}");
        }
        // Through a relay: judged by the connection to the relay, not here.
        let relayed = format!("/ip4/10.244.0.1/tcp/1/p2p/{}/p2p-circuit", peer());
        assert!(!ignored(&ranges, &address(&relayed)));
        assert!(range("0.0.0.0/0").contains("203.0.113.1".parse().unwrap()));
        assert!(!range("0.0.0.0/0").contains("::1".parse().unwrap()));
        for wrong in ["10.244.0.0/33", "42::/129", "nowhere", "10.0.0.0/x"] {
            assert!(wrong.parse::<IpRange>().is_err(), "{wrong}");
        }
    }

    /// Ranges left out aren't published; announced addresses are, whatever they are.
    #[test]
    fn ignored_addresses_are_not_published() {
        let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
        let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1");
        let vpn = parse("/ip4/10.244.3.2/udp/9663/quic-v1");
        let mut book = Book {
            listening: vec![lan.clone(), vpn.clone()],
            ignored: vec!["10.244.0.0/16".parse().unwrap()],
            ..Book::default()
        };
        assert_eq!(book.direct(), [lan]);
        book.announced = vec![vpn.clone()];
        assert_eq!(book.direct(), [vpn]);
    }

    /// A port forwarded by hand: once AutoNAT found the node dialable at the router's
    /// address, that is published with the listen addresses; announced ones still win.
    #[test]
    fn the_address_autonat_confirmed_is_published() {
        let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
        let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1");
        let public = parse("/ip4/91.65.180.254/udp/9663/quic-v1");
        let mut book = Book {
            listening: vec![lan.clone()],
            ..Book::default()
        };
        assert_eq!(book.direct(), [lan.clone()]);
        book.public_address = Some(public.clone().with_p2p(peer()).unwrap());
        assert_eq!(book.direct(), [public.clone(), lan.clone()]);
        let announced = parse("/dns4/example.org/udp/9663/quic-v1");
        book.announced = vec![announced.clone()];
        assert_eq!(book.direct(), [announced]);
    }

    #[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 seed that's a peer ID only is reached through the relays the node knows.
    #[tokio::test]
    async fn a_peer_id_seed_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 (a, _) = memory_node_with(true, vec![r_address.clone()], false);
        let seed = format!("/p2p/{}", b.local_id());
        assert_eq!(a.resolve_seed(&seed).await.unwrap(), b.local_id());
        assert_eq!(a.resolve_seed(b.local_id()).await.unwrap(), b.local_id());
        // Nothing to dial is remembered.
        assert!(!lock(&a.book).addresses.contains_key(&b.peer_id));
        a.send_message(b.local_id(), &status()).await.unwrap();
        let (from, _) = at_b.recv().await.unwrap();
        assert_eq!(from, a.local_id());

        // Without a relay, there's nowhere to look.
        let (c, _) = memory_node();
        assert!(c.resolve_seed(&seed).await.is_err());
    }

    /// b listens nowhere; a, which can be dialled, reaches it through the relay r: b
    /// then dials a directly (identify told it where), and both keep that.
    #[tokio::test]
    async fn a_relayed_connection_is_upgraded_where_one_side_can_be_dialled() {
        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 (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();
        at_b.recv().await.unwrap();
        assert!(
            eventually(|| b.connection(a.local_id()) == Some(Connection::Direct)).await,
            "b didn't dial a directly"
        );
        assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
        // The relayed one goes (`DUPLICATE_GRACE`); the direct one stays.
        let deadline = tokio::time::Instant::now() + DUPLICATE_GRACE * 2;
        while a.stats().libp2p.unwrap().connections.relayed > 0 {
            assert!(
                tokio::time::Instant::now() < deadline,
                "the relayed connection stayed"
            );
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        b.send_message(a.local_id(), &status()).await.unwrap();
        assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
    }

    /// 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,
            upnp: false,
            ignore: Vec::new(),
            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, and a send to
        // it is refused once identify has told (the relay's own refusal comes too late:
        // the protocol goes with the first data, `swarm_config`), saying what to give
        // as a seed instead.
        assert_eq!(
            a.resolve_seed(&r_address.to_string()).await.unwrap(),
            r.local_id()
        );
        assert!(
            eventually(|| lock(&a.book).not_nodes.get(&r.peer_id) == Some(&true)).await,
            "identify didn't tell"
        );
        let refused = a.send_message(r.local_id(), &status()).await.unwrap_err();
        assert!(
            refused.to_string().contains("is a relay, not a node"),
            "{refused:#}"
        );
        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");
    }

    /// Only refusals pause a relay; a slot that ended otherwise is asked for again.
    #[test]
    fn only_refusals_pause_relays() {
        let refused = [
            ("Reservation(Protocol(NoAddressesInReservation))", "x"),
            ("Reservation(Status(ResourceLimitExceeded))", "x"),
            ("Reservation(Status(PermissionDenied))", "x"),
            ("Denied { cause: \"connection gated\" }", "x"),
            ("Upgrade(UnsupportedProtocol)", "x"),
            ("Dial(WrongPeerId { obtained: … })", "x"),
            (
                "Codec(Error(Custom { kind: Other, error: Failed }))",
                "Failed",
            ),
        ];
        for (debug, innermost) in refused {
            assert!(relay_refused(debug, innermost), "{debug}");
        }
        let not = [
            ("Io(Custom { kind: Other, error: Timeout })", "Timeout"),
            (
                "Io(Custom { kind: Other, error: Right(Closed) })",
                "connection is closed",
            ),
            ("Io(…)", "connection reset"),
        ];
        for (debug, innermost) in not {
            assert!(!relay_refused(debug, innermost), "{debug}");
        }
    }

    #[test]
    fn listen_addresses_offered_for_hole_punching() {
        let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
        assert!(offered(&parse("/ip4/192.168.1.2/udp/9661/quic-v1")));
        assert!(offered(&parse("/ip6/fd00::2/tcp/9661")));
        assert!(!offered(&parse("/ip4/127.0.0.1/udp/9661/quic-v1")));
        assert!(!offered(&parse("/ip6/fe80::1/tcp/9661")));
        let relay = PeerId::random();
        assert!(!offered(&parse(&format!(
            "/ip4/192.168.1.2/tcp/9661/p2p/{relay}/p2p-circuit"
        ))));
    }

    /// 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());
        book.slots_here.clear();
        // Once measured, the fastest direct one stays, however young.
        book.rtts
            .insert(ConnectionId::new_unchecked(3), Duration::from_millis(40));
        book.rtts
            .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2));
        assert_eq!(
            book.duplicates(&both, false),
            [
                ConnectionId::new_unchecked(3),
                ConnectionId::new_unchecked(2)
            ]
        );
        assert_eq!(
            book.location(&both).map(|address| address.to_string()),
            Some("/ip4/10.0.0.4/tcp/1".to_owned())
        );
    }
}