From 5994b49412c058079639585501a78e6458fdbb6f Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 19:54:09 +0100 Subject: HyParView (slim) --- src/swarm/bootstrap.rs | 11 +- src/swarm/hyparview.rs | 457 +++++++++++++++++++++++++++++++++++++++++++++ src/swarm/mod.rs | 1 + src/swarm/peer/inbound.rs | 9 +- src/swarm/relay.rs | 4 +- src/swarm/sync.rs | 9 +- src/swarm/topology/mod.rs | 161 +++++++++++++++- src/swarm/working_links.rs | 6 + 8 files changed, 641 insertions(+), 17 deletions(-) create mode 100644 src/swarm/hyparview.rs (limited to 'src/swarm') diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 8665253..76e056d 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -88,6 +88,9 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) }; let recorded = JsonValue::String(swarm_id.clone()); let neighbours = state.neighbours_k; + // Plain names (kebab-case): serializing them can't fail. + let topology = serde_json::to_value(state.configured_ways.0).unwrap_or_default(); + let dissemination = serde_json::to_value(state.configured_ways.1).unwrap_or_default(); state .mutate_doc(move |doc| { store::write_path(doc, &swarm_id_path(), &recorded)?; @@ -95,10 +98,16 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) doc, &crate::swarm::topology::neighbours_path(), &JsonValue::from(neighbours), + )?; + store::write_path(doc, &crate::swarm::topology::topology_path(), &topology)?; + store::write_path( + doc, + &crate::swarm::topology::dissemination_path(), + &dissemination, ) }) .await - .context("Recording the swarm ID and its neighbours")?; + .context("Recording the swarm ID, its neighbours, its topology and dissemination")?; 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/hyparview.rs b/src/swarm/hyparview.rs new file mode 100644 index 0000000..6e363e3 --- /dev/null +++ b/src/swarm/hyparview.rs @@ -0,0 +1,457 @@ +//! HyParView (Leitão, Pereira and Rodrigues, "HyParView: a membership protocol for +//! reliable gossip-based broadcast", 2007), slimmed down, as an experiment +//! (`--topology hyparview`, `swarm::topology::Topology`). +//! +//! Each node keeps an active view: its neighbours, about `k` of them (the swarm's +//! neighbours, as with the member list's graph), each of which keeps it in its own: a +//! link both ends agreed to (NEIGHBOUR, answered by NEIGHBOUR_REPLY), not one every +//! node computes alike. A node with fewer asks others from its passive view, one at a +//! time per missing neighbour; one that has none asks with high priority, and is taken +//! even where that means dropping another (told so: DISCONNECT). A neighbour that fails +//! (judged unreachable: unanswered HELLOs, dormant, not alive lately) leaves the active +//! view, and another is asked. Every neighbour is asked again now and then whether it +//! still is (`REAFFIRM_EVERY`): the paper's links are TCP connections, whose ends know +//! when the other closed them; ours, without, could be left in one end's view. +//! +//! The paper's passive view comes from its own messages (JOIN, FORWARD_JOIN, SHUFFLE), +//! as its nodes know no member list. Ours has one, in the document, so the passive view +//! is drawn from it (`passive_view`): every member we may reach that isn't in the +//! active view. Those messages could replace it later without changing the rest. +//! +//! There is no graph every node shares (`Topology::graph`): what needs one does nothing +//! (writers' trees, recorded links, indirect probes), so changes pass on by a +//! dissemination that needs none (`--dissemination plumtree`), or by announcements. +//! +//! A member being asked counts as a neighbour already: it gets HELLOs, and NEIGHBOUR +//! goes once the handshake is done (a swarm's members only take messages from peers +//! that did one). Refused, or unanswered for `ASK_FOR`, it is not asked again for +//! `REFUSED_FOR`, and stops counting. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use tokio::time::Instant; +use tracing::debug; + +use crate::locks::MutexExt; +use crate::state::{AppState, HelloState}; +use crate::swarm::topology::{self, Inputs, Topology}; +use crate::transport::{NeighbourAnswer, NeighbourRequest, ParsedMessage}; + +/// How long a member asked to be a neighbour has to agree, the handshake before +/// included: over Tor, a few round trips of up to 3 s. +const ASK_FOR: Duration = Duration::from_secs(20); + +/// A member that refused, didn't answer, or left isn't asked again for this long. +const REFUSED_FOR: Duration = Duration::from_secs(60); + +/// A member one neighbour short asks this seldom: in a swarm whose members mostly have +/// `k`, those that refuse are full, and asking each in turn kept NEIGHBOURs going in a +/// quiet swarm (up to 300 in 20 s at 50 members). Two or more short, it asks at once. +const ONE_SHORT_EVERY: Duration = Duration::from_secs(300); + +/// Each neighbour is asked again this often whether it still is (NEIGHBOUR, as a new +/// ask would be): a NEIGHBOUR_REPLY or DISCONNECT lost on a lossy link leaves a link in +/// one end's view only, and nothing else would tell that end: the other still answers +/// it. A neighbour that has us agrees at once; one that doesn't takes us if it has room, +/// else refuses, and we drop it. +const REAFFIRM_EVERY: Duration = Duration::from_secs(60); + +/// A member asked to be a neighbour. +#[derive(Debug)] +struct Asking { + since: Instant, + /// With high priority: we had no neighbour when we asked. + high: bool, + /// NEIGHBOUR sent (after the handshake). + sent: bool, +} + +/// The views. +#[derive(Debug, Default)] +struct Views { + /// Our neighbours: each agreed, or asked us and we agreed. + active: BTreeSet, + /// Asked, not answered yet. + asking: BTreeMap, + /// Refused, unanswered or left lately: since when. + refused: BTreeMap, + /// When we last asked while one neighbour short (`ONE_SHORT_EVERY`). + asked_one_short: Option, + /// When the neighbours were last asked whether they still are (`REAFFIRM_EVERY`). + reaffirmed: Option, +} + +#[derive(Default)] +pub(crate) struct HyParView { + views: Mutex, + /// Bumped whenever the views change what `neighbours` gives (`Topology::version`). + version: AtomicU64, +} + +impl HyParView { + /// The members we could ask to be neighbours: from the member list, every member on + /// one of our transports that we don't judge unreachable, not us. The paper keeps + /// these from its own messages instead (`passive_view`, and the module's docs). + fn reachable_members(local: &str, inputs: &Inputs<'_>) -> BTreeSet { + inputs + .transports + .iter() + .filter(|(member, on)| { + member.as_str() != local + && !on.is_disjoint(inputs.local_transports) + && !(inputs.unreachable)(member) + }) + .map(|(member, _)| member.clone()) + .collect() + } + + /// The passive view: whom we may ask when we lack neighbours. Drawn from the member + /// list (`reachable_members`), not from JOIN and SHUFFLE as in the paper. + fn passive_view(views: &Views, reachable: &BTreeSet) -> Vec { + reachable + .iter() + .filter(|member| { + !views.active.contains(*member) + && !views.asking.contains_key(*member) + && !views.refused.contains_key(*member) + }) + .cloned() + .collect() + } + + fn changed(&self, state: &AppState) { + self.version.fetch_add(1, Ordering::Relaxed); + state.topology_changed.notify_one(); + } + + async fn send(state: &AppState, peer: &str, message: ParsedMessage) { + if let Err(error) = state.transport.send_message(peer, &message).await { + debug!(peer = %peer, error = %format!("{error:#}"), "A HyParView message failed"); + } + } + + fn answer(state: &AppState, agreed: bool) -> NeighbourAnswer { + let (epoch, epoch_seq) = state.epoch_stamp(); + NeighbourAnswer { + epoch, + epoch_seq, + agreed, + } + } +} + +/// `count` of `members`, picked at random. +fn some_of(mut members: Vec, count: usize) -> Vec { + let count = count.min(members.len()); + for picked in 0..count { + let rest = (members.len() - picked) as u64; + let other = picked + (crate::transport::paths::random_nonce() % rest) as usize; + members.swap(picked, other); + } + members.truncate(count); + members +} + +#[async_trait] +impl Topology for HyParView { + fn name(&self) -> &'static str { + "hyparview" + } + + fn graph( + &self, + members: &[String], + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, + ) -> Option>> { + let _ = (members, k, unreachable, stable); + None + } + + /// The active view, and those being asked. + fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet { + let _ = (local, inputs); + let views = self.views.lock_or_recover(); + views + .active + .iter() + .chain(views.asking.keys()) + .cloned() + .collect() + } + + /// Failed neighbours dropped, unanswered asks given up, NEIGHBOUR sent to those + /// asked once the handshake is done, and new members asked while we lack + /// neighbours. + async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) { + let local = state.local_id().to_owned(); + let reachable = Self::reachable_members(&local, inputs); + let now = Instant::now(); + let mut to_leave = Vec::new(); + let (changed, to_ask) = { + let mut views = self.views.lock_or_recover(); + let mut changed = false; + // Told all the same: one we judge unreachable may only be silent towards + // us (a link that works one way), and kept us in its view, it would be our + // neighbour at its end alone. + let failed: Vec = views + .active + .iter() + .filter(|member| !reachable.contains(*member)) + .cloned() + .collect(); + if !failed.is_empty() { + debug!( + dropped = failed.len(), + "Neighbours judged unreachable left the active view" + ); + for member in failed { + views.active.remove(&member); + to_leave.push(member); + } + changed = true; + } + let given_up: Vec = views + .asking + .iter() + .filter(|(member, asking)| { + !reachable.contains(*member) || now.duration_since(asking.since) > ASK_FOR + }) + .map(|(member, _)| member.clone()) + .collect(); + for member in given_up { + // Asked: told we gave up, should it agree after all (its answer lost, + // or still on its way). + if views + .asking + .remove(&member) + .is_some_and(|asking| asking.sent) + { + to_leave.push(member.clone()); + } + views.refused.insert(member, now); + changed = true; + } + views + .refused + .retain(|_, since| now.duration_since(*since) < REFUSED_FOR); + // More than one beyond `k` (asks sent before others asked us came back + // agreed): the excess dropped, at random, and told. + if views.active.len() > inputs.k + 1 { + let excess = views.active.len() - inputs.k; + let all: Vec = views.active.iter().cloned().collect(); + for member in some_of(all, excess) { + views.active.remove(&member); + views.refused.insert(member.clone(), now); + to_leave.push(member); + } + changed = true; + } + let missing = inputs + .k + .saturating_sub(views.active.len() + views.asking.len()); + // One short, with neighbours: not again before `ONE_SHORT_EVERY`. + let one_short = missing == 1 && !views.active.is_empty(); + let waiting = one_short + && views + .asked_one_short + .is_some_and(|at| now.duration_since(at) < ONE_SHORT_EVERY); + if missing > 0 && !waiting { + if one_short { + views.asked_one_short = Some(now); + } + // With high priority only the first, while we have no neighbour: the + // others would make room for us where none is needed. + let mut high = views.active.is_empty() && views.asking.is_empty(); + for member in some_of(Self::passive_view(&views, &reachable), missing) { + views.asking.insert( + member, + Asking { + since: now, + high, + sent: false, + }, + ); + high = false; + changed = true; + } + } + let mut to_ask: Vec<(String, bool)> = views + .asking + .iter() + .filter(|(_, asking)| !asking.sent) + .map(|(member, asking)| (member.clone(), asking.high)) + .collect(); + if views + .reaffirmed + .is_none_or(|at| now.duration_since(at) >= REAFFIRM_EVERY) + { + views.reaffirmed = Some(now); + to_ask.extend(views.active.iter().map(|member| (member.clone(), false))); + } + (changed, to_ask) + }; + if changed { + self.version.fetch_add(1, Ordering::Relaxed); + } + for member in to_leave { + Self::send( + state, + &member, + ParsedMessage::Disconnect(Self::answer(state, false)), + ) + .await; + } + for (member, high) in to_ask { + let ready = match state.get_peer_state(&member).await { + Some(peer_state) => peer_state.lock().await.hello_state == HelloState::Established, + None => false, + }; + if !ready { + continue; + } + // A neighbour asked again isn't asking: nothing to mark. + if let Some(asking) = self.views.lock_or_recover().asking.get_mut(&member) { + asking.sent = true; + } + let (epoch, epoch_seq) = state.epoch_stamp(); + Self::send( + state, + &member, + ParsedMessage::Neighbour(NeighbourRequest { + epoch, + epoch_seq, + high, + }), + ) + .await; + } + } + + async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) { + match message { + ParsedMessage::Neighbour(request) => { + let k = topology::k(state); + let (agreed, dropped) = { + let mut views = self.views.lock_or_recover(); + if views.active.contains(peer) { + (true, None) + } else if request.high || views.active.len() < k { + // Full: room made by dropping another, at random. + let dropped = if views.active.len() >= k { + let others: Vec = views.active.iter().cloned().collect(); + let dropped = some_of(others, 1).pop(); + if let Some(dropped) = &dropped { + views.active.remove(dropped); + views.refused.insert(dropped.clone(), Instant::now()); + } + dropped + } else { + None + }; + views.active.insert(peer.to_owned()); + views.asking.remove(peer); + views.refused.remove(peer); + // Asks not sent yet no longer needed: given up. + let room = k.saturating_sub(views.active.len()); + let unsent: Vec = views + .asking + .iter() + .filter(|(_, asking)| !asking.sent) + .map(|(member, _)| member.clone()) + .collect(); + let sent = views.asking.len() - unsent.len(); + for member in unsent.into_iter().skip(room.saturating_sub(sent)) { + views.asking.remove(&member); + } + (true, dropped) + } else { + (false, None) + } + }; + if let Some(dropped) = dropped { + Self::send( + state, + &dropped, + ParsedMessage::Disconnect(Self::answer(state, false)), + ) + .await; + } + Self::send( + state, + peer, + ParsedMessage::NeighbourReply(Self::answer(state, agreed)), + ) + .await; + if agreed { + self.changed(state); + } + } + ParsedMessage::NeighbourReply(answer) => { + { + let mut views = self.views.lock_or_recover(); + // Not asked: a neighbour asked again whether it still is + // (`REAFFIRM_EVERY`), dropped if it refused; or one we gave up on, + // and told so (`maintain`), which undoes its agreeing. + if views.asking.remove(peer).is_none() { + if answer.agreed || !views.active.remove(peer) { + return; + } + views.refused.insert(peer.to_owned(), Instant::now()); + drop(views); + self.changed(state); + return; + } + if answer.agreed { + views.active.insert(peer.to_owned()); + } else { + views.refused.insert(peer.to_owned(), Instant::now()); + } + } + self.changed(state); + } + ParsedMessage::Disconnect(_) => { + { + let mut views = self.views.lock_or_recover(); + views.active.remove(peer); + views.asking.remove(peer); + views.refused.insert(peer.to_owned(), Instant::now()); + } + self.changed(state); + } + _ => {} + } + } + + fn version(&self) -> u64 { + self.version.load(Ordering::Relaxed) + } + + /// Those we had as neighbours asked first: the handshakes are done, so NEIGHBOUR + /// goes with the next `maintain`. Up to `k` of them, at random, as `k` isn't known + /// here: the most `maintain` asks itself is what we lack. + fn seed(&self, neighbours: &BTreeSet) { + let now = Instant::now(); + let mut views = self.views.lock_or_recover(); + for member in neighbours { + views.asking.insert( + member.clone(), + Asking { + since: now, + high: false, + sent: false, + }, + ); + } + } + + #[cfg(test)] + fn active(&self) -> Option> { + Some(self.views.lock_or_recover().active.clone()) + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index e01c6f1..63ddcfa 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -8,6 +8,7 @@ pub(crate) mod changes; pub(crate) mod dissemination; pub(crate) mod document_file; pub(crate) mod hop; +pub(crate) mod hyparview; pub(crate) mod links; pub(crate) mod members; pub(crate) mod pair; diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index ddcc376..0c41466 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -256,16 +256,21 @@ pub(crate) async fn handle_message( } transport::ParsedMessage::Prune(link) => { state - .dissemination + .dissemination() .linked(state, &peer, link.writer, false) .await; } transport::ParsedMessage::Graft(link) => { state - .dissemination + .dissemination() .linked(state, &peer, link.writer, true) .await; } + message @ (transport::ParsedMessage::Neighbour(_) + | transport::ParsedMessage::NeighbourReply(_) + | transport::ParsedMessage::Disconnect(_)) => { + state.topology().received(state, &peer, &message).await; + } transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) | transport::ParsedMessage::Ping(_) diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index bee8687..5f7b8c0 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -158,7 +158,7 @@ pub(crate) async fn announce( let mut treeless = unauthored || writers.is_empty(); for (writer, changes) in writers { match state - .dissemination + .dissemination() .forward(state, *writer, from, changes) .await { @@ -338,7 +338,7 @@ pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &tran crate::document::changes::missing(&replica.doc, remote.heads.listed()) }; if !missing.is_empty() { - state.dissemination.announced(peer, &missing); + state.dissemination().announced(peer, &missing); let grace = state .lock_rtt() .effective(peer, Instant::now(), state.timing.rtt_fresh_for()) diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 45d3d75..2cbfa11 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -688,7 +688,7 @@ pub(crate) async fn handle_sync( crate::swarm::relay::pulled(state, peer); if state.initial_sync_done.load(Ordering::Acquire) && !just_caught_up { state - .dissemination + .dissemination() .note_arrival(state, peer, writer, carried, &authors) .await; } @@ -706,7 +706,10 @@ pub(crate) async fn handle_sync( // A writer's changes along its tree that we had already: they came another way // first, not saying whose (with an answer to one of our messages, say), so we // passed nothing on then. Our children in its tree may lack them still. - state.dissemination.forward(state, writer, peer, &[]).await; + state + .dissemination() + .forward(state, writer, peer, &[]) + .await; } state @@ -944,7 +947,7 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece let own = crate::swarm::topology::pair_key(state.local_id()); let ours = own_changes(&state, &mut seen_heads).await; let covered = match &ours { - Some(ours) => match state.dissemination.plan(&state, own).await { + Some(ours) => match state.dissemination().plan(&state, own).await { Some(plan) => { for neighbour in &plan.covered { if let Some(peer_state) = state.get_peer_state(neighbour).await { diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index d9f361a..14e9de6 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -33,6 +33,7 @@ use tracing::info; use crate::locks::RwLockExt; use crate::state::{AppState, PeerState}; use crate::swarm::members; +use crate::transport::ParsedMessage; mod graph; @@ -91,12 +92,149 @@ pub(crate) trait Topology: Send + Sync + 'static { let _ = (state, inputs); } + /// A message of the topology's own from `peer` (HyParView's NEIGHBOUR, + /// NEIGHBOUR_REPLY, DISCONNECT). By default ignored. + async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) { + let _ = (state, peer, message); + } + /// Changes whenever the topology's own state changed what `neighbours` gives: the /// neighbours are recomputed then, even though nothing they are chosen from did. By /// default always the same. fn version(&self) -> u64 { 0 } + + /// Taking over from another topology, whose neighbours were `neighbours` (the swarm + /// switched, `swarm_ways`): a topology with state of its own starts from them. By + /// default nothing. + fn seed(&self, neighbours: &BTreeSet) { + let _ = neighbours; + } + + /// The neighbours each agreed to, for a topology that keeps such (HyParView's active + /// view), for the tests and the benchmark to look at. By default none. + #[cfg(test)] + fn active(&self) -> Option> { + None + } +} + +/// The topology and the dissemination in use (`AppState::ways`), and which they are. +pub(crate) struct Ways { + pub(crate) kinds: (Kind, crate::swarm::dissemination::Kind), + pub(crate) topology: Arc, + pub(crate) dissemination: Arc, +} + +impl Ways { + pub(crate) fn new(topology: Kind, dissemination: crate::swarm::dissemination::Kind) -> Self { + Self { + kinds: (topology, dissemination), + topology: topology.make(), + dissemination: dissemination.make(), + } + } +} + +/// Where the swarm records its topology and its dissemination (`swarm_ways`); the +/// genesis node writes its `--topology` and `--dissemination` there +/// (`swarm::bootstrap`). +pub(crate) fn topology_path() -> Vec { + crate::document::meta::path(&[crate::document::meta::TOPOLOGY]) +} + +pub(crate) fn dissemination_path() -> Vec { + crate::document::meta::path(&[crate::document::meta::DISSEMINATION]) +} + +/// The topology and the dissemination the swarm records: its genesis node writes its +/// own (`topology_path`, `dissemination_path`), a protocol writer may change them (every +/// node then swaps to them, `refresh`). Where the document has none (a swarm from +/// before they were recorded), or one this node doesn't know, this node's own (`own`). +pub(crate) fn swarm_ways( + doc: &crate::document::Document, + own: (Kind, crate::swarm::dissemination::Kind), +) -> (Kind, crate::swarm::dissemination::Kind) { + fn recorded( + doc: &crate::document::Document, + path: &[String], + what: &str, + own: T, + ) -> T { + match crate::document::store::read_path(doc, path) { + Ok(Some(value)) => match serde_json::from_value(value.clone()) { + Ok(recorded) => recorded, + Err(_) => { + warn_once(&format!( + "The swarm's {what} ({value}) isn't one this node knows; using --{what}" + )); + own + } + }, + _ => own, + } + } + ( + recorded(doc, &topology_path(), "topology", own.0), + recorded(doc, &dissemination_path(), "dissemination", own.1), + ) +} + +/// Swaps to the topology and the dissemination `kinds` if they aren't the ones in use: +/// fresh, the new topology starting from our neighbours (`Topology::seed`). +fn switch_ways(state: &AppState, kinds: (Kind, crate::swarm::dissemination::Kind)) { + let neighbours: BTreeSet = our_neighbours(state).into_iter().collect(); + let mut ways = state.ways.write_or_recover(); + if ways.kinds == kinds { + return; + } + let before = (ways.topology.name(), ways.dissemination.name()); + if ways.kinds.0 != kinds.0 { + ways.topology = kinds.0.make(); + ways.topology.seed(&neighbours); + } + if ways.kinds.1 != kinds.1 { + ways.dissemination = kinds.1.make(); + } + ways.kinds = kinds; + info!( + from_topology = before.0, + from_dissemination = before.1, + topology = ways.topology.name(), + dissemination = ways.dissemination.name(), + "The swarm switched how neighbours are chosen, or how changes pass on" + ); + if ways + .topology + .graph(&[], 2, &|_| false, &|_| false) + .is_none() + && kinds.1 == crate::swarm::dissemination::Kind::WritersTrees + { + warn_once( + "Writers' trees need a graph every node shares, which this topology has none of: changes pass on by announcements alone (a protocol writer may record dissemination \"plumtree\")", + ); + } +} + +/// The topologies to choose from (`--topology`); one per swarm. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum Kind { + /// The member list's graph (`RingAndPairs`). + #[default] + RingAndPairs, + /// HyParView's active view, slimmed down (`swarm::hyparview`). + Hyparview, +} + +impl Kind { + pub(crate) fn make(self) -> Arc { + match self { + Kind::RingAndPairs => Arc::new(RingAndPairs), + Kind::Hyparview => Arc::new(crate::swarm::hyparview::HyParView::default()), + } + } } /// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by @@ -128,8 +266,8 @@ pub(crate) struct NeighbourView { /// they say they sync with us (`refresh`). outbound: BTreeSet, neighbours: BTreeSet, - /// The topology's version they were computed with (`Topology::version`). - version: u64, + /// The topology they were computed with, and its version (`Topology::version`). + version: (&'static str, u64), /// Every member not judged unreachable is a neighbour (the tests wait for it, /// `is_full_mesh`). Not the swarm being a full mesh: members on a transport we are /// not on, or that we never reached, don't count against it. @@ -173,7 +311,7 @@ pub(crate) fn member_list_graph( .map(|(member, _)| member.clone()) .collect(); let graph = state - .topology + .topology() .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?; for (member, neighbours) in graph { whole.entry(member).or_default().extend(neighbours); @@ -362,6 +500,7 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] pub(crate) fn intended_graph(state: &AppState) -> Option>> { member_list_graph(state)?; + let topology = state.topology(); let view = state.neighbours.read_or_recover(); let view = view.as_ref()?; let graph = view @@ -370,7 +509,7 @@ pub(crate) fn intended_graph(state: &AppState) -> Option( /// of notifications coalesces into one recomputation. pub(crate) async fn run_worker(state: AppState) { info!( - topology = state.topology.name(), - dissemination = state.dissemination.name(), + topology = state.topology().name(), + dissemination = state.dissemination().name(), "Choosing neighbours" ); loop { @@ -471,6 +610,8 @@ pub(crate) fn affects_neighbours(path: &[String]) -> bool { path.is_empty() || path.starts_with(&members::members_dir()) || path.starts_with(&neighbours_path()) + || path.starts_with(&topology_path()) + || path.starts_with(&dissemination_path()) } /// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`). @@ -543,6 +684,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { let replica = state.doc.read().await; members::apply(state, &replica.doc); let k = swarm_neighbours(&replica.doc, state.neighbours_k); + switch_ways(state, swarm_ways(&replica.doc, state.configured_ways)); let current = members::members(&replica.doc); // Those that said they are alive within the electorate window. let alive: BTreeSet = members::electorate( @@ -646,8 +788,9 @@ pub(crate) async fn refresh(state: &AppState) -> Option { unreachable: &unreachable_of, stable: &stable_of, }; - state.topology.maintain(state, &inputs).await; - let version = state.topology.version(); + let topology = state.topology(); + topology.maintain(state, &inputs).await; + let version = (topology.name(), topology.version()); let previous = { let view = state.neighbours.read_or_recover(); if view.as_ref().is_some_and(|view| { @@ -694,7 +837,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option { Some(previous) => previous, None => state.peer_states.read().await.keys().cloned().collect(), }; - let mut computed = state.topology.neighbours(state.local_id(), &inputs); + let mut computed = topology.neighbours(state.local_id(), &inputs); computed.extend(outbound.iter().cloned()); #[cfg(test)] let listed_neighbours = computed.len(); diff --git a/src/swarm/working_links.rs b/src/swarm/working_links.rs index d0ea41a..b384919 100644 --- a/src/swarm/working_links.rs +++ b/src/swarm/working_links.rs @@ -150,6 +150,12 @@ pub(crate) async fn check(state: &AppState) { } let me = state.local_id().to_owned(); let Some(graph) = topology::member_list_graph(state) else { + // No shared graph (`Topology::graph`): nothing to differ from; counted as a + // check all the same (`settled`). + #[cfg(test)] + { + state.working_links.lock_or_recover().checks += 1; + } return; }; let Some(ours) = graph.get(&me) else { -- cgit v1.3.1