topology: explode!

1 files changed, 359 insertions, 0 deletions

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;