aboutsummaryrefslogtreecommitdiffstats
path: root/src/simtest/hyparview.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/simtest/hyparview.rs')
-rw-r--r--src/simtest/hyparview.rs239
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;
+}