1 files changed, 294 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))
+}