find reachable neighbours

1 files changed, 130 insertions, 55 deletions

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<String>,
     k: usize,
+    unreachable: BTreeSet<String>,
     neighbours: BTreeSet<String>,
 }
 
@@ -52,22 +56,31 @@ pub(crate) struct NeighbourChange {
     pub(crate) removed: Vec<String>,
 }
 
-/// 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<NeighbourChange> {
     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 &current {
+        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<NeighbourChange> {
         Some(previous) => previous,
         None => state.peer_states.read().await.keys().cloned().collect(),
     };
-    let computed = neighbours(&current, state.local_id(), k);
+    let computed = neighbours(&current, 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<NeighbourChange> {
     };
     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<NeighbourChange> {
         .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<NeighbourChange> {
 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<String> {
+/// 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<String> {
     let mut all: Vec<String> = 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<String, BTreeSet<String>> {
+/// 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<String, BTreeSet<String>> {
     let k = k.max(2);
     let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
-    let mut links: BTreeMap<String, BTreeSet<String>> = all
+    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("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<String, BTreeSet<S
     links
 }
 
-/// Ring predecessor and successor of every member, by key hash.
-fn ring_links<'a>(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<String, BTreeSet<String>> {
+        super::graph(members, k, &all_reachable)
+    }
+
+    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()
@@ -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<String> = 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<String> = 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)))
+        );
+    }
 }