From 278ca3cd21397ffad07f0874b80b77f41088a3f2 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 04:15:59 +0100 Subject: swarm neighbours --- src/api/stats.rs | 2 +- src/config/mod.rs | 2 +- src/document/meta.rs | 3 ++ src/swarm/bootstrap.rs | 12 +++++-- src/swarm/topology/mod.rs | 76 ++++++++++++++++++++++++++++++++++++++++++--- src/swarm/topology/tests.rs | 22 +++++++++++++ 6 files changed, 109 insertions(+), 8 deletions(-) (limited to 'src') diff --git a/src/api/stats.rs b/src/api/stats.rs index 306b6a2..2af6549 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -377,7 +377,7 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR blob_requests_dropped: node.blob_requests_dropped, }, topology: TopologyStats { - k: state.neighbours_k, + k: topology::k(state), view: topology::summary(state).map(|s| TopologyView { members: s.members, reachable: s.reachable, diff --git a/src/config/mod.rs b/src/config/mod.rs index a6f70aa..93ae7fd 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -275,7 +275,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::Integer, "COUNT", Some("12"), - "Neighbours each node keeps (at least 2). Swarms of up to this many + 1 nodes are a full mesh; larger ones sync over several hops.", + "Neighbours each member keeps (at least 2), on a genesis node: recorded in the document as the swarm's, which every node uses (a protocol writer may change it there). Elsewhere it counts only in a swarm that recorded none (one from before), where it must be the same on every node. Swarms of up to this many + 1 nodes are a full mesh; larger ones sync over several hops.", ), option( "electorate-window", diff --git a/src/document/meta.rs b/src/document/meta.rs index 24ed7c7..af1cd50 100644 --- a/src/document/meta.rs +++ b/src/document/meta.rs @@ -7,6 +7,8 @@ //! * `votes/`: the votes register (`remake::records`). //! * `epochs/`: epoch metadata, written by snapshots (`remake::snapshot`). //! * `swarm`: the swarm ID, written once by the genesis node (`swarm::bootstrap`). +//! * `neighbours`: the neighbours each member keeps, written by the genesis node, and +//! changed by a protocol writer if need be (`swarm::topology::swarm_neighbours`). //! * `relays`: relays for the swarm's nodes, listed by an operator //! (`swarm::members::relays`). @@ -18,6 +20,7 @@ pub(crate) const REMAKES: &str = "remakes"; pub(crate) const VOTES: &str = "votes"; pub(crate) const EPOCHS: &str = "epochs"; pub(crate) const SWARM: &str = "swarm"; +pub(crate) const NEIGHBOURS: &str = "neighbours"; pub(crate) const RELAYS: &str = "relays"; /// The store path `_continuum/`. diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 4f6c072..8665253 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -87,10 +87,18 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) } }; let recorded = JsonValue::String(swarm_id.clone()); + let neighbours = state.neighbours_k; state - .mutate_doc(move |doc| store::write_path(doc, &swarm_id_path(), &recorded)) + .mutate_doc(move |doc| { + store::write_path(doc, &swarm_id_path(), &recorded)?; + store::write_path( + doc, + &crate::swarm::topology::neighbours_path(), + &JsonValue::from(neighbours), + ) + }) .await - .context("Recording the swarm ID")?; + .context("Recording the swarm ID and its neighbours")?; state.initial_sync_done.store(true, Ordering::Release); state.mark_ready(swarm_id).await?; // Remembered members, if any, to find those still running. 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 { + 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 = 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 = 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 { 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 = members::electorate( @@ -320,10 +383,15 @@ pub(crate) async fn refresh(state: &AppState) -> Option { .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 = 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())); +} -- cgit v1.3.1