libp2p: explode!

1 files changed, 657 insertions, 0 deletions

diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs
new file mode 100644
index 0000000..86e7734
--- /dev/null
+++ b/src/transport/libp2p/mod.rs
@@ -0,0 +1,657 @@
+//! libp2p as a transport (`--transports libp2p`, in builds with the `libp2p` feature):
+//! direct connections over TCP (Noise, Yamux) and QUIC, to addresses this node knows.
+//!
+//! A node's transport identity is its libp2p peer ID; where it is, its listen addresses
+//! (multiaddrs ending in the peer ID, `locations`). Seeds are such multiaddrs, or ones
+//! without the peer ID, which a connection finds out. Messages go one per stream of the
+//! `/continuum/1` protocol: their length (4 bytes, big endian), then the message as
+//! `wire` encodes it. A send is done once the stream is written and closed; a connection
+//! that can't be made, or a stream that can't be written, refuses it. A stream's protocol
+//! goes with its first data, without waiting for the other side to agree
+//! (`swarm_config`): opening a stream costs no round trip.
+//!
+//! Connected peers tell each other where they listen (libp2p's identify protocol), so a
+//! node learns the addresses of peers that dialled it. A node publishes where it can be
+//! reached (`local_addresses`): the addresses it announces (`--libp2p-announce`, e.g. a
+//! public address forwarded to it), or else those it listens on, loopback left out.
+//!
+//! A node nobody can dial (behind NAT, or a closed port) can still be reached through a
+//! relay: a node of the swarm that relays (`--libp2p-relay`), or one configured
+//! (`--libp2p-relays`), on which it reserves a slot; its address through the relay
+//! (`<relay>/p2p-circuit/p2p/<it>`) is published with the others. Over a relayed connection, both sides then try to
+//! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until
+//! that works, or for good if it doesn't (punched again now and then, `PUNCH_RETRIES`),
+//! within its limits (`RELAY_CIRCUIT_DURATION`,
+//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). Hole
+//! punching needs both sides to listen (`HolePunching`: at their LAN addresses too);
+//! where only one does, the other dials it directly once it knows where (`upgrade`).
+//!
+//! Of several connections to one peer (both dialled at once, or AutoNAT dialled back),
+//! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`),
+//! by the side with the smaller peer ID: it keeps the direct one with the shortest round
+//! trip (libp2p's ping, on each connection), so a path over the LAN wins over one
+//! through a VPN, whichever opened first.
+//!
+//! A relay says so where it publishes its addresses: next to each, the same address
+//! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the
+//! member list. A node that needs a relay picks up to `AUTO_RELAYS` relays
+//! (`relay_check`): those configured first, then those listed in the document
+//! (`learn_relays`), then members that relay. It needs one when
+//! it has no address others could dial, or when AutoNAT says it can't be dialled:
+//! AutoNAT asks connected peers (nodes of the swarm) to dial it back. Once AutoNAT says
+//! it can be dialled, it lets the relays it picked go.
+//!
+//! With `--libp2p-ignore`, address ranges are left out (VPNs): not published, not
+//! offered for hole punching, not dialled, and connections there refused (`Ignoring`).
+//!
+//! With `--libp2p-upnp`, a node asks its router to forward its listen ports (UPnP), and
+//! publishes the router's public address for them with its own (`Book::direct`): then
+//! it can be dialled from elsewhere, whatever the other side's NAT.
+//!
+//! With `--libp2p-mdns`, a node finds libp2p peers on its LAN (multicast DNS). Their
+//! LAN addresses are noted, and those that turn out to speak continuum (identify says
+//! so) are seeds while the node joins (`discovered_seeds`): nodes on one LAN find each
+//! other without configured seeds.
+//!
+//! It uses public libp2p infrastructure only with `--libp2p-public`: then it joins the
+//! IPFS DHT (Kademlia, through the IPFS bootstrap nodes), where it finds addresses of
+//! peers the member list doesn't have, and public peers that relay; those are picked
+//! after members that relay. Public relays keep circuits short (by default 2 minutes and
+//! 128 KiB): enough to punch holes through NATs, not to carry a swarm's traffic. Without
+//! it, relays are those configured, or members that say they relay.
+
+use std::collections::HashMap;
+use std::path::Path;
+use std::str::FromStr;
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use ::libp2p::multiaddr::Protocol;
+use ::libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
+use ::libp2p::swarm::{ConnectionId, SwarmEvent};
+use ::libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm, identity::Keypair};
+use anyhow::{Context, Result, anyhow, bail};
+use async_trait::async_trait;
+use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
+use tokio::sync::{Semaphore, mpsc, oneshot, watch};
+use tracing::{debug, info, trace, warn};
+
+use crate::locks::MutexExt;
+use crate::transport::stats::{
+    Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats,
+};
+use crate::transport::{
+    Addressed, Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport,
+    hand_on_probe, is_probe, wire,
+};
+
+mod behaviour;
+mod book;
+mod config;
+mod driver;
+mod errors;
+mod ignore;
+mod relays;
+
+use behaviour::*;
+use book::*;
+pub(crate) use config::{Libp2pConfig, check};
+use driver::*;
+use errors::*;
+use ignore::*;
+use relays::*;
+
+/// The scheme of this transport's addresses (`transport::paths`).
+pub(crate) const SCHEME: &str = "libp2p";
+
+const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1");
+/// Sends taking this long are logged with where the time went (at debug).
+const SLOW_SEND: Duration = Duration::from_millis(5);
+/// The largest message read: anyone who can connect can send anything, so this bounds
+/// what one stream costs. Sync messages above 16 chunks travel as blobs.
+const MAX_MESSAGE: usize = 128 << 20;
+/// How long a connection may take to come up.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+/// A connection nobody uses is closed after this; neighbours keep theirs busy.
+const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
+/// Addresses of one dial tried at once (libp2p's default: 8). Hole punching (`HolePunching`)
+/// dials every address both sides offer, where they listen (LAN, VPNs, public IPv6) and
+/// where others saw them, often more than 8, while the other side punches towards all of
+/// them at the same moment: an address dialled only after others failed (each after
+/// QUIC's 5 s handshake timeout) misses its punch, and the LAN's is often among the
+/// last. Dials of the member list's addresses (`ADDRESSES_PER_PEER`) are unchanged.
+const DIAL_CONCURRENCY: std::num::NonZeroU8 = std::num::NonZeroU8::new(32).unwrap();
+/// How long a stream may take to open and to carry a message, before what its size adds
+/// (`stream_timeout`).
+const STREAM_TIMEOUT: Duration = Duration::from_secs(10);
+/// The slowest a message is expected to go, in bytes per second.
+const MIN_RATE: u64 = 1 << 20;
+/// Inbound streams read at once.
+const READS: usize = 32;
+/// Addresses kept per peer, and peers with addresses at most.
+const ADDRESSES_PER_PEER: usize = 8;
+const MAX_PEERS: usize = 4096;
+
+/// A relayed connection (circuit) through this node lasts at most this long and carries
+/// at most this much: generous, since the swarm's own relays may carry sync traffic for
+/// good where hole punching fails (libp2p's defaults, 2 minutes and 128 KiB, are for
+/// the punching only).
+const RELAY_CIRCUIT_DURATION: Duration = Duration::from_secs(3600);
+const RELAY_CIRCUIT_BYTES: u64 = 1 << 30;
+/// Circuits through this node at once, and per peer (either end: a node behind the
+/// relay is an end of every circuit to it). libp2p's rate limits on circuits and slots
+/// (a burst, then one every minute or two, per peer and per IP address: every node
+/// behind one NAT shares one) are left out: they're for public relays serving
+/// strangers, and refused circuits used them up too.
+const RELAY_CIRCUITS: usize = 1024;
+const RELAY_CIRCUITS_PER_PEER: usize = 64;
+/// A configured relay that couldn't be reached to find out its peer ID is tried again
+/// after this.
+const RELAY_RETRY: Duration = Duration::from_secs(30);
+/// Relays picked among the members, at most, and how often a node checks whether it
+/// needs them.
+const AUTO_RELAYS: usize = 2;
+const RELAY_CHECK: Duration = Duration::from_secs(30);
+/// A relay that refused a slot (`relay_refused`: none left, not allowed, not a relay,
+/// another peer at its address: a relay restarted without its libp2p key) isn't picked
+/// again for this long. One that didn't grant it for another reason (it doesn't answer,
+/// the connection closed) isn't for `RELAY_RETRY`; a slot lost is reserved again at
+/// once.
+const RELAY_PAUSE: Duration = Duration::from_secs(600);
+/// A peer connected only through a relay is dialled directly where it can be reached
+/// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to
+/// `RELAY_PAUSE`.
+const UPGRADE_RETRY: Duration = Duration::from_secs(30);
+/// A peer nobody mentioned for this long (the member list, identify, mDNS, a
+/// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted
+/// without their libp2p key, whose new ones the member list has since. Members are
+/// mentioned every few seconds (`swarm::topology`). Checked every `PRUNE_EVERY`.
+const STALE_AFTER: Duration = Duration::from_secs(600);
+const PRUNE_EVERY: Duration = Duration::from_secs(60);
+/// Several connections to one peer (both dialled at once, or AutoNAT dialled back to
+/// see whether it can) are cut to one, the fastest (`Book::duplicates`), this long after
+/// the last one opened: long enough for what it was dialled for, and for each to be
+/// pinged (at once, as it opens). Looked for every `DEDUPE_EVERY`.
+const DUPLICATE_GRACE: Duration = Duration::from_secs(10);
+/// The same while a peer is connected only through relays: a new relayed connection is
+/// a hole punch under way (`PUNCH_RETRIES`), which takes its tries (up to a QUIC
+/// handshake's 5 s each, three times) on that connection.
+const PUNCH_GRACE: Duration = Duration::from_secs(60);
+/// After libp2p's hole punching gave up on a peer (three tries, as the relayed
+/// connection opened), it tries again after these, the last one over and over: what
+/// failed may have been passing (a NAT's mapping, a busy link). Each time through a new
+/// connection through the relay, on which the peer starts the punch (`retry_punches`).
+const PUNCH_RETRIES: [Duration; 3] = [
+    Duration::from_secs(60),
+    Duration::from_secs(300),
+    Duration::from_secs(1800),
+];
+const DEDUPE_EVERY: Duration = Duration::from_secs(5);
+
+/// How long a message of `bytes` may take on a stream.
+fn stream_timeout(bytes: usize) -> Duration {
+    STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE)
+}
+
+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>,
+}
+
+impl Libp2pTransport {
+    /// Builds the swarm, listens, and runs it in a task of its own.
+    pub(crate) fn start(config: Libp2pConfig) -> Result<Self> {
+        let peer_id = config.key.public().to_peer_id();
+        let features = Features::of(&config);
+        let mut swarm = if config.memory {
+            memory_swarm(config.key, features, config.ignore.clone())?
+        } else {
+            network_swarm(config.key, features, config.ignore.clone())?
+        };
+        for address in config.listen {
+            swarm
+                .listen_on(address.clone())
+                .with_context(|| format!("Listening on {address}"))?;
+        }
+        for address in &config.announce {
+            // Identify tells peers these.
+            swarm.add_external_address(address.clone());
+        }
+        let mut control = swarm.behaviour().stream.new_control();
+        let incoming = if config.continuum {
+            Some(
+                control
+                    .accept(PROTOCOL)
+                    .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?,
+            )
+        } else {
+            None
+        };
+        let (commands, receiver) = mpsc::unbounded_channel();
+        let mut book = Book {
+            announced: config.announce,
+            ignored: config.ignore.clone(),
+            serving: config.relay,
+            ..Book::default()
+        };
+        for relay in config.relays {
+            if let Some(address) = book.name_relay(relay, Named::Configured, peer_id) {
+                let _ = commands.send(Command::FindRelay { address });
+            }
+        }
+        let book = Arc::new(Mutex::new(book));
+        let (stopping, stopped) = watch::channel(false);
+        tokio::spawn(run(
+            swarm,
+            receiver,
+            commands.clone(),
+            Arc::clone(&book),
+            Relaying {
+                serving: config.relay,
+                public: config.public,
+                local: peer_id,
+            },
+            stopping,
+        ));
+        info!(peer_id = %peer_id, "Transport started");
+        Ok(Self {
+            local_id: peer_id.to_string(),
+            peer_id,
+            control,
+            commands,
+            book,
+            incoming: tokio::sync::Mutex::new(incoming),
+            stopped,
+        })
+    }
+
+    /// Connected to `peer`, if it can be had in time.
+    async fn connect(&self, peer: PeerId) -> Result<()> {
+        let addresses = {
+            let book = lock(&self.book);
+            let mut known = book
+                .addresses
+                .get(&peer)
+                .map(|known| dialable(known, self.peer_id))
+                .unwrap_or_default();
+            known.retain(|address| !ignored(&book.ignored, address));
+            if known.is_empty() {
+                book.through_relays(peer, self.peer_id)
+            } else {
+                known
+            }
+        };
+        let (done, connected) = oneshot::channel();
+        self.commands
+            .send(Command::Connect {
+                peer,
+                addresses,
+                done,
+            })
+            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
+        match tokio::time::timeout(CONNECT_TIMEOUT, connected).await {
+            Ok(Ok(Ok(()))) => Ok(()),
+            Ok(Ok(Err(error))) => bail!("couldn't connect to {peer}: {error}"),
+            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
+            Err(_) => bail!("connecting to {peer} took longer than {CONNECT_TIMEOUT:?}"),
+        }
+    }
+}
+
+/// One message from an inbound stream.
+async fn read_message(mut stream: ::libp2p::Stream) -> Result<Vec<u8>> {
+    let mut length = [0u8; 4];
+    tokio::time::timeout(STREAM_TIMEOUT, stream.read_exact(&mut length))
+        .await
+        .map_err(|_| anyhow!("no length within {STREAM_TIMEOUT:?}"))??;
+    let length = u32::from_be_bytes(length) as usize;
+    if length > MAX_MESSAGE {
+        bail!("a message of {length} bytes; at most {MAX_MESSAGE} are read");
+    }
+    let mut body = vec![0u8; length];
+    tokio::time::timeout(stream_timeout(length), stream.read_exact(&mut body))
+        .await
+        .map_err(|_| {
+            anyhow!(
+                "{length} bytes took longer than {:?}",
+                stream_timeout(length)
+            )
+        })??;
+    Ok(body)
+}
+
+#[async_trait]
+impl Transport for Libp2pTransport {
+    fn local_id(&self) -> &str {
+        &self.local_id
+    }
+
+    fn is_peer_id(&self, id: &str) -> bool {
+        PeerId::from_str(id).is_ok()
+    }
+
+    /// The longest a message may take on a stream.
+    fn delivery_limit(&self) -> Duration {
+        stream_timeout(MAX_MESSAGE)
+    }
+
+    /// Where we can be reached, each with our peer ID: the announced addresses, or else
+    /// those we listen on (`published`), and those through relays. A relay adds each
+    /// direct one again, ending in `/p2p-circuit`: where to reserve a slot.
+    fn local_addresses(&self) -> Vec<String> {
+        let book = lock(&self.book);
+        let direct: Vec<Multiaddr> = book
+            .direct()
+            .into_iter()
+            .filter_map(|address| address.with_p2p(self.peer_id).ok())
+            .collect();
+        let relayed = book
+            .listening
+            .iter()
+            .filter(|address| is_relayed(address))
+            .filter_map(|address| address.clone().with_p2p(self.peer_id).ok());
+        let offers: Vec<Multiaddr> = if book.serving {
+            direct
+                .iter()
+                .map(|address| address.clone().with(Protocol::P2pCircuit))
+                .collect()
+        } else {
+            Vec::new()
+        };
+        direct
+            .into_iter()
+            .chain(relayed)
+            .chain(offers)
+            .map(|address| address.to_string())
+            .collect()
+    }
+
+    /// A peer ID, or a multiaddr ending in one (noted as where that peer is).
+    fn locate(&self, target: &str) -> Option<String> {
+        if let Ok(peer) = PeerId::from_str(target) {
+            return Some(peer.to_string());
+        }
+        let mut address = Multiaddr::from_str(target).ok()?;
+        // A relay saying where to reserve a slot: `<address>/p2p/<relay>/p2p-circuit`.
+        if let Some(Protocol::P2pCircuit) = address.iter().last() {
+            address.pop();
+            let relay = split_peer(&address)?;
+            if relay != self.peer_id {
+                let new = {
+                    let mut book = lock(&self.book);
+                    let new = book.relays.get(&relay) != Some(&address);
+                    if new && (book.relays.len() < MAX_PEERS || book.relays.contains_key(&relay)) {
+                        book.relays.insert(relay, address.clone());
+                    }
+                    book.remember(relay, address);
+                    new
+                };
+                if new {
+                    let _ = self.commands.send(Command::CheckRelays);
+                }
+            }
+            return Some(relay.to_string());
+        }
+        let peer = split_peer(&address)?;
+        address.pop();
+        // A peer ID only says nothing about where.
+        if peer != self.peer_id && !address.is_empty() {
+            lock(&self.book).remember(peer, address);
+        }
+        Some(peer.to_string())
+    }
+
+    /// Multiaddrs, with or without the relay's peer ID; others are ignored.
+    fn learn_relays(&self, targets: &[String]) {
+        let listed = targets
+            .iter()
+            .filter_map(|target| Multiaddr::from_str(target).ok())
+            .collect();
+        let Some(found) = lock(&self.book).learn_listed(listed, self.peer_id) else {
+            return;
+        };
+        for address in found {
+            let _ = self.commands.send(Command::FindRelay { address });
+        }
+        let _ = self.commands.send(Command::CheckRelays);
+    }
+
+    fn connection(&self, target: &str) -> Option<Connection> {
+        let peer = PeerId::from_str(target).ok()?;
+        lock(&self.book).connection(&peer)
+    }
+
+    fn punching(&self, target: &str) -> Option<crate::transport::Punching> {
+        let peer = PeerId::from_str(target).ok()?;
+        let (failed, next) = *lock(&self.book).punching.get(&peer)?;
+        Some(crate::transport::Punching {
+            failed,
+            next_in_secs: next
+                .saturating_duration_since(tokio::time::Instant::now())
+                .as_secs(),
+        })
+    }
+
+    fn location(&self, target: &str) -> Option<String> {
+        let peer = PeerId::from_str(target).ok()?;
+        let address = lock(&self.book).location(&peer)?.clone();
+        Some(
+            address
+                .with_p2p(peer)
+                .unwrap_or_else(|address| address)
+                .to_string(),
+        )
+    }
+
+    fn stats(&self) -> TransportStats {
+        TransportStats {
+            libp2p: Some(lock(&self.book).stats()),
+            ..TransportStats::default()
+        }
+    }
+
+    /// Nodes found on the LAN (mDNS) that speak continuum.
+    fn discovered_seeds(&self) -> Vec<String> {
+        std::mem::take(&mut lock(&self.book).discovered)
+            .iter()
+            .map(Multiaddr::to_string)
+            .collect()
+    }
+
+    /// Where `target` is, each address with its peer ID.
+    fn locations(&self, target: &str) -> Vec<String> {
+        let Ok(peer) = PeerId::from_str(target) else {
+            return Vec::new();
+        };
+        lock(&self.book)
+            .addresses
+            .get(&peer)
+            .into_iter()
+            .flatten()
+            .filter_map(|address| address.clone().with_p2p(peer).ok())
+            .map(|address| address.to_string())
+            .collect()
+    }
+
+    /// A multiaddr ending in a peer ID is that peer; one without is connected to, to
+    /// find out. A bare peer ID resolves if this transport knows where it is.
+    async fn resolve_seed(&self, seed: &str) -> Result<String> {
+        let address = match PeerId::from_str(seed) {
+            Ok(peer) => Multiaddr::empty().with(Protocol::P2p(peer)),
+            Err(_) => {
+                Multiaddr::from_str(seed).with_context(|| format!("{seed} is no multiaddr"))?
+            }
+        };
+        // A peer ID only (`/p2p/<peer ID>`): reached where it's known, or through the
+        // relays we know.
+        if let [Protocol::P2p(peer)] = address.iter().collect::<Vec<_>>()[..] {
+            let book = lock(&self.book);
+            if book.addresses.contains_key(&peer)
+                || !book.through_relays(peer, self.peer_id).is_empty()
+            {
+                return Ok(peer.to_string());
+            }
+            bail!(
+                "no address known for libp2p peer {peer}, and no relay to reach it through; \
+                 give a multiaddr, or a relay (libp2p-relays)"
+            );
+        }
+        if let Some(peer) = self.locate(seed) {
+            return Ok(peer);
+        }
+        let (done, identified) = oneshot::channel();
+        self.commands
+            .send(Command::Identify { address, done })
+            .map_err(|_| anyhow!("the libp2p swarm stopped"))?;
+        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
+            Ok(Ok(Ok(peer))) => Ok(peer.to_string()),
+            Ok(Ok(Err(error))) => bail!("couldn't connect to {seed}: {error}"),
+            Ok(Err(_)) => bail!("the libp2p swarm stopped"),
+            Err(_) => bail!("connecting to {seed} took longer than {CONNECT_TIMEOUT:?}"),
+        }
+    }
+
+    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+        let body = wire::encode(message)?;
+        Ok(EncodedMessage {
+            peer: peer.to_owned(),
+            address: peer.to_owned(),
+            payload_len: body.len(),
+            body: body.into(),
+            kind: message.kind(),
+            message: None,
+        })
+    }
+
+    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
+        let peer = PeerId::from_str(&message.peer)
+            .with_context(|| format!("{} is no libp2p peer ID", message.peer))?;
+        let started = std::time::Instant::now();
+        self.connect(peer).await?;
+        // The protocol goes with the first data (`swarm_config`), so a peer that doesn't
+        // speak it seems to take the message: once identify has told, say so.
+        if let Some(relays) = lock(&self.book).not_nodes.get(&peer).copied() {
+            if relays {
+                bail!(
+                    "{peer} is a relay, not a node: give a node behind it as the seed \
+                     (<relay>/p2p-circuit/p2p/<peer ID>, or /p2p/<peer ID> with libp2p-relays)"
+                );
+            }
+            bail!("{peer} doesn't speak continuum");
+        }
+        let connected = started.elapsed();
+        let mut control = self.control.clone();
+        let body = message.body.clone();
+        let write = async move {
+            let mut stream = control.open_stream(peer, PROTOCOL).await?;
+            let opened = started.elapsed();
+            stream.write_all(&(body.len() as u32).to_be_bytes()).await?;
+            stream.write_all(&body).await?;
+            stream.close().await?;
+            // Kept until the peer has closed its side too. Over QUIC, dropping a stream
+            // stops its reading half (STOP_SENDING), and the peer's answer to the
+            // protocol (sent with the first data, `swarm_config`) then fails, and with it
+            // the stream, before the peer has read the message: identify, which reads
+            // its answers, works, while every message is lost. Yamux (TCP) doesn't mind.
+            // Off the send, which is done.
+            tokio::spawn(async move {
+                let mut rest = [0u8; 64];
+                let _ = tokio::time::timeout(STREAM_TIMEOUT, async {
+                    while let Ok(read) = stream.read(&mut rest).await {
+                        if read == 0 {
+                            break;
+                        }
+                    }
+                })
+                .await;
+            });
+            Ok::<_, anyhow::Error>(opened)
+        };
+        let timeout = stream_timeout(message.body.len());
+        let opened = tokio::time::timeout(timeout, write)
+            .await
+            .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))?
+            .with_context(|| format!("Sending to {peer}"))?;
+        let took = started.elapsed();
+        if took >= SLOW_SEND {
+            // Where a send's time goes: the swarm task (connecting), the stream's protocol
+            // negotiation (opening), or the connection (writing).
+            debug!(
+                peer = %Addressed(SCHEME, &peer),
+                kind = ?message.kind,
+                bytes = message.body.len(),
+                connect_ms = connected.as_secs_f64() * 1000.0,
+                open_ms = (opened - connected).as_secs_f64() * 1000.0,
+                write_ms = (took - opened).as_secs_f64() * 1000.0,
+                "Slow send"
+            );
+        }
+        Ok(())
+    }
+
+    /// Reads inbound streams (`READS` at once) and handles their messages one at a
+    /// time, in the order they were read; PINGs and PONGs at once (`is_probe`).
+    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        let Some(mut incoming) = self.incoming.lock().await.take() else {
+            bail!("libp2p consumer already started, or this node only relays");
+        };
+        let handler: Arc<HandlerFn> = Arc::from(handler);
+        let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS);
+        let reads = Arc::new(Semaphore::new(READS));
+        let probes = Arc::clone(&handler);
+        tokio::spawn(async move {
+            while let Some((peer, stream)) = incoming.next().await {
+                let Ok(permit) = Arc::clone(&reads).acquire_owned().await else {
+                    break;
+                };
+                let sender = sender.clone();
+                let probes = Arc::clone(&probes);
+                tokio::spawn(async move {
+                    match read_message(stream).await {
+                        Ok(body) => match wire::decode(&body) {
+                            Ok(message) if is_probe(&message) => {
+                                hand_on_probe(
+                                    SCHEME,
+                                    &probes,
+                                    peer.to_string(),
+                                    message,
+                                    body.len(),
+                                );
+                            }
+                            Ok(message) => {
+                                let _ = sender.send((peer, message, body.len())).await;
+                            }
+                            Err(error) => {
+                                debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Dropping a message that doesn't decode");
+                            }
+                        },
+                        Err(error) => {
+                            debug!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Dropping an unreadable stream");
+                        }
+                    }
+                    drop(permit);
+                });
+            }
+        });
+        while let Some((peer, message, payload_len)) = received.recv().await {
+            if let Err(error) = handler(peer.to_string(), message, payload_len).await {
+                warn!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Failed to process inbound message");
+            }
+        }
+        Ok(())
+    }
+
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        let mut stopped = self.stopped.clone();
+        let _ = stopped.wait_for(|stopped| *stopped).await;
+        Err(std::io::Error::other("the libp2p swarm stopped"))
+    }
+}
+
+#[cfg(test)]
+mod tests;