HyParView (slim)

4 files changed, 310 insertions, 9 deletions

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 \