diff options
| -rw-r--r-- | src/swarm/members.rs | 22 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 910 | ||||
| -rw-r--r-- | src/swarm/topology/graph.rs | 236 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 359 | ||||
| -rw-r--r-- | src/swarm/topology/tests.rs | 313 |
5 files changed, 930 insertions, 910 deletions
diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 3bfb9cf..94db4c9 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -135,6 +135,28 @@ pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec<String> { } } +/// What the member list tells this node besides who its neighbours are: where members +/// are (`learn_addresses`), what they call themselves (`set_aliases`), and the relays +/// listed (`relays`). Done whenever the neighbours are recomputed (`topology::refresh`). +pub(crate) fn apply(state: &AppState, doc: &AutoCommit) { + learn_addresses(state, doc); + set_aliases(state, aliases(doc)); + state.transport.learn_relays(&relays(doc)); +} + +/// Hands the members' `aliases` to the logger, under every name that stands for each: +/// its node ID, and its addresses (`logging`, which renames them bare too). +fn set_aliases(state: &AppState, aliases: HashMap<String, String>) { + let mut names = HashMap::new(); + for (member, alias) in aliases { + for address in state.transport.identities_of(&member) { + names.insert(address, alias.clone()); + } + names.insert(member, alias); + } + crate::logging::set_aliases(names); +} + /// Hands every other member's addresses to the transport (`Transport::learn_addresses`): /// the member list is how a node finds where members it never met are. pub(crate) fn learn_addresses(state: &AppState, doc: &AutoCommit) { diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs deleted file mode 100644 index c861101..0000000 --- a/src/swarm/topology.rs +++ /dev/null @@ -1,910 +0,0 @@ -//! 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 sha2::{Digest, Sha256}; -use tracing::info; - -use crate::locks::RwLockExt; -use crate::state::AppState; -use crate::swarm::members; - -/// 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()) -} - -/// Hands the members' `aliases` to the logger, under every name that stands for each: -/// its node ID, and its addresses (`logging`, which renames them bare too). -fn set_aliases(state: &AppState, aliases: std::collections::HashMap<String, String>) { - let mut names = std::collections::HashMap::new(); - for (member, alias) in aliases { - for address in state.transport.identities_of(&member) { - names.insert(address, alias.clone()); - } - names.insert(member, alias); - } - crate::logging::set_aliases(names); -} - -/// 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::learn_addresses(state, &replica.doc); - set_aliases(state, members::aliases(&replica.doc)); - state.transport.learn_relays(&members::relays(&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 ¤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, - }); - Some(change) -} - -const RING_LABEL: &[u8] = b"continuum/topology-ring/1"; -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. -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`. -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. -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. -fn mix_pair(a: u64, b: u64) -> u64 { - let (low, high) = if a <= b { (a, b) } else { (b, a) }; - mix(low ^ mix(high)) -} - -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) -} - -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() -} - -#[cfg(test)] -mod tests { - 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); - } - } -} 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 ¤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, + }); + 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); + } +} |