topology: explode!

3 files changed, 908 insertions, 0 deletions

diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs
new file mode 100644
index 0000000..fb33170
--- /dev/null
+++ b/src/swarm/topology/graph.rs
@@ -0,0 +1,236 @@
+//! The graph itself, from the members alone: a ring by hash, plus each member's
+//! lowest-scoring pairs, stable members preferred; per transport, and across them. Pure:
+//! the same members give every node the same graph.
+
+use sha2::{Digest, Sha256};
+
+use super::*;
+
+pub(super) const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
+
+pub(super) const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2";
+
+/// How many members may prefer the same stable member, as a multiple of `k`. Without a
+/// bound, 20 stable members among 2000 get about 500 neighbours each: every
+/// announcement in the swarm lands on them, and the backbone saturates first. Members
+/// that find a stable member full prefer the next one, then go by pair score, so the
+/// graph stays connected and symmetric.
+pub(super) const STABLE_FANIN_FACTOR: usize = 4;
+
+/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`);
+/// the empty one for an address without (a transport whose identities are addresses).
+pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
+    addresses
+        .iter()
+        .map(|address| {
+            address
+                .split_once(':')
+                .map_or("", |(scheme, _)| scheme)
+                .to_owned()
+        })
+        .collect()
+}
+
+/// The neighbours of `local` among `members` (by the transports each is on), on each of
+/// `local_transports`: on each, its neighbours among the members on it
+/// (`neighbours`), all together.
+pub(crate) fn neighbours_across(
+    members: &BTreeMap<String, BTreeSet<String>>,
+    local: &str,
+    local_transports: &BTreeSet<String>,
+    k: usize,
+    unreachable: &dyn Fn(&str) -> bool,
+    stable: &dyn Fn(&str) -> bool,
+) -> BTreeSet<String> {
+    let mut all = BTreeSet::new();
+    for transport in local_transports {
+        let on_it: Vec<String> = members
+            .iter()
+            .filter(|(member, transports)| *member != local && transports.contains(transport))
+            .map(|(member, _)| member.clone())
+            .collect();
+        all.extend(neighbours(&on_it, local, k, unreachable, stable));
+    }
+    all
+}
+
+/// The neighbours of `local` among `members` (which may or may not contain `local`),
+/// skipping members for which `unreachable` holds.
+pub(crate) fn neighbours(
+    members: &[String],
+    local: &str,
+    k: usize,
+    unreachable: &dyn Fn(&str) -> bool,
+    stable: &dyn Fn(&str) -> bool,
+) -> BTreeSet<String> {
+    let mut all: Vec<String> = members.to_vec();
+    all.push(local.to_owned());
+    graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
+        .remove(local)
+        .unwrap_or_default()
+}
+
+/// Every reachable member's neighbours. Hashes each member once and scores pairs by
+/// mixing the two hashes (`pair_score`), so the N² scores are a few nanoseconds each:
+/// tens of milliseconds for thousands of members. `stable` members (`swarm::stability`)
+/// are preferred for half of each member's scored slots, each by at most
+/// `STABLE_FANIN_FACTOR * k` members.
+///
+/// Each reachable member walks the ring (by key hash) both ways to the first reachable
+/// member, which gives a ring over the reachable members, and adds the `k - 2`
+/// reachable members with the lowest pair score. A link exists if either end chose it.
+/// With at most `k + 1` reachable members, the reachable members form a full mesh.
+///
+/// Each node judges reachability by itself, so nodes may briefly disagree; that only
+/// adds links (whoever chose a link initiates, the other answers), it never cuts one.
+/// The preference budget is spent in key order, which every node shares, so nodes that
+/// agree on reachability compute the same links.
+pub(crate) fn graph(
+    members: &[String],
+    k: usize,
+    unreachable: &dyn Fn(&str) -> bool,
+    stable: &dyn Fn(&str) -> bool,
+) -> BTreeMap<String, BTreeSet<String>> {
+    let k = k.max(2);
+    let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
+    let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect();
+    let mut links: BTreeMap<String, BTreeSet<String>> = live
+        .iter()
+        .map(|m| ((*m).to_owned(), BTreeSet::new()))
+        .collect();
+
+    let mut link = |a: &str, b: &str| {
+        if a != b {
+            links.get_mut(a).expect("live member").insert(b.to_owned());
+            links.get_mut(b).expect("live member").insert(a.to_owned());
+        }
+    };
+
+    if live.len() <= k + 1 {
+        for a in &live {
+            for b in &live {
+                link(a, b);
+            }
+        }
+        return links;
+    }
+
+    // The ring over all members, so its order doesn't change when one goes offline.
+    let mut ring: Vec<([u8; 32], &str)> =
+        all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect();
+    ring.sort_unstable();
+    let n = ring.len();
+    let position: BTreeMap<&str, usize> =
+        ring.iter().enumerate().map(|(i, (_, m))| (*m, i)).collect();
+    let first_live = |from: usize, step: usize| {
+        (1..n)
+            .map(|d| ring[(from + d * step) % n].1)
+            .find(|m| live.contains(m))
+            .expect("more than k + 1 live members")
+    };
+
+    // Live members in key order, with their pair keys; and those of them that are stable.
+    let keyed: Vec<(&str, u64)> = live.iter().map(|m| (*m, pair_key(m))).collect();
+    let stable_keyed: Vec<(&str, u64)> = keyed.iter().copied().filter(|(m, _)| stable(m)).collect();
+
+    // How many members have preferred each stable member so far (`keyed` is in key
+    // order, the same on every node).
+    let budget = STABLE_FANIN_FACTOR * k;
+    let mut preferred_by: BTreeMap<&str, usize> = BTreeMap::new();
+    // Half the slots (rounded up) prefer stable members, so long-running nodes form a
+    // backbone; the rest go by pair score alone, keeping paths short. A stable member
+    // enough others prefer already is skipped, so no hub grows with the swarm.
+    let stable_slots = (k - 2).div_ceil(2);
+
+    for &(member, key) in &keyed {
+        let at = position[member];
+        let next = first_live(at, 1);
+        let prev = first_live(at, n - 1);
+        link(member, next);
+        link(member, prev);
+        let skip = [member, next, prev];
+        // The stable candidates in score order (there are few).
+        let mut stable_candidates = scored(&stable_keyed, key, skip);
+        stable_candidates.sort_unstable();
+        let preferred: Vec<&str> = stable_candidates
+            .iter()
+            .map(|(_, other)| *other)
+            .filter(|other| preferred_by.get(other).copied().unwrap_or(0) < budget)
+            .take(stable_slots)
+            .collect();
+        for other in preferred.iter().copied() {
+            *preferred_by.entry(other).or_default() += 1;
+        }
+        // The rest: the lowest scores among the others that aren't preferred, which are
+        // among the `need + preferred` lowest overall. Selecting those is linear; only
+        // they are sorted.
+        let need = k - 2 - preferred.len();
+        let mut candidates = scored(&keyed, key, skip);
+        let lowest = (need + preferred.len()).min(candidates.len());
+        if lowest < candidates.len() {
+            candidates.select_nth_unstable(lowest);
+            candidates.truncate(lowest);
+        }
+        candidates.sort_unstable();
+        let rest = candidates
+            .iter()
+            .map(|(_, other)| *other)
+            .filter(|other| !preferred.contains(other))
+            .take(need);
+        for partner in preferred.iter().copied().chain(rest) {
+            link(member, partner);
+        }
+    }
+    links
+}
+
+/// `others` but those in `skip`, scored as pairs with the member whose key is `key`.
+pub(super) fn scored<'a>(
+    others: &[(&'a str, u64)],
+    key: u64,
+    skip: [&str; 3],
+) -> Vec<(u64, &'a str)> {
+    others
+        .iter()
+        .filter(|(other, _)| !skip.contains(other))
+        .map(|&(other, other_key)| (mix_pair(key, other_key), other))
+        .collect()
+}
+
+/// Symmetric: the same for (a, b) and (b, a).
+pub(crate) fn pair_score(a: &str, b: &str) -> u64 {
+    mix_pair(pair_key(a), pair_key(b))
+}
+
+/// What a member contributes to its pair scores: computed once per member.
+pub(super) fn pair_key(member: &str) -> u64 {
+    let digest = hash(PAIR_LABEL, &[member]);
+    u64::from_be_bytes(digest[..8].try_into().expect("8 bytes"))
+}
+
+/// The score of a pair from its members' keys, symmetric: SplitMix64's finalizer over
+/// the lower key and the mixed higher one. Keys are uniform (SHA-256), so for each member
+/// the others' scores are in an order unrelated to anyone else's.
+pub(super) fn mix_pair(a: u64, b: u64) -> u64 {
+    let (low, high) = if a <= b { (a, b) } else { (b, a) };
+    mix(low ^ mix(high))
+}
+
+pub(super) fn mix(mut z: u64) -> u64 {
+    z = z.wrapping_add(0x9e37_79b9_7f4a_7c15);
+    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
+    z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
+    z ^ (z >> 31)
+}
+
+pub(super) fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] {
+    let mut hasher = Sha256::new();
+    for field in [label]
+        .into_iter()
+        .chain(fields.iter().map(|f| f.as_bytes()))
+    {
+        hasher.update((field.len() as u32).to_be_bytes());
+        hasher.update(field);
+    }
+    hasher.finalize().into()
+}
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
new file mode 100644
index 0000000..ea71712
--- /dev/null
+++ b/src/swarm/topology/mod.rs
@@ -0,0 +1,359 @@
+//! Which members a node syncs with: its neighbours.
+//!
+//! In a full mesh every change reaches every node from every other node (Automerge
+//! decides per pair, so nobody knows that a third peer already passed it on), so the
+//! copies of one write grow with N². Above `k` members, each node keeps only a few
+//! neighbours, and changes travel over several hops.
+//!
+//! Every node computes the graph from the member list and which members it can reach,
+//! with no coordination:
+//! * a ring ordered by key hash: each member's nearest reachable member in both
+//!   directions, which keeps the reachable members connected;
+//! * plus, for each member, the `k - 2` reachable others with the lowest pair score (a
+//!   hash of both keys, so both ends agree on it), which keeps paths short.
+//!
+//! A link exists if either end chose it, so links are symmetric and a node has at least
+//! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh:
+//! small swarms behave exactly as before. A member joining, leaving or going offline
+//! changes only links near it.
+//!
+//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
+//! among the members on it (as their addresses in the member list say), and a node's
+//! neighbours are the union of its neighbours on each transport it is on
+//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
+//! two transports is in both: as long as members on several transports link the
+//! transports, the swarm stays connected, and such a member has about `k` neighbours per
+//! transport. With one transport, it is the one graph as before.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use tracing::info;
+
+use crate::locks::RwLockExt;
+use crate::state::AppState;
+use crate::swarm::members;
+
+mod graph;
+
+pub(crate) use graph::pair_score;
+use graph::*;
+
+/// This node's current neighbours, and the member list they were computed from.
+#[derive(Debug, Default)]
+pub(crate) struct NeighbourView {
+    members: Vec<String>,
+    k: usize,
+    unreachable: BTreeSet<String>,
+    stable: BTreeSet<String>,
+    /// The transports (address schemes) of each member, and ours.
+    transports: BTreeMap<String, BTreeSet<String>>,
+    local_transports: BTreeSet<String>,
+    /// Peers that proved they are of our swarm but aren't in our member list (yet): their
+    /// registration hasn't reached us. They are neighbours too (`refresh`).
+    unlisted: BTreeSet<String>,
+    /// Members the member list gives no address for on our transports, that reach us
+    /// anyway: they only connect out (no listen address). They are neighbours too while
+    /// they say they sync with us (`refresh`).
+    outbound: BTreeSet<String>,
+    neighbours: BTreeSet<String>,
+    /// Every reachable member is a neighbour.
+    full_mesh: bool,
+}
+
+/// Whether the neighbours are computed from the member list yet (before, everyone
+/// counts as a neighbour).
+pub(crate) fn view_known(state: &AppState) -> bool {
+    state.neighbours.read_or_recover().is_some()
+}
+
+/// Whether every reachable member is our neighbour (also while the view is unknown).
+pub(crate) fn is_full_mesh(state: &AppState) -> bool {
+    state
+        .neighbours
+        .read_or_recover()
+        .as_ref()
+        .is_none_or(|view| view.full_mesh)
+}
+
+/// The current view in numbers, once known.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Summary {
+    /// Registered members, us included.
+    pub(crate) members: usize,
+    /// Members other than us not currently judged unreachable.
+    pub(crate) reachable: usize,
+    pub(crate) stable: usize,
+    pub(crate) neighbours: usize,
+    pub(crate) stable_neighbours: usize,
+    pub(crate) full_mesh: bool,
+}
+
+pub(crate) fn summary(state: &AppState) -> Option<Summary> {
+    let local = state.local_id();
+    state
+        .neighbours
+        .read_or_recover()
+        .as_ref()
+        .map(|view| Summary {
+            members: view.members.len(),
+            reachable: view
+                .members
+                .iter()
+                .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
+                .count(),
+            stable: view.stable.len(),
+            neighbours: view.neighbours.len(),
+            stable_neighbours: view
+                .neighbours
+                .iter()
+                .filter(|m| view.stable.contains(*m))
+                .count(),
+            full_mesh: view.full_mesh,
+        })
+}
+
+/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
+/// openings. Everyone is a neighbour until the view is known (while joining, or before
+/// any member registered), so joining works as in a full mesh. We always *answer*
+/// anyone who contacts us.
+pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
+    match &*state.neighbours.read_or_recover() {
+        Some(view) => view.neighbours.contains(peer),
+        None => true,
+    }
+}
+
+/// Every member's neighbours as the member list makes them, as this node sees it (whom
+/// it can reach, who is stable): the graph each node computes its own part of. Ours as
+/// they are (with members that only connect out, or aren't listed yet). For a map of
+/// the swarm (the dashboard); None until the view is known. A link exists if either end
+/// chose it: draw one if either lists the other. About `members²` work: compute it when
+/// the member list may have changed, not on every look.
+#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
+pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> {
+    let view = state.neighbours.read_or_recover();
+    let view = view.as_ref()?;
+    let unreachable = |member: &str| view.unreachable.contains(member);
+    let stable = |member: &str| view.stable.contains(member);
+    let graph = view
+        .members
+        .iter()
+        .map(|member| {
+            let neighbours = if member == state.local_id() {
+                view.neighbours.clone()
+            } else {
+                let own = view.transports.get(member).cloned().unwrap_or_default();
+                neighbours_across(
+                    &view.transports,
+                    member,
+                    &own,
+                    view.k,
+                    &unreachable,
+                    &stable,
+                )
+            };
+            (member.clone(), neighbours)
+        })
+        .collect();
+    Some(graph)
+}
+
+/// How the neighbours changed.
+pub(crate) struct NeighbourChange {
+    pub(crate) added: Vec<String>,
+    pub(crate) removed: Vec<String>,
+}
+
+/// Recomputes the neighbours whenever something they depend on may have changed
+/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
+/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
+/// fallback. A burst of notifications coalesces into one recomputation.
+pub(crate) async fn run_worker(state: AppState) {
+    loop {
+        state.topology_changed.notified().await;
+        if let Some(change) = refresh(&state).await {
+            crate::swarm::peer::apply_neighbour_change(&state, change).await;
+        }
+    }
+}
+
+/// Whether a change at `path` may change the neighbours: one in the member list, or the
+/// root (the whole document replaced, as after an epoch switch).
+pub(crate) fn affects_neighbours(path: &[String]) -> bool {
+    path.is_empty() || path.starts_with(&members::members_dir())
+}
+
+/// Recomputes the neighbours once ready, if the member list, the reachable members or
+/// `k` changed.
+pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
+    if !state.is_ready().await {
+        return None;
+    }
+    let (current, stable, transports, alive) = {
+        let replica = state.doc.read().await;
+        members::apply(state, &replica.doc);
+        let current = members::members(&replica.doc);
+        // Those that said they are alive within the electorate window.
+        let alive: BTreeSet<String> = members::electorate(
+            &replica.doc,
+            state.clock.now_millis().0,
+            state.electorate_window,
+        )
+        .into_iter()
+        .collect();
+        let stable: BTreeSet<String> = current
+            .iter()
+            .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
+            .cloned()
+            .collect();
+        let transports: BTreeMap<String, BTreeSet<String>> = current
+            .iter()
+            .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
+            .collect();
+        (current, stable, transports, alive)
+    };
+    let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
+    let k = state.neighbours_k;
+    // Those that don't answer, those on none of our transports, and those that haven't
+    // said they are alive for longer than the electorate window, unless we heard from
+    // them since we started (one that comes back says so to us first: its HELLO).
+    let mut unreachable = BTreeSet::new();
+    for member in &current {
+        if member == state.local_id() {
+            continue;
+        }
+        if !state.transport.can_reach(member) {
+            unreachable.insert(member.clone());
+            continue;
+        }
+        let (gone, heard) = match state.get_peer_state(member).await {
+            Some(peer_state) => {
+                let peer_state = peer_state.lock().await;
+                (peer_state.is_unreachable(), peer_state.last_seen.is_some())
+            }
+            None => (false, false),
+        };
+        if gone || (!heard && !alive.contains(member)) {
+            unreachable.insert(member.clone());
+        }
+    }
+    // A node that joined while the genesis node hadn't registered yet has the genesis
+    // document without the genesis node in it, and registers itself there: then neither
+    // lists the other, and with each the other's neighbour only by the member list,
+    // neither would ever sync with the other, and their registrations never pass. So a
+    // peer that proved itself a member of our swarm counts until the list says otherwise:
+    // syncing brings each the other's registration.
+    let mut unlisted = BTreeSet::new();
+    for (peer, peer_state) in state.peers_snapshot().await {
+        if current.contains(&peer) || peer == state.local_id() {
+            continue;
+        }
+        let peer_state = peer_state.lock().await;
+        if matches!(peer_state.standing, crate::state::Standing::Member { .. })
+            && !peer_state.is_unreachable()
+        {
+            unlisted.insert(peer);
+        }
+    }
+    // A member that lists no address on our transports, but reached us anyway, only
+    // connects out: it isn't in the graph on our side (it's on none of our transports, as
+    // far as the list goes), while on its side we are, so it would sync with us while we
+    // never sent it our changes (only answers to its). It picks its neighbours; those it
+    // picked keep it, as it says (STATUS).
+    let mut outbound = BTreeSet::new();
+    for member in &current {
+        if member == state.local_id()
+            || unreachable.contains(member)
+            || transports
+                .get(member)
+                .is_some_and(|on| !on.is_disjoint(&local_transports))
+        {
+            continue;
+        }
+        if let Some(peer_state) = state.get_peer_state(member).await {
+            let peer_state = peer_state.lock().await;
+            if matches!(peer_state.standing, crate::state::Standing::Member { .. })
+                && peer_state.syncs_with_us != Some(false)
+            {
+                outbound.insert(member.clone());
+            }
+        }
+    }
+    let previous = {
+        let view = state.neighbours.read_or_recover();
+        if view.as_ref().is_some_and(|view| {
+            view.members == current
+                && view.unlisted == unlisted
+                && view.outbound == outbound
+                && view.k == k
+                && view.unreachable == unreachable
+                && view.stable == stable
+                && view.transports == transports
+                && view.local_transports == local_transports
+        }) {
+            return None;
+        }
+        view.as_ref().map(|view| view.neighbours.clone())
+    };
+
+    // Before the first view everyone known counted as a neighbour.
+    let previous = match previous {
+        Some(previous) => previous,
+        None => state.peer_states.read().await.keys().cloned().collect(),
+    };
+    let mut computed = neighbours_across(
+        &transports,
+        state.local_id(),
+        &local_transports,
+        k,
+        &|m| unreachable.contains(m),
+        &|m| stable.contains(m),
+    );
+    computed.extend(outbound.iter().cloned());
+    let listed_neighbours = computed.len();
+    computed.extend(unlisted.iter().cloned());
+    let change = NeighbourChange {
+        added: computed
+            .iter()
+            .filter(|m| !previous.contains(*m))
+            .cloned()
+            .collect(),
+        removed: previous
+            .iter()
+            .filter(|m| !computed.contains(*m))
+            .cloned()
+            .collect(),
+    };
+    info!(
+        members = current.len(),
+        unreachable = unreachable.len(),
+        stable = stable.len(),
+        k,
+        neighbours = computed.len(),
+        unlisted = unlisted.len(),
+        outbound = outbound.len(),
+        added = change.added.len(),
+        removed = change.removed.len(),
+        "Neighbours computed from the member list"
+    );
+    let reachable_others = current
+        .iter()
+        .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
+        .count();
+    *state.neighbours.write_or_recover() = Some(NeighbourView {
+        full_mesh: listed_neighbours >= reachable_others,
+        members: current,
+        k,
+        unreachable,
+        stable,
+        transports,
+        local_transports,
+        unlisted,
+        outbound,
+        neighbours: computed,
+    });
+    Some(change)
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/src/swarm/topology/tests.rs b/src/swarm/topology/tests.rs
new file mode 100644
index 0000000..d1b755f
--- /dev/null
+++ b/src/swarm/topology/tests.rs
@@ -0,0 +1,313 @@
+use super::*;
+use std::collections::VecDeque;
+
+fn members(n: usize) -> Vec<String> {
+    (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect()
+}
+
+fn all_reachable(_: &str) -> bool {
+    false
+}
+
+fn none_stable(_: &str) -> bool {
+    false
+}
+
+fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
+    super::graph(members, k, &all_reachable, &none_stable)
+}
+
+fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool {
+    let Some(start) = g.keys().next() else {
+        return true;
+    };
+    let mut seen = BTreeSet::from([start.clone()]);
+    let mut queue = VecDeque::from([start.clone()]);
+    while let Some(m) = queue.pop_front() {
+        for next in &g[&m] {
+            if seen.insert(next.clone()) {
+                queue.push_back(next.clone());
+            }
+        }
+    }
+    seen.len() == g.len()
+}
+
+fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> {
+    graph
+        .iter()
+        .flat_map(|(a, ns)| {
+            ns.iter().map(move |b| {
+                if a < b {
+                    (a.clone(), b.clone())
+                } else {
+                    (b.clone(), a.clone())
+                }
+            })
+        })
+        .collect()
+}
+
+fn hex(bytes: [u8; 32]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+/// Over random swarms (size, `k`, who is reachable, who is stable): links only
+/// between reachable members, never to oneself, symmetric; a full mesh while small,
+/// otherwise at least `k` links each and connected; independent of the order members
+/// are listed in; and each member's own view (`neighbours`) agrees with the graph.
+#[test]
+fn random_swarms() {
+    crate::testrand::for_seeds(300, |rng| {
+        let n = rng.between(1, 60);
+        let k = rng.between(2, 14);
+        let mut all: Vec<String> = (0..n).map(|_| hex(rng.bytes32())).collect();
+        let unreachable: BTreeSet<String> =
+            all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
+        let stable: BTreeSet<String> = all.iter().filter(|_| rng.chance(0.3)).cloned().collect();
+        let is_unreachable = |m: &str| unreachable.contains(m);
+        let is_stable = |m: &str| stable.contains(m);
+
+        let g = super::graph(&all, k, &is_unreachable, &is_stable);
+        // Owned: `all` is shuffled below.
+        let live: BTreeSet<String> = all
+            .iter()
+            .filter(|m| !unreachable.contains(*m))
+            .cloned()
+            .collect();
+        assert_eq!(g.keys().cloned().collect::<BTreeSet<_>>(), live);
+        for (a, ns) in &g {
+            assert!(!ns.contains(a), "{a} links to itself");
+            for b in ns {
+                assert!(live.contains(b), "{a} links to unreachable {b}");
+                assert!(g[b].contains(a), "{a} -> {b} but not back");
+            }
+        }
+        if live.len() <= k + 1 {
+            assert!(
+                g.values().all(|ns| ns.len() + 1 == live.len()),
+                "not a full mesh"
+            );
+        } else {
+            assert!(
+                g.values().all(|ns| ns.len() >= k),
+                "a member with fewer than k links"
+            );
+            assert!(connected(&g), "not connected");
+        }
+
+        rng.shuffle(&mut all);
+        assert_eq!(super::graph(&all, k, &is_unreachable, &is_stable), g);
+
+        if let Some(local) = live.iter().nth(rng.below(live.len())) {
+            let others: Vec<String> = all.iter().filter(|m| *m != local).cloned().collect();
+            assert_eq!(
+                neighbours(&others, local, k, &is_unreachable, &is_stable),
+                g[local]
+            );
+        }
+    });
+}
+
+#[test]
+fn small_swarms_are_a_full_mesh() {
+    let ms = members(13);
+    for m in &ms {
+        assert_eq!(
+            neighbours(&ms, m, 12, &all_reachable, &none_stable).len(),
+            12
+        );
+    }
+}
+
+#[test]
+fn links_are_symmetric_and_degree_at_least_k() {
+    let ms = members(250);
+    let g = graph(&ms, 8);
+    for (a, ns) in &g {
+        assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len());
+        assert!(!ns.contains(a));
+        for b in ns {
+            assert!(g[b].contains(a), "{a} -> {b} not symmetric");
+        }
+    }
+}
+
+#[test]
+fn graph_is_connected() {
+    for (n, k) in [(250, 8), (250, 2), (60, 3)] {
+        assert!(connected(&graph(&members(n), k)), "n={n} k={k}");
+    }
+}
+
+#[test]
+fn one_member_joining_changes_few_links() {
+    let ms = members(200);
+    let before = edges(&graph(&ms, 8));
+    let mut more = ms.clone();
+    more.push(format!("{:064x}", 999_999));
+    let after = edges(&graph(&more, 8));
+    let changed = before.symmetric_difference(&after).count();
+    assert!(changed <= 4 * 8, "{changed} links changed");
+}
+
+#[test]
+fn reachable_members_stay_connected_when_most_are_offline() {
+    let ms = members(100);
+    // Every tenth member is online.
+    let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect();
+    let offline = |m: &str| !online.contains(m);
+    let g = super::graph(&ms, 12, &offline, &none_stable);
+    assert_eq!(g.len(), 10);
+    // Ten reachable members and k = 12: a full mesh among them.
+    assert!(g.values().all(|ns| ns.len() == 9));
+
+    let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect();
+    let offline = |m: &str| !online.contains(m);
+    let g = super::graph(&ms, 8, &offline, &none_stable);
+    assert_eq!(g.len(), 25);
+    assert!(connected(&g));
+    assert!(
+        g.values()
+            .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n)))
+    );
+}
+
+#[test]
+fn stable_members_become_hubs_with_bounded_degree() {
+    let ms = members(250);
+    let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect();
+    let is_stable = |m: &str| stable.contains(m);
+    let g = super::graph(&ms, 12, &all_reachable, &is_stable);
+    assert!(connected(&g));
+
+    let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) {
+        let degrees: Vec<usize> = g
+            .iter()
+            .filter(|(m, _)| pred(m))
+            .map(|(_, ns)| ns.len())
+            .collect();
+        let max = *degrees.iter().max().unwrap();
+        (
+            max,
+            degrees.iter().sum::<usize>() as f64 / degrees.len() as f64,
+        )
+    };
+    let (stable_max, stable_mean) = degree(&|m| stable.contains(m));
+    let (_, transient_mean) = degree(&|m| !stable.contains(m));
+    // 230 transient members put about 5 of their 10 scored slots on 20 stable ones.
+    assert!(
+        stable_mean > 3.0 * transient_mean,
+        "{stable_mean} vs {transient_mean}"
+    );
+    assert!(
+        stable_max <= 100,
+        "a stable member has {stable_max} neighbours"
+    );
+}
+
+/// A hub's degree in a swarm of `n` with `stable_count` stable members stays within
+/// the bound.
+fn hub_degree_is_bounded(n: usize, stable_count: usize) {
+    // Few stable members among many: without the fan-in budget each would be chosen
+    // by most of the swarm (about 130 neighbours with 10 of 250). With it, a hub has
+    // at most the budget plus the links others choose by pair score and the ring.
+    let k = 12;
+    let ms = members(n);
+    let stable: BTreeSet<String> = ms.iter().step_by(n / stable_count).cloned().collect();
+    let is_stable = |m: &str| stable.contains(m);
+    let g = super::graph(&ms, k, &all_reachable, &is_stable);
+    assert!(connected(&g), "n={n} stable={stable_count}");
+    let hub_max = g
+        .iter()
+        .filter(|(m, _)| stable.contains(*m))
+        .map(|(_, ns)| ns.len())
+        .max()
+        .unwrap();
+    assert!(
+        hub_max <= 8 * k,
+        "n={n} stable={stable_count}: a stable member has {hub_max} neighbours"
+    );
+}
+
+/// Two transports with members on one or the other, and two members on both: every
+/// link is over a shared transport, and the two members on both connect the rest.
+#[test]
+fn members_on_two_transports_bridge_the_others() {
+    let k = 8;
+    let ms = members(201);
+    let on = |names: &[&str]| -> BTreeSet<String> {
+        names.iter().map(|name| (*name).to_owned()).collect()
+    };
+    let transports: BTreeMap<String, BTreeSet<String>> = ms
+        .iter()
+        .enumerate()
+        .map(|(i, m)| {
+            let transports = match i {
+                0 | 1 => on(&["a", "b"]),
+                i if i % 2 == 0 => on(&["a"]),
+                _ => on(&["b"]),
+            };
+            (m.clone(), transports)
+        })
+        .collect();
+    let no_stable = |_: &str| false;
+    let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+    for (m, own) in &transports {
+        for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) {
+            assert!(
+                !transports[&n].is_disjoint(own),
+                "{m} and {n} share no transport"
+            );
+            graph.entry(m.clone()).or_default().insert(n.clone());
+            graph.entry(n).or_default().insert(m.clone());
+        }
+    }
+    assert_eq!(graph.len(), ms.len());
+    assert!(connected(&graph));
+    let bridge = &ms[0];
+    assert!(graph[bridge].iter().any(|n| !transports[n].contains("b")));
+    assert!(graph[bridge].iter().any(|n| !transports[n].contains("a")));
+
+    // One transport: the one graph, as without transports.
+    let one: BTreeMap<String, BTreeSet<String>> =
+        ms.iter().map(|m| (m.clone(), on(&[""]))).collect();
+    for m in &ms[..5] {
+        assert_eq!(
+            neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable),
+            neighbours(&ms, m, k, &all_reachable, &no_stable)
+        );
+    }
+}
+
+#[test]
+fn schemes_of_addresses() {
+    let addresses = |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() };
+    assert_eq!(
+        schemes(&addresses(&[
+            "mycelium:ab",
+            "libp2p:/ip4/1.2.3.4/tcp/1",
+            "libp2p:/ip6/::1/tcp/1"
+        ])),
+        BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()])
+    );
+    assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()]));
+    assert!(schemes(&[]).is_empty());
+}
+
+#[test]
+fn hub_degree_is_bounded_however_few_stable_members() {
+    for (n, stable_count) in [(250, 10), (400, 5)] {
+        hub_degree_is_bounded(n, stable_count);
+    }
+}
+
+/// The same at the sizes the bound was designed for; slow in debug builds, so run
+/// on request (`cargo test --release -- --ignored`).
+#[test]
+#[ignore]
+fn hub_degree_is_bounded_in_large_swarms() {
+    for (n, stable_count) in [(1000, 10), (1000, 20), (2000, 20)] {
+        hub_degree_is_bounded(n, stable_count);
+    }
+}