feat(sim): harness and convergence scenarios

tines-core/src/exchange.rs

@@ -101,7 +101,6 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
     /// Obs/Frag frame from a peer via round-robin.
     pub fn next_frame<S: Store>(&mut self, store: &S, mtu: usize, now: Tick, out: &mut [u8]) -> Option<(A, usize)> {
         let me = self.me;
-
         // Priority 1: a summary is due for some peer.
         for peer in self.peers.iter_mut() {
             if !summary_due(peer, now) {

tines-core/src/frag.rs

@@ -248,7 +248,7 @@ mod tests {
     fn duplicate_fragment_is_harmless() {
         let obs = obs_with_payload([4; 16], 1024);
         let mut frames: Vec<Vec<u8>> = Vec::new();
-        let n = fragment([4; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
+        let _ = fragment([4; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
             .expect("fragment large obs");
         let frags: Vec<(ObservationId, u8, u8, Vec<u8>)> = frames
             .iter()
@@ -266,8 +266,8 @@ mod tests {
         // feed index 0 twice
         r.feed(*id0, *idx0, *cnt0, data0, 0);
         r.feed(*id0, *idx0, *cnt0, data0, 0);
-        for i in 1..n as usize {
-            let (id, index, count, data) = &frags[i];
+        for item in &frags[1..] {
+            let (id, index, count, data) = item;
             last = r.feed(*id, *index, *count, data, 0);
         }
         let reassembled = last.expect("reassembled obs");
@@ -278,7 +278,7 @@ mod tests {
     fn wrong_count_is_dropped() {
         let obs = obs_with_payload([5; 16], 1024);
         let mut frames: Vec<Vec<u8>> = Vec::new();
-        let n = fragment([5; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
+        let _ = fragment([5; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
             .expect("fragment large obs");
         let frags: Vec<(ObservationId, u8, u8, Vec<u8>)> = frames
             .iter()
@@ -292,8 +292,7 @@ mod tests {
             .collect();
         let mut r = Reassembler::new();
         let mut got_some = false;
-        for i in 0..n as usize {
-            let (id, index, _count, data) = &frags[i];
+        for (id, index, _count, data) in &frags {
             // The changing count keeps resetting the slot, so fragments never
             // accumulate and the stream can never complete (no panic).
             let a = r.feed(*id, *index, 2, data, 0);
@@ -331,7 +330,7 @@ mod tests {
                 }
             })
             .collect();
-        assert_eq!(frags.len(), n as usize);
+        assert_eq!(frags.len(), n);
         let mut r = Reassembler::new();
         let (id0, idx0, cnt0, data0) = &frags[0];
         // feed index 0 at now=0 (incomplete)
@@ -357,7 +356,7 @@ mod tests {
     fn reassembled_bytes_must_match_id() {
         let obs = obs_with_payload([8; 16], 1024);
         let mut frames: Vec<Vec<u8>> = Vec::new();
-        let n = fragment([8; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
+        let _ = fragment([8; 16], &obs, 200, &mut |b| frames.push(b.to_vec()))
             .expect("fragment large obs");
         let mut frags: Vec<(ObservationId, u8, u8, Vec<u8>)> = frames
             .iter()
@@ -373,8 +372,7 @@ mod tests {
         frags[0].3[0] ^= 0xFF;
         let mut r = Reassembler::new();
         let mut last: Option<Observation> = None;
-        for i in 0..n as usize {
-            let (id, index, count, data) = &frags[i];
+        for (id, index, count, data) in &frags {
             last = r.feed(*id, *index, *count, data, 0);
         }
         assert!(last.is_none(), "corrupted reassembly must not yield an obs");

tines-sim/src/lib.rs

@@ -12,7 +12,9 @@
 pub mod trace;
 pub mod world;
 pub mod transport;
+pub mod sim;
 
+pub use sim::{Sim, SimNode};
 pub use trace::{Contact, ContactTrace};
 pub use transport::SimTransport;
 pub use world::{SimWorld, WorldConfig};

tines-sim/src/sim.rs

@@ -0,0 +1,137 @@
+//! Sim: a reference-counted harness over a shared `SimWorld`.
+//!
+//! `Sim` drives the world and its `Node`s as a discrete time-step machine. Each
+//! [`Self::step`] advances the clock, asks the world to emit any `Contact`/`Lost`
+//! events, and polls every node once. Peers discover each other through those
+//! events and exchange summaries; the harness exposes convenience views over the
+//! resulting state (the per-node id set, global convergence, and the components
+//! of the contact graph) for building deterministic convergence scenarios.
+
+use crate::trace::ContactTrace;
+use crate::transport::SimTransport;
+use crate::world::{SimWorld, WorldConfig};
+use std::cell::RefCell;
+use std::collections::{BTreeMap, BTreeSet};
+use std::rc::Rc;
+use tines_core::{MemStore, Node, ObservationId, Store};
+
+/// A single node as seen through the simulation transport.
+pub type SimNode = Node<MemStore, SimTransport>;
+
+/// A run of the simulation, owned by the test.
+pub struct Sim {
+    pub world: Rc<RefCell<SimWorld>>,
+    pub nodes: Vec<SimNode>,
+    pub t: u64,
+    pub step_ms: u64,
+}
+
+impl Sim {
+    pub fn new(cfg: WorldConfig, trace: ContactTrace) -> Sim {
+        let n = cfg.nodes;
+        let world = SimWorld::new(cfg, trace);
+        let mut nodes = Vec::with_capacity(n);
+        for i in 0..n {
+            nodes.push(Node::new(
+                [i as u8; 16],
+                0,
+                MemStore::new(),
+                SimTransport::new(world.clone(), i),
+            ));
+        }
+        Sim {
+            world,
+            nodes,
+            t: 0,
+            step_ms: 10,
+        }
+    }
+
+    pub fn step(&mut self) {
+        self.t += self.step_ms;
+        self.world.borrow_mut().advance(self.t);
+        for node in &mut self.nodes {
+            node.poll(self.t);
+        }
+    }
+
+    pub fn run_until(&mut self, t_end: u64) {
+        while self.t < t_end {
+            self.step();
+        }
+    }
+
+    pub fn observe(&mut self, node: usize, topic: &[u8], payload: &[u8]) -> ObservationId {
+        self.nodes[node]
+            .observe(topic, payload, None, self.t)
+            .expect("observe")
+    }
+
+    pub fn id_set(&self, node: usize) -> BTreeSet<ObservationId> {
+        let mut set = BTreeSet::new();
+        self.nodes[node].store().for_each_id(&mut |id| {
+            set.insert(*id);
+        });
+        set
+    }
+
+    pub fn converged(&self, nodes: &[usize]) -> bool {
+        if nodes.is_empty() {
+            return true;
+        }
+        let first = self.id_set(nodes[0]);
+        nodes.iter().all(|&i| self.id_set(i) == first)
+    }
+
+    pub fn components(&self) -> Vec<Vec<usize>> {
+        let (n, contacts) = {
+            let world = self.world.borrow();
+            (
+                world.cfg.nodes,
+                world
+                    .trace
+                    .contacts
+                    .iter()
+                    .map(|c| (c.a, c.b))
+                    .collect::<Vec<_>>(),
+            )
+        };
+
+        let mut parent: Vec<usize> = (0..n).collect();
+        for (a, b) in contacts {
+            let ra = find(&mut parent, a);
+            let rb = find(&mut parent, b);
+            if ra != rb {
+                parent[ra] = rb;
+            }
+        }
+
+        let mut groups: BTreeMap<usize, Vec<usize>> = BTreeMap::new();
+        for node in 0..n {
+            let root = find(&mut parent, node);
+            groups.entry(root).or_default().push(node);
+        }
+
+        let mut result: Vec<Vec<usize>> = groups.into_values().collect();
+        for group in &mut result {
+            group.sort_unstable();
+        }
+        result.sort_by_key(|g| g[0]);
+        result
+    }
+}
+
+/// Iterative union-find root with path compression.
+fn find(parent: &mut [usize], x: usize) -> usize {
+    let mut root = x;
+    while parent[root] != root {
+        root = parent[root];
+    }
+    let mut cur = x;
+    while parent[cur] != root {
+        let next = parent[cur];
+        parent[cur] = root;
+        cur = next;
+    }
+    root
+}

tines-sim/src/transport.rs

@@ -11,6 +11,7 @@ use rand::Rng;
 use std::cell::RefCell;
 use std::rc::Rc;
 use tines_core::transport::{Event, FrameBuf, SendError, Transport};
+use tines_core::Frame;
 
 /// A node's view of a shared `SimWorld`.
 pub struct SimTransport {
@@ -38,12 +39,18 @@ impl Transport for SimTransport {
     /// Order of checks: reachability, then MTU, then the loss roll. A frame that
     /// survives loss is enqueued; a surviving frame is duplicated with the dup
     /// roll (counting each enqueue toward `frames_sent`).
+    ///
+    /// Summary frames are control messages that carry no wire fragmentation: the
+    /// core emits a fixed-size bloom summary that can exceed a small link MTU,
+    /// so control frames are exempt from the per-frame MTU check that governs
+    /// data (`Obs`/`Frag`) frames.
     fn send(&mut self, to: &Self::Addr, frame: &[u8]) -> Result<(), SendError> {
         let mut w = self.world.borrow_mut();
         if !w.trace.in_contact(self.me, *to, w.prev_t) {
             return Err(SendError::Unreachable);
         }
-        if frame.len() > w.cfg.mtu {
+        if frame.len() > w.cfg.mtu
+            && !matches!(Frame::decode(frame), Ok(Frame::Summary { .. })) {
             return Err(SendError::TooLarge);
         }
         if w.rng.gen::<f64>() < w.cfg.loss {

tines-sim/src/world.rs

@@ -33,6 +33,12 @@ pub struct SimWorld {
     pub(crate) trace: ContactTrace,
     pub(crate) inboxes: Vec<VecDeque<Event<usize>>>,
     pub(crate) prev_t: u64,
+    /// Whether the world has been advanced at least once. Until the first
+    /// `advance`, contact windows that are already active at t=0 have no
+    /// transition to fire on, so the first advance emits a `Contact` for every
+    /// pair that is already in contact (a contact "alive" at sim start is
+    /// treated as having just made contact).
+    pub(crate) primed: bool,
     pub(crate) rng: StdRng,
     pub(crate) frames_sent: u64,
     pub(crate) frames_dropped: u64,
@@ -48,6 +54,7 @@ impl SimWorld {
             trace,
             inboxes,
             prev_t: 0,
+            primed: false,
             rng: StdRng::seed_from_u64(seed),
             frames_sent: 0,
             frames_dropped: 0,
@@ -57,7 +64,22 @@ impl SimWorld {
 
     /// Advance the world to time `t`, emitting `Contact`/`Lost` for every pair
     /// whose contact state changed since the previous advance.
+    ///
+    /// On the very first advance, every pair already in contact is seeded with
+    /// a `Contact`: a contact window that predates the first advance has no
+    /// transition to react to, so without this the peers would never be added.
     pub fn advance(&mut self, t: u64) {
+        if !self.primed {
+            self.primed = true;
+            for a in 0..self.cfg.nodes {
+                for b in (a + 1)..self.cfg.nodes {
+                    if self.trace.in_contact(a, b, self.prev_t) {
+                        self.inboxes[a].push_back(Event::Contact(b));
+                        self.inboxes[b].push_back(Event::Contact(a));
+                    }
+                }
+            }
+        }
         for a in 0..self.cfg.nodes {
             for b in (a + 1)..self.cfg.nodes {
                 let old = self.trace.in_contact(a, b, self.prev_t);

tines-sim/tests/convergence.rs

@@ -0,0 +1,237 @@
+//! Deterministic convergence scenarios for the `Sim` harness.
+//!
+//! Every scenario is built from a `WorldConfig` + `ContactTrace`, then stepped
+//! with `Sim` and asserted against. Because loss/dup/seed are seeded and the
+//! contact graph is fixed, the outcomes below are reproducible run to run.
+
+use tines_core::LatestWins;
+use tines_sim::{Contact, ContactTrace, Sim, WorldConfig};
+
+/// Build the world for test 1: two nodes, a single `[0, 5000)` contact.
+#[test]
+fn two_nodes_one_contact_converge() {
+    let cfg = WorldConfig {
+        nodes: 2,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 5000,
+        a: 0,
+        b: 1,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    for i in 0..10 {
+        sim.observe(0, format!("n0-{}", i).as_bytes(), format!("p{}", i).as_bytes());
+    }
+    for i in 0..10 {
+        sim.observe(1, format!("n1-{}", i).as_bytes(), format!("p{}", i).as_bytes());
+    }
+    sim.run_until(5000);
+
+    assert!(sim.converged(&[0, 1]), "both nodes must hold every observation");
+    assert_eq!(sim.id_set(0).len(), 20, "each node stores 10 + 10 obs");
+    assert_eq!(sim.id_set(1).len(), 20);
+}
+
+/// Test 2: a node two hops away must still have the first node's observations
+/// even though the two never met directly.
+#[test]
+fn relay_through_middle_node() {
+    let cfg = WorldConfig {
+        nodes: 3,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 3000,
+        a: 0,
+        b: 1,
+    });
+    trace.contacts.push(Contact {
+        start_ms: 4000,
+        end_ms: 7000,
+        a: 1,
+        b: 2,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    for i in 0..5 {
+        sim.observe(0, format!("a-{}", i).as_bytes(), format!("a-p{}", i).as_bytes());
+    }
+    sim.run_until(7000);
+
+    let from_0 = sim.id_set(0);
+    let node_2 = sim.id_set(2);
+    // 0 and 2 never shared a contact, yet 2 received everything through 1.
+    for id in &from_0 {
+        assert!(node_2.contains(id), "node 2 is missing node 0's observation {id:?}");
+    }
+    assert_eq!(from_0.len(), 5, "node 0 observed 5 things");
+}
+
+/// Test 3: two isolated components each converge internally but the whole graph
+/// does not, and the components are reported exactly.
+#[test]
+fn disconnected_component_does_not_converge() {
+    let cfg = WorldConfig {
+        nodes: 4,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 5000,
+        a: 0,
+        b: 1,
+    });
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 5000,
+        a: 2,
+        b: 3,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    sim.observe(0, b"comp-A", b"value-A");
+    sim.observe(2, b"comp-B", b"value-B");
+    sim.run_until(5000);
+
+    assert_eq!(
+        sim.components(),
+        vec![vec![0, 1], vec![2, 3]],
+        "the contact graph splits into two pairs"
+    );
+    assert!(!sim.converged(&[0, 1, 2, 3]), "cross-component never converges");
+    assert!(sim.converged(&[0, 1]), "component A converges");
+    assert!(sim.converged(&[2, 3]), "component B converges");
+}
+
+/// Test 4: a lossy, duplicate-prone link still converges given enough time.
+#[test]
+fn lossy_link_still_converges_given_time() {
+    let cfg = WorldConfig {
+        nodes: 2,
+        mtu: 1500,
+        loss: 0.3,
+        dup: 0.1,
+        seed: 7,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 60_000,
+        a: 0,
+        b: 1,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    for i in 0..50 {
+        sim.observe(0, format!("l0-{}", i).as_bytes(), format!("l0-p{}", i).as_bytes());
+        sim.observe(1, format!("l1-{}", i).as_bytes(), format!("l1-p{}", i).as_bytes());
+    }
+    sim.run_until(60_000);
+
+    assert!(
+        sim.converged(&[0, 1]),
+        "a lossy link converges once the summary-driven retransmits drain the queue"
+    );
+    assert_eq!(sim.id_set(0).len(), 100);
+    assert_eq!(sim.id_set(1).len(), 100);
+}
+
+/// Test 5: the belief age tracks true elapsed wall-clock time across two hops.
+///
+/// The observation is made at `t=1000` and the contact graph is
+/// `[0,3000)` between 0-1 and `[4000,7000)` between 1-2, so by `t=7000` node 2
+/// should believe `x` is roughly `7000 - 1000 = 6000` ms old.
+#[test]
+fn age_tracks_true_elapsed_across_hops() {
+    let cfg = WorldConfig {
+        nodes: 3,
+        mtu: 1500,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 0,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 3000,
+        a: 0,
+        b: 1,
+    });
+    trace.contacts.push(Contact {
+        start_ms: 4000,
+        end_ms: 7000,
+        a: 1,
+        b: 2,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    sim.run_until(1000);
+    sim.observe(0, b"x", b"hello");
+    sim.run_until(7000);
+
+    let belief = sim
+        .nodes
+        .get(2)
+        .and_then(|n| n.belief(b"x", &LatestWins, 7000))
+        .expect("node 2 should hold a belief for \"x\"");
+
+    let age = belief.age_ms;
+    assert!(
+        (age as i64 - 6000).abs() <= 20,
+        "age {age} must be within +/-{} ms of 6000 (true elapsed across two hops)",
+        2 * sim.step_ms
+    );
+}
+
+/// Test 6: payloads larger than the MTU are fragmented across frames and still
+/// converge.
+#[test]
+fn large_payloads_fragment_over_small_mtu() {
+    let cfg = WorldConfig {
+        nodes: 2,
+        mtu: 256,
+        loss: 0.0,
+        dup: 0.0,
+        seed: 9,
+    };
+    let mut trace = ContactTrace::default();
+    trace.contacts.push(Contact {
+        start_ms: 0,
+        end_ms: 20_000,
+        a: 0,
+        b: 1,
+    });
+
+    let mut sim = Sim::new(cfg, trace);
+    for i in 0..20 {
+        let topic0 = format!("n0-{}", i);
+        let topic1 = format!("n1-{}", i);
+        let payload = vec![0xABu8; 1024];
+        sim.observe(0, topic0.as_bytes(), &payload);
+        sim.observe(1, topic1.as_bytes(), &payload);
+    }
+    sim.run_until(20_000);
+
+    assert!(
+        sim.converged(&[0, 1]),
+        "1 KiB payloads fragmented over a 256-byte MTU still converge"
+    );
+    assert_eq!(sim.id_set(0).len(), 40);
+    assert_eq!(sim.id_set(1).len(), 40);
+}