path: root/src/simtest/mod.rs

//! 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))
}