sync with neighbours

1 files changed, 91 insertions, 0 deletions

diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 0ec8b14..f097cdd 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -18,6 +18,97 @@
 use std::collections::{BTreeMap, BTreeSet};
 
 use sha2::{Digest, Sha256};
+use tracing::info;
+
+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,
+    neighbours: BTreeSet<String>,
+}
+
+/// 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()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+    {
+        Some(view) => view.neighbours.contains(peer),
+        None => true,
+    }
+}
+
+/// How the neighbours changed.
+pub(crate) struct NeighbourChange {
+    pub(crate) added: Vec<String>,
+    pub(crate) removed: Vec<String>,
+}
+
+/// Recomputes the neighbours once ready, if the member list (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 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)
+        {
+            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 computed = neighbours(&current, state.local_id(), k);
+    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(),
+        k,
+        neighbours = computed.len(),
+        added = change.added.len(),
+        removed = change.removed.len(),
+        "Neighbours computed from the member list"
+    );
+    *state
+        .neighbours
+        .write()
+        .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView {
+        members: current,
+        k,
+        neighbours: computed,
+    });
+    Some(change)
+}
 
 const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
 const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1";