//! 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, k: usize, unreachable: BTreeSet, stable: BTreeSet, /// The transports (address schemes) of each member, and ours. transports: BTreeMap>, local_transports: BTreeSet, /// 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, /// 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, neighbours: BTreeSet, /// 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 { 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>> { 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, pub(crate) removed: Vec, } /// 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( 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 { 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 = members::electorate( &replica.doc, state.clock.now_millis().0, state.electorate_window, ) .into_iter() .collect(); let stable: BTreeSet = current .iter() .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable) .cloned() .collect(); let transports: BTreeMap> = current .iter() .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m)))) .collect(); (current, stable, transports, alive) }; let local_transports: BTreeSet = 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 ¤t { 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 ¤t { 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;