swarm neighbours

2 files changed, 94 insertions, 4 deletions

diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index 37b2f71..179cf83 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -290,7 +290,69 @@ pub(crate) async fn run_worker(state: AppState) {
 /// 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())
+    path.is_empty()
+        || path.starts_with(&members::members_dir())
+        || path.starts_with(&neighbours_path())
+}
+
+/// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`).
+const MAX_NEIGHBOURS: u64 = 256;
+
+/// Where the swarm records its neighbours (`swarm_neighbours`); the genesis node writes
+/// its `--neighbours` there (`swarm::bootstrap`).
+pub(crate) fn neighbours_path() -> Vec<String> {
+    crate::document::meta::path(&[crate::document::meta::NEIGHBOURS])
+}
+
+/// The neighbours each member keeps, as the swarm records it: its genesis node writes
+/// its `--neighbours` (`neighbours_path`), a protocol writer may change it (every
+/// node then recomputes its neighbours, and the writers' trees with them). The same on
+/// every node, as the member list's graph must be (`member_list_graph`). Where the
+/// document has none (a swarm from before it was recorded), or one that isn't a number
+/// from 2 to `MAX_NEIGHBOURS`, this node's own (`own`).
+pub(crate) fn swarm_neighbours(doc: &crate::document::Document, own: usize) -> usize {
+    match crate::document::store::read_path(doc, &neighbours_path()) {
+        Ok(Some(value)) => match value.as_u64() {
+            Some(k) if (2..=MAX_NEIGHBOURS).contains(&k) => k as usize,
+            _ => {
+                warn_once(&format!(
+                    "The swarm's neighbours ({value}) isn't a number from 2 to {MAX_NEIGHBOURS}; using --neighbours ({own})"
+                ));
+                own
+            }
+        },
+        _ => own,
+    }
+}
+
+/// Logs `message` at info level, unless it was the last one logged so.
+fn info_once(message: &str) {
+    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
+    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if *last != message {
+        info!("{message}");
+        *last = message.to_owned();
+    }
+}
+
+/// Logs `message` at warn level, unless it was the last one logged so.
+fn warn_once(message: &str) {
+    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
+    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if *last != message {
+        tracing::warn!("{message}");
+        *last = message.to_owned();
+    }
+}
+
+/// The neighbours each member keeps, as the view was last computed with
+/// (`swarm_neighbours`); this node's `--neighbours` before.
+pub(crate) fn k(state: &AppState) -> usize {
+    state
+        .neighbours
+        .read_or_recover()
+        .as_ref()
+        .map_or(state.neighbours_k, |view| view.k)
 }
 
 /// Recomputes the neighbours once ready, if the member list, the reachable members or
@@ -299,9 +361,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     if !state.is_ready().await {
         return None;
     }
-    let (current, stable, transports, alive) = {
+    let (current, stable, transports, alive, k) = {
         let replica = state.doc.read().await;
         members::apply(state, &replica.doc);
+        let k = swarm_neighbours(&replica.doc, state.neighbours_k);
         let current = members::members(&replica.doc);
         // Those that said they are alive within the electorate window.
         let alive: BTreeSet<String> = members::electorate(
@@ -320,10 +383,15 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             .iter()
             .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
             .collect();
-        (current, stable, transports, alive)
+        (current, stable, transports, alive, k)
     };
+    if k != state.neighbours_k {
+        info_once(&format!(
+            "The swarm keeps {k} neighbours per member (recorded by its genesis node, or a protocol writer); --neighbours ({}) counts only on a genesis node",
+            state.neighbours_k
+        ));
+    }
     let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
-    let k = state.neighbours_k;
     // Those that don't answer, those on none of our transports, and those that haven't
     // said they are alive for longer than the electorate window, unless we heard from
     // them since we started (one that comes back says so to us first: its HELLO).
diff --git a/src/swarm/topology/tests.rs b/src/swarm/topology/tests.rs
index 4069c0e..1b9c75d 100644
--- a/src/swarm/topology/tests.rs
+++ b/src/swarm/topology/tests.rs
@@ -467,3 +467,25 @@ fn bench_writer_trees() {
         }
     }
 }
+
+/// The neighbours a swarm records count; without one, or with one out of range, the
+/// node's own.
+#[test]
+fn the_swarms_neighbours_count() {
+    use crate::document::testing::change;
+    use crate::document::{Document, store};
+    let mut doc = Document::new();
+    assert_eq!(swarm_neighbours(&doc, 12), 12);
+    for (recorded, expected) in [
+        (serde_json::json!(8), 8),
+        (serde_json::json!(1), 12),
+        (serde_json::json!(1000), 12),
+        (serde_json::json!("eight"), 12),
+    ] {
+        change(&mut doc, |doc| {
+            store::write_path(doc, &neighbours_path(), &recorded).unwrap()
+        });
+        assert_eq!(swarm_neighbours(&doc, 12), expected, "{recorded}");
+    }
+    assert!(affects_neighbours(&neighbours_path()));
+}