HyParView (slim)

1 files changed, 239 insertions, 0 deletions

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;
+}