path: root/src/transport/p2p.rs

//! 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.
//!
//! 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`), on which it reserves a slot
//! (`--libp2p-relays`); its address through the relay (`<relay>/p2p-circuit/p2p/<it>`)
//! is published with the others. Over a relayed connection, both sides then try to
//! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until
//! that works, or for good if it doesn't, within its limits (`RELAY_CIRCUIT_DURATION`,
//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send).
//!
//! 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 and has none configured picks up to
//! `AUTO_RELAYS` of those (`relay_check`). It needs one when it has no address others
//! could dial, or when AutoNAT says it can't be dialled: AutoNAT asks connected peers
//! (nodes of the swarm) to dial it back. Once AutoNAT says it can be dialled, it lets
//! the relays it picked go.
//!
//! With `--libp2p-mdns`, a node finds libp2p peers on its LAN (multicast DNS). Their
//! LAN addresses are noted, and those that turn out to speak continuum (identify says
//! so) are seeds while the node joins (`discovered_seeds`): nodes on one LAN find each
//! other without configured seeds.
//!
//! It never uses public libp2p infrastructure: 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, warn};

use crate::locks::MutexExt;
use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport, wire};

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

/// A relayed connection (circuit) through this node lasts at most this long and carries
/// at most this much: generous, since the swarm's own relays may carry sync traffic for
/// good where hole punching fails (libp2p's defaults, 2 minutes and 128 KiB, are for
/// the punching only).
const RELAY_CIRCUIT_DURATION: Duration = Duration::from_secs(3600);
const RELAY_CIRCUIT_BYTES: u64 = 1 << 30;
/// Circuits through this node at once, and per peer.
const RELAY_CIRCUITS: usize = 64;
const RELAY_CIRCUITS_PER_PEER: usize = 8;
/// A reservation on a relay that ended (the relay restarted, say) is made 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);

/// 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,
    /// Find peers on the LAN (mDNS).
    pub(crate) mdns: bool,
    /// libp2p's in-memory transport instead of TCP and QUIC: for tests.
    pub(crate) memory: bool,
}

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

enum Command {
    /// Connect to `peer` (at `addresses`) unless connected.
    Connect {
        peer: PeerId,
        addresses: Vec<Multiaddr>,
        done: oneshot::Sender<Result<(), String>>,
    },
    /// Connect to `address` to find out which peer is there.
    Identify {
        address: Multiaddr,
        done: oneshot::Sender<Result<PeerId, String>>,
    },
    /// See whether relays among the members are needed now (one was just learned of).
    CheckRelays,
    /// Reserve a slot on the relay at `relay` (again). `configured`: the address as
    /// configured, for later attempts (without a peer ID, the relay may since have
    /// another).
    Reserve {
        relay: Multiaddr,
        configured: Multiaddr,
    },
}

/// Where peers are, and where we listen.
#[derive(Default)]
struct Book {
    addresses: HashMap<PeerId, Vec<Multiaddr>>,
    listening: Vec<Multiaddr>,
    /// `Libp2pConfig::announce`.
    announced: Vec<Multiaddr>,
    /// `Libp2pConfig::relay`: this node relays, and says so.
    serving: bool,
    /// Members that say they relay, and where (with the relay's peer ID).
    relays: HashMap<PeerId, Multiaddr>,
    /// Peers found on the LAN (mDNS) not known yet to speak continuum.
    lan: std::collections::HashSet<PeerId>,
    /// Peers that speak continuum (identify says so).
    continuum: std::collections::HashSet<PeerId>,
    /// Peers found on the LAN that speak continuum, with their address: seeds not
    /// handed out yet (`discovered_seeds`).
    discovered: Vec<Multiaddr>,
}

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

    /// Where others could dial us directly: the announced addresses, or else those we
    /// listen on (`published`), without the peer ID.
    fn direct(&self) -> Vec<Multiaddr> {
        if !self.announced.is_empty() {
            return self.announced.clone();
        }
        let direct: Vec<Multiaddr> = self
            .listening
            .iter()
            .filter(|address| !is_relayed(address))
            .cloned()
            .collect();
        published(&direct)
    }
}

/// Whether `address` only reaches this host (or this link): no use to other hosts, and
/// dialled by one, it reaches that host itself.
fn is_local(address: &Multiaddr) -> bool {
    match address.iter().next() {
        Some(Protocol::Ip4(ip)) => ip.is_loopback() || ip.is_link_local() || ip.is_unspecified(),
        Some(Protocol::Ip6(ip)) => {
            ip.is_loopback() || ip.is_unicast_link_local() || ip.is_unspecified()
        }
        _ => false,
    }
}

/// The addresses of `all` worth publishing: all but local ones, unless they're all
/// local (a test on one host).
fn published(all: &[Multiaddr]) -> Vec<Multiaddr> {
    let reachable: Vec<Multiaddr> = all
        .iter()
        .filter(|address| !is_local(address))
        .cloned()
        .collect();
    if reachable.is_empty() {
        all.to_vec()
    } else {
        reachable
    }
}

impl Book {
    /// `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;
        }
        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))
}

/// `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 mut swarm = if config.memory {
            memory_swarm(config.key, config.relay, config.mdns)?
        } else {
            network_swarm(config.key, config.relay, config.mdns)?
        };
        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 = control
            .accept(PROTOCOL)
            .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?;
        let configured_relays = !config.relays.is_empty();
        let book = Arc::new(Mutex::new(Book {
            announced: config.announce,
            serving: config.relay,
            ..Book::default()
        }));
        let (commands, receiver) = mpsc::unbounded_channel();
        for relay in config.relays {
            let _ = commands.send(Command::Reserve {
                configured: relay.clone(),
                relay,
            });
        }
        let (stopping, stopped) = watch::channel(false);
        tokio::spawn(run(
            swarm,
            receiver,
            commands.clone(),
            Arc::clone(&book),
            Relaying {
                serving: config.relay,
                configured: configured_relays,
                local: peer_id,
            },
            stopping,
        ));
        info!(peer_id = %peer_id, "libp2p transport started");
        Ok(Self {
            local_id: peer_id.to_string(),
            peer_id,
            control,
            commands,
            book,
            incoming: tokio::sync::Mutex::new(Some(incoming)),
            stopped,
        })
    }

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

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

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

impl Behaviour {
    fn new(
        key: &Keypair,
        relay_client: ::libp2p::relay::client::Behaviour,
        relay: bool,
        mdns: bool,
    ) -> std::io::Result<Self> {
        let peer_id = key.public().to_peer_id();
        let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
        let relay = relay.then(|| {
            ::libp2p::relay::Behaviour::new(
                peer_id,
                ::libp2p::relay::Config {
                    max_circuits: RELAY_CIRCUITS,
                    max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER,
                    max_circuit_duration: RELAY_CIRCUIT_DURATION,
                    max_circuit_bytes: RELAY_CIRCUIT_BYTES,
                    ..Default::default()
                },
            )
        });
        let mdns = match mdns {
            true => Some(::libp2p::mdns::tokio::Behaviour::new(
                ::libp2p::mdns::Config::default(),
                peer_id,
            )?),
            false => None,
        };
        Ok(Self {
            stream: libp2p_stream::Behaviour::new(),
            identify: ::libp2p::identify::Behaviour::new(identify),
            relay_client,
            relay: relay.into(),
            dcutr: ::libp2p::dcutr::Behaviour::new(peer_id),
            autonat: ::libp2p::autonat::Behaviour::new(
                peer_id,
                ::libp2p::autonat::Config {
                    // Swarms on private networks (a LAN, WireGuard) count too.
                    only_global_ips: false,
                    ..Default::default()
                },
            ),
            mdns: mdns.into(),
        })
    }
}

fn network_swarm(key: Keypair, relay: bool, mdns: bool) -> 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, relay, mdns)
                .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>)
        })?
        .with_swarm_config(|config| config.with_idle_connection_timeout(IDLE_TIMEOUT))
        .build())
}

fn memory_swarm(key: Keypair, relay: bool, mdns: bool) -> 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, relay, mdns)?,
        peer_id,
        ::libp2p::swarm::Config::with_tokio_executor().with_idle_connection_timeout(IDLE_TIMEOUT),
    ))
}

/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
/// transport is dropped.
/// How a node relates to relays.
struct Relaying {
    /// It relays for others.
    serving: bool,
    /// Relays to use are configured: it doesn't pick any.
    configured: 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 among the members, 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");
        for listener in std::mem::take(picked).into_keys() {
            swarm.remove_listener(listener);
        }
        return;
    }
    let (direct, offers) = {
        let book = lock(book);
        (!book.direct().is_empty(), book.relays.clone())
    };
    let using: Vec<PeerId> = picked.values().copied().collect();
    for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) {
        match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
            Ok(listener) => {
                info!(relay = %relay, address = %address, "Picked a member as libp2p relay");
                picked.insert(listener, relay);
            }
            Err(error) => {
                debug!(relay = %relay, error = %error, "Couldn't use a member as libp2p relay");
            }
        }
    }
}

/// `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 among the members, 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);
    // Relays we listen through, by listener, as configured.
    let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> =
        HashMap::new();
    let reserve_later = |configured: Multiaddr| {
        let again = again.clone();
        tokio::spawn(async move {
            tokio::time::sleep(RELAY_RETRY).await;
            let _ = again.send(Command::Reserve {
                relay: configured.clone(),
                configured,
            });
        });
    };
    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();
    loop {
        tokio::select! {
            _ = check.tick(), if !relaying.configured => {
                pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
            }
            command = commands.recv() => match command {
                None => break,
                Some(Command::CheckRelays) => {
                    if !relaying.configured {
                        pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
                    }
                }
                Some(Command::Connect { peer, addresses, done }) => {
                    if swarm.is_connected(&peer) {
                        let _ = done.send(Ok(()));
                        continue;
                    }
                    let waiting = connecting.entry(peer).or_default();
                    waiting.push(done);
                    if waiting.len() > 1 {
                        continue; // Already dialing.
                    }
                    let opts = DialOpts::peer_id(peer)
                        .addresses(addresses)
                        .condition(PeerCondition::DisconnectedAndNotDialing)
                        .build();
                    if let Err(error) = swarm.dial(opts) {
                        for done in connecting.remove(&peer).into_iter().flatten() {
                            let _ = done.send(Err(error.to_string()));
                        }
                    }
                }
                Some(Command::Reserve { relay, configured }) if split_peer(&relay).is_none() => {
                    // Which peer relays there first.
                    let again = again.clone();
                    tokio::spawn(async move {
                        let (done, identified) = oneshot::channel();
                        if again
                            .send(Command::Identify { address: relay.clone(), done })
                            .is_err()
                        {
                            return;
                        }
                        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
                            Ok(Ok(Ok(peer))) => {
                                if let Ok(relay) = relay.with_p2p(peer) {
                                    let _ = again.send(Command::Reserve { relay, configured });
                                }
                            }
                            _ => {
                                warn!(relay = %relay, "Couldn't reach a libp2p relay; trying again later");
                                tokio::time::sleep(RELAY_RETRY).await;
                                let _ = again.send(Command::Reserve { relay, configured });
                            }
                        }
                    });
                }
                Some(Command::Reserve { relay, configured }) => {
                    match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) {
                        Ok(listener) => {
                            debug!(relay = %relay, "Reserving a slot on a libp2p relay");
                            reservations.insert(listener, configured);
                        }
                        Err(error) => {
                            warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later");
                            reserve_later(configured);
                        }
                    }
                }
                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, .. } => {
                    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::OutgoingConnectionError { connection_id, peer_id, error } => {
                    if let Some((_, done)) = identifying.remove(&connection_id) {
                        let _ = done.send(Err(error.to_string()));
                    }
                    if let Some(peer) = peer_id
                        && !swarm.is_connected(&peer)
                    {
                        for done in connecting.remove(&peer).into_iter().flatten() {
                            let _ = done.send(Err(error.to_string()));
                        }
                    }
                }
                SwarmEvent::NewListenAddr { address, .. } => {
                    info!(address = %address, "libp2p 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, "libp2p 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));
                    if let Some(relay) = reservations.remove(&listener_id) {
                        warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later");
                        reserve_later(relay);
                    }
                    // A relay picked among the members: the next check picks again.
                    if let Some(relay) = picked.remove(&listener_id) {
                        debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying");
                    }
                }
                SwarmEvent::ListenerError { error, .. } => {
                    warn!(error = %error, "libp2p listener failed");
                }
                SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                    ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. },
                )) => {
                    if !renewal {
                        info!(relay = %relay_peer_id, "Reachable through a libp2p relay");
                    }
                }
                SwarmEvent::Behaviour(BehaviourEvent::Autonat(
                    ::libp2p::autonat::Event::StatusChanged { old, new },
                )) => {
                    info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)");
                    nat = new;
                }
                SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
                    debug!(event = ?event, "libp2p relay");
                }
                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                    Ok(_) => info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"),
                    Err(error) => debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"),
                },
                SwarmEvent::Behaviour(BehaviourEvent::Identify(
                    ::libp2p::identify::Event::Received { peer_id, info, .. },
                )) => {
                    debug!(
                        peer = %peer_id,
                        listen = ?info.listen_addrs,
                        observed = %info.observed_addr,
                        agent = %info.agent_version,
                        "libp2p peer identified"
                    );
                    let mut book = lock(&book);
                    // Last first: `remember` puts each in front.
                    for address in published(&info.listen_addrs).into_iter().rev() {
                        book.remember(peer_id, address);
                    }
                    if info.protocols.contains(&PROTOCOL) {
                        book.continuum.insert(peer_id);
                        if book.lan.remove(&peer_id) {
                            book.discover(peer_id);
                        }
                    }
                }
                SwarmEvent::Behaviour(BehaviourEvent::Mdns(
                    ::libp2p::mdns::Event::Discovered(found),
                )) => {
                    for (peer, address) in found {
                        if peer == relaying.local {
                            continue;
                        }
                        debug!(peer = %peer, address = %address, "libp2p peer on the LAN");
                        let continuum = {
                            let mut book = lock(&book);
                            book.remember(peer, address.clone());
                            book.continuum.contains(&peer)
                        };
                        if continuum {
                            lock(&book).discover(peer);
                        } else if lock(&book).lan.len() < MAX_PEERS {
                            // Connected, identify says whether it speaks continuum.
                            lock(&book).lan.insert(peer);
                            let _ = swarm.dial(
                                DialOpts::peer_id(peer)
                                    .addresses(vec![address])
                                    .condition(PeerCondition::DisconnectedAndNotDialing)
                                    .build(),
                            );
                        }
                    }
                }
                _ => {}
            },
        }
    }
    let _ = stopping.send(true);
}

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

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

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

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

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

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

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

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

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

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

    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))?;
        self.connect(peer).await?;
        let mut control = self.control.clone();
        let body = message.body.clone();
        let write = async move {
            let mut stream = control.open_stream(peer, PROTOCOL).await?;
            stream.write_all(&(body.len() as u32).to_be_bytes()).await?;
            stream.write_all(&body).await?;
            stream.close().await?;
            Ok::<(), anyhow::Error>(())
        };
        let timeout = stream_timeout(message.body.len());
        tokio::time::timeout(timeout, write)
            .await
            .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))?
            .with_context(|| format!("Sending to {peer}"))
    }

    /// Reads inbound streams (`READS` at once) and handles their messages one at a
    /// time, in the order they were read.
    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
        let Some(mut incoming) = self.incoming.lock().await.take() else {
            bail!("libp2p consumer already started");
        };
        let (sender, mut received) = mpsc::channel::<(PeerId, Vec<u8>)>(READS);
        let reads = Arc::new(Semaphore::new(READS));
        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();
                tokio::spawn(async move {
                    match read_message(stream).await {
                        Ok(body) => {
                            let _ = sender.send((peer, body)).await;
                        }
                        Err(error) => {
                            debug!(peer = %peer, error = %format!("{error:#}"), "Dropping an unreadable libp2p stream");
                        }
                    }
                    drop(permit);
                });
            }
        });
        while let Some((peer, body)) = received.recv().await {
            match wire::decode(&body) {
                Ok(message) => {
                    if let Err(error) = handler(peer.to_string(), message, body.len()).await {
                        warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message");
                    }
                }
                Err(error) => {
                    debug!(peer = %peer, error = %error, "Dropping a libp2p message that doesn't decode");
                }
            }
        }
        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, Keepalive};

    /// 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,
            mdns: 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 ping() -> ParsedMessage {
        ParsedMessage::Ping(Keepalive {
            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(), &ping()).await.unwrap();
        let (from, message) = at_a.recv().await.unwrap();
        assert_eq!(from, b.local_id());
        assert!(matches!(message, ParsedMessage::Ping(keepalive) if keepalive.epoch_seq == 3));

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

    #[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(), &ping()).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,
            mdns: false,
            memory: true,
        })
        .unwrap();
        assert_eq!(
            transport.local_addresses(),
            [announced.with_p2p(peer).unwrap().to_string()]
        );
    }

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

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

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

    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()]);
    }

    #[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, &ping()).await.is_err());
        assert!(b.resolve_seed(&stranger).await.is_err());
    }
}