diff options
| -rw-r--r-- | src/main.rs | 2 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 284 | ||||
| -rw-r--r-- | src/simtest/timing.rs | 219 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 5 | ||||
| -rw-r--r-- | src/transport/mod.rs | 2 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 28 | ||||
| -rw-r--r-- | src/transport/sim.rs | 444 |
7 files changed, 976 insertions, 8 deletions
diff --git a/src/main.rs b/src/main.rs index 0a978fe..6d2329f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,6 +21,8 @@ mod force; mod logging; mod node; mod remake; +#[cfg(test)] +mod simtest; mod state; mod swarm; mod transport; 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); +} diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index bd15a34..5e9ac06 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -351,8 +351,11 @@ async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String peer_state.epoch, )); } + // Awaited outside the macro: its arguments aren't Send, and a spawned bootstrap + // (the simulator's) must be. + let local_swarm = state.swarm_id().await; warn!( - local_swarm = ?state.swarm_id().await, + local_swarm = ?local_swarm, local_epoch = %state.local_epoch(), initial_sync_done = state.initial_sync_done.load(Ordering::Acquire), unresolved_seeds = ?unresolved, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 564aada..c3db962 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -11,6 +11,8 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +#[cfg(test)] +pub(crate) mod sim; pub mod traffic; /// Random identifier of one run of a node (from start to exit). diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index f519928..efab475 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -561,13 +561,7 @@ impl Transport for MyceliumTransport { } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - let (outbound, payload_len) = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - kind: message.kind(), - payload_len, - }) + encode_message(peer, message) } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { @@ -712,6 +706,26 @@ impl Transport for MyceliumTransport { } } +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let (outbound, payload_len) = format_message(peer, message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + body: serde_json::to_vec(&outbound)?.into(), + kind: message.kind(), + payload_len, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + fn encode_topic(topic: &str) -> String { BASE64.encode(topic.as_bytes()) } diff --git a/src/transport/sim.rs b/src/transport/sim.rs new file mode 100644 index 0000000..3e0163f --- /dev/null +++ b/src/transport/sim.rs @@ -0,0 +1,444 @@ +//! A simulated network for tests: many nodes in one process, each with a +//! [`SimTransport`] on a shared [`SimNet`]. +//! +//! Messages go through the Mycelium wire encoding (`mycelium::encode_message`), so the +//! codec is exercised as in production. The net delays, loses and duplicates them per +//! link, from a seeded generator; nodes can be taken down (everything to and from them +//! is lost) and brought up again; and a test can inject a message as if a node had sent +//! it, e.g. one that arrives late. +//! +//! Time is tokio's real clock, not a paused one: a node's document work runs on the +//! blocking pool (`spawn_blocking`), which a paused clock doesn't wait for, so timers +//! would fire while the work is still running. Tests use a short keepalive instead. +//! Delays and losses come from the seed; when a message's turn comes still depends on +//! the scheduler, so a run is reproducible in its network decisions, not in its +//! interleavings. + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use async_trait::async_trait; +use bytes::Bytes; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, warn}; + +use super::mycelium::{decode_message, encode_message}; +use super::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; + +/// How one direction of a link behaves. +#[derive(Clone, Debug)] +pub(crate) struct LinkParams { + /// Probability that a message is lost. + pub(crate) loss: f64, + /// Probability that a message arrives twice (each copy with its own delay). + pub(crate) duplicate: f64, + /// Delay, uniform in this range. Overlapping delays reorder messages. + pub(crate) delay: (Duration, Duration), +} + +impl Default for LinkParams { + /// A clean, fast link: nothing lost, a few milliseconds. + fn default() -> Self { + Self { + loss: 0.0, + duplicate: 0.0, + delay: (Duration::from_millis(1), Duration::from_millis(5)), + } + } +} + +/// A message on its way into a node. +struct Delivery { + from: String, + body: Bytes, + payload_len: usize, + /// Told once the node has handled the message (injections wait for it). + handled: Option<oneshot::Sender<()>>, +} + +/// What the net counted, for assertions. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct NetCounts { + pub(crate) sent: u64, + pub(crate) delivered: u64, + pub(crate) lost: u64, + pub(crate) duplicated: u64, +} + +struct NetInner { + rng: SplitMix64, + default_link: LinkParams, + links: HashMap<(String, String), LinkParams>, + down: HashSet<String>, + inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>, + counts: NetCounts, +} + +/// The network all simulated nodes share. Cloning is cheap. +#[derive(Clone)] +pub(crate) struct SimNet { + inner: Arc<Mutex<NetInner>>, +} + +impl SimNet { + pub(crate) fn new(seed: u64) -> Self { + Self { + inner: Arc::new(Mutex::new(NetInner { + rng: SplitMix64(seed), + default_link: LinkParams::default(), + links: HashMap::new(), + down: HashSet::new(), + inboxes: HashMap::new(), + counts: NetCounts::default(), + })), + } + } + + fn lock(&self) -> MutexGuard<'_, NetInner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// A transport for a new node `id` on this net. + pub(crate) fn transport(&self, id: &str) -> Arc<SimTransport> { + let (tx, rx) = mpsc::unbounded_channel(); + let previous = self.lock().inboxes.insert(id.to_owned(), tx); + assert!(previous.is_none(), "node {id} is already on the net"); + Arc::new(SimTransport { + id: id.to_owned(), + net: self.clone(), + inbox: tokio::sync::Mutex::new(Some(rx)), + }) + } + + /// How links without their own parameters behave. + pub(crate) fn set_default_link(&self, params: LinkParams) { + self.lock().default_link = params; + } + + /// How messages from `from` to `to` are carried (one direction). + pub(crate) fn set_link(&self, from: &str, to: &str, params: LinkParams) { + self.lock() + .links + .insert((from.to_owned(), to.to_owned()), params); + } + + /// Everything to and from `id` is lost until [`bring_up`](Self::bring_up), including + /// messages already on their way. The node itself keeps running. + pub(crate) fn take_down(&self, id: &str) { + self.lock().down.insert(id.to_owned()); + } + + pub(crate) fn bring_up(&self, id: &str) { + self.lock().down.remove(id); + } + + pub(crate) fn counts(&self) -> NetCounts { + self.lock().counts + } + + /// Hands a message to the net: decides now whether and when it arrives. + fn send(&self, from: &str, message: &EncodedMessage) { + let delays = { + let mut guard = self.lock(); + let inner = &mut *guard; + inner.counts.sent += 1; + if inner.down.contains(from) || inner.down.contains(&message.peer) { + inner.counts.lost += 1; + return; + } + let params = inner + .links + .get(&(from.to_owned(), message.peer.clone())) + .unwrap_or(&inner.default_link) + .clone(); + if inner.rng.chance(params.loss) { + inner.counts.lost += 1; + return; + } + let copies = if inner.rng.chance(params.duplicate) { + inner.counts.duplicated += 1; + 2 + } else { + 1 + }; + (0..copies) + .map(|_| inner.rng.duration_in(params.delay)) + .collect::<Vec<_>>() + }; + for delay in delays { + let net = self.clone(); + let delivery = Delivery { + from: from.to_owned(), + body: message.body.clone(), + payload_len: message.payload_len, + handled: None, + }; + let to = message.peer.clone(); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + net.arrive(&to, delivery); + }); + } + } + + /// The message reaches `to`, unless either end went down meanwhile. + fn arrive(&self, to: &str, delivery: Delivery) { + let mut guard = self.lock(); + let inner = &mut *guard; + if inner.down.contains(to) || inner.down.contains(&delivery.from) { + inner.counts.lost += 1; + return; + } + // An unknown recipient (an address no node has) loses the message, as + // Mycelium would. + let Some(inbox) = inner.inboxes.get(to) else { + inner.counts.lost += 1; + return; + }; + if inbox.send(delivery).is_ok() { + inner.counts.delivered += 1; + } else { + inner.counts.lost += 1; + } + } + + /// Delivers `message` to `to` as if `from` had sent it, now, bypassing links and + /// `take_down`; returns once `to` has handled it. `from` needn't be a node on the + /// net. For messages a real node wouldn't send at that moment, e.g. one sent long + /// ago and delivered late. + pub(crate) async fn inject(&self, from: &str, to: &str, message: &ParsedMessage) -> Result<()> { + let encoded = encode_message(to, message)?; + let (handled_tx, handled_rx) = oneshot::channel(); + let delivery = Delivery { + from: from.to_owned(), + body: encoded.body, + payload_len: encoded.payload_len, + handled: Some(handled_tx), + }; + { + let inner = self.lock(); + let Some(inbox) = inner.inboxes.get(to) else { + bail!("no node {to} on the net"); + }; + if inbox.send(delivery).is_err() { + bail!("node {to} has no consumer running"); + } + } + handled_rx.await?; + Ok(()) + } + + fn is_known(&self, id: &str) -> bool { + self.lock().inboxes.contains_key(id) + } +} + +/// One node's end of the [`SimNet`]. Its identity is the ID it was created with; seeds +/// are node IDs. +pub(crate) struct SimTransport { + id: String, + net: SimNet, + inbox: tokio::sync::Mutex<Option<mpsc::UnboundedReceiver<Delivery>>>, +} + +#[async_trait] +impl Transport for SimTransport { + fn local_id(&self) -> &str { + &self.id + } + + /// A seed is a node ID; one no node has doesn't resolve (yet). + async fn resolve_seed(&self, seed: &str) -> Result<String> { + if self.net.is_known(seed) { + Ok(seed.to_owned()) + } else { + bail!("no node {seed} on the net") + } + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + encode_message(peer, message) + } + + /// Always accepted: the net loses messages silently, like Mycelium once it has + /// taken them. + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + self.net.send(&self.id, message); + Ok(()) + } + + /// Handles messages one at a time, in arrival order, as the Mycelium consumer does. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let Some(mut inbox) = self.inbox.lock().await.take() else { + bail!("consumer of {} already started", self.id); + }; + while let Some(delivery) = inbox.recv().await { + match decode_message(&delivery.body) { + Ok(message) => { + debug!(node = %self.id, from = %delivery.from, message = ?message, "Simulated delivery"); + if let Err(error) = handler(delivery.from, message, delivery.payload_len).await + { + warn!(node = %self.id, error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Err(error) => { + warn!(node = %self.id, error = %format!("{error:#}"), "Undecodable simulated message"); + } + } + if let Some(handled) = delivery.handled { + let _ = handled.send(()); + } + } + Ok(()) + } + + /// A simulated node has no daemon that could exit. + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + std::future::pending().await + } +} + +/// A small, seedable generator (SplitMix64): enough for network decisions, and no +/// dependency. +struct SplitMix64(u64); + +impl SplitMix64 { + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) + } + + /// Uniform in [0, 1). + fn next_f64(&mut self) -> f64 { + (self.next_u64() >> 11) as f64 / (1_u64 << 53) as f64 + } + + fn chance(&mut self, probability: f64) -> bool { + probability > 0.0 && self.next_f64() < probability + } + + fn duration_in(&mut self, (min, max): (Duration, Duration)) -> Duration { + if max <= min { + return min; + } + min + (max - min).mul_f64(self.next_f64()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::document::replica::Epoch; + use crate::transport::{HeadsSummary, Keepalive}; + + fn ping() -> ParsedMessage { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: true, + syncs_with_you: false, + }) + } + + /// Collects what a node receives. + fn record(transport: Arc<SimTransport>) -> Arc<Mutex<Vec<String>>> { + let received = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&received); + tokio::spawn(async move { + transport + .start_consumer(Box::new(move |from, _message, _len| { + sink.lock().unwrap().push(from); + Box::pin(async { Ok(()) }) + })) + .await + }); + received + } + + #[tokio::test] + async fn delivers_decodes_and_loses_when_down() { + let net = SimNet::new(1); + let a = net.transport("a"); + let received = record(net.transport("b")); + + let settle = || tokio::time::sleep(Duration::from_millis(30)); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.take_down("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.bring_up("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + + assert_eq!(*received.lock().unwrap(), vec!["a", "a"]); + let counts = net.counts(); + assert_eq!((counts.sent, counts.delivered, counts.lost), (3, 2, 1)); + assert_eq!(a.resolve_seed("b").await.unwrap(), "b"); + assert!(a.resolve_seed("c").await.is_err()); + } + + #[tokio::test] + async fn loss_and_duplication_follow_the_seed() { + let run = |seed| async move { + let net = SimNet::new(seed); + net.set_default_link(LinkParams { + loss: 0.3, + duplicate: 0.2, + ..LinkParams::default() + }); + let a = net.transport("a"); + let _b = net.transport("b"); + for _ in 0..200 { + a.send_message("b", &ping()).await.unwrap(); + } + tokio::time::sleep(Duration::from_millis(50)).await; + net.counts() + }; + let first = run(7).await; + assert_eq!(first, run(7).await); + assert!(first.lost > 30 && first.lost < 90, "{first:?}"); + assert!(first.duplicated > 10, "{first:?}"); + } + + #[tokio::test] + async fn a_link_overrides_the_default_in_one_direction() { + let net = SimNet::new(1); + net.set_link( + "a", + "b", + LinkParams { + loss: 1.0, + ..LinkParams::default() + }, + ); + let a = net.transport("a"); + let b = net.transport("b"); + let at_a = record(Arc::clone(&a)); + let at_b = record(Arc::clone(&b)); + + a.send_message("b", &ping()).await.unwrap(); + b.send_message("a", &ping()).await.unwrap(); + tokio::time::sleep(Duration::from_millis(30)).await; + + assert!(at_b.lock().unwrap().is_empty()); + assert_eq!(*at_a.lock().unwrap(), vec!["b"]); + } + + #[tokio::test] + async fn inject_returns_once_handled() { + let net = SimNet::new(1); + let received = record(net.transport("b")); + net.inject("ghost", "b", &ping()).await.unwrap(); + assert_eq!(*received.lock().unwrap(), vec!["ghost"]); + assert!(net.inject("ghost", "nobody", &ping()).await.is_err()); + } +} |