HyParView (slim)

8 files changed, 641 insertions, 17 deletions

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<String>,
+    /// Asked, not answered yet.
+    asking: BTreeMap<String, Asking>,
+    /// Refused, unanswered or left lately: since when.
+    refused: BTreeMap<String, Instant>,
+    /// When we last asked while one neighbour short (`ONE_SHORT_EVERY`).
+    asked_one_short: Option<Instant>,
+    /// When the neighbours were last asked whether they still are (`REAFFIRM_EVERY`).
+    reaffirmed: Option<Instant>,
+}
+
+#[derive(Default)]
+pub(crate) struct HyParView {
+    views: Mutex<Views>,
+    /// 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<String> {
+        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<String>) -> Vec<String> {
+        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<String>, count: usize) -> Vec<String> {
+    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<BTreeMap<String, BTreeSet<String>>> {
+        let _ = (members, k, unreachable, stable);
+        None
+    }
+
+    /// The active view, and those being asked.
+    fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
+        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<String> = 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<String> = 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<String> = 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<String> = 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<String> = 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<String>) {
+        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<BTreeSet<String>> {
+        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<String>) {
+        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<BTreeSet<String>> {
+        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<dyn Topology>,
+    pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>,
+}
+
+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<String> {
+    crate::document::meta::path(&[crate::document::meta::TOPOLOGY])
+}
+
+pub(crate) fn dissemination_path() -> Vec<String> {
+    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<T: serde::de::DeserializeOwned>(
+        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<String> = 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<dyn Topology> {
+        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<String>,
     neighbours: BTreeSet<String>,
-    /// 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<BTreeMap<String, BTreeSet<String>>> {
     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<BTreeMap<String, BTreeS
         .map(|member| {
             (
                 member.clone(),
-                view.neighbours_of(&*state.topology, state.local_id(), member),
+                view.neighbours_of(&*topology, state.local_id(), member),
             )
         })
         .collect();
@@ -453,8 +592,8 @@ pub(crate) fn change_peer<T>(
 /// 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<NeighbourChange> {
         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<String> = members::electorate(
@@ -646,8 +788,9 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         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<NeighbourChange> {
         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 {