diff options
Diffstat (limited to 'src/simtest')
| -rw-r--r-- | src/simtest/mod.rs | 294 | ||||
| -rw-r--r-- | src/simtest/soak.rs | 101 | ||||
| -rw-r--r-- | src/simtest/timing.rs | 285 |
3 files changed, 680 insertions, 0 deletions
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs new file mode 100644 index 0000000..e8e6e5b --- /dev/null +++ b/src/simtest/mod.rs @@ -0,0 +1,294 @@ +//! Tests of whole nodes on the simulated network (`transport::sim`): several nodes in +//! one process, built and started as `main` does (`node`), talking through the +//! Mycelium wire encoding. +//! +//! The clock is real (see `transport::sim` for why), so the nodes run with a short +//! keepalive and tests wait for conditions with a timeout instead of advancing time. +//! Timeouts are generous: a test fails only if something doesn't happen at all. +//! +//! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs +//! (all nodes interleaved; messages name peers by ID). + +mod soak; +mod timing; + +use std::collections::HashSet; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use automerge::{AutoCommit, ChangeHash}; +use serde_json::Value as JsonValue; +use tokio::time::Instant; + +use crate::document::replica::Epoch; +use crate::document::store; +use crate::files::Files; +use crate::node::{self, Node, NodeConfig}; +use crate::state::{AppState, PeerState, Timing}; +use crate::swarm::bootstrap::{self, Start}; +use crate::transport::sim::SimNet; + +/// Keepalive of simulated nodes: dormancy after 1 s, probes of dormant peers every +/// 2.5 s (`Timing::new`). +pub(crate) const KEEPALIVE: Duration = Duration::from_millis(250); +/// First wait for a sync answer before sending again (10 s in production, where large +/// messages take a while): a lost sync message costs a test 1 s, a heal 2 s. +pub(crate) const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(1); +/// How long the simulated net may take to deliver a message: well above its delays. +/// A catch-up source that stops sending sync messages is given up after this. +pub(crate) const DELIVERY_LIMIT: Duration = Duration::from_secs(3); + +/// A simulated swarm: the net and the nodes started on it. +pub(crate) struct Sim { + pub(crate) net: SimNet, + /// Every node started, for the description of a failed wait. + nodes: std::sync::Mutex<Vec<(String, AppState)>>, +} + +impl Sim { + pub(crate) fn new(seed: u64) -> Self { + init_logging(); + Self { + net: SimNet::new(seed), + nodes: Default::default(), + } + } + + /// Starts node `id` with the test defaults: builds it, starts its workers, and + /// bootstraps it in the background (through `seeds`, other nodes' IDs). Returns at + /// once; [`wait_ready`] waits for it. + pub(crate) fn start(&self, id: &str, seeds: &[&str], start: Start) -> AppState { + self.start_with(id, seeds, start, |_| {}) + } + + /// [`start`](Self::start), with the configuration adjusted first. + pub(crate) fn start_with( + &self, + id: &str, + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { + let mut config = NodeConfig { + transport: self.net.transport(id), + initial_epoch: Epoch::GENESIS, + initial_doc: AutoCommit::new(), + earlier_actors: Vec::new(), + timing: Timing { + delivery_limit: DELIVERY_LIMIT, + ..Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT) + }, + neighbours_k: 8, + alias: Some(id.to_owned()), + electorate_window: Duration::from_secs(30 * 24 * 3600), + blob_chunk_bytes: 64 * 1024, + swarm_key: None, + protocol_writers: HashSet::new(), + files: Arc::new(Files::default()), + state_file: None, + document_file: None, + config_report: JsonValue::Null, + clock: Default::default(), + }; + adjust(&mut config); + let state = Node::new(config).spawn_workers(); + self.nodes + .lock() + .unwrap() + .push((id.to_owned(), state.clone())); + + let seeds: Vec<String> = seeds.iter().map(|seed| (*seed).to_owned()).collect(); + let bootstrapping = state.clone(); + let id = id.to_owned(); + tokio::spawn(async move { + if let Err(error) = bootstrap::bootstrap(&bootstrapping, &seeds, start).await { + panic!("bootstrapping {id} failed: {error:#}"); + } + node::spawn_ready_workers(&bootstrapping); + }); + state + } +} + +/// Logs to the test output if `RUST_LOG` is set. +fn init_logging() { + if std::env::var_os("RUST_LOG").is_some() { + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .compact() + .try_init(); + } +} + +impl Sim { + /// Waits until `condition` holds, checking every 20 ms. After `timeout`, panics with + /// a description of every node (`describe`). + pub(crate) async fn wait_until<F, Fut>(&self, what: &str, timeout: Duration, mut condition: F) + where + F: FnMut() -> Fut, + Fut: Future<Output = bool>, + { + let deadline = Instant::now() + timeout; + loop { + if condition().await { + return; + } + if Instant::now() >= deadline { + panic!( + "timed out after {timeout:?} waiting until {what}\n{}", + self.describe().await + ); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + /// Waits until every node is ready. + pub(crate) async fn wait_ready(&self, nodes: &[&AppState], timeout: Duration) { + self.wait_until("every node is ready", timeout, move || async move { + for node in nodes { + if !node.is_ready().await { + return false; + } + } + true + }) + .await; + } + + /// Waits until every node has the same heads. + pub(crate) async fn wait_converged(&self, nodes: &[&AppState], timeout: Duration) { + self.wait_until( + "every node has the same heads", + timeout, + move || async move { + let first = heads(nodes[0]).await; + for node in &nodes[1..] { + if heads(node).await != first { + return false; + } + } + true + }, + ) + .await; + } + + /// Every node's state as far as sync and liveness go, and what the net counted. + pub(crate) async fn describe(&self) -> String { + use std::fmt::Write; + + let nodes = self.nodes.lock().unwrap().clone(); + let now = Instant::now(); + let ago = |at: Option<Instant>| at.map(|at| now.saturating_duration_since(at)); + let mut out = format!("net: {:?}\n", self.net.counts()); + for (id, node) in &nodes { + let heads: Vec<String> = heads(node) + .await + .iter() + .map(|head| head.to_string()[..8].to_owned()) + .collect(); + let _ = writeln!( + out, + "node {id}: ready={} epoch={} heads={heads:?} initial_sync_done={}", + node.is_ready().await, + node.local_epoch(), + node.initial_sync_done + .load(std::sync::atomic::Ordering::Acquire), + ); + let _ = writeln!( + out, + " view: {:?}", + *node + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + ); + let peers: Vec<_> = node + .peer_states + .read() + .await + .iter() + .map(|(peer, state)| (peer.clone(), Arc::clone(state))) + .collect(); + for (peer, peer_state) in peers { + let p = peer_state.lock().await; + let _ = writeln!( + out, + " peer {peer}: neighbour={} leads={} standing={:?} liveness={:?} hello={:?} \ + reports_ready={:?} epoch={:?} last_seen_ago={:?} last_sync_rx_ago={:?} \ + unreachable={} their_heads={:?} in_flight={} sending={} reply_due_in={:?} \ + retry_at={:?}", + crate::swarm::topology::is_neighbour(node, &peer), + p.leads, + p.standing, + p.liveness, + p.hello_state, + p.reports_ready, + p.epoch, + ago(p.last_seen), + ago(p.last_sync_rx), + p.is_unreachable(&node.timing, now), + p.sync.state.their_heads.as_ref().map(|heads| heads + .iter() + .map(|head| head.to_string()[..8].to_owned()) + .collect::<Vec<_>>()), + p.sync.state.in_flight, + p.sync.sending.is_some(), + p.sync + .reply + .as_ref() + .map(|wait| wait.deadline.saturating_duration_since(now)), + p.sync.retry_at.map(|at| at.saturating_duration_since(now)), + ); + } + } + out + } +} + +fn path(path: &str) -> Vec<String> { + path.split('/').map(str::to_owned).collect() +} + +/// Writes `value` at `path` (`a/b/c`) as one change, as the API does. +pub(crate) async fn write(node: &AppState, at: &str, value: JsonValue) { + let at = path(at); + node.mutate_doc(move |doc| store::write_path(doc, &at, &value)) + .await + .expect("write"); +} + +/// The value at `path` (`a/b/c`), if any. +pub(crate) async fn read(node: &AppState, at: &str) -> Option<JsonValue> { + store::read_path(&node.doc.read().await.doc, &path(at)).expect("read") +} + +pub(crate) async fn heads(node: &AppState) -> Vec<ChangeHash> { + let mut heads = node.doc.read().await.heads().to_vec(); + heads.sort_unstable(); + heads +} + +/// Whether `node` has `peer` among its neighbours as computed from the member list. +/// Before that list is known everyone counts as a neighbour, so this waits for it. +pub(crate) fn has_neighbour(node: &AppState, peer: &str) -> bool { + node.neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .is_some() + && crate::swarm::topology::is_neighbour(node, peer) +} + +/// Something about `node`'s state for `peer`; None if it doesn't know the peer. +pub(crate) async fn peer<T>( + node: &AppState, + peer: &str, + f: impl FnOnce(&PeerState) -> T, +) -> Option<T> { + let peer_state = node.get_peer_state(peer).await?; + let peer_state = peer_state.lock().await; + Some(f(&peer_state)) +} diff --git a/src/simtest/soak.rs b/src/simtest/soak.rs new file mode 100644 index 0000000..6906238 --- /dev/null +++ b/src/simtest/soak.rs @@ -0,0 +1,101 @@ +//! Many nodes under realistic conditions for a while. Ignored by default (a few +//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, write}; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; +use crate::swarm::members; +use crate::swarm::stats::Propagation; +use crate::transport::sim::LinkParams; + +const NODES: usize = 25; +/// One per second, round robin over the nodes. +const WRITES: usize = 60; +/// 99% of changes reach a node within this, from the writer's commit. Measured: p99 +/// within the 2.5 s bucket, mean about 0.7 s (October 2026); the bound is the next +/// bucket up. +const P99_BOUND_MS: u64 = 5_000; + +/// 25 nodes (a partial mesh: 8 neighbours each) over links with 2% loss, 1% +/// duplication and 50–500 ms delay; one write per second for a minute. All converge, +/// and 99% of changes reach each node within `P99_BOUND_MS`. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +#[ignore] +async fn soak_25_nodes() { + let sim = Sim::new(25); + sim.net.set_default_link(LinkParams { + loss: 0.02, + duplicate: 0.01, + delay: (Duration::from_millis(50), Duration::from_millis(500)), + ..LinkParams::default() + }); + let ids: Vec<String> = (0..NODES).map(|i| format!("n{i:02}")).collect(); + let mut nodes = vec![sim.start(&ids[0], &[], Start::Genesis)]; + for id in &ids[1..] { + nodes.push(sim.start(id, &[ids[0].as_str()], Start::Join)); + } + let all: Vec<&AppState> = nodes.iter().collect(); + let all = all.as_slice(); + sim.wait_ready(all, Duration::from_secs(120)).await; + sim.wait_until( + "every node knows every member", + Duration::from_secs(60), + move || async move { + for node in all { + if members::members(&node.doc.read().await.doc).len() < NODES { + return false; + } + } + true + }, + ) + .await; + + for i in 0..WRITES { + write(&nodes[i % NODES], &format!("soak/{i}"), json!(i)).await; + tokio::time::sleep(Duration::from_secs(1)).await; + } + sim.wait_converged(all, Duration::from_secs(60)).await; + + let mut total = Propagation::default(); + for node in all { + let propagation = node.stats.snapshot().propagation; + for (sum, count) in total.buckets.iter_mut().zip(propagation.buckets) { + *sum += count; + } + total.count += propagation.count; + total.sum_ms += propagation.sum_ms; + } + let p99 = percentile_bound(&total, 0.99); + println!( + "propagation: {} changes, mean {} ms, p99 within {}, buckets {:?}; net {:?}", + total.count, + total.sum_ms / total.count.max(1), + p99.map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")), + total.buckets().collect::<Vec<_>>(), + sim.net.counts(), + ); + assert!(total.count > 0); + assert!( + p99.is_some_and(|ms| ms <= P99_BOUND_MS), + "p99 propagation above {P99_BOUND_MS} ms" + ); +} + +/// The upper bound of the bucket in which the `quantile` of changes is reached; None if +/// only in the last (unbounded) one. +fn percentile_bound(propagation: &Propagation, quantile: f64) -> Option<u64> { + let wanted = (propagation.count as f64 * quantile).ceil() as u64; + let mut seen = 0; + for (bound, count) in propagation.buckets() { + seen += count; + if seen >= wanted { + return bound; + } + } + None +} diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs new file mode 100644 index 0000000..7810f28 --- /dev/null +++ b/src/simtest/timing.rs @@ -0,0 +1,285 @@ +//! The timing rules of `peers.md` on whole nodes: joining and converging, dormancy and +//! return, and late messages that must not undo what newer ones said. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, has_neighbour, heads, peer, read, write}; +use crate::document::replica::Epoch; +use crate::state::{AppState, Liveness}; +use crate::swarm::bootstrap::Start; +use crate::transport::sim::LinkParams; +use crate::transport::{HeadsSummary, HelloRequest, Keepalive, MessageKind, ParsedMessage, RunId}; + +/// Long enough for anything that should happen within a few keepalives. +const SOON: Duration = Duration::from_secs(10); +/// With lost messages: several reply timeouts (1 s, doubling) and heals in a row. +const LOSSY: Duration = Duration::from_secs(30); + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn three_nodes_join_and_converge() { + let sim = Sim::new(1); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + + let swarm = a.swarm_id().await; + assert!(swarm.is_some()); + assert_eq!(b.swarm_id().await, swarm); + assert_eq!(c.swarm_id().await, swarm); + + // A write anywhere reaches everyone. + write(b, "users/bob", json!({"name": "Bob"})).await; + sim.wait_until("c has b's write", SOON, move || async move { + read(c, "users/bob/name").await == Some(json!("Bob")) + }) + .await; + + // Concurrent writes merge to one document. + write(a, "users/alice", json!(1)).await; + write(c, "users/carol", json!(3)).await; + sim.wait_converged(&[a, b, c], SOON).await; + for node in [a, b, c] { + assert_eq!(read(node, "users/alice").await, Some(json!(1))); + assert_eq!(read(node, "users/carol").await, Some(json!(3))); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn converges_with_duplicated_and_reordered_messages() { + let sim = Sim::new(2); + sim.net.set_default_link(LinkParams { + duplicate: 0.3, + delay: (Duration::from_millis(1), Duration::from_millis(40)), + ..LinkParams::default() + }); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + + for round in 0..10 { + write(a, &format!("a/{round}"), json!(round)).await; + write(b, &format!("b/{round}"), json!(round)).await; + write(c, &format!("c/{round}"), json!(round)).await; + } + sim.wait_converged(&[a, b, c], SOON).await; + for node in [a, b, c] { + for prefix in ["a", "b", "c"] { + assert_eq!(read(node, &format!("{prefix}/9")).await, Some(json!(9))); + } + } + assert!(sim.net.counts().duplicated > 0); +} + +/// Lost messages are made up for: sync messages by the reply timeout, sessions out of +/// step by the keepalive heal. Joining under loss too: HELLOs are retried. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn converges_over_lossy_links() { + let sim = Sim::new(5); + sim.net.set_default_link(LinkParams { + loss: 0.15, + duplicate: 0.05, + delay: (Duration::from_millis(1), Duration::from_millis(30)), + ..LinkParams::default() + }); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], LOSSY).await; + + for round in 0..10 { + write(a, &format!("a/{round}"), json!(round)).await; + write(b, &format!("b/{round}"), json!(round)).await; + write(c, &format!("c/{round}"), json!(round)).await; + tokio::time::sleep(Duration::from_millis(50)).await; + } + sim.wait_converged(&[a, b, c], LOSSY).await; + for node in [a, b, c] { + for prefix in ["a", "b", "c"] { + assert_eq!(read(node, &format!("{prefix}/9")).await, Some(json!(9))); + } + } + assert!(sim.net.counts().lost > 0); +} + +/// R24: a catch-up source that is alive (it answers handshakes and keepalives) but whose +/// sync messages never arrive is given up after three unanswered sync messages and +/// `delivery_limit` without one, and another peer takes over. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stalled_catch_up_source_is_replaced() { + let sim = Sim::new(6); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + write(a, "data", json!("x")).await; + sim.wait_converged(&[a, b], SOON).await; + + // c joins through a, so a becomes its source (the first ready peer it hears + // from); b it learns of from a. + sim.net.set_link( + "a", + "c", + LinkParams { + drop: vec![MessageKind::Sync], + ..LinkParams::default() + }, + ); + let c = sim.start("c", &["a"], Start::Join); + let c = &c; + sim.wait_ready(&[c], LOSSY).await; + + assert_eq!(read(c, "data").await, Some(json!("x"))); + let sources = c.stats.snapshot().catch_up_sources; + assert!(sources >= 2, "{sources} catch-up sources"); +} + +/// A peer that hears nothing for `dormant_after` is dormant; once messages flow again +/// it is active, re-handshakes, and syncs what it missed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn peer_goes_dormant_and_returns() { + let sim = Sim::new(3); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + // Only neighbours are judged, and a node joins the member list (which neighbours + // are computed from) within 5 s of becoming ready: until then the other doesn't + // count it as a neighbour. + sim.wait_until( + "a and b are each other's neighbours", + SOON, + move || async move { has_neighbour(a, "b") && has_neighbour(b, "a") }, + ) + .await; + + sim.net.take_down("b"); + sim.wait_until("a finds b dormant", SOON, move || is_dormant(a, "b")) + .await; + sim.wait_until("b finds a dormant", SOON, move || is_dormant(b, "a")) + .await; + let dormancies = peer(a, "b", |p| p.incidents.dormancies).await.unwrap(); + assert!(dormancies >= 1); + + // Written while apart. + write(a, "while/apart", json!(true)).await; + assert_ne!(heads(a).await, heads(b).await); + + sim.net.bring_up("b"); + sim.wait_until("a finds b active", SOON, move || async move { + !is_dormant(a, "b").await + }) + .await; + sim.wait_until("b finds a active", SOON, move || async move { + !is_dormant(b, "a").await + }) + .await; + sim.wait_until("b has what a wrote meanwhile", SOON, move || async move { + read(b, "while/apart").await == Some(json!(true)) + }) + .await; + sim.wait_converged(&[a, b], SOON).await; +} + +async fn is_dormant(node: &AppState, other: &str) -> bool { + peer(node, other, |p| { + matches!(p.liveness, Liveness::Dormant { .. }) + }) + .await + == Some(true) +} + +/// R17: a message sent before the peer's last epoch change and delivered late doesn't +/// flip the epoch we recorded for it back; a new run of the peer starts its count over. +/// The peer is a "ghost" whose messages are injected, so their order is the test's. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn late_epoch_does_not_flip_a_peers_epoch() { + let sim = Sim::new(4); + let a = sim.start("a", &[], Start::Genesis); + sim.wait_ready(&[&a], SOON).await; + let swarm_id = a.swarm_id().await.unwrap(); + + let epoch = |generation: u64| Epoch { + generation, + id: [generation as u8; 32], + }; + let hello = |run: RunId, epoch: Epoch, epoch_seq: u32| { + ParsedMessage::Hello(HelloRequest { + swarm_id: Some(swarm_id.clone()), + epoch, + epoch_seq, + auth: Vec::new(), + ready: true, + run_id: run, + }) + }; + let ping = |epoch: Epoch, epoch_seq: u32| { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch, + epoch_seq, + ready: true, + syncs_with_you: false, + }) + }; + let a = &a; + let recorded = move || async move { + peer(a, "ghost", |p| (p.epoch, p.epoch_seq)) + .await + .expect("a knows the ghost") + }; + + let first_run: RunId = [1; 16]; + sim.net + .inject("ghost", "a", &hello(first_run, Epoch::GENESIS, 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(Epoch::GENESIS), Some(0))); + + // The ghost switched twice; we hear of the second switch first. + sim.net + .inject("ghost", "a", &ping(epoch(2), 2)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + + // The PING from between the switches arrives late: ignored. + sim.net + .inject("ghost", "a", &ping(epoch(1), 1)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + // So is one from before both. + sim.net + .inject("ghost", "a", &ping(Epoch::GENESIS, 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + + // The ghost restarted: its new run counts from 0 again. + let second_run: RunId = [2; 16]; + sim.net + .inject("ghost", "a", &hello(second_run, epoch(2), 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(0))); + assert_eq!( + peer(a, "ghost", |p| p.run_id).await.unwrap(), + Some(second_run) + ); + sim.net + .inject("ghost", "a", &ping(epoch(3), 1)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(3)), Some(1))); + + // One ahead peer isn't enough for a ready node to rejoin. + assert_eq!(a.local_epoch(), Epoch::GENESIS); + assert!(a.is_ready().await); +} |