From c2468964b11bd3606f03375c370fd181a1723113 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 24 Oct 2026 04:20:21 +0200 Subject: find reachable neighbours --- src/state.rs | 30 ++++++++ src/swarm/peer.rs | 2 + src/swarm/topology.rs | 185 +++++++++++++++++++++++++++++++++++--------------- 3 files changed, 162 insertions(+), 55 deletions(-) diff --git a/src/state.rs b/src/state.rs index 64fe6a4..7574e35 100644 --- a/src/state.rs +++ b/src/state.rs @@ -137,8 +137,18 @@ pub(crate) struct PeerState { /// How many heads its most recent keepalive listed (None if it sent a digest). pub(crate) reported_heads: Option, + + /// HELLOs sent in the current handshake attempt without hearing anything back. + pub(crate) hello_attempts: u32, + /// Since when we consider the peer unreachable (three unanswered HELLOs, or + /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the + /// peer becomes a neighbour candidate again and gets tried once more. + pub(crate) unreachable_since: Option, } +/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). +const UNREACHABLE_AFTER_HELLOS: u32 = 3; + /// One Automerge sync with a peer, from a reset to the next. Everything a reset must /// forget lives here, so a reset replaces it whole and can't miss a field. #[derive(Clone)] @@ -284,6 +294,8 @@ impl PeerState { reports_ready: None, syncs_with_us: None, reported_heads: None, + hello_attempts: 0, + unreachable_since: None, } } @@ -309,6 +321,8 @@ impl PeerState { /// Records inbound activity. Returns true if the peer was dormant. pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { self.last_seen = Some(now); + self.hello_attempts = 0; + self.unreachable_since = None; if matches!(self.liveness, Liveness::Dormant { .. }) { self.liveness = Liveness::Active; @@ -321,6 +335,21 @@ impl PeerState { false } + /// A HELLO went out (or failed to). After the third unanswered one, the peer counts + /// as unreachable. + pub(crate) fn hello_sent(&mut self, now: Instant) { + self.hello_attempts += 1; + if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { + self.unreachable_since.get_or_insert(now); + } + } + + /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). + pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool { + self.unreachable_since + .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe) + } + /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight /// finishes; a later handshake starts over. pub(crate) fn stop_hello_retries(&mut self) { @@ -336,6 +365,7 @@ impl PeerState { } self.liveness = Liveness::Dormant { since: now }; + self.unreachable_since.get_or_insert(now); self.reset_sync(); if self.hello_state != HelloState::Sending { diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 7ab704f..a42f5a4 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -220,6 +220,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> // A new handshake: retries start over. peer_state.hello_state = HelloState::Sending; peer_state.hello_backoff.reset(); + peer_state.hello_attempts = 0; true } else { trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending"); @@ -282,6 +283,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::Sending { + peer_state.hello_sent(Instant::now()); let wait = match peer_state.liveness { Liveness::Dormant { .. } => state.timing.dormant_probe, Liveness::Active => peer_state.hello_backoff.next(), diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index f097cdd..b5880cf 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -5,19 +5,22 @@ //! 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 same graph from the member list alone, with no coordination: -//! * a ring ordered by key hash: each member's predecessor and successor, which keeps -//! the graph connected; -//! * plus, for each member, the `k - 2` others with the lowest pair score (a hash of -//! both keys, so both ends agree on it), which keeps paths short. +//! 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` members the graph is the full mesh: small swarms -//! behave exactly as before. A member joining or leaving changes only links near it. +//! `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. use std::collections::{BTreeMap, BTreeSet}; use sha2::{Digest, Sha256}; +use tokio::time::Instant; use tracing::info; use crate::state::AppState; @@ -28,6 +31,7 @@ use crate::swarm::members; pub(crate) struct NeighbourView { members: Vec, k: usize, + unreachable: BTreeSet, neighbours: BTreeSet, } @@ -52,22 +56,31 @@ pub(crate) struct NeighbourChange { pub(crate) removed: Vec, } -/// Recomputes the neighbours once ready, if the member list (or `k`) changed. +/// 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 = members::members(&state.doc.read().await.doc); let k = state.neighbours_k; + let now = Instant::now(); + let mut unreachable = BTreeSet::new(); + for member in ¤t { + if let Some(peer_state) = state.get_peer_state(member).await { + if peer_state.lock().await.is_unreachable(&state.timing, now) { + unreachable.insert(member.clone()); + } + } + } let previous = { let view = state .neighbours .read() .unwrap_or_else(|poisoned| poisoned.into_inner()); - if view - .as_ref() - .is_some_and(|view| view.members == current && view.k == k) - { + if view.as_ref().is_some_and(|view| { + view.members == current && view.k == k && view.unreachable == unreachable + }) { return None; } view.as_ref().map(|view| view.neighbours.clone()) @@ -78,7 +91,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { Some(previous) => previous, None => state.peer_states.read().await.keys().cloned().collect(), }; - let computed = neighbours(¤t, state.local_id(), k); + let computed = neighbours(¤t, state.local_id(), k, &|m| unreachable.contains(m)); let change = NeighbourChange { added: computed .iter() @@ -93,6 +106,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { }; info!( members = current.len(), + unreachable = unreachable.len(), k, neighbours = computed.len(), added = change.added.len(), @@ -105,6 +119,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView { members: current, k, + unreachable, neighbours: computed, }); Some(change) @@ -113,48 +128,84 @@ pub(crate) async fn refresh(state: &AppState) -> Option { const RING_LABEL: &[u8] = b"continuum/topology-ring/1"; const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1"; -/// The neighbours of `local` among `members` (which may or may not contain `local`). -pub(crate) fn neighbours(members: &[String], local: &str, k: usize) -> BTreeSet { +/// 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, +) -> BTreeSet { let mut all: Vec = members.to_vec(); all.push(local.to_owned()); - graph(&all, k).remove(local).unwrap_or_default() + graph(&all, k, &|m: &str| m != local && unreachable(m)) + .remove(local) + .unwrap_or_default() } -/// Every member's neighbours. Costs about N² hashes (a few milliseconds for hundreds of -/// members); computed again only when the member list changes. -pub(crate) fn graph(members: &[String], k: usize) -> BTreeMap> { +/// Every reachable member's neighbours. Costs about N² hashes (a few milliseconds for +/// hundreds of 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. +pub(crate) fn graph( + members: &[String], + k: usize, + unreachable: &dyn Fn(&str) -> bool, +) -> BTreeMap> { let k = k.max(2); let all: BTreeSet<&str> = members.iter().map(String::as_str).collect(); - let mut links: BTreeMap> = all + let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect(); + let mut links: BTreeMap> = 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("member").insert(b.to_owned()); - links.get_mut(b).expect("member").insert(a.to_owned()); + links.get_mut(a).expect("live member").insert(b.to_owned()); + links.get_mut(b).expect("live member").insert(a.to_owned()); } }; - if all.len() <= k + 1 { - for a in &all { - for b in &all { + if live.len() <= k + 1 { + for a in &live { + for b in &live { link(a, b); } } return links; } - let ring = ring_links(&all); - for (member, [prev, next]) in &ring { - link(member, prev); + // 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") + }; + + for member in &live { + let at = position[member]; + let next = first_live(at, 1); + let prev = first_live(at, n - 1); link(member, next); - // Its k - 2 partners with the lowest pair score, besides its ring links. - let mut candidates: Vec<([u8; 32], &str)> = all + link(member, prev); + let mut candidates: Vec<([u8; 32], &str)> = live .iter() .copied() - .filter(|other| other != member && other != prev && other != next) + .filter(|other| other != member && *other != next && *other != prev) .map(|other| (pair_score(member, other), other)) .collect(); candidates.sort_unstable(); @@ -165,17 +216,6 @@ pub(crate) fn graph(members: &[String], k: usize) -> BTreeMap(all: &BTreeSet<&'a str>) -> BTreeMap<&'a str, [&'a str; 2]> { - let mut ring: Vec<([u8; 32], &'a str)> = - all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect(); - ring.sort_unstable(); - let n = ring.len(); - (0..n) - .map(|i| (ring[i].1, [ring[(i + n - 1) % n].1, ring[(i + 1) % n].1])) - .collect() -} - /// Symmetric: the same for (a, b) and (b, a). fn pair_score(a: &str, b: &str) -> [u8; 32] { let (low, high) = if a <= b { (a, b) } else { (b, a) }; @@ -203,6 +243,30 @@ mod tests { (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect() } + fn all_reachable(_: &str) -> bool { + false + } + + fn graph(members: &[String], k: usize) -> BTreeMap> { + super::graph(members, k, &all_reachable) + } + + fn connected(g: &BTreeMap>) -> 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>) -> BTreeSet<(String, String)> { graph .iter() @@ -222,7 +286,7 @@ mod tests { fn small_swarms_are_a_full_mesh() { let ms = members(13); for m in &ms { - assert_eq!(neighbours(&ms, m, 12).len(), 12); + assert_eq!(neighbours(&ms, m, 12, &all_reachable).len(), 12); } } @@ -242,18 +306,7 @@ mod tests { #[test] fn graph_is_connected() { for (n, k) in [(250, 8), (250, 2), (60, 3)] { - let ms = members(n); - let g = graph(&ms, k); - let mut seen = BTreeSet::from([ms[0].clone()]); - let mut queue = VecDeque::from([ms[0].clone()]); - while let Some(m) = queue.pop_front() { - for next in &g[&m] { - if seen.insert(next.clone()) { - queue.push_back(next.clone()); - } - } - } - assert_eq!(seen.len(), n, "n={n} k={k}"); + assert!(connected(&graph(&members(n), k)), "n={n} k={k}"); } } @@ -267,4 +320,26 @@ mod tests { 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 = ms.iter().step_by(10).cloned().collect(); + let offline = |m: &str| !online.contains(m); + let g = super::graph(&ms, 12, &offline); + 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 = ms.iter().step_by(4).cloned().collect(); + let offline = |m: &str| !online.contains(m); + let g = super::graph(&ms, 8, &offline); + assert_eq!(g.len(), 25); + assert!(connected(&g)); + assert!( + g.values() + .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n))) + ); + } } -- cgit v1.3.1