diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 71308d5..a3114ba 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -101,7 +101,6 @@ impl Exchange { /// Obs/Frag frame from a peer via round-robin. pub fn next_frame(&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) { diff --git a/tines-core/src/frag.rs b/tines-core/src/frag.rs index b6af1ba..1ae5b4c 100644 --- a/tines-core/src/frag.rs +++ b/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::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)> = 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::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)> = 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::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)> = frames .iter() @@ -373,8 +372,7 @@ mod tests { frags[0].3[0] ^= 0xFF; let mut r = Reassembler::new(); let mut last: Option = 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"); diff --git a/tines-sim/src/lib.rs b/tines-sim/src/lib.rs index 2b7ba6a..336f220 100644 --- a/tines-sim/src/lib.rs +++ b/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}; diff --git a/tines-sim/src/sim.rs b/tines-sim/src/sim.rs new file mode 100644 index 0000000..70d922e --- /dev/null +++ b/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; + +/// A run of the simulation, owned by the test. +pub struct Sim { + pub world: Rc>, + pub nodes: Vec, + 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 { + 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> { + let (n, contacts) = { + let world = self.world.borrow(); + ( + world.cfg.nodes, + world + .trace + .contacts + .iter() + .map(|c| (c.a, c.b)) + .collect::>(), + ) + }; + + let mut parent: Vec = (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> = BTreeMap::new(); + for node in 0..n { + let root = find(&mut parent, node); + groups.entry(root).or_default().push(node); + } + + let mut result: Vec> = 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 +} diff --git a/tines-sim/src/transport.rs b/tines-sim/src/transport.rs index c5f76e9..6d4612a 100644 --- a/tines-sim/src/transport.rs +++ b/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::() < w.cfg.loss { diff --git a/tines-sim/src/world.rs b/tines-sim/src/world.rs index 6851e36..c459e45 100644 --- a/tines-sim/src/world.rs +++ b/tines-sim/src/world.rs @@ -33,6 +33,12 @@ pub struct SimWorld { pub(crate) trace: ContactTrace, pub(crate) inboxes: Vec>>, 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); diff --git a/tines-sim/tests/convergence.rs b/tines-sim/tests/convergence.rs new file mode 100644 index 0000000..8747cec --- /dev/null +++ b/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); +}