nodes prefer stable neighbours

1 files changed, 89 insertions, 10 deletions

diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index dddf81b..8e25a2a 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -32,6 +32,7 @@ pub(crate) struct NeighbourView {
     members: Vec<String>,
     k: usize,
     unreachable: BTreeSet<String>,
+    stable: BTreeSet<String>,
     neighbours: BTreeSet<String>,
     /// Every reachable member is a neighbour.
     full_mesh: bool,
@@ -74,7 +75,16 @@ 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 (current, stable) = {
+        let replica = state.doc.read().await;
+        let current = members::members(&replica.doc);
+        let stable: BTreeSet<String> = current
+            .iter()
+            .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
+            .cloned()
+            .collect();
+        (current, stable)
+    };
     let k = state.neighbours_k;
     let now = Instant::now();
     let mut unreachable = BTreeSet::new();
@@ -91,7 +101,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             .read()
             .unwrap_or_else(|poisoned| poisoned.into_inner());
         if view.as_ref().is_some_and(|view| {
-            view.members == current && view.k == k && view.unreachable == unreachable
+            view.members == current
+                && view.k == k
+                && view.unreachable == unreachable
+                && view.stable == stable
         }) {
             return None;
         }
@@ -103,7 +116,13 @@ 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, &|m| unreachable.contains(m));
+    let computed = neighbours(
+        &current,
+        state.local_id(),
+        k,
+        &|m| unreachable.contains(m),
+        &|m| stable.contains(m),
+    );
     let change = NeighbourChange {
         added: computed
             .iter()
@@ -119,6 +138,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     info!(
         members = current.len(),
         unreachable = unreachable.len(),
+        stable = stable.len(),
         k,
         neighbours = computed.len(),
         added = change.added.len(),
@@ -137,6 +157,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         members: current,
         k,
         unreachable,
+        stable,
         neighbours: computed,
     });
     Some(change)
@@ -152,16 +173,18 @@ pub(crate) fn neighbours(
     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))
+    graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
         .remove(local)
         .unwrap_or_default()
 }
 
 /// Every reachable member's neighbours. Costs about N² hashes (a few milliseconds for
-/// hundreds of members).
+/// hundreds of members). `stable` members (`swarm::stability`) are preferred for half of
+/// each member's scored slots.
 ///
 /// 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`
@@ -174,6 +197,7 @@ 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();
@@ -226,7 +250,22 @@ pub(crate) fn graph(
             .map(|other| (pair_score(member, other), other))
             .collect();
         candidates.sort_unstable();
-        for (_, partner) in candidates.into_iter().take(k - 2) {
+        // 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 and the load
+        // on stable members bounded.
+        let stable_slots = (k - 2).div_ceil(2);
+        let preferred: Vec<&str> = candidates
+            .iter()
+            .map(|(_, other)| *other)
+            .filter(|other| stable(other))
+            .take(stable_slots)
+            .collect();
+        let rest = candidates
+            .iter()
+            .map(|(_, other)| *other)
+            .filter(|other| !preferred.contains(other))
+            .take(k - 2 - preferred.len());
+        for partner in preferred.iter().copied().chain(rest) {
             link(member, partner);
         }
     }
@@ -264,8 +303,12 @@ mod tests {
         false
     }
 
+    fn none_stable(_: &str) -> bool {
+        false
+    }
+
     fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
-        super::graph(members, k, &all_reachable)
+        super::graph(members, k, &all_reachable, &none_stable)
     }
 
     fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool {
@@ -303,7 +346,10 @@ mod tests {
     fn small_swarms_are_a_full_mesh() {
         let ms = members(13);
         for m in &ms {
-            assert_eq!(neighbours(&ms, m, 12, &all_reachable).len(), 12);
+            assert_eq!(
+                neighbours(&ms, m, 12, &all_reachable, &none_stable).len(),
+                12
+            );
         }
     }
 
@@ -344,14 +390,14 @@ mod tests {
         // 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);
+        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);
+        let g = super::graph(&ms, 8, &offline, &none_stable);
         assert_eq!(g.len(), 25);
         assert!(connected(&g));
         assert!(
@@ -359,4 +405,37 @@ mod tests {
                 .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"
+        );
+    }
 }