dashboard: without LINKS; LINKS removed

The dashboard reads every member's links from the members themselves;
LINKS and LINKS_REPLY, swarm::links and their wire tags go (wire
version bumped). Nodes the browser can't reach are drawn from what the
reachable ones measure.

Auth plan M3.

4 files changed, 2 insertions, 1140 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
deleted file mode 100644
index 2234783..0000000
--- a/src/swarm/links.rs
+++ /dev/null
@@ -1,1132 +0,0 @@
-//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a
-//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its
-//! neighbours and the peers it holds dormant (`own_links`), each with the transport its
-//! messages go over, whether only through a relay, the round trip and whether it is
-//! dormant: as the member measures them, which no other node can.
-//!
-//! Every node answers, whatever its build: one with a dashboard asks any member, which
-//! may have none. Only members (in the member list) are answered, as the links name
-//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing
-//! is written to the document: asked for while someone looks, forgotten after.
-//!
-//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again
-//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a
-//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone
-//! subscribes (round trips change without telling), until the subscription runs out
-//! (the asker renews it with its next LINKS, so one that is gone stops it by not
-//! renewing) or a LINKS with 0 ends it.
-//!
-//! A member the asker can't reach (on none of its transports: one on Tor alone, asked
-//! from one on libp2p alone; or one that doesn't answer it, as behind NAT: as choosing
-//! neighbours judges it, `topology::reaches`) is asked through a peer that can
-//! (`relay_for`): a LINKS
-//! naming the members (`of`) asks the receiver to ask them on the sender's behalf, and
-//! to pass their links on (`relayed`). The receiver asks each itself, or through a peer
-//! of its own while hops are left (`hops`), for as long as the sender asked, and holds
-//! what arrives a while to send it together: the window (`batch_ms`) is the asker's for
-//! the whole way (at most `MAX_BATCH`), shared out among the hops that may be left, so
-//! the links reach the asker within it however long the chain (every hop holds
-//! `MIN_BATCH` at least). Such a
-//! subscription isn't ended by a message: it runs out, at every hop, unless the asker
-//! renews it, so one that is gone (or no longer looks) stops it within a lease.
-//!
-//! A way of asking a member that brings none of its links (`FAILED_AFTER`: we don't
-//! reach it after all, or the relay doesn't) is left out a while for that member, and
-//! the next ask (the renewal) goes the next way: so the old way runs out by itself, at
-//! no cost but the wait. Every hop does so for what it asks on.
-
-use std::collections::{HashMap, HashSet};
-use std::sync::Mutex;
-use std::time::Duration;
-
-use tokio::time::Instant;
-use tracing::debug;
-
-use crate::locks::MutexExt;
-use crate::state::AppState;
-use crate::swarm::{members, topology};
-use crate::transport::paths::{self, ClockOffset};
-use crate::transport::{
-    Connection, Destination, Epoch, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage,
-    RelayedLinks,
-};
-
-/// Links one reply lists at most (a member has about `--neighbours`, more if others
-/// prefer it).
-pub(crate) const MAX_LINKS: usize = 256;
-/// Other transports in a link at most.
-pub(crate) const MAX_OTHERS: usize = 8;
-/// A member answers each other one at most this often.
-const ANSWER_EVERY: Duration = Duration::from_secs(5);
-/// Replies kept from members at most (one each), and requests out.
-const MAX_KEPT: usize = 4096;
-/// Subscriptions a member keeps at most, and how long one lasts at most.
-const MAX_SUBSCRIBERS: usize = 64;
-const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300);
-/// How often a member with subscribers looks whether its links changed.
-const WATCH_EVERY: Duration = Duration::from_secs(2);
-/// A reply to a LINKS that doesn't subscribe is taken for this long.
-const REPLY_WITHIN: Duration = Duration::from_secs(30);
-/// Members a relayed LINKS names at most, and that one LINKS_REPLY passes on at most.
-const MAX_RELAYED: usize = 64;
-/// Bytes of relayed links one LINKS_REPLY carries at most, as encoded (one member's
-/// alone may be more): what a member reports is usually small (about 12 links of 150
-/// bytes), but may be up to `MAX_LINKS` links, and 64 such would make a message of
-/// megabytes.
-const MAX_RELAYED_BYTES: usize = 256 * 1024;
-/// How long relayed links are held to be sent together, along the whole way: the
-/// asker's window (at most `MAX_BATCH`), each hop holding at least `MIN_BATCH`.
-pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250);
-const MIN_BATCH: Duration = Duration::from_millis(10);
-const MAX_BATCH: Duration = Duration::from_secs(5);
-/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and
-/// on average this many per second (`RELAY_RATE`): a dashboard asking every member
-/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped.
-const RELAY_BURST: f64 = 512.0;
-const RELAY_RATE: f64 = 32.0;
-/// Nodes a relayed LINKS goes through at most, the first included.
-const MAX_HOPS: u8 = 3;
-/// A way of asking a member (directly, or through a peer) that brought none of its
-/// links for this long after we asked has failed it (`split`): the next ask goes
-/// another way. More than a window (`MAX_BATCH`) along a few hops of slow transports.
-const FAILED_AFTER: Duration = Duration::from_secs(20);
-/// How long a way that failed a member is left out for that member.
-const FAILED_KEPT: Duration = Duration::from_secs(300);
-
-/// The requests out and the replies in, and when each asker was last answered.
-#[derive(Default)]
-pub(crate) struct Links {
-    inner: Mutex<Inner>,
-    /// Someone subscribed (`answer`): `watch` looks again.
-    subscribed: tokio::sync::Notify,
-}
-
-#[derive(Default)]
-struct Inner {
-    /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its
-    /// subscription's end, or `REPLY_WITHIN`).
-    asked: HashMap<String, (u64, Instant)>,
-    /// The latest LINKS_REPLY of each member: when, and its links.
-    replies: HashMap<String, (Instant, Vec<Link>)>,
-    /// Each member's epoch, as its latest LINKS_REPLY said (`LinksReply::epoch`).
-    epochs: HashMap<String, Epoch>,
-    /// When we last answered each member.
-    answered: HashMap<String, Instant>,
-    /// Who subscribed to our links: the nonce to answer with, and until when.
-    subscribers: HashMap<String, (u64, Instant)>,
-    /// The links subscribers were last sent (`watch`).
-    /// And in which epoch: a new one is news too (a dashboard shows who entered it).
-    published: Option<(Epoch, Vec<Link>)>,
-    /// Our relayed LINKS out, by the peer asked to relay: its nonce, until when replies
-    /// to it are taken, and the members asked for through it.
-    via: HashMap<String, (u64, Instant, HashSet<String>)>,
-    /// What we ask on others' behalf, by member asked for: who asked (`Relaying`).
-    relaying: HashMap<String, HashMap<String, Relaying>>,
-    /// Relayed links due to an asker, by asker: the members whose links changed.
-    due: HashMap<String, HashSet<String>>,
-    /// How many members each peer may still have us ask for it (`RELAY_BURST`,
-    /// `RELAY_RATE`), as of when.
-    relay_budget: HashMap<String, (f64, Instant)>,
-    /// How we ask each member (`split`).
-    routes: HashMap<String, Route>,
-}
-
-/// How we ask a member: the way we last did, since when, whether its links came since,
-/// and the ways that failed it lately (`FAILED_AFTER`), each since when.
-struct Route {
-    /// Through this peer; None: directly.
-    via: Option<String>,
-    asked: Instant,
-    arrived: bool,
-    failed: HashMap<Option<String>, Instant>,
-}
-
-/// One asker's relayed subscription to a member.
-#[derive(Clone, Copy)]
-struct Relaying {
-    /// The nonce to answer with.
-    nonce: u64,
-    /// Until when, if still wanted.
-    until: Instant,
-    /// How long its links are held to be sent with others.
-    window: Duration,
-    /// Asked for once (no subscription): ends with the first links passed on.
-    once: bool,
-}
-
-impl Links {
-    /// The latest links each member replied with, and how long ago.
-    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-    pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> {
-        let now = Instant::now();
-        self.inner
-            .lock_or_recover()
-            .replies
-            .iter()
-            .map(|(member, (at, links))| {
-                (
-                    member.clone(),
-                    now.saturating_duration_since(*at),
-                    links.clone(),
-                )
-            })
-            .collect()
-    }
-
-    /// The epoch each member was in, as its latest links said (`replies`).
-    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-    pub(crate) fn epochs(&self) -> HashMap<String, Epoch> {
-        self.inner.lock_or_recover().epochs.clone()
-    }
-
-    /// Who is subscribed to our links (not run out).
-    #[cfg(test)]
-    pub(crate) fn subscribers(&self) -> Vec<String> {
-        let now = Instant::now();
-        self.inner
-            .lock_or_recover()
-            .subscribers
-            .iter()
-            .filter(|(_, (_, until))| now < *until)
-            .map(|(peer, _)| peer.clone())
-            .collect()
-    }
-
-    /// The members we ask on others' behalf (relayed, not run out), and who asked.
-    #[cfg(test)]
-    pub(crate) fn relaying(&self) -> Vec<(String, Vec<String>)> {
-        let now = Instant::now();
-        self.inner
-            .lock_or_recover()
-            .relaying
-            .iter()
-            .map(|(member, askers)| {
-                let askers: Vec<String> = askers
-                    .iter()
-                    .filter(|(_, relaying)| now < relaying.until)
-                    .map(|(asker, _)| asker.clone())
-                    .collect();
-                (member.clone(), askers)
-            })
-            .filter(|(_, askers)| !askers.is_empty())
-            .collect()
-    }
-
-    /// The members our subscriptions are with, and the peers relaying for us (not run
-    /// out).
-    #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-    pub(crate) fn subscribed(&self) -> Vec<String> {
-        let now = Instant::now();
-        let inner = self.inner.lock_or_recover();
-        inner
-            .asked
-            .iter()
-            .filter(|(_, (_, until))| now < *until)
-            .map(|(member, _)| member.clone())
-            .chain(
-                inner
-                    .via
-                    .iter()
-                    .filter(|(_, (_, until, _))| now < *until)
-                    .map(|(relay, _)| relay.clone()),
-            )
-            .collect()
-    }
-}
-
-/// This node's links: its neighbours, and the peers it holds dormant, as it measures
-/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its
-/// link stays listed, dormant, until it is heard from again: left out, the link would
-/// show only as that peer last reported it, from before.
-pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
-    let mut links = Vec::new();
-    for (peer, peer_state) in state.peers_snapshot().await {
-        let dormant = peer_state.lock().await.liveness.is_dormant();
-        if !dormant && !topology::is_neighbour(state, &peer) {
-            continue;
-        }
-        let destination = state.transport.destination(&peer);
-        let transport = destination
-            .as_ref()
-            .and_then(|destination| scheme_of(&destination.address));
-        links.push(Link {
-            others: others(destination.as_ref(), transport.as_deref()),
-            transport,
-            relayed: destination
-                .as_ref()
-                .is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
-            // Ranked up before down: the first is down only if all are.
-            down: destination
-                .as_ref()
-                .and_then(|destination| destination.paths.first())
-                .is_some_and(|path| path.down),
-            // The path in use's (`Transport::rtt`, the first of `destination.paths`).
-            rtt_us: state
-                .transport
-                .rtt(&peer)
-                .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
-            dormant,
-            clock: state.transport.clock_offset(&peer),
-            peer,
-        });
-    }
-    links.sort_by(|a, b| a.peer.cmp(&b.peer));
-    links.truncate(MAX_LINKS);
-    links
-}
-
-/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it
-/// has one.
-fn scheme_of(address: &str) -> Option<String> {
-    Some(paths::scheme_of(address))
-        .filter(|scheme| !scheme.is_empty())
-        .map(str::to_owned)
-}
-
-/// The transports besides `used` the peer proved itself at, with a path up: each with
-/// the round trip of its fastest path, if any is measured, and whether all of them go
-/// through a relay.
-fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
-    let mut others: Vec<OtherPath> = Vec::new();
-    for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
-        let Some(transport) = scheme_of(&path.address) else {
-            continue;
-        };
-        if path.down || used == Some(transport.as_str()) {
-            continue;
-        }
-        let rtt_us = path
-            .rtt_ms
-            .filter(|ms| ms.is_finite() && *ms >= 0.0)
-            .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
-        let relayed = path.connection == Some(Connection::Relayed);
-        match others.iter_mut().find(|other| other.transport == transport) {
-            Some(other) => {
-                other.rtt_us = match (other.rtt_us, rtt_us) {
-                    (Some(a), Some(b)) => Some(a.min(b)),
-                    (a, b) => a.or(b),
-                };
-                other.relayed &= relayed;
-            }
-            None => others.push(OtherPath {
-                transport,
-                rtt_us,
-                relayed,
-            }),
-        }
-    }
-    others.sort_by(|a, b| a.transport.cmp(&b.transport));
-    others.truncate(MAX_OTHERS);
-    others
-}
-
-/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for
-/// that long; its replies land in `AppState::links`.
-#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) {
-    let nonce = {
-        let now = Instant::now();
-        let mut inner = state.links.inner.lock_or_recover();
-        if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) {
-            inner.asked.clear();
-        }
-        let until = now + subscribe.max(REPLY_WITHIN);
-        // The same nonce while its replies are taken: a renewal doesn't drop a reply on
-        // its way under the last one (in a chain of relays, every hop renews at once,
-        // `relay`), as `ask_via` keeps one per relay.
-        match inner.asked.get_mut(member) {
-            Some((nonce, taken)) if now < *taken => {
-                *taken = (*taken).max(until);
-                *nonce
-            }
-            _ => {
-                let nonce = crate::transport::paths::random_nonce();
-                inner.asked.insert(member.to_owned(), (nonce, until));
-                nonce
-            }
-        }
-    };
-    let (epoch, epoch_seq) = state.epoch_stamp();
-    let request = ParsedMessage::Links(LinksRequest {
-        epoch,
-        epoch_seq,
-        nonce,
-        subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
-        of: Vec::new(),
-        batch_ms: 0,
-        hops: 0,
-    });
-    if let Err(error) = state.transport.send_message(member, &request).await {
-        debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
-    }
-}
-
-/// Asks `members` for their links (`ask`): those we can reach ourselves, the others
-/// through peers that can (`relay_for`, `ask_via`), with our window for what they pass
-/// on (`DEFAULT_BATCH`).
-#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) {
-    let (direct, through) = split(state, members, None, true).await;
-    for member in direct {
-        let state = state.clone();
-        tokio::spawn(async move { ask(&state, &member, subscribe).await });
-    }
-    for (relay, members) in through {
-        ask_via(
-            state,
-            &relay,
-            members,
-            subscribe,
-            DEFAULT_BATCH,
-            MAX_HOPS - 1,
-        )
-        .await;
-    }
-}
-
-/// `members` we can reach ourselves, and the others by the peer to ask them through
-/// (none for those no peer leads to, nor without `through_peers`); not `exclude` (who
-/// asked us). A way that brought none of a member's links since we last asked it that
-/// way (`FAILED_AFTER`) is left out for a while (`FAILED_KEPT`): directly, the next
-/// relay, and so on; once every way failed, they are all tried again.
-async fn split(
-    state: &AppState,
-    members: Vec<String>,
-    exclude: Option<&str>,
-    through_peers: bool,
-) -> (Vec<String>, HashMap<String, Vec<String>>) {
-    let mut direct = Vec::new();
-    let mut through: HashMap<String, Vec<String>> = HashMap::new();
-    for member in members {
-        if member == state.local_id() || Some(member.as_str()) == exclude {
-            continue;
-        }
-        let failed = failed_ways(state, &member);
-        let mut via = way(state, &member, exclude, through_peers, &failed).await;
-        if via.is_none() && !failed.is_empty() {
-            forget_failed(state, &member);
-            via = way(state, &member, exclude, through_peers, &HashMap::new()).await;
-        }
-        let Some(via) = via else {
-            continue;
-        };
-        asked_by(state, &member, via.clone());
-        match via {
-            None => direct.push(member),
-            Some(relay) => through.entry(relay).or_default().push(member),
-        }
-    }
-    (direct, through)
-}
-
-/// How to ask `member`, but the `failed` ways: Some(None) directly, Some(Some(peer))
-/// through `peer` (if `through_peers`); None if no way is left.
-async fn way(
-    state: &AppState,
-    member: &str,
-    exclude: Option<&str>,
-    through_peers: bool,
-    failed: &HashMap<Option<String>, Instant>,
-) -> Option<Option<String>> {
-    if !failed.contains_key(&None) && topology::reaches(state, member).await {
-        return Some(None);
-    }
-    if !through_peers {
-        return None;
-    }
-    relay_for(state, member, exclude, failed).await.map(Some)
-}
-
-/// The ways that failed `member` lately; the one we last asked it by among them if it
-/// brought nothing within `FAILED_AFTER`.
-fn failed_ways(state: &AppState, member: &str) -> HashMap<Option<String>, Instant> {
-    let now = Instant::now();
-    let mut inner = state.links.inner.lock_or_recover();
-    let Some(route) = inner.routes.get_mut(member) else {
-        return HashMap::new();
-    };
-    route
-        .failed
-        .retain(|_, since| now.saturating_duration_since(*since) < FAILED_KEPT);
-    if !route.arrived && now.saturating_duration_since(route.asked) >= FAILED_AFTER {
-        debug!(member = %member, via = ?route.via, "LINKS brought nothing that way; asking another");
-        route.failed.insert(route.via.clone(), now);
-    }
-    route.failed.clone()
-}
-
-/// Every way may be tried again for `member`.
-fn forget_failed(state: &AppState, member: &str) {
-    if let Some(route) = state.links.inner.lock_or_recover().routes.get_mut(member) {
-        route.failed.clear();
-    }
-}
-
-/// We ask `member` `via` now: the wait for its links starts over, unless we asked it so
-/// already and are still waiting.
-fn asked_by(state: &AppState, member: &str, via: Option<String>) {
-    let now = Instant::now();
-    let mut inner = state.links.inner.lock_or_recover();
-    if inner.routes.len() >= MAX_KEPT && !inner.routes.contains_key(member) {
-        inner.routes.clear();
-    }
-    let route = inner
-        .routes
-        .entry(member.to_owned())
-        .or_insert_with(|| Route {
-            via: via.clone(),
-            asked: now,
-            arrived: false,
-            failed: HashMap::new(),
-        });
-    if route.via != via || route.arrived {
-        route.via = via;
-        route.asked = now;
-        route.arrived = false;
-    }
-}
-
-/// The peer to ask `member` through, which we can't reach: an active member we reach
-/// (that answers us) that, by the member list, is on a transport `member` is on; else one on a transport
-/// we aren't on (it may know one that is). Of several, the one with the lowest pair score
-/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`, nor one
-/// that failed `member` (`failed`).
-async fn relay_for(
-    state: &AppState,
-    member: &str,
-    exclude: Option<&str>,
-    failed: &HashMap<Option<String>, Instant>,
-) -> Option<String> {
-    let theirs = topology::transports_of(state, member);
-    let ours = topology::transports_of(state, state.local_id());
-    let mut best: Option<(bool, u64, String)> = None;
-    for (peer, peer_state) in state.peers_snapshot().await {
-        if peer == member
-            || Some(peer.as_str()) == exclude
-            || failed.contains_key(&Some(peer.clone()))
-            || !state.transport.can_reach(&peer)
-        {
-            continue;
-        }
-        {
-            let peer_state = peer_state.lock().await;
-            if peer_state.is_unreachable()
-                || peer_state.liveness.is_dormant()
-                || !matches!(peer_state.standing, crate::state::Standing::Member { .. })
-            {
-                continue;
-            }
-        }
-        let on = topology::transports_of(state, &peer);
-        let shares = !on.is_disjoint(&theirs);
-        if !shares && on.is_subset(&ours) {
-            continue;
-        }
-        let score = topology::pair_score(&peer, member);
-        // Sharing a transport with the member first, then the lowest score.
-        let better = best.as_ref().is_none_or(|(best_shares, best_score, _)| {
-            (shares, std::cmp::Reverse(score)) > (*best_shares, std::cmp::Reverse(*best_score))
-        });
-        if better {
-            best = Some((shares, score, peer));
-        }
-    }
-    best.map(|(_, _, peer)| peer)
-}
-
-/// Asks `relay` to ask `members` on our behalf (a LINKS naming them), with `subscribe`,
-/// passing their links on within `window`, through `hops` more nodes at most: in LINKS
-/// of `MAX_RELAYED` members at most. One nonce per relay while its replies are taken, so
-/// its replies for every member we asked it for are taken.
-async fn ask_via(
-    state: &AppState,
-    relay: &str,
-    members: Vec<String>,
-    subscribe: Duration,
-    window: Duration,
-    hops: u8,
-) {
-    let nonce = {
-        let now = Instant::now();
-        let mut inner = state.links.inner.lock_or_recover();
-        if inner.via.len() >= MAX_KEPT && !inner.via.contains_key(relay) {
-            inner.via.retain(|_, (_, until, _)| now < *until);
-        }
-        let until = now + subscribe.max(REPLY_WITHIN);
-        let entry = inner.via.entry(relay.to_owned()).or_insert_with(|| {
-            (
-                crate::transport::paths::random_nonce(),
-                until,
-                HashSet::new(),
-            )
-        });
-        if now >= entry.1 {
-            entry.0 = crate::transport::paths::random_nonce();
-            entry.2.clear();
-        }
-        entry.1 = entry.1.max(until);
-        entry.2.extend(members.iter().cloned());
-        entry.0
-    };
-    let (epoch, epoch_seq) = state.epoch_stamp();
-    for members in members.chunks(MAX_RELAYED) {
-        let request = ParsedMessage::Links(LinksRequest {
-            epoch,
-            epoch_seq,
-            nonce,
-            subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX),
-            of: members.to_vec(),
-            batch_ms: u32::try_from(window.as_millis()).unwrap_or(u32::MAX),
-            hops,
-        });
-        if let Err(error) = state.transport.send_message(relay, &request).await {
-            debug!(relay = %relay, error = %format!("{error:#}"), "Relayed LINKS failed");
-        }
-    }
-}
-
-/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any
-/// more. A peer relaying for us isn't told: what it asks for us runs out by itself.
-#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
-pub(crate) async fn unsubscribe(state: &AppState, member: &str) {
-    let nonce = {
-        let mut inner = state.links.inner.lock_or_recover();
-        if inner.via.remove(member).is_some() && !inner.asked.contains_key(member) {
-            return;
-        }
-        match inner.asked.remove(member) {
-            Some((nonce, _)) => nonce,
-            None => return,
-        }
-    };
-    let (epoch, epoch_seq) = state.epoch_stamp();
-    let request = ParsedMessage::Links(LinksRequest {
-        epoch,
-        epoch_seq,
-        nonce,
-        subscribe_secs: 0,
-        of: Vec::new(),
-        batch_ms: 0,
-        hops: 0,
-    });
-    if let Err(error) = state.transport.send_message(member, &request).await {
-        debug!(member = %member, error = %format!("{error:#}"), "LINKS failed");
-    }
-}
-
-/// A LINKS from `peer`: answered if it's a member, and not answered lately.
-pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) {
-    // From the neighbours' view, kept as the member list changes: no document read per
-    // request.
-    let member = match crate::swarm::topology::listed(state, peer) {
-        Some(listed) => listed,
-        None => {
-            let replica = state.doc.read().await;
-            members::members(&replica.doc)
-                .iter()
-                .any(|member| member == peer)
-        }
-    };
-    if !member {
-        debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member");
-        return;
-    }
-    if !request.of.is_empty() {
-        relay(state, peer, request).await;
-        return;
-    }
-    {
-        let now = Instant::now();
-        let mut inner = state.links.inner.lock_or_recover();
-        // A subscription is taken, renewed or ended even when the reply is due later.
-        if request.subscribe_secs == 0 {
-            inner.subscribers.remove(peer);
-        } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer)
-        {
-            let lasts =
-                Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
-            inner
-                .subscribers
-                .insert(peer.to_owned(), (request.nonce, now + lasts));
-            state.links.subscribed.notify_one();
-        }
-        if inner
-            .answered
-            .get(peer)
-            .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY)
-        {
-            return;
-        }
-        if inner.answered.len() >= MAX_KEPT {
-            inner
-                .answered
-                .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY);
-        }
-        inner.answered.insert(peer.to_owned(), now);
-    }
-    let (epoch, epoch_seq) = state.epoch_stamp();
-    let reply = ParsedMessage::LinksReply(LinksReply {
-        epoch,
-        epoch_seq,
-        nonce: request.nonce,
-        links: own_links(state).await,
-        relayed: Vec::new(),
-    });
-    if let Err(error) = state.transport.send_message(peer, &reply).await {
-        debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
-    }
-}
-
-/// A relayed LINKS from `asker` (`answer`): the members it names are asked on its
-/// behalf, ourselves or through a peer of ours while hops are left, and what we know of
-/// them already is passed on at once (within its window); the rest as it arrives.
-async fn relay(state: &AppState, asker: &str, request: LinksRequest) {
-    let now = Instant::now();
-    let lease = Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION);
-    let once = lease.is_zero();
-    // What is left of the asker's window, shared out among this hop and those that may
-    // follow: ours is a share, the rest goes on.
-    let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH);
-    let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH);
-    let rest = left.saturating_sub(window);
-    let mut members: Vec<String> = request
-        .of
-        .into_iter()
-        .filter(|member| member != asker && member != state.local_id())
-        .take(MAX_RELAYED)
-        .collect();
-    let known: Vec<String> = {
-        let mut inner = state.links.inner.lock_or_recover();
-        // Within the asker's budget: each member costs one.
-        if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) {
-            inner.relay_budget.retain(|_, (budget, at)| {
-                *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE
-                    < RELAY_BURST
-            });
-        }
-        let (budget, at) = inner
-            .relay_budget
-            .entry(asker.to_owned())
-            .or_insert((RELAY_BURST, now));
-        *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE)
-            .min(RELAY_BURST);
-        *at = now;
-        let allowed = (*budget as usize).min(members.len());
-        *budget -= allowed as f64;
-        if allowed < members.len() {
-            debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it");
-            members.truncate(allowed);
-        }
-        for member in &members {
-            if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) {
-                inner
-                    .relaying
-                    .retain(|_, askers| askers.values().any(|relaying| now < relaying.until));
-            }
-            let askers = inner.relaying.entry(member.clone()).or_default();
-            askers.retain(|_, relaying| now < relaying.until);
-            if askers.len() >= MAX_SUBSCRIBERS && !askers.contains_key(asker) {
-                continue;
-            }
-            askers.insert(
-                asker.to_owned(),
-                Relaying {
-                    nonce: request.nonce,
-                    // Asked once: for as long as its reply is awaited.
-                    until: now + if once { REPLY_WITHIN } else { lease },
-                    window,
-                    once,
-                },
-            );
-        }
-        members
-            .iter()
-            .filter(|member| {
-                inner
-                    .replies
-                    .get(*member)
-                    .is_some_and(|(at, _)| now.saturating_duration_since(*at) < REPLY_WITHIN)
-            })
-            .cloned()
-            .collect()
-    };
-    for member in known {
-        forward(state, &member);
-    }
-    let (direct, through) = split(state, members, Some(asker), request.hops > 0).await;
-    for member in direct {
-        // Asked once while we subscribe anyway (our own dashboard, or another asker's):
-        // what comes is passed on, and a LINKS with 0 would end that subscription.
-        let subscribed = once
-            && state
-                .links
-                .inner
-                .lock_or_recover()
-                .asked
-                .get(&member)
-                .is_some_and(|(_, until)| now + REPLY_WITHIN < *until);
-        if !subscribed {
-            ask(state, &member, lease).await;
-        }
-    }
-    for (next, members) in through {
-        ask_via(state, &next, members, lease, rest, request.hops - 1).await;
-    }
-}
-
-/// `member`'s links changed (or arrived): due to those who asked us for them
-/// (`relaying`), each sent with what else is due to it at the end of its window.
-fn forward(state: &AppState, member: &str) {
-    let now = Instant::now();
-    let mut starting = Vec::new();
-    {
-        let mut inner = state.links.inner.lock_or_recover();
-        let askers: Vec<(String, Duration)> = match inner.relaying.get_mut(member) {
-            Some(askers) => {
-                askers.retain(|_, relaying| now < relaying.until);
-                askers
-                    .iter()
-                    .map(|(asker, relaying)| (asker.clone(), relaying.window))
-                    .collect()
-            }
-            None => return,
-        };
-        for (asker, window) in askers {
-            let due = inner.due.entry(asker.clone()).or_default();
-            if due.is_empty() {
-                starting.push((asker, window));
-            }
-            due.insert(member.to_owned());
-        }
-    }
-    for (asker, window) in starting {
-        let state = state.clone();
-        tokio::spawn(async move {
-            tokio::time::sleep(window).await;
-            pass_on(&state, &asker).await;
-        });
-    }
-}
-
-/// What is due to `asker`: the links of the members it asked for, as we have them, in
-/// as few LINKS_REPLYs as `MAX_RELAYED` allows (one per nonce it asked with).
-async fn pass_on(state: &AppState, asker: &str) {
-    let now = Instant::now();
-    let mut by_nonce: HashMap<u64, Vec<RelayedLinks>> = HashMap::new();
-    {
-        let mut inner = state.links.inner.lock_or_recover();
-        let due = inner.due.remove(asker).unwrap_or_default();
-        for member in due {
-            let Some(relaying) = inner
-                .relaying
-                .get(&member)
-                .and_then(|askers| askers.get(asker))
-                .copied()
-                .filter(|relaying| now < relaying.until)
-            else {
-                continue;
-            };
-            let Some((at, links)) = inner.replies.get(&member) else {
-                continue;
-            };
-            let Some(epoch) = inner.epochs.get(&member).copied() else {
-                continue;
-            };
-            let age_ms =
-                u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX);
-            by_nonce
-                .entry(relaying.nonce)
-                .or_default()
-                .push(RelayedLinks {
-                    member: member.clone(),
-                    age_ms,
-                    epoch,
-                    links: links.clone(),
-                });
-            if relaying.once
-                && let Some(askers) = inner.relaying.get_mut(&member)
-            {
-                askers.remove(asker);
-            }
-        }
-    }
-    let (epoch, epoch_seq) = state.epoch_stamp();
-    for (nonce, relayed) in by_nonce {
-        for part in by_size(relayed) {
-            let reply = ParsedMessage::LinksReply(LinksReply {
-                epoch,
-                epoch_seq,
-                nonce,
-                links: Vec::new(),
-                relayed: part,
-            });
-            if let Err(error) = state.transport.send_message(asker, &reply).await {
-                debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed");
-            }
-        }
-    }
-}
-
-/// `relayed` in parts for a LINKS_REPLY each: `MAX_RELAYED` members' links at most, and
-/// `MAX_RELAYED_BYTES` as encoded (but for one member's alone).
-fn by_size(relayed: Vec<RelayedLinks>) -> Vec<Vec<RelayedLinks>> {
-    let mut parts: Vec<Vec<RelayedLinks>> = Vec::new();
-    let mut bytes = 0;
-    for links in relayed {
-        let size = rmp_serde::encode::to_vec(&links).map_or(0, |encoded| encoded.len());
-        let full = parts.last().is_none_or(|part| {
-            part.len() >= MAX_RELAYED || (!part.is_empty() && bytes + size > MAX_RELAYED_BYTES)
-        });
-        if full {
-            parts.push(Vec::new());
-            bytes = 0;
-        }
-        bytes += size;
-        parts.last_mut().expect("just made").push(links);
-    }
-    parts
-}
-
-/// A LINKS_REPLY from `peer`: its own links kept if it answers our latest LINKS to it,
-/// and those it relays if it answers our latest relayed one, while their replies are
-/// taken (a subscription's go on); then passed on to whoever asked us for them.
-pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) {
-    let now = Instant::now();
-    let mut arrived = Vec::new();
-    {
-        let mut inner = state.links.inner.lock_or_recover();
-        let own = inner
-            .asked
-            .get(peer)
-            .is_some_and(|(nonce, until)| *nonce == reply.nonce && now < *until);
-        let via: Option<HashSet<String>> = inner
-            .via
-            .get(peer)
-            .filter(|(nonce, until, _)| *nonce == reply.nonce && now < *until)
-            .map(|(_, _, members)| members.clone());
-        let mut keep = |member: String, age: Duration, epoch: Epoch, mut links: Vec<Link>| {
-            links.truncate(MAX_LINKS);
-            for link in &mut links {
-                link.others.truncate(MAX_OTHERS);
-            }
-            if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(&member) {
-                inner.replies.clear();
-                inner.epochs.clear();
-            }
-            let at = now.checked_sub(age).unwrap_or(now);
-            // Arrived, if older news: the way works.
-            if let Some(route) = inner.routes.get_mut(&member) {
-                route.arrived = true;
-            }
-            // Not older news over newer.
-            let known = inner.replies.get(&member);
-            if known.is_some_and(|(known, _)| *known > at) {
-                return;
-            }
-            // The same links as we have, lately: no news, so not passed on. A copy that
-            // went round between relays (each asking the other for the member) comes
-            // back a little newer each time, as its age counts from when each sent it:
-            // passed on, it would go round for good.
-            let news = known.is_none_or(|(known_at, known_links)| {
-                *known_links != links || now.saturating_duration_since(*known_at) >= REPLY_WITHIN
-            }) || inner.epochs.get(&member) != Some(&epoch);
-            inner.replies.insert(member.clone(), (at, links));
-            inner.epochs.insert(member.clone(), epoch);
-            if news {
-                arrived.push(member);
-            }
-        };
-        if own {
-            keep(peer.to_owned(), Duration::ZERO, reply.epoch, reply.links);
-        }
-        if let Some(members) = via {
-            for relayed in reply.relayed.into_iter().take(MAX_RELAYED) {
-                if members.contains(&relayed.member) {
-                    keep(
-                        relayed.member,
-                        Duration::from_millis(u64::from(relayed.age_ms)),
-                        relayed.epoch,
-                        relayed.links,
-                    );
-                }
-            }
-        }
-    }
-    for member in arrived {
-        forward(state, &member);
-    }
-}
-
-/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's
-/// links; sent to each of them when they changed since they were last sent. Without
-/// any, it waits for one. Runs for the node's life.
-pub(crate) async fn watch(state: AppState) {
-    let mut every = tokio::time::interval(WATCH_EVERY);
-    every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    loop {
-        every.tick().await;
-        let subscribers: Vec<(String, u64)> = {
-            let now = Instant::now();
-            let mut inner = state.links.inner.lock_or_recover();
-            inner.subscribers.retain(|_, (_, until)| now < *until);
-            if inner.subscribers.is_empty() {
-                inner.published = None;
-            }
-            inner
-                .subscribers
-                .iter()
-                .map(|(peer, (nonce, _))| (peer.clone(), *nonce))
-                .collect()
-        };
-        if subscribers.is_empty() {
-            // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`):
-            // the next look is due an interval later.
-            state.links.subscribed.notified().await;
-            every.reset();
-            continue;
-        }
-        let links = own_links(&state).await;
-        let (epoch, epoch_seq) = state.epoch_stamp();
-        {
-            let mut inner = state.links.inner.lock_or_recover();
-            if inner
-                .published
-                .as_ref()
-                .is_some_and(|(published_in, published)| {
-                    *published_in == epoch && !changed(published, &links)
-                })
-            {
-                continue;
-            }
-            inner.published = Some((epoch, links.clone()));
-        }
-        for (peer, nonce) in subscribers {
-            let reply = ParsedMessage::LinksReply(LinksReply {
-                epoch,
-                epoch_seq,
-                nonce,
-                links: links.clone(),
-                relayed: Vec::new(),
-            });
-            if let Err(error) = state.transport.send_message(&peer, &reply).await {
-                debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed");
-            }
-        }
-    }
-}
-
-/// Whether `new` says something `old` didn't: another peer, transport, relay or
-/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
-/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
-/// moves aren't news.
-pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool {
-    let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
-        (Some(a), Some(b)) => {
-            a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
-        }
-        (a, b) => a.is_some() != b.is_some(),
-    };
-    let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
-        (Some(a), Some(b)) => {
-            let by = a.abs_diff(b);
-            by > 1000 && by > a.max(b) / 5
-        }
-        (a, b) => a.is_some() != b.is_some(),
-    };
-    old.len() != new.len()
-        || old.iter().zip(new).any(|(a, b)| {
-            a.peer != b.peer
-                || a.transport != b.transport
-                || a.relayed != b.relayed
-                || a.down != b.down
-                || a.dormant != b.dormant
-                || moved(a.rtt_us, b.rtt_us)
-                || clock_moved(a.clock, b.clock)
-                || a.others.len() != b.others.len()
-                || a.others.iter().zip(&b.others).any(|(x, y)| {
-                    x.transport != y.transport
-                        || x.relayed != y.relayed
-                        || moved(x.rtt_us, y.rtt_us)
-                })
-        })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn relayed_links_are_sent_in_parts_by_count_and_size() {
-        let relayed = |links: usize| RelayedLinks {
-            member: "m".into(),
-            age_ms: 0,
-            epoch: Epoch {
-                generation: 1,
-                id: [0; 32],
-            },
-            links: (0..links)
-                .map(|i| link(&format!("peer-{i:0>70}"), Some(1500)))
-                .collect(),
-        };
-        // Small: by count.
-        let parts = by_size((0..130).map(|_| relayed(2)).collect());
-        assert_eq!(parts.iter().map(Vec::len).collect::<Vec<_>>(), [64, 64, 2]);
-        // Large: by size, each part within the bytes but for one alone.
-        let big = rmp_serde::encode::to_vec(&relayed(MAX_LINKS))
-            .unwrap()
-            .len();
-        assert!(20 * big > MAX_RELAYED_BYTES, "{big}");
-        let parts = by_size((0..20).map(|_| relayed(MAX_LINKS)).collect());
-        assert_eq!(parts.iter().map(Vec::len).sum::<usize>(), 20);
-        for part in &parts {
-            let bytes: usize = part
-                .iter()
-                .map(|links| rmp_serde::encode::to_vec(links).unwrap().len())
-                .sum();
-            assert!(part.len() == 1 || bytes <= MAX_RELAYED_BYTES, "{bytes}");
-        }
-        assert!(parts.len() > 1);
-    }
-
-    fn link(peer: &str, rtt_us: Option<u32>) -> Link {
-        Link {
-            peer: peer.into(),
-            transport: Some("libp2p".into()),
-            relayed: false,
-            down: false,
-            rtt_us,
-            dormant: false,
-            others: Vec::new(),
-            clock: None,
-        }
-    }
-
-    #[test]
-    fn only_news_is_a_change() {
-        let old = [link("a", Some(10_000))];
-        assert!(!changed(&old, &[link("a", Some(11_000))]));
-        assert!(changed(&old, &[link("a", Some(13_000))]));
-        assert!(changed(&old, &[link("a", None)]));
-        assert!(changed(&old, &[link("b", Some(10_000))]));
-        assert!(changed(&old, &[]));
-        let mut dormant = link("a", Some(10_000));
-        dormant.dormant = true;
-        assert!(changed(&old, &[dormant]));
-        let mut down = link("a", Some(10_000));
-        down.down = true;
-        assert!(changed(&old, &[down]));
-        // Small round trips: a millisecond's wobble isn't news.
-        assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
-        // A clock: news when measured, and when it moved beyond both bounds.
-        let clocked = |offset_ms| Link {
-            clock: Some(ClockOffset {
-                offset_ms,
-                within_ms: 50,
-            }),
-            ..link("a", Some(10_000))
-        };
-        assert!(changed(&old, &[clocked(0)]));
-        assert!(!changed(&[clocked(0)], &[clocked(80)]));
-        assert!(changed(&[clocked(0)], &[clocked(5_000)]));
-    }
-}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index ccffaa0..7b5b09c 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -10,7 +10,6 @@ pub(crate) mod document_file;
 pub(crate) mod flood;
 pub(crate) mod hop;
 pub(crate) mod hyparview;
-pub(crate) mod links;
 pub(crate) mod members;
 pub(crate) mod pair;
 pub(crate) mod peer;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 0c41466..d25c7de 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -242,12 +242,6 @@ pub(crate) async fn handle_message(
             heard_heads(state, &peer, &remote);
             on_status(state, &peer, &remote, reactivated).await;
         }
-        transport::ParsedMessage::Links(request) => {
-            crate::swarm::links::answer(state, &peer, request).await;
-        }
-        transport::ParsedMessage::LinksReply(reply) => {
-            crate::swarm::links::received(state, &peer, reply);
-        }
         transport::ParsedMessage::Probe(request) => {
             crate::swarm::working_links::answer(state, &peer, request).await;
         }
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index c23d2b2..2d812e5 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -530,7 +530,8 @@ pub(crate) async fn reaches(state: &AppState, peer: &str) -> bool {
 }
 
 /// The transports `member` is on, by the member list (the schemes of its addresses);
-/// ours as we are on them. None known: empty.
+/// ours as we are on them. None known: empty. For the tests.
+#[cfg(test)]
 pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> {
     let view = state.neighbours.read_or_recover();
     match view.as_ref() {