HyParView (slim)

25 files changed, 1161 insertions, 41 deletions

diff --git a/docs/development.md b/docs/development.md
index 0617562..9296e8b 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -32,7 +32,8 @@ summary of each workload's mean delay and messages per write, run by run (runs
 differ: one can't tell a small change from chance); with
 `SIM_DISSEMINATION=plumtree`, every simulated node passes changes on by PlumTree
 (`--dissemination`), the benchmark's views of the trees still being the writers'
-trees'; all at once with
+trees'; with `SIM_TOPOLOGY=hyparview`, every simulated node chooses its neighbours by
+HyParView (`--topology`; with PlumTree, as writers' trees need the shared graph); all at once with
 `BENCH_PARALLEL=1`, faster on many cores, though sharing them makes delays come out
 somewhat higher; and in `bench_bridge`, large values from a writer on a fast transport
 through a bridge to a node on a slow one: what that node gets as sync messages and as
diff --git a/docs/example.toml b/docs/example.toml
index 6f4ccbe..de32ada 100644
--- a/docs/example.toml
+++ b/docs/example.toml
@@ -190,10 +190,19 @@ htpasswd = "/etc/continuum/htpasswd"
 
 # How a change reaches every member: writers-trees (computed alike on every node from
 # the member list) or plumtree (eager links pruned where they deliver twice, grafted
-# where announcements had to bring a change). For experiments: every node of a swarm
-# must use the same one.
+# where announcements had to bring a change). A genesis node records its own in the
+# document; every node follows what the swarm records, which a protocol writer may
+# change (_continuum/dissemination), and uses its own only where the swarm records none
+# it knows.
 # dissemination = "writers-trees"
 
+# How a node's neighbours are chosen: ring-and-pairs (the graph every node computes
+# alike from the member list) or hyparview (an active view of neighbours each agreed
+# to, the others asked from the member list as neighbours fail). Recorded and followed
+# as dissemination is (_continuum/topology); with hyparview, what needs the shared
+# graph does nothing (writers' trees: use dissemination = "plumtree").
+# topology = "ring-and-pairs"
+
 # Warn in the log when this node's clock differs from most of its peers' by more than
 # this, for sure (beyond half the round trip it was measured over, from their PONGs); 0:
 # never.
diff --git a/docs/measurements.md b/docs/measurements.md
index e134ce2..a444862 100644
--- a/docs/measurements.md
+++ b/docs/measurements.md
@@ -180,7 +180,8 @@ clocks"); setting them is the operating system's job (NTP).
 Messages and payload bytes by direction (`sent`, `received`, `refused`) and kind
 (`hello`, `helloReply`, `sync`, `status`, `statusReply`, `blob`, `identity` for WHO and
 ME, `paths` for PING and PONG, `links` for LINKS and their replies, `probe` for PROBE
-and PROBE_REPLY, `plumtree` for PRUNE and GRAFT). The layers below the node count WHO and ME, and the
+and PROBE_REPLY, `plumtree` for PRUNE and GRAFT, `hyparview` for NEIGHBOUR,
+NEIGHBOUR_REPLY and DISCONNECT). The layers below the node count WHO and ME, and the
 paths layer's own PINGs and PONGs, under the address they went to or came from; a peer
 (`/peers`) has those of its addresses too, where it is known to be at them. Payload bytes are what a transport's message carries (the wire
 encoding): not the transport's framing (for Mycelium, its topic and the base64 of its
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 5ea3c19..0193495 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1707,7 +1707,7 @@ components:
 
     ByKind:
       type: object
-      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, links, probe, plumtree]
+      required: [hello, helloReply, sync, status, statusReply, blob, identity, paths, links, probe, plumtree, hyparview]
       properties:
         hello:
           $ref: '#/components/schemas/Counter'
@@ -1737,6 +1737,9 @@ components:
         plumtree:
           $ref: '#/components/schemas/Counter'
           description: PRUNE and GRAFT, with --dissemination plumtree.
+        hyparview:
+          $ref: '#/components/schemas/Counter'
+          description: NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT, with --topology hyparview.
 
     Traffic:
       type: object
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 11fde59..bcf4033 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -193,6 +193,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
             (direction, "links", by.links),
             (direction, "probe", by.probe),
             (direction, "plumtree", by.plumtree),
+            (direction, "hyparview", by.hyparview),
         ]
     };
     let mut series = Vec::new();
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 5f6634f..b7c7ff2 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 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.",
+        "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. With topology hyparview, the active view's size, 5 unless set (docs/topology.md).",
     ),
     option(
         "electorate-window",
@@ -285,11 +285,18 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). At least 7h. Members renew their stamps every seventh of the shortest window any of them publishes, hourly at most, so nodes may differ. Matters on the node that coordinates a remake.",
     ),
     option(
+        "topology",
+        Kind::Text,
+        "ring-and-pairs|hyparview",
+        Some("ring-and-pairs"),
+        "How a node's neighbours are chosen: by the graph every node computes alike from the member list, or by HyParView (an active view of neighbours each agreed to, the others asked from the member list as neighbours fail). A genesis node records its own in the document; every node follows what the swarm records, which a protocol writer may change (_continuum/topology), and uses its own only where the swarm records none it knows. With hyparview, what needs the shared graph does nothing (writers' trees: use plumtree).",
+    ),
+    option(
         "dissemination",
         Kind::Text,
         "writers-trees|plumtree",
         Some("writers-trees"),
-        "How a change reaches every member: along writers' trees computed alike on every node from the member list, or by PlumTree (eager links pruned where they deliver twice, grafted where announcements had to bring a change). For experiments: every node of a swarm must use the same one.",
+        "How a change reaches every member: along writers' trees computed alike on every node from the member list, or by PlumTree (eager links pruned where they deliver twice, grafted where announcements had to bring a change). A genesis node records its own in the document; every node follows what the swarm records, which a protocol writer may change (_continuum/dissemination), and uses its own only where the swarm records none it knows. Default: plumtree with topology hyparview, else writers-trees (docs/topology.md).",
     ),
     option(
         "superseded-writes",
@@ -583,6 +590,8 @@ pub(crate) struct Settings {
     )]
     pub(crate) electorate_window: Duration,
     #[serde(default)]
+    pub(crate) topology: crate::swarm::topology::Kind,
+    #[serde(default)]
     pub(crate) dissemination: crate::swarm::dissemination::Kind,
     #[serde(default)]
     pub(crate) superseded_writes: crate::remake::Superseded,
@@ -1538,6 +1547,67 @@ pub(crate) fn load() -> Result<Loaded> {
     Ok(loaded)
 }
 
+/// HyParView's active view, where `neighbours` isn't set: the paper's, for swarms of up
+/// to about 10 000 members (its fanout of 4, plus one).
+const HYPARVIEW_NEIGHBOURS: usize = 5;
+
+/// Defaults that depend on other options, for those not `set`: with `topology
+/// hyparview`, `dissemination plumtree` (writers' trees need the graph every node shares,
+/// which HyParView has none of) and `neighbours` `HYPARVIEW_NEIGHBOURS`.
+fn derive_defaults(settings: &mut Settings, set: &dyn Fn(&str) -> bool) {
+    use crate::swarm::dissemination::Kind as Dissemination;
+    use crate::swarm::topology::Kind as Topology;
+    if settings.topology == Topology::Hyparview {
+        if !set("dissemination") {
+            settings.dissemination = Dissemination::Plumtree;
+            info!("topology hyparview: dissemination plumtree, as none is set");
+        }
+        if !set("neighbours") {
+            settings.neighbours = HYPARVIEW_NEIGHBOURS;
+            info!(
+                neighbours = HYPARVIEW_NEIGHBOURS,
+                "topology hyparview: an active view of HyParView's size, as neighbours isn't set"
+            );
+        }
+    }
+}
+
+/// Combinations of options that work, but not as well as others (`docs/topology.md`):
+/// warned of at start.
+fn warn_combinations(settings: &Settings, set: &dyn Fn(&str) -> bool) {
+    use crate::swarm::dissemination::Kind as Dissemination;
+    use crate::swarm::topology::Kind as Topology;
+    match (settings.topology, settings.dissemination) {
+        (Topology::Hyparview, Dissemination::WritersTrees) => warn!(
+            "topology hyparview with dissemination writers-trees: writers' trees need the graph every node shares, which HyParView has none of, so changes pass on by announcements alone, a round and a pull per hop; dissemination plumtree is what goes with it"
+        ),
+        (Topology::RingAndPairs, Dissemination::Plumtree) => warn!(
+            "dissemination plumtree with topology ring-and-pairs: an experiment; writers' trees, on the graph every node shares, were faster in every scenario benchmarked (docs/topology.md)"
+        ),
+        _ => {}
+    }
+    if settings.topology == Topology::RingAndPairs && settings.neighbours == 2 {
+        warn!(
+            "neighbours 2: the member list's graph is its ring alone, and a change takes about half as many hops as there are members"
+        );
+    }
+    if settings.topology == Topology::Hyparview && settings.neighbours == 2 {
+        warn!(
+            "neighbours 2 with topology hyparview: a member whose two neighbours fail at once is cut off until it asks others"
+        );
+    }
+    if !settings.genesis {
+        for name in ["neighbours", "topology", "dissemination"] {
+            if set(name) {
+                info!(
+                    option = name,
+                    "Set on a node that isn't the genesis node: the swarm's own (recorded by its genesis node) counts, this only where it records none"
+                );
+            }
+        }
+    }
+}
+
 fn resolve(merged: Layer, files: Vec<PathBuf>) -> Result<Loaded> {
     // An override is for the start it was given for; in a file it would silently apply
     // to every start, e.g. replacing the document file with a backup each time.
@@ -1568,6 +1638,9 @@ fn resolve(merged: Layer, files: Vec<PathBuf>) -> Result<Loaded> {
     settings.alias = settings.alias.map(|alias| alias.trim().to_owned());
     settings.transports = settings.ordered_transports();
     settings.check().context("invalid configuration")?;
+    let set = |name: &str| merged.get(name).is_some_and(|entry| !entry.value.is_null());
+    derive_defaults(&mut settings, &set);
+    warn_combinations(&settings, &set);
     // No dashboard to be reached there: none is published (and one published before is
     // removed, `swarm::members`).
     if !cfg!(feature = "dashboard")
diff --git a/src/document/meta.rs b/src/document/meta.rs
index af1cd50..a60efc1 100644
--- a/src/document/meta.rs
+++ b/src/document/meta.rs
@@ -11,6 +11,9 @@
 //!   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`).
+//! * `topology`, `dissemination`: how members' neighbours are chosen, and how a change
+//!   reaches every member, written by the genesis node, and changed by a protocol
+//!   writer if need be (`swarm::topology::swarm_ways`).
 
 /// The key every protocol path starts with.
 pub(crate) const ROOT: &str = "_continuum";
@@ -22,6 +25,8 @@ pub(crate) const EPOCHS: &str = "epochs";
 pub(crate) const SWARM: &str = "swarm";
 pub(crate) const NEIGHBOURS: &str = "neighbours";
 pub(crate) const RELAYS: &str = "relays";
+pub(crate) const TOPOLOGY: &str = "topology";
+pub(crate) const DISSEMINATION: &str = "dissemination";
 
 /// The store path `_continuum/<parts…>`.
 pub(crate) fn path(parts: &[&str]) -> Vec<String> {
diff --git a/src/main.rs b/src/main.rs
index 002ce7f..0a626e2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -225,6 +225,7 @@ async fn main() -> Result<()> {
         clock_skew_warn: args.clock_skew_warn,
         stream_heartbeat: args.stream_heartbeat,
         superseded_writes: args.superseded_writes,
+        topology: args.topology,
         dissemination: args.dissemination,
         // At most 4MiB (`Settings::check`).
         blob_chunk_bytes: args.blob_chunk as u32,
diff --git a/src/node.rs b/src/node.rs
index a9d99b2..28ce095 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -53,6 +53,8 @@ pub(crate) struct NodeConfig {
     pub(crate) stream_heartbeat: Duration,
     /// `--superseded-writes`.
     pub(crate) superseded_writes: crate::remake::Superseded,
+    /// `--topology`.
+    pub(crate) topology: crate::swarm::topology::Kind,
     /// `--dissemination`.
     pub(crate) dissemination: crate::swarm::dissemination::Kind,
     pub(crate) blob_chunk_bytes: u32,
@@ -102,8 +104,11 @@ impl Node {
                 discovered_swarm_id: None,
             })),
             peer_states: Arc::new(RwLock::new(HashMap::new())),
-            topology: Arc::new(crate::swarm::topology::RingAndPairs),
-            dissemination: config.dissemination.make(),
+            ways: Arc::new(std::sync::RwLock::new(crate::swarm::topology::Ways::new(
+                config.topology,
+                config.dissemination,
+            ))),
+            configured_ways: (config.topology, config.dissemination),
             // Counted above the firewall (in the paths layer, under the identity layer):
             // what it drops outbound counts as sent (it plays a lossy network), what it
             // drops inbound never arrived. Without the identity layer (tests) there is
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 2530015..dac7a48 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -491,7 +491,7 @@ impl Totals {
 }
 
 /// The kinds, in the order of the report.
-const KINDS: [&str; 11] = [
+const KINDS: [&str; 12] = [
     "SYNC",
     "STATUS",
     "STATUS_REPLY",
@@ -503,9 +503,10 @@ const KINDS: [&str; 11] = [
     "LINKS",
     "PROBE",
     "PRUNE/GRAFT",
+    "NEIGHBOUR",
 ];
 
-fn counters(by: &ByKind) -> [Counter; 11] {
+fn counters(by: &ByKind) -> [Counter; 12] {
     [
         by.sync,
         by.status,
@@ -518,10 +519,11 @@ fn counters(by: &ByKind) -> [Counter; 11] {
         by.links,
         by.probe,
         by.plumtree,
+        by.hyparview,
     ]
 }
 
-fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
+fn counters_mut(by: &mut ByKind) -> [&mut Counter; 12] {
     [
         &mut by.sync,
         &mut by.status,
@@ -534,6 +536,7 @@ fn counters_mut(by: &mut ByKind) -> [&mut Counter; 11] {
         &mut by.links,
         &mut by.probe,
         &mut by.plumtree,
+        &mut by.hyparview,
     ]
 }
 
@@ -547,8 +550,11 @@ async fn run(seed: u64, scenario: &'static Scenario) -> (String, Vec<(&'static s
     let reply_timeout = scenario.reply_timeout.unwrap_or(SYNC_REPLY_TIMEOUT);
     let timing = move |config: &mut crate::node::NodeConfig| {
         config.timing = Timing::new(KEEPALIVE, reply_timeout, DELIVERY_LIMIT);
-        // As in production (`--neighbours`).
-        config.neighbours_k = 12;
+        // As in production (`--neighbours`): 12, or HyParView's 5 (`config::derive_defaults`).
+        config.neighbours_k = match config.topology {
+            crate::swarm::topology::Kind::Hyparview => 5,
+            crate::swarm::topology::Kind::RingAndPairs => 12,
+        };
     };
     let mut nodes = vec![sim.start_on_with(&ids[0], (scenario.on)(0), &[], Start::Genesis, timing)];
     // Those there from the start; the others join during the workloads.
@@ -620,7 +626,7 @@ async fn run(seed: u64, scenario: &'static Scenario) -> (String, Vec<(&'static s
         scenario.name,
         writes(),
         1000 / WRITE_EVERY.as_millis(),
-        all[0].dissemination.name()
+        all[0].dissemination().name()
     );
     let mut joined = initial;
     for (w, (workload, writer)) in [
@@ -685,6 +691,7 @@ async fn run(seed: u64, scenario: &'static Scenario) -> (String, Vec<(&'static s
         trees(&mut out, &nodes, &sim.net).await;
         sent_kept(&mut out, &nodes).await;
         eager_links(&mut out, &nodes, &sim.net);
+        active_views(&mut out, &nodes);
         if let Some(writer) = writer {
             slowest(&mut out, &nodes, &nodes[writer], &sim.net).await;
         }
@@ -801,6 +808,37 @@ pub(super) async fn sent_kept(out: &mut String, nodes: &[AppState]) {
     );
 }
 
+/// HyParView's active views after the workload (`swarm::hyparview`): how many neighbours
+/// each agreed to, the fewest and the most, and the links only one end has in its view
+/// (a DISCONNECT lost, or one end gave up on the other). Nothing for a topology that
+/// keeps no such views.
+pub(super) fn active_views(out: &mut String, nodes: &[AppState]) {
+    let mut views = HashMap::new();
+    for node in nodes {
+        let Some(active) = node.topology().active() else {
+            return;
+        };
+        views.insert(node.local_id().to_owned(), active);
+    }
+    let sizes: Vec<usize> = views.values().map(|active| active.len()).collect();
+    let one_ended = views
+        .iter()
+        .flat_map(|(node, active)| active.iter().map(move |other| (node, other)))
+        .filter(|(node, other)| {
+            !views
+                .get(*other)
+                .is_some_and(|theirs| theirs.contains(*node))
+        })
+        .count();
+    let _ = writeln!(
+        out,
+        "    HyParView's active views at the end: {:.1} neighbours on average, {} to {}; {one_ended} links in one end's view only",
+        sizes.iter().sum::<usize>() as f64 / sizes.len().max(1) as f64,
+        sizes.iter().min().copied().unwrap_or(0),
+        sizes.iter().max().copied().unwrap_or(0),
+    );
+}
+
 /// A link with a longer round trip is slow (`eager_links`): over Tor, a slow circuit
 /// (`tor_like`).
 const SLOW_LINK_MS: u64 = 1000;
@@ -822,12 +860,16 @@ fn mean(values: &[u64]) -> u64 {
 /// along eager links alone, by the links' latencies (the fastest such path, as the
 /// first copy of a push takes it), against writers' trees' (`trees`): how deep the
 /// tree is. Nothing for a way that keeps no such links.
-fn eager_links(out: &mut String, nodes: &[AppState], net: &crate::transport::sim::SimNet) {
+pub(super) fn eager_links(
+    out: &mut String,
+    nodes: &[AppState],
+    net: &crate::transport::sim::SimNet,
+) {
     let mut pushes: HashMap<String, Vec<String>> = HashMap::new();
     // The round trips of eager and of lazy links, as each node measured them.
     let (mut eager_rtts, mut lazy_rtts) = (Vec::new(), Vec::new());
     for node in nodes {
-        let Some(lazy) = node.dissemination.lazy() else {
+        let Some(lazy) = node.dissemination().lazy() else {
             return;
         };
         let (eager, others): (Vec<String>, Vec<String>) =
diff --git a/src/simtest/hyparview.rs b/src/simtest/hyparview.rs
new file mode 100644
index 0000000..2dc5cb7
--- /dev/null
+++ b/src/simtest/hyparview.rs
@@ -0,0 +1,239 @@
+//! HyParView's active views (`swarm::hyparview`, `--topology hyparview`): links both
+//! ends agreed to, kept as members fail.
+
+use std::collections::BTreeSet;
+use std::time::Duration;
+
+use serde_json::json;
+
+use super::{Sim, write};
+use crate::state::AppState;
+use crate::swarm::bootstrap::Start;
+
+const SOON: Duration = Duration::from_secs(60);
+const K: usize = 3;
+
+/// Every node by HyParView, changes passed on by PlumTree, each keeping `K` neighbours.
+fn hyparview(config: &mut crate::node::NodeConfig) {
+    config.topology = crate::swarm::topology::Kind::Hyparview;
+    config.dissemination = crate::swarm::dissemination::Kind::Plumtree;
+    config.neighbours_k = K;
+}
+
+/// Each node's active view, by node ID.
+fn views(nodes: &[AppState]) -> Vec<(String, BTreeSet<String>)> {
+    nodes
+        .iter()
+        .map(|node| {
+            (
+                node.local_id().to_owned(),
+                node.topology().active().expect("an active view"),
+            )
+        })
+        .collect()
+}
+
+/// Every node has a neighbour, none more than `K` and the ones that made room for a
+/// node that had none, and every link is in both ends' views.
+fn settled(nodes: &[AppState], up: &BTreeSet<String>) -> bool {
+    unsettled(nodes, up).is_empty()
+}
+
+/// What keeps the views from being `settled`, one line each: empty or too large views,
+/// links in one end's view only.
+fn unsettled(nodes: &[AppState], up: &BTreeSet<String>) -> Vec<String> {
+    let views = views(nodes);
+    let mut wrong = Vec::new();
+    for (node, active) in views.iter().filter(|(node, _)| up.contains(node)) {
+        if active.is_empty() || active.len() > K + 1 {
+            wrong.push(format!("{node}: {} neighbours", active.len()));
+        }
+        for other in active {
+            if !views
+                .iter()
+                .any(|(id, theirs)| id == other && theirs.contains(node))
+            {
+                wrong.push(format!("{node} has {other}, which hasn't it"));
+            }
+        }
+    }
+    wrong
+}
+
+/// Waits until the views are `settled` among `up`, and `also` holds; else fails with
+/// each node's active view (by its name in `names`, as `nodes` are in order) and what
+/// keeps them unsettled: `Sim::describe` shows the neighbours being asked as well.
+async fn wait_settled<F: Fn(&[AppState]) -> bool>(
+    nodes: &[AppState],
+    names: &[String],
+    up: &BTreeSet<String>,
+    what: &str,
+    timeout: Duration,
+    also: F,
+) {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        if settled(nodes, up) && also(nodes) {
+            return;
+        }
+        if tokio::time::Instant::now() >= deadline {
+            let name = |id: &str| {
+                nodes
+                    .iter()
+                    .position(|node| node.local_id() == id)
+                    .map_or(id.to_owned(), |at| names[at].clone())
+            };
+            let views = views(nodes)
+                .into_iter()
+                .map(|(node, active)| {
+                    format!(
+                        "  {}: {}",
+                        name(&node),
+                        active
+                            .iter()
+                            .map(|id| name(id))
+                            .collect::<Vec<_>>()
+                            .join(", ")
+                    )
+                })
+                .collect::<Vec<_>>()
+                .join("\n");
+            let wrong = unsettled(nodes, up)
+                .into_iter()
+                .map(|line| {
+                    let mut line = line;
+                    for node in nodes {
+                        line = line.replace(node.local_id(), &name(node.local_id()));
+                    }
+                    format!("  {line}")
+                })
+                .collect::<Vec<_>>()
+                .join("\n");
+            let ways = nodes
+                .iter()
+                .map(|node| {
+                    format!(
+                        "{} {}/{}",
+                        name(node.local_id()),
+                        node.topology().name(),
+                        node.dissemination().name()
+                    )
+                })
+                .collect::<Vec<_>>()
+                .join(", ");
+            panic!(
+                "timed out after {timeout:?} waiting until {what}\nactive views:\n{views}\nunsettled:\n{wrong}\nways: {ways}"
+            );
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}
+
+/// Twelve members, each keeping three neighbours: the active views settle, symmetric,
+/// and a write reaches every member. A member that goes away leaves the views of those
+/// that had it, which take others, and the next write reaches every member still up.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn active_views_settle_and_heal() {
+    let sim = Sim::with_identity(88);
+    let ids: Vec<String> = (0..12).map(|i| format!("h{i:02}")).collect();
+    let mut nodes = vec![sim.start_with(&ids[0], &[], Start::Genesis, hyparview)];
+    for id in &ids[1..] {
+        nodes.push(sim.start_with(id, &[ids[0].as_str()], Start::Join, hyparview));
+    }
+    let all: Vec<&AppState> = nodes.iter().collect();
+    sim.wait_ready(&all, SOON).await;
+    // By node ID, as the views hold them (`ids` are the names on the simulated net).
+    let up: BTreeSet<String> = nodes.iter().map(|n| n.local_id().to_owned()).collect();
+    wait_settled(&nodes, &ids, &up, "the active views settle", SOON, |_| true).await;
+    write(&nodes[0], "x", json!(1)).await;
+    sim.wait_converged(&all, SOON).await;
+
+    // The member most others have in their views goes away (by its index: its name on
+    // the net, its ID in the views).
+    let gone_at = {
+        let views = views(&nodes);
+        (0..nodes.len())
+            .max_by_key(|i| {
+                views
+                    .iter()
+                    .filter(|(_, active)| active.contains(nodes[*i].local_id()))
+                    .count()
+            })
+            .expect("a member")
+    };
+    sim.net.take_down(&ids[gone_at]);
+    let gone = nodes[gone_at].local_id().to_owned();
+    let up: BTreeSet<String> = nodes
+        .iter()
+        .map(|n| n.local_id().to_owned())
+        .filter(|id| *id != gone)
+        .collect();
+    wait_settled(
+        &nodes,
+        &ids,
+        &up,
+        "the views heal without the member gone",
+        SOON * 2,
+        |nodes| {
+            views(nodes)
+                .iter()
+                .all(|(node, active)| *node == gone || !active.contains(&gone))
+        },
+    )
+    .await;
+    let remaining: Vec<&AppState> = nodes.iter().filter(|n| *n.local_id() != gone).collect();
+    write(remaining[0], "y", json!(2)).await;
+    sim.wait_converged(&remaining, SOON).await;
+}
+
+/// A swarm on the member list's graph and writers' trees switches, as a protocol writer
+/// records HyParView and PlumTree: every node swaps, starting from the neighbours it had,
+/// the active views settle, and a write reaches every member.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn a_swarm_switches_to_hyparview_at_runtime() {
+    let sim = Sim::with_identity(89);
+    let three = |config: &mut crate::node::NodeConfig| config.neighbours_k = K;
+    let ids: Vec<String> = (0..8).map(|i| format!("s{i:02}")).collect();
+    let mut nodes = vec![sim.start_with(&ids[0], &[], Start::Genesis, three)];
+    for id in &ids[1..] {
+        nodes.push(sim.start_with(id, &[ids[0].as_str()], Start::Join, three));
+    }
+    let all: Vec<&AppState> = nodes.iter().collect();
+    sim.wait_ready(&all, SOON).await;
+    assert!(nodes[0].topology().active().is_none());
+
+    nodes[0]
+        .mutate_doc(|doc| {
+            crate::document::store::write_path(
+                doc,
+                &crate::swarm::topology::topology_path(),
+                &json!("hyparview"),
+            )?;
+            crate::document::store::write_path(
+                doc,
+                &crate::swarm::topology::dissemination_path(),
+                &json!("plumtree"),
+            )
+        })
+        .await
+        .expect("recorded");
+    let up: BTreeSet<String> = nodes.iter().map(|n| n.local_id().to_owned()).collect();
+    // Longer than a start: a node that switched first asks others still on the member
+    // list's graph, which ignore NEIGHBOUR; it gives up on them after 20 s, and asks
+    // them again only a minute later (or they ask it, once switched).
+    wait_settled(
+        &nodes,
+        &ids,
+        &up,
+        "every node switched, and the active views settle",
+        SOON * 2,
+        |nodes| {
+            nodes.iter().all(|node| {
+                node.topology().name() == "hyparview" && node.dissemination().name() == "plumtree"
+            })
+        },
+    )
+    .await;
+    write(&nodes[3], "z", json!(3)).await;
+    sim.wait_converged(&all, SOON).await;
+}
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index c3b2368..dadc3e8 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -11,6 +11,7 @@
 
 mod bench;
 mod deltas;
+mod hyparview;
 mod identity;
 mod links;
 mod reach;
@@ -193,9 +194,18 @@ impl Sim {
             clock_skew_warn: Duration::from_secs(2),
             stream_heartbeat: Duration::from_secs(30),
             superseded_writes: Default::default(),
+            // `SIM_TOPOLOGY=hyparview`: every simulated node chooses its neighbours so.
+            topology: match std::env::var("SIM_TOPOLOGY").as_deref() {
+                Ok("hyparview") => crate::swarm::topology::Kind::Hyparview,
+                _ => Default::default(),
+            },
             // `SIM_DISSEMINATION=plumtree`: every simulated node passes changes on so.
+            // Unset with `SIM_TOPOLOGY=hyparview`: PlumTree, as `--dissemination` is then.
             dissemination: match std::env::var("SIM_DISSEMINATION").as_deref() {
                 Ok("plumtree") => crate::swarm::dissemination::Kind::Plumtree,
+                Err(_) if std::env::var("SIM_TOPOLOGY").as_deref() == Ok("hyparview") => {
+                    crate::swarm::dissemination::Kind::Plumtree
+                }
                 _ => Default::default(),
             },
             blob_chunk_bytes: 64 * 1024,
diff --git a/src/simtest/soak.rs b/src/simtest/soak.rs
index c9c9c19..2a9272a 100644
--- a/src/simtest/soak.rs
+++ b/src/simtest/soak.rs
@@ -140,6 +140,11 @@ async fn soak_with_churn() {
             SYNC_REPLY_TIMEOUT,
             crate::transport::sim::DELIVERY_LIMIT,
         );
+        // With `SIM_TOPOLOGY=hyparview`, HyParView's 5, as in production
+        // (`config::derive_defaults`).
+        if config.topology == crate::swarm::topology::Kind::Hyparview {
+            config.neighbours_k = 5;
+        }
     };
     let genesis = ids[0].as_str();
     let mut nodes = vec![sim.start_with(genesis, &[], Start::Genesis, timing)];
@@ -289,14 +294,19 @@ async fn soak_with_churn() {
         .metrics()
         .num_alive_tasks();
     println!(
-        "soak with churn: {writes} writes, {away_turn} times away, {restart_turn} restarts; \
+        "soak with churn ({}, {}): {writes} writes, {away_turn} times away, {restart_turn} restarts; \
          propagation: {} changes, mean {} ms, p99 within {}",
+        nodes[0].topology().name(),
+        nodes[0].dissemination().name(),
         propagation.count,
         propagation.sum_ms / propagation.count.max(1),
         percentile_bound(&propagation, 0.99)
             .map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")),
     );
     super::bench::sent_kept(&mut out, &nodes).await;
+    // For PlumTree and HyParView: their links and views at the end (nothing otherwise).
+    super::bench::eager_links(&mut out, &nodes, &sim.net);
+    super::bench::active_views(&mut out, &nodes);
     print!("{out}");
     println!(
         "    per node at most: {peers} peers, round trips of {rtts}, throughput of \
diff --git a/src/state/mod.rs b/src/state/mod.rs
index d5c18b3..3dfc6b1 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -68,10 +68,17 @@ pub(crate) struct AppState {
 
     pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>,
     pub(crate) transport: Arc<dyn transport::Transport>,
-    /// How members' neighbours are chosen (`swarm::topology::Topology`).
-    pub(crate) topology: Arc<dyn crate::swarm::topology::Topology>,
-    /// How a change reaches every member (`swarm::dissemination::Dissemination`).
-    pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>,
+    /// How members' neighbours are chosen (`swarm::topology::Topology`), and how a change
+    /// reaches every member (`swarm::dissemination::Dissemination`): as the swarm
+    /// records them (`swarm::topology::swarm_ways`), swapped when that changes;
+    /// `--topology` and `--dissemination` until then (`configured_ways`).
+    pub(crate) ways: Arc<std::sync::RwLock<crate::swarm::topology::Ways>>,
+    /// `--topology` and `--dissemination`: what a genesis node records, and what counts
+    /// where the swarm records nothing this node knows.
+    pub(crate) configured_ways: (
+        crate::swarm::topology::Kind,
+        crate::swarm::dissemination::Kind,
+    ),
     pub(crate) firewall: Arc<Firewall>,
     /// Messages and bytes exchanged, per peer (`transport::traffic`).
     pub(crate) traffic: Arc<transport::traffic::Traffic>,
@@ -253,6 +260,16 @@ impl AppState {
             .collect()
     }
 
+    /// How members' neighbours are chosen now (`ways`).
+    pub(crate) fn topology(&self) -> Arc<dyn crate::swarm::topology::Topology> {
+        Arc::clone(&self.ways.read_or_recover().topology)
+    }
+
+    /// How a change reaches every member now (`ways`).
+    pub(crate) fn dissemination(&self) -> Arc<dyn crate::swarm::dissemination::Dissemination> {
+        Arc::clone(&self.ways.read_or_recover().dissemination)
+    }
+
     pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> {
         self.rtt.lock_or_recover()
     }
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 {
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 52b544a..8cbf412 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -251,6 +251,12 @@ pub(crate) enum ParsedMessage {
     Prune(TreeLink),
     /// Asks the recipient to send a writer's changes at once (`swarm::plumtree`).
     Graft(TreeLink),
+    /// Asks the recipient to be neighbours (`swarm::hyparview`).
+    Neighbour(NeighbourRequest),
+    /// Answers a NEIGHBOUR.
+    NeighbourReply(NeighbourAnswer),
+    /// Tells the recipient we are neighbours no more (`swarm::hyparview`).
+    Disconnect(NeighbourAnswer),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -397,6 +403,9 @@ pub(crate) enum MessageKind {
     ProbeReply,
     Prune,
     Graft,
+    Neighbour,
+    NeighbourReply,
+    Disconnect,
 }
 
 impl ParsedMessage {
@@ -418,6 +427,9 @@ impl ParsedMessage {
             ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply,
             ParsedMessage::Prune(_) => MessageKind::Prune,
             ParsedMessage::Graft(_) => MessageKind::Graft,
+            ParsedMessage::Neighbour(_) => MessageKind::Neighbour,
+            ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply,
+            ParsedMessage::Disconnect(_) => MessageKind::Disconnect,
         }
     }
 
@@ -441,6 +453,10 @@ impl ParsedMessage {
             ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
                 Some((link.epoch, link.epoch_seq))
             }
+            ParsedMessage::Neighbour(request) => Some((request.epoch, request.epoch_seq)),
+            ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
+                Some((answer.epoch, answer.epoch_seq))
+            }
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
@@ -461,6 +477,29 @@ pub(crate) struct TreeLink {
     pub(crate) writer: u64,
 }
 
+/// NEIGHBOUR: the sender asks the recipient to be neighbours (`swarm::hyparview`), as
+/// HyParView's NEIGHBOR; `high` when the sender has none (then the recipient makes room
+/// if it must).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct NeighbourRequest {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    pub(crate) high: bool,
+}
+
+/// NEIGHBOUR_REPLY (whether the recipient of a NEIGHBOUR agreed) and DISCONNECT (never
+/// agreed: neighbours no more) (`swarm::hyparview`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct NeighbourAnswer {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    pub(crate) agreed: bool,
+}
+
 /// PROBE: whether the recipient reaches `member` (`swarm::working_links`), as SWIM
 /// asks others before it suspects a member. Answered from what the recipient knows,
 /// without contacting the member.
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 501c77b..291a76b 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -57,6 +57,8 @@ pub(crate) struct ByKind {
     pub(crate) probe: Counter,
     /// PRUNE and GRAFT (`swarm::plumtree`).
     pub(crate) plumtree: Counter,
+    /// NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT (`swarm::hyparview`).
+    pub(crate) hyparview: Counter,
 }
 
 impl ByKind {
@@ -73,6 +75,7 @@ impl ByKind {
             MessageKind::Links,
             MessageKind::Probe,
             MessageKind::Prune,
+            MessageKind::Neighbour,
         ] {
             let theirs = *other.get(kind);
             self.get_mut(kind).absorb(theirs);
@@ -92,6 +95,9 @@ impl ByKind {
             MessageKind::Links | MessageKind::LinksReply => &self.links,
             MessageKind::Probe | MessageKind::ProbeReply => &self.probe,
             MessageKind::Prune | MessageKind::Graft => &self.plumtree,
+            MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
+                &self.hyparview
+            }
         }
     }
 
@@ -108,6 +114,9 @@ impl ByKind {
             MessageKind::Links | MessageKind::LinksReply => &mut self.links,
             MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe,
             MessageKind::Prune | MessageKind::Graft => &mut self.plumtree,
+            MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
+                &mut self.hyparview
+            }
         }
     }
 }
@@ -302,6 +311,9 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
         MessageKind::ProbeReply => ("PROBE_REPLY", "probe"),
         MessageKind::Prune => ("PRUNE", "plumtree"),
         MessageKind::Graft => ("GRAFT", "plumtree"),
+        MessageKind::Neighbour => ("NEIGHBOUR", "hyparview"),
+        MessageKind::NeighbourReply => ("NEIGHBOUR_REPLY", "hyparview"),
+        MessageKind::Disconnect => ("DISCONNECT", "hyparview"),
     }
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 729acd3..0733d9c 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 37;
+pub(crate) const VERSION: u8 = 38;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -35,6 +35,9 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::ProbeReply => 13,
         MessageKind::Prune => 14,
         MessageKind::Graft => 15,
+        MessageKind::Neighbour => 16,
+        MessageKind::NeighbourReply => 17,
+        MessageKind::Disconnect => 18,
     }
 }
 
@@ -94,6 +97,10 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
         ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
             rmp_serde::encode::write(&mut payload, link)?
         }
+        ParsedMessage::Neighbour(request) => rmp_serde::encode::write(&mut payload, request)?,
+        ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
+            rmp_serde::encode::write(&mut payload, answer)?
+        }
     }
     Ok(payload)
 }
@@ -171,6 +178,12 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Prune(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Graft) {
         ParsedMessage::Graft(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Neighbour) {
+        ParsedMessage::Neighbour(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::NeighbourReply) {
+        ParsedMessage::NeighbourReply(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Disconnect) {
+        ParsedMessage::Disconnect(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -551,6 +564,21 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             writer: 5,
         }),
+        ParsedMessage::Neighbour(crate::transport::NeighbourRequest {
+            epoch,
+            epoch_seq: 1,
+            high: true,
+        }),
+        ParsedMessage::NeighbourReply(crate::transport::NeighbourAnswer {
+            epoch,
+            epoch_seq: 1,
+            agreed: true,
+        }),
+        ParsedMessage::Disconnect(crate::transport::NeighbourAnswer {
+            epoch,
+            epoch_seq: 1,
+            agreed: false,
+        }),
     ]
 }
 
@@ -582,7 +610,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 16, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 19, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();