add simulation test

2 files changed, 503 insertions, 0 deletions

diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
new file mode 100644
index 0000000..2e33310
--- /dev/null
+++ b/src/simtest/mod.rs
@@ -0,0 +1,284 @@
+//! 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 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};
+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::from_keepalive`). Sync reply timeouts and heals don't scale with it
+/// (10 s and 20 s at least).
+pub(crate) const KEEPALIVE: Duration = Duration::from_millis(250);
+
+/// 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(),
+            keepalive: KEEPALIVE,
+            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,
+        };
+        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/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);
+}