rename transport::p2p to transport::libp2p

1 files changed, 0 insertions, 1836 deletions

diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
deleted file mode 100644
index 30fdb23..0000000
--- a/src/transport/p2p.rs
+++ /dev/null
@@ -1,1836 +0,0 @@
-//! 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 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::{
-    Connection, 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);
-/// A relay that lost the slot before granting it (it doesn't answer, or another peer
-/// answers at its address: a relay restarted without its libp2p key) isn't picked again
-/// for this long.
-const RELAY_PAUSE: Duration = Duration::from_secs(600);
-/// A peer nobody mentioned for this long (the member list, identify, mDNS, a
-/// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted
-/// without their libp2p key, whose new ones the member list has since. Members are
-/// mentioned every few seconds (`swarm::topology`). Checked every `PRUNE_EVERY`.
-const STALE_AFTER: Duration = Duration::from_secs(600);
-const PRUNE_EVERY: Duration = Duration::from_secs(60);
-
-/// 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,
-    /// Use public infrastructure: the IPFS DHT, public relays.
-    pub(crate) public: bool,
-    /// libp2p's in-memory transport instead of TCP and QUIC: for tests.
-    pub(crate) memory: bool,
-}
-
-impl Libp2pConfig {
-    /// The key in `path` (32 raw bytes, an Ed25519 secret key), or a new one each start
-    /// without a file: the peer ID only says where a node is, so a new one costs nothing
-    /// but the WHO that finds it again (`transport::identity`).
-    pub(crate) fn load_key(path: Option<&Path>) -> Result<Keypair> {
-        let Some(path) = path else {
-            return Ok(Keypair::generate_ed25519());
-        };
-        let mut bytes = std::fs::read(path)
-            .with_context(|| format!("Reading the libp2p key from {}", path.display()))?;
-        crate::swarm::auth::warn_if_readable_by_others(path, "libp2p key");
-        if bytes.len() != 32 {
-            bail!(
-                "The libp2p key in {} has {} bytes; it has 32",
-                path.display(),
-                bytes.len()
-            );
-        }
-        Keypair::ed25519_from_bytes(&mut bytes)
-            .map_err(|error| anyhow!("The libp2p key in {}: {error}", path.display()))
-    }
-}
-
-enum Command {
-    /// Connect to `peer` (at `addresses`) unless connected.
-    Connect {
-        peer: PeerId,
-        addresses: Vec<Multiaddr>,
-        done: oneshot::Sender<Result<(), String>>,
-    },
-    /// Connect to `address` to find out which peer is there.
-    Identify {
-        address: Multiaddr,
-        done: oneshot::Sender<Result<PeerId, String>>,
-    },
-    /// See whether relays 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>,
-    /// With `Libp2pConfig::public`: other peers that relay (identify says so), and
-    /// where.
-    public_relays: HashMap<PeerId, Multiaddr>,
-    /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`).
-    failed_relays: HashMap<PeerId, tokio::time::Instant>,
-    /// Relays that granted us a slot we hold.
-    slots: std::collections::HashSet<PeerId>,
-    /// Open connections: to whom, and whether through a relay.
-    connections: HashMap<ConnectionId, (PeerId, bool)>,
-    /// What AutoNAT says, and the address it confirmed while public.
-    reachability: Reachability,
-    public_address: Option<Multiaddr>,
-    /// Since start (`stats`).
-    counts: Counts,
-    /// When each peer was last mentioned (`STALE_AFTER`).
-    seen: HashMap<PeerId, tokio::time::Instant>,
-    /// Peers found on the LAN (mDNS) not known yet to speak continuum.
-    lan: std::collections::HashSet<PeerId>,
-    /// Peers that speak continuum (identify says so).
-    continuum: std::collections::HashSet<PeerId>,
-    /// Peers found on the LAN that speak continuum, with their address: seeds not
-    /// handed out yet (`discovered_seeds`).
-    discovered: Vec<Multiaddr>,
-}
-
-/// What `Libp2pStats` counts since start.
-#[derive(Default)]
-struct Counts {
-    slots_granted: u64,
-    slots_lost: u64,
-    slots_not_granted: u64,
-    hole_punches: HolePunches,
-    dial_failures: DialFailures,
-}
-
-/// Counts a dial that failed with `error` (not one that wasn't needed).
-fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) {
-    use ::libp2p::swarm::DialError;
-    match error {
-        DialError::DialPeerConditionFalse(_) => {}
-        DialError::Transport(_) => failures.unreachable += 1,
-        DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1,
-        DialError::NoAddresses => failures.no_addresses += 1,
-        _ => failures.other += 1,
-    }
-}
-
-impl Book {
-    /// How we are connected to `peer`: directly if any connection is direct.
-    fn connection(&self, peer: &PeerId) -> Option<Connection> {
-        let mut found = None;
-        for (connected, relayed) in self.connections.values() {
-            if connected == peer {
-                if !relayed {
-                    return Some(Connection::Direct);
-                }
-                found = Some(Connection::Relayed);
-            }
-        }
-        found
-    }
-
-    fn stats(&self) -> Libp2pStats {
-        let mut connections = Connections::default();
-        for (_, relayed) in self.connections.values() {
-            if *relayed {
-                connections.relayed += 1;
-            } else {
-                connections.direct += 1;
-            }
-        }
-        Libp2pStats {
-            reachability: self.reachability,
-            public_address: self.public_address.as_ref().map(Multiaddr::to_string),
-            connections,
-            relay_slots: RelaySlots {
-                held: self.slots.len(),
-                paused: self
-                    .failed_relays
-                    .values()
-                    .filter(|at| at.elapsed() < RELAY_PAUSE)
-                    .count(),
-                granted: self.counts.slots_granted,
-                lost: self.counts.slots_lost,
-                not_granted: self.counts.slots_not_granted,
-            },
-            hole_punches: self.counts.hole_punches,
-            dial_failures: self.counts.dial_failures,
-        }
-    }
-
-    /// `peer` was mentioned just now.
-    fn touch(&mut self, peer: PeerId) {
-        self.seen.insert(peer, tokio::time::Instant::now());
-    }
-
-    /// Forgets the peers not mentioned within `STALE_AFTER` before `now`, but those
-    /// `connected`. Returns how many.
-    fn prune(&mut self, now: tokio::time::Instant, connected: impl Fn(&PeerId) -> bool) -> usize {
-        let stale: Vec<PeerId> = self
-            .addresses
-            .keys()
-            .chain(self.relays.keys())
-            .chain(self.public_relays.keys())
-            .chain(self.continuum.iter())
-            .chain(self.lan.iter())
-            .filter(|peer| {
-                !connected(peer)
-                    && self
-                        .seen
-                        .get(peer)
-                        .is_none_or(|at| now.saturating_duration_since(*at) >= STALE_AFTER)
-            })
-            .copied()
-            .collect::<std::collections::HashSet<_>>()
-            .into_iter()
-            .collect();
-        for peer in &stale {
-            self.addresses.remove(peer);
-            self.relays.remove(peer);
-            self.public_relays.remove(peer);
-            self.continuum.remove(peer);
-            self.lan.remove(peer);
-            self.failed_relays.remove(peer);
-            self.seen.remove(peer);
-        }
-        stale.len()
-    }
-
-    /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first.
-    fn discover(&mut self, peer: PeerId) {
-        let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else {
-            return;
-        };
-        if let Ok(address) = address.clone().with_p2p(peer)
-            && !self.discovered.contains(&address)
-            && self.discovered.len() < MAX_PEERS
-        {
-            self.discovered.push(address);
-        }
-    }
-
-    /// Where others could dial us directly: the announced addresses, or else those we
-    /// listen on (`published`), without the peer ID.
-    fn direct(&self) -> Vec<Multiaddr> {
-        if !self.announced.is_empty() {
-            return self.announced.clone();
-        }
-        let direct: Vec<Multiaddr> = self
-            .listening
-            .iter()
-            .filter(|address| !is_relayed(address))
-            .cloned()
-            .collect();
-        published(&direct)
-    }
-}
-
-/// Whether `address` only reaches this host (or this link): no use to other hosts, and
-/// dialled by one, it reaches that host itself.
-fn is_local(address: &Multiaddr) -> bool {
-    match address.iter().next() {
-        Some(Protocol::Ip4(ip)) => ip.is_loopback() || ip.is_link_local() || ip.is_unspecified(),
-        Some(Protocol::Ip6(ip)) => {
-            ip.is_loopback() || ip.is_unicast_link_local() || ip.is_unspecified()
-        }
-        _ => false,
-    }
-}
-
-/// The addresses of `all` worth publishing: all but local ones, unless they're all
-/// local (a test on one host).
-fn published(all: &[Multiaddr]) -> Vec<Multiaddr> {
-    let reachable: Vec<Multiaddr> = all
-        .iter()
-        .filter(|address| !is_local(address))
-        .cloned()
-        .collect();
-    if reachable.is_empty() {
-        all.to_vec()
-    } else {
-        reachable
-    }
-}
-
-impl Book {
-    /// Whether all of `addresses` are known of `peer` (`remember`).
-    fn knows(&self, peer: PeerId, addresses: &[Multiaddr]) -> bool {
-        let Some(known) = self.addresses.get(&peer) else {
-            return false;
-        };
-        addresses.iter().all(|address| {
-            let mut address = address.clone();
-            if let Some(Protocol::P2p(_)) = address.iter().last() {
-                address.pop();
-            }
-            known.contains(&address)
-        })
-    }
-
-    /// `peer` is at `address` (without its `/p2p` part): kept first.
-    fn remember(&mut self, peer: PeerId, mut address: Multiaddr) {
-        if let Some(Protocol::P2p(_)) = address.iter().last() {
-            address.pop();
-        }
-        if !self.addresses.contains_key(&peer) && self.addresses.len() >= MAX_PEERS {
-            return;
-        }
-        self.touch(peer);
-        let addresses = self.addresses.entry(peer).or_default();
-        addresses.retain(|known| *known != address);
-        addresses.insert(0, address);
-        addresses.truncate(ADDRESSES_PER_PEER);
-    }
-}
-
-/// Whether `address` goes through a relay.
-fn is_relayed(address: &Multiaddr) -> bool {
-    address
-        .iter()
-        .any(|protocol| matches!(protocol, Protocol::P2pCircuit))
-}
-
-/// The relay `address` goes through, if it goes through one: the peer before
-/// `/p2p-circuit`.
-fn relay_of(address: &Multiaddr) -> Option<PeerId> {
-    let mut relay = None;
-    for protocol in address.iter() {
-        match protocol {
-            Protocol::P2p(peer) => relay = Some(peer),
-            Protocol::P2pCircuit => return relay,
-            _ => {}
-        }
-    }
-    None
-}
-
-/// The addresses of `known` this node can dial: not those through itself as the relay
-/// (a relay reaches the nodes it relays for only while they are connected to it).
-fn dialable(known: &[Multiaddr], local: PeerId) -> Vec<Multiaddr> {
-    known
-        .iter()
-        .filter(|address| relay_of(address) != Some(local))
-        .cloned()
-        .collect()
-}
-
-/// `address` with the peer ID it ends in, if it ends in one.
-fn split_peer(address: &Multiaddr) -> Option<PeerId> {
-    match address.iter().last() {
-        Some(Protocol::P2p(peer)) => Some(peer),
-        _ => None,
-    }
-}
-
-pub(crate) struct Libp2pTransport {
-    local_id: String,
-    peer_id: PeerId,
-    control: libp2p_stream::Control,
-    commands: mpsc::UnboundedSender<Command>,
-    book: Arc<Mutex<Book>>,
-    incoming: tokio::sync::Mutex<Option<libp2p_stream::IncomingStreams>>,
-    stopped: watch::Receiver<bool>,
-}
-
-fn lock(book: &Mutex<Book>) -> MutexGuard<'_, Book> {
-    book.lock_or_recover()
-}
-
-impl Libp2pTransport {
-    /// Builds the swarm, listens, and runs it in a task of its own.
-    pub(crate) fn start(config: Libp2pConfig) -> Result<Self> {
-        let peer_id = config.key.public().to_peer_id();
-        let features = Features::of(&config);
-        let mut swarm = if config.memory {
-            memory_swarm(config.key, features)?
-        } else {
-            network_swarm(config.key, features)?
-        };
-        for address in config.listen {
-            swarm
-                .listen_on(address.clone())
-                .with_context(|| format!("Listening on {address}"))?;
-        }
-        for address in &config.announce {
-            // Identify tells peers these.
-            swarm.add_external_address(address.clone());
-        }
-        let mut control = swarm.behaviour().stream.new_control();
-        let incoming = 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,
-                public: config.public,
-                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)
-            .map(|known| dialable(known, self.peer_id))
-            .unwrap_or_default();
-        let (done, connected) = oneshot::channel();
-        self.commands
-            .send(Command::Connect {
-                peer,
-                addresses,
-                done,
-            })
-            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
-        match tokio::time::timeout(CONNECT_TIMEOUT, connected).await {
-            Ok(Ok(Ok(()))) => Ok(()),
-            Ok(Ok(Err(error))) => bail!("couldn't connect to {peer}: {error}"),
-            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
-            Err(_) => bail!("connecting to {peer} took longer than {CONNECT_TIMEOUT:?}"),
-        }
-    }
-}
-
-/// The protocol version identify says (informational: nothing checks it).
-const IDENTIFY_VERSION: &str = "/continuum/1";
-
-#[derive(::libp2p::swarm::NetworkBehaviour)]
-struct Behaviour {
-    stream: libp2p_stream::Behaviour,
-    identify: ::libp2p::identify::Behaviour,
-    relay_client: ::libp2p::relay::client::Behaviour,
-    /// Only with `Libp2pConfig::relay`.
-    relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
-    dcutr: ::libp2p::dcutr::Behaviour,
-    autonat: ::libp2p::autonat::Behaviour,
-    /// Only with `Libp2pConfig::mdns`.
-    mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
-    /// Only with `Libp2pConfig::public`: the IPFS DHT.
-    kad: ::libp2p::swarm::behaviour::toggle::Toggle<
-        ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>,
-    >,
-}
-
-/// What the behaviour has besides what every node has.
-#[derive(Clone, Copy)]
-struct Features {
-    relay: bool,
-    mdns: bool,
-    public: bool,
-}
-
-impl Features {
-    fn of(config: &Libp2pConfig) -> Self {
-        Self {
-            relay: config.relay,
-            mdns: config.mdns,
-            public: config.public,
-        }
-    }
-}
-
-/// The IPFS DHT's protocol, and its bootstrap nodes (all at `IPFS_BOOTSTRAP_ADDRESS`).
-const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");
-const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io";
-const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [
-    "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
-    "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
-    "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
-    "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
-];
-/// How often a node on the DHT refreshes its routing table.
-const DHT_BOOTSTRAP: Duration = Duration::from_secs(300);
-
-impl Behaviour {
-    fn new(
-        key: &Keypair,
-        relay_client: ::libp2p::relay::client::Behaviour,
-        features: Features,
-    ) -> std::io::Result<Self> {
-        let Features {
-            relay,
-            mdns,
-            public,
-        } = features;
-        let peer_id = key.public().to_peer_id();
-        let kad = public.then(|| {
-            let mut dht = ::libp2p::kad::Behaviour::with_config(
-                peer_id,
-                ::libp2p::kad::store::MemoryStore::new(peer_id),
-                ::libp2p::kad::Config::new(IPFS_DHT),
-            );
-            let bootstrap: Multiaddr = IPFS_BOOTSTRAP_ADDRESS.parse().expect("a multiaddr");
-            for peer in IPFS_BOOTSTRAP_PEERS {
-                let peer: PeerId = peer.parse().expect("a peer ID");
-                let _ = dht.add_address(&peer, bootstrap.clone());
-            }
-            dht
-        });
-        let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
-            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
-        let relay = relay.then(|| {
-            ::libp2p::relay::Behaviour::new(
-                peer_id,
-                ::libp2p::relay::Config {
-                    max_circuits: RELAY_CIRCUITS,
-                    max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER,
-                    max_circuit_duration: RELAY_CIRCUIT_DURATION,
-                    max_circuit_bytes: RELAY_CIRCUIT_BYTES,
-                    ..Default::default()
-                },
-            )
-        });
-        let mdns = match mdns {
-            true => Some(::libp2p::mdns::tokio::Behaviour::new(
-                ::libp2p::mdns::Config::default(),
-                peer_id,
-            )?),
-            false => None,
-        };
-        Ok(Self {
-            stream: libp2p_stream::Behaviour::new(),
-            identify: ::libp2p::identify::Behaviour::new(identify),
-            relay_client,
-            relay: relay.into(),
-            dcutr: ::libp2p::dcutr::Behaviour::new(peer_id),
-            autonat: ::libp2p::autonat::Behaviour::new(
-                peer_id,
-                ::libp2p::autonat::Config {
-                    // Swarms on private networks (a LAN, WireGuard) count too.
-                    only_global_ips: false,
-                    ..Default::default()
-                },
-            ),
-            mdns: mdns.into(),
-            kad: kad.into(),
-        })
-    }
-}
-
-fn network_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
-    Ok(::libp2p::SwarmBuilder::with_existing_identity(key)
-        .with_tokio()
-        .with_tcp(
-            ::libp2p::tcp::Config::default(),
-            ::libp2p::noise::Config::new,
-            ::libp2p::yamux::Config::default,
-        )?
-        .with_quic()
-        .with_dns()?
-        .with_relay_client(
-            ::libp2p::noise::Config::new,
-            ::libp2p::yamux::Config::default,
-        )?
-        .with_behaviour(|key, relay_client| {
-            Behaviour::new(key, relay_client, features)
-                .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>)
-        })?
-        .with_swarm_config(|config| config.with_idle_connection_timeout(IDLE_TIMEOUT))
-        .build())
-}
-
-fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
-    use ::libp2p::Transport as _;
-    use ::libp2p::core::transport::{MemoryTransport, OrTransport};
-    use ::libp2p::core::upgrade::Version;
-
-    let peer_id = key.public().to_peer_id();
-    let (relay_transport, relay_client) = ::libp2p::relay::client::new(peer_id);
-    let transport = OrTransport::new(relay_transport, MemoryTransport::default())
-        .boxed()
-        .upgrade(Version::V1)
-        .authenticate(::libp2p::noise::Config::new(&key)?)
-        .multiplex(::libp2p::yamux::Config::default())
-        .timeout(CONNECT_TIMEOUT)
-        .boxed();
-    Ok(Swarm::new(
-        transport,
-        Behaviour::new(&key, relay_client, features)?,
-        peer_id,
-        ::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,
-    /// 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 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");
-        let mut book = lock(book);
-        for (listener, relay) in std::mem::take(picked) {
-            swarm.remove_listener(listener);
-            book.slots.remove(&relay);
-        }
-        return;
-    }
-    let (direct, offers, public) = {
-        let mut book = lock(book);
-        book.failed_relays
-            .retain(|_, at| at.elapsed() < RELAY_PAUSE);
-        let failed = book.failed_relays.clone();
-        let usable = |offers: &HashMap<PeerId, Multiaddr>| -> HashMap<PeerId, Multiaddr> {
-            offers
-                .iter()
-                .filter(|(relay, _)| !failed.contains_key(relay))
-                .map(|(relay, address)| (*relay, address.clone()))
-                .collect()
-        };
-        let public = if relaying.public {
-            usable(&book.public_relays)
-        } else {
-            HashMap::new()
-        };
-        (!book.direct().is_empty(), usable(&book.relays), public)
-    };
-    let mut using: Vec<PeerId> = picked.values().copied().collect();
-    // Members first, then public ones for what's left.
-    let mut wanted = relay_check(relaying.local, direct, nat, &offers, &using);
-    using.extend(wanted.iter().map(|(relay, _)| *relay));
-    wanted.extend(relay_check(relaying.local, direct, nat, &public, &using));
-    for (relay, address) in wanted {
-        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");
-            }
-        }
-    }
-}
-
-/// 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);
-    }
-}
-
-/// `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);
-    let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP);
-    dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    // Relays we listen through, by listener: as configured, and their peer.
-    let mut reservations: HashMap<::libp2p::core::transport::ListenerId, (Multiaddr, PeerId)> =
-        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();
-    let mut prune = tokio::time::interval(PRUNE_EVERY);
-    prune.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    loop {
-        tokio::select! {
-            _ = prune.tick() => {
-                let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer));
-                if forgotten > 0 {
-                    debug!(forgotten, "Forgot libp2p 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(), 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 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::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 }) => {
-                    let Some(relay_peer) = split_peer(&relay) else {
-                        continue; // The arm above takes those.
-                    };
-                    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, relay_peer));
-                        }
-                        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, endpoint, .. } => {
-                    {
-                        let mut book = lock(&book);
-                        book.touch(peer_id);
-                        let relayed = endpoint.is_relayed();
-                        book.connections.insert(connection_id, (peer_id, relayed));
-                        debug!(
-                            peer = %peer_id,
-                            connection = %connection_id,
-                            dialled = endpoint.is_dialer(),
-                            address = %endpoint.get_remote_address(),
-                            relayed,
-                            "libp2p connection opened"
-                        );
-                    }
-                    for done in connecting.remove(&peer_id).into_iter().flatten() {
-                        let _ = done.send(Ok(()));
-                    }
-                    if let Some((address, done)) = identifying.remove(&connection_id) {
-                        lock(&book).remember(peer_id, address);
-                        let _ = done.send(Ok(peer_id));
-                    }
-                }
-                SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => {
-                    lock(&book).connections.remove(&connection_id);
-                    debug!(
-                        peer = %peer_id,
-                        connection = %connection_id,
-                        cause = ?cause,
-                        "libp2p 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(error.to_string()));
-                    }
-                    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(error.to_string()));
-                            }
-                            find_on_dht(&mut swarm, peer);
-                        }
-                    }
-                }
-                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, relay_peer)) = reservations.remove(&listener_id) {
-                        {
-                            let mut book = lock(&book);
-                            if book.slots.remove(&relay_peer) {
-                                book.counts.slots_lost += 1;
-                            } else {
-                                book.counts.slots_not_granted += 1;
-                            }
-                        }
-                        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, another
-                    // one if this one never granted the slot.
-                    if let Some(relay) = picked.remove(&listener_id) {
-                        let lost = {
-                            let mut book = lock(&book);
-                            let lost = book.slots.remove(&relay);
-                            if lost {
-                                book.counts.slots_lost += 1;
-                            } else {
-                                book.counts.slots_not_granted += 1;
-                                book.failed_relays.insert(relay, tokio::time::Instant::now());
-                            }
-                            lost
-                        };
-                        if lost {
-                            debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying");
-                        } else {
-                            info!(
-                                relay = %relay,
-                                pause = ?RELAY_PAUSE,
-                                "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while"
-                            );
-                        }
-                    }
-                }
-                SwarmEvent::ListenerError { error, .. } => {
-                    warn!(error = %error, "libp2p 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 = %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)");
-                    {
-                        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)) => {
-                    debug!(event = ?event, "libp2p relay");
-                }
-                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
-                    Ok(_) => {
-                        lock(&book).counts.hole_punches.succeeded += 1;
-                        info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay");
-                    }
-                    Err(error) => {
-                        lock(&book).counts.hole_punches.failed += 1;
-                        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 { 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 = %peer_id, connection = %connection_id, "libp2p peer identified again");
-                    } else {
-                        debug!(
-                            peer = %peer_id,
-                            connection = %connection_id,
-                            listen = ?info.listen_addrs,
-                            observed = %info.observed_addr,
-                            agent = %info.agent_version,
-                            "libp2p peer identified"
-                        );
-                    }
-                    // Last first: `remember` puts each in front.
-                    for address in listen.into_iter().rev() {
-                        book.remember(peer_id, address);
-                    }
-                    if relaying.public
-                        && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME)
-                        && (book.public_relays.len() < MAX_PEERS
-                            || book.public_relays.contains_key(&peer_id))
-                        && let Some(address) = book
-                            .addresses
-                            .get(&peer_id)
-                            .and_then(|known| known.first())
-                            .and_then(|address| address.clone().with_p2p(peer_id).ok())
-                    {
-                        book.public_relays.insert(peer_id, address);
-                    }
-                    if info.protocols.contains(&PROTOCOL) {
-                        book.continuum.insert(peer_id);
-                        if book.lan.remove(&peer_id) {
-                            book.discover(peer_id);
-                        }
-                    }
-                }
-                SwarmEvent::Behaviour(BehaviourEvent::Mdns(
-                    ::libp2p::mdns::Event::Discovered(found),
-                )) => {
-                    for (peer, address) in found {
-                        if peer == relaying.local {
-                            continue;
-                        }
-                        debug!(peer = %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())
-    }
-
-    fn connection(&self, target: &str) -> Option<Connection> {
-        let peer = PeerId::from_str(target).ok()?;
-        lock(&self.book).connection(&peer)
-    }
-
-    fn stats(&self) -> TransportStats {
-        TransportStats {
-            libp2p: Some(lock(&self.book).stats()),
-            ..TransportStats::default()
-        }
-    }
-
-    /// Nodes found on the LAN (mDNS) that speak continuum.
-    fn discovered_seeds(&self) -> Vec<String> {
-        std::mem::take(&mut lock(&self.book).discovered)
-            .iter()
-            .map(Multiaddr::to_string)
-            .collect()
-    }
-
-    /// Where `target` is, each address with its peer ID.
-    fn locations(&self, target: &str) -> Vec<String> {
-        let Ok(peer) = PeerId::from_str(target) else {
-            return Vec::new();
-        };
-        lock(&self.book)
-            .addresses
-            .get(&peer)
-            .into_iter()
-            .flatten()
-            .filter_map(|address| address.clone().with_p2p(peer).ok())
-            .map(|address| address.to_string())
-            .collect()
-    }
-
-    /// A multiaddr ending in a peer ID is that peer; one without is connected to, to
-    /// find out. A bare peer ID resolves if this transport knows where it is.
-    async fn resolve_seed(&self, seed: &str) -> Result<String> {
-        if let Ok(peer) = PeerId::from_str(seed) {
-            if lock(&self.book).addresses.contains_key(&peer) {
-                return Ok(peer.to_string());
-            }
-            bail!("no address known for libp2p peer {peer}; give a multiaddr");
-        }
-        let address =
-            Multiaddr::from_str(seed).with_context(|| format!("{seed} is no multiaddr"))?;
-        if let Some(peer) = self.locate(seed) {
-            return Ok(peer);
-        }
-        let (done, identified) = oneshot::channel();
-        self.commands
-            .send(Command::Identify { address, done })
-            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
-        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
-            Ok(Ok(Ok(peer))) => Ok(peer.to_string()),
-            Ok(Ok(Err(error))) => bail!("couldn't connect to {seed}: {error}"),
-            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
-            Err(_) => bail!("connecting to {seed} took longer than {CONNECT_TIMEOUT:?}"),
-        }
-    }
-
-    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        let body = wire::encode(message)?;
-        Ok(EncodedMessage {
-            peer: peer.to_owned(),
-            address: peer.to_owned(),
-            payload_len: body.len(),
-            body: body.into(),
-            kind: message.kind(),
-            message: None,
-        })
-    }
-
-    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
-        let peer = PeerId::from_str(&message.peer)
-            .with_context(|| format!("{} is no libp2p peer ID", message.peer))?;
-        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, 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,
-            mdns: false,
-            public: false,
-            memory: true,
-        })
-        .unwrap();
-        let address = if listen {
-            address.with_p2p(peer).unwrap()
-        } else {
-            Multiaddr::empty().with(Protocol::P2p(peer))
-        };
-        (transport, address)
-    }
-
-    fn memory_node() -> (Libp2pTransport, Multiaddr) {
-        memory_node_with(true, Vec::new(), false)
-    }
-
-    fn status() -> ParsedMessage {
-        ParsedMessage::Status(Status {
-            heads: HeadsSummary::of(Vec::new()),
-            epoch: Epoch::GENESIS,
-            epoch_seq: 3,
-            ready: true,
-            syncs_with_you: false,
-        })
-    }
-
-    /// Starts `transport`'s consumer, forwarding what it receives.
-    fn received(
-        transport: Arc<Libp2pTransport>,
-    ) -> mpsc::UnboundedReceiver<(String, ParsedMessage)> {
-        let (sender, receiver) = mpsc::unbounded_channel();
-        tokio::spawn(async move {
-            transport
-                .start_consumer(Box::new(move |from, message, _| {
-                    let sender = sender.clone();
-                    Box::pin(async move {
-                        let _ = sender.send((from, message));
-                        Ok(())
-                    })
-                }))
-                .await
-        });
-        receiver
-    }
-
-    #[tokio::test]
-    async fn messages_both_ways_over_one_connection() {
-        let (a, a_address) = memory_node();
-        let (b, _) = memory_node();
-        let (a, b) = (Arc::new(a), Arc::new(b));
-        let mut at_a = received(Arc::clone(&a));
-        let mut at_b = received(Arc::clone(&b));
-
-        assert_eq!(
-            b.resolve_seed(&a_address.to_string()).await.unwrap(),
-            a.local_id()
-        );
-        b.send_message(a.local_id(), &status()).await.unwrap();
-        let (from, message) = at_a.recv().await.unwrap();
-        assert_eq!(from, b.local_id());
-        assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3));
-
-        // a knows no address of b, but b's connection is there.
-        a.send_message(b.local_id(), &status()).await.unwrap();
-        let (from, _) = at_b.recv().await.unwrap();
-        assert_eq!(from, a.local_id());
-
-        assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
-        assert_eq!(b.connection(a.local_id()), Some(Connection::Direct));
-        let stats = a.stats().libp2p.unwrap();
-        assert!(stats.connections.direct >= 1);
-        assert_eq!(stats.connections.relayed, 0);
-    }
-
-    #[tokio::test]
-    async fn a_seed_without_a_peer_id_is_connected_to_to_find_out() {
-        let (a, a_address) = memory_node();
-        let (b, _) = memory_node();
-        let mut bare = a_address.clone();
-        bare.pop();
-        assert_eq!(
-            b.resolve_seed(&bare.to_string()).await.unwrap(),
-            a.local_id()
-        );
-        // And b now knows where a is.
-        assert_eq!(b.locations(a.local_id()), [a_address.to_string()]);
-    }
-
-    #[tokio::test]
-    async fn listen_addresses_carry_the_peer_id() {
-        let (a, a_address) = memory_node();
-        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
-        while a.local_addresses().is_empty() && tokio::time::Instant::now() < deadline {
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-        assert_eq!(a.local_addresses(), [a_address.to_string()]);
-        assert_eq!(
-            a.locate(&a_address.to_string()).as_deref(),
-            Some(a.local_id())
-        );
-    }
-
-    /// Waits up to 5 s for `condition`.
-    async fn eventually(mut condition: impl FnMut() -> bool) -> bool {
-        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
-        while !condition() {
-            if tokio::time::Instant::now() >= deadline {
-                return false;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-        true
-    }
-
-    /// A node learns where a peer that dialled it listens (identify), and can then
-    /// pass it on.
-    #[tokio::test]
-    async fn a_dialled_node_learns_where_the_dialler_listens() {
-        let (a, a_address) = memory_node();
-        let (b, b_address) = memory_node();
-        let (a, b) = (Arc::new(a), Arc::new(b));
-        let _at_a = received(Arc::clone(&a));
-        b.resolve_seed(&a_address.to_string()).await.unwrap();
-        b.send_message(a.local_id(), &status()).await.unwrap();
-        let b_id = b.local_id().to_owned();
-        assert!(
-            eventually(|| a.locations(&b_id) == [b_address.to_string()]).await,
-            "a knows {:?} of b",
-            a.locations(&b_id)
-        );
-    }
-
-    #[tokio::test]
-    async fn announced_addresses_are_published_instead() {
-        let key = Keypair::generate_ed25519();
-        let peer = key.public().to_peer_id();
-        let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
-        let announced: Multiaddr = "/ip4/203.0.113.7/tcp/9661".parse().unwrap();
-        let transport = Libp2pTransport::start(Libp2pConfig {
-            key,
-            listen: vec![format!("/memory/{port}").parse().unwrap()],
-            announce: vec![announced.clone()],
-            relays: Vec::new(),
-            relay: false,
-            mdns: false,
-            public: false,
-            memory: true,
-        })
-        .unwrap();
-        assert_eq!(
-            transport.local_addresses(),
-            [announced.with_p2p(peer).unwrap().to_string()]
-        );
-    }
-
-    #[test]
-    fn local_addresses_are_left_out_unless_there_are_only_local_ones() {
-        let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
-        let loopback = parse("/ip4/127.0.0.1/tcp/1");
-        let link = parse("/ip6/fe80::1/tcp/1");
-        let lan = parse("/ip4/192.168.1.2/tcp/1");
-        let wireguard = parse("/ip6/fd00::2/tcp/1");
-        assert_eq!(
-            published(&[
-                loopback.clone(),
-                link.clone(),
-                lan.clone(),
-                wireguard.clone()
-            ]),
-            [lan, wireguard]
-        );
-        assert_eq!(
-            published(&[loopback.clone(), link.clone()]),
-            [loopback, link]
-        );
-    }
-
-    /// b listens nowhere but through the relay r; a reaches it there.
-    #[tokio::test]
-    async fn a_node_nobody_can_dial_is_reached_through_a_relay() {
-        let (_r, r_address) = memory_node_with(true, Vec::new(), true);
-        let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
-        let b = Arc::new(b);
-        let mut at_b = received(Arc::clone(&b));
-        assert!(
-            eventually(|| !b.local_addresses().is_empty()).await,
-            "b got no slot on the relay"
-        );
-        let relayed = b.local_addresses()[0].clone();
-        assert!(
-            relayed.starts_with(&format!("{r_address}/p2p-circuit/")),
-            "{relayed}"
-        );
-
-        assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await);
-        let slots = b.stats().libp2p.unwrap().relay_slots;
-        assert_eq!((slots.granted, slots.not_granted), (1, 0));
-
-        let (a, _) = memory_node();
-        assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
-        a.send_message(b.local_id(), &status()).await.unwrap();
-        let (from, _) = at_b.recv().await.unwrap();
-        assert_eq!(from, a.local_id());
-        assert!(a.connection(b.local_id()).is_some());
-    }
-
-    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 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 start = tokio::time::Instant::now();
-        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);
-        let later = start + 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.
-    }
-
-    /// 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;
-            book.connections
-                .insert(ConnectionId::new_unchecked(id), (peer, relayed));
-        };
-        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);
-        let connections = book.stats().connections;
-        assert_eq!((connections.direct, connections.relayed), (1, 2));
-    }
-}