diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/bootstrap.rs | 12 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 76 | ||||
| -rw-r--r-- | src/swarm/topology/tests.rs | 22 |
3 files changed, 104 insertions, 6 deletions
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<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())); +} |
