path: root/src/swarm/topology/mod.rs

//! 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, PeerState};
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>,
}

/// Runs `change` on a peer's state, and has the neighbours recomputed (`run_worker`) if
/// it changed what they are chosen from (`PeerState::neighbour_inputs`). Changes to a
/// peer that may matter to the neighbours go through here.
pub(crate) fn change_peer<T>(
    state: &AppState,
    peer_state: &mut PeerState,
    change: impl FnOnce(&mut PeerState) -> T,
) -> T {
    let before = peer_state.neighbour_inputs();
    let result = change(peer_state);
    if peer_state.neighbour_inputs() != before {
        state.topology_changed.notify_one();
    }
    result
}

/// Recomputes the neighbours whenever something they depend on may have changed
/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
/// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). 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,
    });
    // New neighbours are kept alive and judged.
    state.peers_due.now();
    Some(change)
}

#[cfg(test)]
mod tests;