add simulation test

3 files changed, 467 insertions, 7 deletions

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());
+    }
+}