1 files changed, 82 insertions, 0 deletions

diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 57856b6..0dae7e9 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -107,6 +107,26 @@ pub(crate) struct NeighbourChange {
     pub(crate) removed: Vec<String>,
 }
 
+/// Recomputes the neighbours whenever something they depend on may have changed
+/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`),
+/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
+/// fallback, which also covers unreachability expiring after `dormant_probe`. 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> {
@@ -397,6 +417,68 @@ mod tests {
             .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);