From 260c100bd973fefa11478cb61b1d8b4e08308d76 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 21:44:04 +0100 Subject: add simulation test --- src/simtest/timing.rs | 219 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 219 insertions(+) create mode 100644 src/simtest/timing.rs (limited to 'src/simtest/timing.rs') diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs new file mode 100644 index 0000000..2633235 --- /dev/null +++ b/src/simtest/timing.rs @@ -0,0 +1,219 @@ +//! 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, ParsedMessage, RunId}; + +/// Long enough for anything that should happen within a few keepalives. +const SOON: Duration = Duration::from_secs(10); + +#[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); +} + +/// 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); +} -- cgit v1.3.1