1 files changed, 285 insertions, 0 deletions
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);
+}