From ab9b892d946ec2c46f01324e439b7c82f8ee4a6d Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Sat, 17 Oct 2026 21:48:37 -0700 Subject: [PATCH] chore: ci script and README --- README.md | 104 ++++++++++++++++++++++++++++ scripts/ci.sh | 6 ++ tines-core/src/belief.rs | 24 +++++-- tines-core/src/bloom.rs | 10 ++- tines-core/src/exchange.rs | 120 +++++++++++++++++++++++++++------ tines-core/src/frag.rs | 90 +++++++++++++++++++++---- tines-core/src/lib.rs | 8 +-- tines-core/src/mem_store.rs | 57 ++++++++++------ tines-core/src/node.rs | 18 ++--- tines-core/src/observation.rs | 3 +- tines-core/src/store.rs | 7 +- tines-core/src/wire.rs | 2 +- tines-core/tests/pair.rs | 21 ++++-- tines-sim/src/lib.rs | 6 +- tines-sim/src/trace.rs | 7 +- tines-sim/src/transport.rs | 3 +- tines-sim/src/world.rs | 22 +++--- tines-sim/tests/convergence.rs | 45 ++++++++++--- tines-sim/tests/props.rs | 52 +++++++------- 19 files changed, 468 insertions(+), 137 deletions(-) create mode 100644 README.md create mode 100755 scripts/ci.sh diff --git a/README.md b/README.md new file mode 100644 index 0000000..49bf330 --- /dev/null +++ b/README.md @@ -0,0 +1,104 @@ +# tines — opportunistic shared memory for mostly-offline robots + +Small robots doing their work collect facts about the world. They are mostly +offline, so when two of them happen to meet in range they swap what they know so +that each robot's picture of the world improves over time. No robot *depends* on +receiving anything — it can do its job on its own observations — but there is no +reason not to share. + +`tines` is the library that does this: a store of timestamped-by-age +observations, a belief layer derived from them, and a delay-tolerant exchange +protocol that runs over whatever datagram link the robot has. The name is from +Vinge's *A Fire Upon the Deep*: the Tines only form a mind when close enough to +hear each other. + +## Architecture + +Three layers, each a separate concern, top to bottom: + +``` + ┌──────────────────────────────────────────┐ + │ memory — "my priors": beliefs per │ derived, local, never synced + │ topic, computed from │ + │ storage under a policy │ + ├──────────────────────────────────────────┤ + │ storage — append-only log of raw │ synced by set-union + │ observations; local │ + │ trimming policy │ + ├──────────────────────────────────────────┤ + │ comms — contact-driven exchange │ PHY-agnostic; transport + │ of observations │ trait + adapters + └──────────────────────────────────────────┘ +``` + +Only observations cross the wire. Memory is a pure function of storage plus a +policy and is never transmitted; storage is synced by set-union (order- and +path-independent); comms is a contact-driven exchange of observations over a +`Transport` trait. + +## Usage + +`tines-core` exposes a single application-facing facade, `Node`, generic +over any `Store` and `Transport`. Persist `Node::next_seq()` across reboots +(`boot_count << 32 | counter`) and pass it back as the initial `first_seq`; a +repeated `seq` after reboot collides in the content-id hash and dedupes silently. + +```rust +use tines_core::{LatestWins, MemStore, Node, NodeId}; + +let me: NodeId = [7; 16]; +let boot_count = 0u64; +let counter = 3u64; +let store = MemStore::new(); +let transport = /* any Transport implementation */; + +// first_seq must survive reboot: boot_count << 32 | counter +let mut node: Node = Node::new( + me, + boot_count << 32 | counter, + store, + transport, +); + +// Record something you saw. observed_at is an optional absolute clock +// (ms since epoch); `now` is a local monotonic Tick on every call. +let id = node + .observe(b"door/3", b"open", Some(1_700_000_000_000), 0) + .expect("observation fits the size limits"); + +// Drive the exchange in your loop. Returns how many new observations were +// stored this round. +let stored = node.poll(500); + +// Ask what you believe about a topic prefix, under a belief policy +// (e.g. the youngest observation wins). +let belief = node.belief(b"door", &LatestWins, 500); +``` + +## Crate layout + +``` +tines/ + tines-core/ no_std core: Observation, Store / Transport / BeliefPolicy + traits, the exchange state machine, and the `Node` facade. + `std` is optional; `tines-core` builds for `thumbv7em-none-eabihf` + with `--no-default-features`. + tines-sim/ std: the simulator, contact traces, and the convergence / + order-independence property tests. +``` + +## CI + +`scripts/ci.sh` runs the full gate — formatting, clippy, the test suite, and the +`no_std` build — and is safe to drop into a CI config: + +```sh +sh scripts/ci.sh +``` + +## Where to read more + +The spec (authoritative) and the implementation plan that argues from it: + +- **Spec:** `docs/superpowers/specs/2026-10-16-tines-design.md` +- **Plan:** `docs/superpowers/plans/2026-10-16-tines-core-and-sim.md` diff --git a/scripts/ci.sh b/scripts/ci.sh new file mode 100755 index 0000000..70915ba --- /dev/null +++ b/scripts/ci.sh @@ -0,0 +1,6 @@ +#!/bin/sh +set -e +cargo fmt --all -- --check +cargo clippy --workspace --all-targets -- -D warnings +cargo test --workspace +sh scripts/check-no-std.sh diff --git a/tines-core/src/belief.rs b/tines-core/src/belief.rs index 7113888..d13f7c5 100644 --- a/tines-core/src/belief.rs +++ b/tines-core/src/belief.rs @@ -14,7 +14,12 @@ pub trait BeliefPolicy { /// Streaming fold: called once per matching observation (already /// age-adjusted to `now`), in no particular order. `acc` is `None` before /// the first call. Return the new accumulator. - fn step(&self, me: NodeId, acc: Option, obs: &Observation) -> Option; + fn step( + &self, + me: NodeId, + acc: Option, + obs: &Observation, + ) -> Option; } /// Latest observation wins: the observation with the smallest `age_ms`; ties @@ -29,12 +34,18 @@ pub struct SelfFirst; impl BeliefPolicy for LatestWins { type Belief = Observation; - fn step(&self, _me: NodeId, acc: Option, obs: &Observation) -> Option { + fn step( + &self, + _me: NodeId, + acc: Option, + obs: &Observation, + ) -> Option { match acc { None => Some(obs.clone()), Some(cur) => { if obs.age_ms < cur.age_ms - || (obs.age_ms == cur.age_ms && (obs.observer, obs.seq) < (cur.observer, cur.seq)) + || (obs.age_ms == cur.age_ms + && (obs.observer, obs.seq) < (cur.observer, cur.seq)) { Some(obs.clone()) } else { @@ -48,7 +59,12 @@ impl BeliefPolicy for LatestWins { impl BeliefPolicy for SelfFirst { type Belief = Observation; - fn step(&self, me: NodeId, acc: Option, obs: &Observation) -> Option { + fn step( + &self, + me: NodeId, + acc: Option, + obs: &Observation, + ) -> Option { match acc { None => Some(obs.clone()), Some(cur) => { diff --git a/tines-core/src/bloom.rs b/tines-core/src/bloom.rs index ca50a15..ae19d7c 100644 --- a/tines-core/src/bloom.rs +++ b/tines-core/src/bloom.rs @@ -142,7 +142,10 @@ mod tests { bloom.insert(&id); } for id in ids { - assert!(bloom.might_contain(&id), "inserted id must be reported present"); + assert!( + bloom.might_contain(&id), + "inserted id must be reported present" + ); } } @@ -152,7 +155,10 @@ mod tests { let mut rng = Rng::new(2); let ids: [ObservationId; 100] = std::array::from_fn(|_| rng.id()); for id in ids { - assert!(!bloom.might_contain(&id), "empty filter must report nothing"); + assert!( + !bloom.might_contain(&id), + "empty filter must report nothing" + ); } } diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index a3114ba..68ed157 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -80,7 +80,12 @@ impl Exchange { /// /// Returns `Some(obs)` when an inbound observation is newly stored (the /// caller may react); otherwise `None`. - pub fn on_event(&mut self, ev: Event, store: &mut S, now: Tick) -> Option { + pub fn on_event( + &mut self, + ev: Event, + store: &mut S, + now: Tick, + ) -> Option { match ev { Event::Contact(addr) => { self.add_peer(addr); @@ -99,7 +104,13 @@ impl Exchange { /// /// Priority: (1) send a Summary to any peer due for one, then (2) emit one /// 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)> { + 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() { @@ -173,7 +184,13 @@ impl Exchange { None } - fn on_frame(&mut self, addr: A, bytes: &[u8], store: &mut S, now: Tick) -> Option { + fn on_frame( + &mut self, + addr: A, + bytes: &[u8], + store: &mut S, + now: Tick, + ) -> Option { let frame = match Frame::decode(bytes) { Ok(frame) => frame, Err(_) => return None, @@ -194,7 +211,13 @@ impl Exchange { Ok(false) | Err(_) => None, }; } - Frame::Frag { from: _, id, index, count, data } => { + Frame::Frag { + from: _, + id, + index, + count, + data, + } => { if let Some(obs) = peer.reassembler.feed(id, index, count, &data, now) { let returned = obs.clone(); return match store.insert(obs, now) { @@ -220,7 +243,14 @@ fn summary_due(peer: &Peer, now: Tick) -> bool { /// Emits at most one frame for `peer`: continuing an in-flight fragment if one /// is pending, otherwise draining the send queue (skipping ids no longer /// stored). -fn emit_one(me: NodeId, peer: &mut Peer, store: &S, mtu: usize, now: Tick, out: &mut [u8]) -> Option<(A, usize)> { +fn emit_one( + me: NodeId, + peer: &mut Peer, + store: &S, + mtu: usize, + now: Tick, + out: &mut [u8], +) -> Option<(A, usize)> { // in_flight takes precedence: continue the current fragmentation. if let Some((id, index)) = peer.in_flight { match store.get(&id, now) { @@ -333,13 +363,19 @@ fn rebuild_send_queue(peer: &mut Peer, store: &S, now: Tick) { peer.send_queue.clear(); for (id, _) in heap { - peer.send_queue.push(id).expect("heap is capped at SEND_QUEUE_CAP"); + peer.send_queue + .push(id) + .expect("heap is capped at SEND_QUEUE_CAP"); } } /// Inserts `(id, age)` into the age-ascending top-K buffer, keeping it sorted /// and capped at `SEND_QUEUE_CAP`. -fn insert_youngest(heap: &mut Vec<(ObservationId, u64), { consts::SEND_QUEUE_CAP }>, id: ObservationId, age: u64) { +fn insert_youngest( + heap: &mut Vec<(ObservationId, u64), { consts::SEND_QUEUE_CAP }>, + id: ObservationId, + age: u64, +) { if heap.len() < consts::SEND_QUEUE_CAP { let pos = insert_pos(heap, age); let _ = heap.insert(pos, (id, age)); @@ -372,7 +408,14 @@ mod tests { const ME: NodeId = [42; 16]; - fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option, topic: &[u8], payload: &[u8]) -> Observation { + fn obs( + observer: [u8; 16], + seq: u64, + age: u64, + at: Option, + topic: &[u8], + payload: &[u8], + ) -> Observation { Observation { observer, seq, @@ -397,7 +440,9 @@ mod tests { ex.on_event(Event::Contact(1), &mut store, 0); let mut buf = [0u8; consts::MAX_FRAME]; - let (addr, n) = ex.next_frame(&store, consts::MAX_FRAME, 0, &mut buf).expect("summary on contact"); + let (addr, n) = ex + .next_frame(&store, consts::MAX_FRAME, 0, &mut buf) + .expect("summary on contact"); assert_eq!(addr, 1); match Frame::decode(&buf[..n]).unwrap() { Frame::Summary { from, .. } => assert_eq!(from, ME), @@ -450,10 +495,14 @@ mod tests { assert_obs_id(&emitted[1], id_c); assert_obs_id(&emitted[2], id_a); assert!( - !emitted.iter().any(|f| matches!(f, Frame::Obs { obs, .. } if obs.id() == id_b)), + !emitted + .iter() + .any(|f| matches!(f, Frame::Obs { obs, .. } if obs.id() == id_b)), "B must be omitted" ); - assert!(ex.next_frame(&store, consts::MAX_FRAME, 0, &mut [0u8; consts::MAX_FRAME]).is_none()); + assert!(ex + .next_frame(&store, consts::MAX_FRAME, 0, &mut [0u8; consts::MAX_FRAME]) + .is_none()); } fn assert_obs_id(frame: &Frame, id: ObservationId) { @@ -473,7 +522,9 @@ mod tests { from: [2; 16], obs: obs.clone(), }; - let returned = ex.on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0).expect("obs returned"); + let returned = ex + .on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0) + .expect("obs returned"); assert_eq!(returned, obs); assert!(store.contains(&obs.id())); } @@ -488,12 +539,16 @@ mod tests { from: [2; 16], obs: obs.clone(), }; - assert!(ex.on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0).is_some()); + assert!(ex + .on_event(Event::Frame(1, frame_bytes(&frame)), &mut store, 0) + .is_some()); let dup = Frame::Obs { from: [2; 16], obs: obs.clone(), }; - assert!(ex.on_event(Event::Frame(1, frame_bytes(&dup)), &mut store, 0).is_none()); + assert!(ex + .on_event(Event::Frame(1, frame_bytes(&dup)), &mut store, 0) + .is_none()); } #[test] @@ -504,7 +559,11 @@ mod tests { let before = store.len(); let bogus = Event::Frame(1, heapless::Vec::from_slice(&[0xFFu8; 50]).unwrap()); assert_eq!(ex.on_event(bogus, &mut store, 0), None); - assert_eq!(store.len(), before, "store must be unchanged by a malformed frame"); + assert_eq!( + store.len(), + before, + "store must be unchanged by a malformed frame" + ); } #[test] @@ -554,7 +613,9 @@ mod tests { ex.on_event(Event::Contact(1), &mut store, 0); let mut buf = [0u8; consts::MAX_FRAME]; - let (_, n) = ex.next_frame(&store, consts::MAX_FRAME, 0, &mut buf).expect("first summary"); + let (_, n) = ex + .next_frame(&store, consts::MAX_FRAME, 0, &mut buf) + .expect("first summary"); match Frame::decode(&buf[..n]).unwrap() { Frame::Summary { salt, .. } => assert_eq!(salt, 0), other => panic!("expected a Summary, got {other:?}"), @@ -568,7 +629,12 @@ mod tests { // After the interval: another Summary, now rotated to salt 1. let mut buf = [0u8; consts::MAX_FRAME]; let (_, n) = ex - .next_frame(&store, consts::MAX_FRAME, consts::SUMMARY_INTERVAL_MS, &mut buf) + .next_frame( + &store, + consts::MAX_FRAME, + consts::SUMMARY_INTERVAL_MS, + &mut buf, + ) .expect("second summary"); match Frame::decode(&buf[..n]).unwrap() { Frame::Summary { salt, .. } => assert_eq!(salt, 1), @@ -589,7 +655,9 @@ mod tests { fn summary_salt(ex: &mut Exchange, store: &MemStore, now: Tick) -> u32 { let mut buf = [0u8; consts::MAX_FRAME]; - let (_, n) = ex.next_frame(store, consts::MAX_FRAME, now, &mut buf).expect("summary"); + let (_, n) = ex + .next_frame(store, consts::MAX_FRAME, now, &mut buf) + .expect("summary"); match Frame::decode(&buf[..n]).unwrap() { Frame::Summary { salt, .. } => salt, other => panic!("expected a Summary, got {other:?}"), @@ -629,15 +697,25 @@ mod tests { } } - assert!(!indices.is_empty(), "the large observation must be fragmented"); + assert!( + !indices.is_empty(), + "the large observation must be fragmented" + ); let count = indices[0].1; - assert!(count >= 2, "a 1024-byte payload must span multiple fragments"); + assert!( + count >= 2, + "a 1024-byte payload must span multiple fragments" + ); for (i, &(index, c)) in indices.iter().enumerate() { assert_eq!(index, i as u8, "fragments must carry increasing indices"); assert_eq!(c, count, "all fragments share the same count"); } let last = indices.last().copied().unwrap(); - assert_eq!(last.0 + 1, last.1, "fragment indices are contiguous to count"); + assert_eq!( + last.0 + 1, + last.1, + "fragment indices are contiguous to count" + ); assert_eq!(indices.len(), count as usize); // After draining, in_flight clears and the queue drops the id. assert!(ex.peers()[0].in_flight.is_none()); diff --git a/tines-core/src/frag.rs b/tines-core/src/frag.rs index 1ae5b4c..3e00a69 100644 --- a/tines-core/src/frag.rs +++ b/tines-core/src/frag.rs @@ -1,7 +1,7 @@ //! Fragmentation and reassembly of `Observation`s over a byte-oriented link. use crate::consts; -use crate::{Frame, Observation, NodeId, ObservationId, Tick, WireError}; +use crate::{Frame, NodeId, Observation, ObservationId, Tick, WireError}; use heapless::Vec; /// Header allowance per fragment (bytes), pinned. @@ -85,7 +85,14 @@ impl Reassembler { } } - pub fn feed(&mut self, id: ObservationId, index: u8, count: u8, data: &[u8], now: Tick) -> Option { + pub fn feed( + &mut self, + id: ObservationId, + index: u8, + count: u8, + data: &[u8], + now: Tick, + ) -> Option { // 1. Drop malformed indices: count must be in 1..=64 and index < count. // `index >= 64` is guarded too so the `1 << index` bitmap never // shifts out of bounds on adversarial wire input. @@ -115,13 +122,19 @@ impl Reassembler { if self.buf.len().saturating_add(data.len()) > self.buf.capacity() { return None; } - self.buf.extend_from_slice(data).expect("buf capacity check above"); + self.buf + .extend_from_slice(data) + .expect("buf capacity check above"); self.offs[index as usize] = start as u16; self.lens[index as usize] = data.len() as u16; self.received |= 1u64 << index; // 4. Not all fragments have arrived yet. - let complete: u64 = if count >= 64 { u64::MAX } else { (1u64 << count) - 1 }; + let complete: u64 = if count >= 64 { + u64::MAX + } else { + (1u64 << count) - 1 + }; if self.received != complete { return None; } @@ -219,12 +232,22 @@ mod tests { let mut r = Reassembler::new(); let mut last: Option = None; for f in frames.iter() { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { last = r.feed(id, index, count, &data, 0); } } let reassembled = last.expect("reassembled obs"); - assert_eq!(reassembled, obs, "reassembled observation must equal original"); + assert_eq!( + reassembled, obs, + "reassembled observation must equal original" + ); } #[test] @@ -236,7 +259,14 @@ mod tests { let mut r = Reassembler::new(); let mut last: Option = None; for f in frames.iter().rev() { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { last = r.feed(id, index, count, &data, 0); } } @@ -253,7 +283,14 @@ mod tests { let frags: Vec<(ObservationId, u8, u8, Vec)> = frames .iter() .map(|f| { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { (id, index, count, data.to_vec()) } else { panic!("expected Frag frames"); @@ -283,7 +320,14 @@ mod tests { let frags: Vec<(ObservationId, u8, u8, Vec)> = frames .iter() .map(|f| { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { (id, index, count, data.to_vec()) } else { panic!("expected Frag frames"); @@ -323,7 +367,14 @@ mod tests { let frags: Vec<(ObservationId, u8, u8, Vec)> = frames .iter() .map(|f| { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { (id, index, count, data.to_vec()) } else { panic!("expected Frag frames"); @@ -337,12 +388,18 @@ mod tests { r.feed(*id0, *idx0, *cnt0, data0, 0); // feed index 1 at now=timeout+1 -> slot restarts with only index 1, returns None let (id1, idx1, cnt1, data1) = &frags[1]; - assert!(r.feed(*id1, *idx1, *cnt1, data1, consts::REASSEMBLY_TIMEOUT_MS + 1).is_none()); + assert!(r + .feed(*id1, *idx1, *cnt1, data1, consts::REASSEMBLY_TIMEOUT_MS + 1) + .is_none()); // re-feed index 0 late: proves index 0 from now=0 was discarded, but the // restarted slot now holds index 1, so the stream completes to the original let (idc, idxc, cntc, datac) = &frags[0]; let late = r.feed(*idc, *idxc, *cntc, datac, consts::REASSEMBLY_TIMEOUT_MS + 2); - assert_eq!(late, Some(obs.clone()), "late re-feed must complete the restarted stream"); + assert_eq!( + late, + Some(obs.clone()), + "late re-feed must complete the restarted stream" + ); // sanity: a fresh feed of the whole set from now=0 also completes let mut r2 = Reassembler::new(); let mut complete: Option = None; @@ -361,7 +418,14 @@ mod tests { let mut frags: Vec<(ObservationId, u8, u8, Vec)> = frames .iter() .map(|f| { - if let Frame::Frag { id, index, count, data, .. } = Frame::decode(f).unwrap() { + if let Frame::Frag { + id, + index, + count, + data, + .. + } = Frame::decode(f).unwrap() + { (id, index, count, data.to_vec()) } else { panic!("expected Frag frames"); diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index a2e41e7..98dc374 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -9,11 +9,11 @@ pub mod bloom; pub mod consts; pub mod exchange; pub mod frag; -pub mod observation; -pub mod store; #[cfg(feature = "alloc")] pub mod mem_store; pub mod node; +pub mod observation; +pub mod store; pub mod transport; pub mod wire; @@ -22,8 +22,8 @@ pub use bloom::Bloom; pub use exchange::{Exchange, Peer}; pub use frag::{fragment, Reassembler}; pub use node::{Node, ObserveError}; -pub use observation::{Observation, NodeId, ObservationId, Payload, Tick, Topic}; -pub use store::{Store, TrimPolicy, StoreError}; +pub use observation::{NodeId, Observation, ObservationId, Payload, Tick, Topic}; +pub use store::{Store, StoreError, TrimPolicy}; pub use transport::{Event, FrameBuf, SendError, Transport}; pub use wire::{Frame, WireError}; diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs index 9f59a33..a39595c 100644 --- a/tines-core/src/mem_store.rs +++ b/tines-core/src/mem_store.rs @@ -8,14 +8,14 @@ #[cfg(feature = "alloc")] extern crate alloc; -#[cfg(feature = "alloc")] -use alloc::collections::{BTreeMap, BTreeSet}; -#[cfg(feature = "alloc")] -use alloc::vec::Vec; #[cfg(feature = "alloc")] use crate::observation::{Observation, ObservationId, Tick}; #[cfg(feature = "alloc")] use crate::store::{current_age, Store, StoreError, TrimPolicy}; +#[cfg(feature = "alloc")] +use alloc::collections::{BTreeMap, BTreeSet}; +#[cfg(feature = "alloc")] +use alloc::vec::Vec; #[cfg(feature = "alloc")] struct Held { @@ -60,7 +60,13 @@ impl Store for MemStore { // Dedupe: keep the first-stored entry and its age lineage. return Ok(false); } - self.map.insert(id, Held { obs, stored_at: now }); + self.map.insert( + id, + Held { + obs, + stored_at: now, + }, + ); Ok(true) } @@ -84,12 +90,7 @@ impl Store for MemStore { } } - fn for_each_topic( - &self, - prefix: &[u8], - now: Tick, - f: &mut dyn FnMut(&Observation), - ) { + fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) { for held in self.map.values() { if held.obs.topic.starts_with(prefix) { let elapsed = now.saturating_sub(held.stored_at); @@ -107,7 +108,10 @@ impl Store for MemStore { let mut groups: BTreeMap<&[u8], Vec> = BTreeMap::new(); for (id, held) in self.map.iter() { if !removed.contains(id) { - groups.entry(held.obs.topic.as_slice()).or_default().push(*id); + groups + .entry(held.obs.topic.as_slice()) + .or_default() + .push(*id); } } for member_ids in groups.values() { @@ -116,8 +120,11 @@ impl Store for MemStore { } let mut ids = member_ids.clone(); ids.sort_by(|&a, &b| { - current_age(&self.map[&a].obs, self.map[&a].stored_at, now) - .cmp(¤t_age(&self.map[&b].obs, self.map[&b].stored_at, now)) + current_age(&self.map[&a].obs, self.map[&a].stored_at, now).cmp(¤t_age( + &self.map[&b].obs, + self.map[&b].stored_at, + now, + )) }); for id in ids.into_iter().skip(keep) { removed.insert(id); @@ -128,9 +135,7 @@ impl Store for MemStore { // (b) max_age_ms: drop anything whose current age exceeds the cutoff. if let Some(max_age) = policy.max_age_ms { for (id, held) in self.map.iter() { - if !removed.contains(id) - && current_age(&held.obs, held.stored_at, now) > max_age - { + if !removed.contains(id) && current_age(&held.obs, held.stored_at, now) > max_age { removed.insert(*id); } } @@ -147,8 +152,11 @@ impl Store for MemStore { .collect(); if remaining.len() > max_count { remaining.sort_by(|&a, &b| { - current_age(&self.map[&a].obs, self.map[&a].stored_at, now) - .cmp(¤t_age(&self.map[&b].obs, self.map[&b].stored_at, now)) + current_age(&self.map[&a].obs, self.map[&a].stored_at, now).cmp(¤t_age( + &self.map[&b].obs, + self.map[&b].stored_at, + now, + )) }); let over = remaining.len() - max_count; for id in remaining.into_iter().rev().take(over) { @@ -255,7 +263,8 @@ mod tests { fn trim_keep_per_topic_keeps_youngest() { let mut s = MemStore::new(); for (i, age) in [10u64, 20, 30, 40, 50].iter().enumerate() { - s.insert(obs([1; 16], i as u64, *age, None, b"t"), 0).unwrap(); + s.insert(obs([1; 16], i as u64, *age, None, b"t"), 0) + .unwrap(); } let policy = TrimPolicy { keep_per_topic: Some(2), @@ -293,7 +302,8 @@ mod tests { fn trim_max_count_drops_oldest() { let mut s = MemStore::new(); for i in 0..10u64 { - s.insert(obs([1; 16], i, (i + 1) * 10, None, b"t"), 0).unwrap(); + s.insert(obs([1; 16], i, (i + 1) * 10, None, b"t"), 0) + .unwrap(); } let policy = TrimPolicy { keep_per_topic: None, @@ -306,7 +316,10 @@ mod tests { s.for_each_topic(b"t", 0, &mut |o| { ages.insert(o.age_ms); }); - assert_eq!(ages, std::collections::BTreeSet::from_iter([10u64, 20, 30, 40])); + assert_eq!( + ages, + std::collections::BTreeSet::from_iter([10u64, 20, 30, 40]) + ); } #[test] diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 7495ee3..a3d797a 100644 --- a/tines-core/src/node.rs +++ b/tines-core/src/node.rs @@ -11,8 +11,8 @@ //! transport frames are dropped by the [`Exchange`]. use crate::{ - BeliefPolicy, consts, Exchange, NodeId, Observation, ObservationId, Payload, Store, StoreError, - TrimPolicy, Tick, Topic, Transport, + consts, BeliefPolicy, Exchange, NodeId, Observation, ObservationId, Payload, Store, StoreError, + Tick, Topic, Transport, TrimPolicy, }; /// Errors returned by [`Node::observe`]. @@ -115,7 +115,10 @@ impl Node { // Hoist the MTU, then borrow `exchange`/`store`, then borrow // `transport` for the send — disjoint field borrows, never overlapping. let mtu = self.transport.mtu(); - if let Some((addr, n)) = self.exchange.next_frame(&self.store, mtu, now, &mut self.out) { + if let Some((addr, n)) = self + .exchange + .next_frame(&self.store, mtu, now, &mut self.out) + { let _ = self.transport.send(&addr, &self.out[..n]); } stored @@ -134,11 +137,10 @@ impl Node { now: Tick, ) -> Option { let mut acc: Option = None; - self.store - .for_each_topic(topic_prefix, now, &mut |obs| { - let prev = acc.take(); - acc = policy.step(self.me, prev, obs); - }); + self.store.for_each_topic(topic_prefix, now, &mut |obs| { + let prev = acc.take(); + acc = policy.step(self.me, prev, obs); + }); acc } diff --git a/tines-core/src/observation.rs b/tines-core/src/observation.rs index 23e7e80..cc988f6 100644 --- a/tines-core/src/observation.rs +++ b/tines-core/src/observation.rs @@ -118,7 +118,8 @@ mod tests { assert_eq!( o.id(), [ - 0x92, 0x37, 0x37, 0xa1, 0x06, 0x9b, 0xd9, 0x5e, 0x08, 0xc2, 0xe6, 0x39, 0x7d, 0x4a, 0x8f, 0xd0 + 0x92, 0x37, 0x37, 0xa1, 0x06, 0x9b, 0xd9, 0x5e, 0x08, 0xc2, 0xe6, 0x39, 0x7d, 0x4a, + 0x8f, 0xd0 ] ); } diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs index 5e6fc3d..4ce03f4 100644 --- a/tines-core/src/store.rs +++ b/tines-core/src/store.rs @@ -62,12 +62,7 @@ pub trait Store { /// Visit every stored observation whose topic starts with `prefix` /// (an empty prefix matches all), each age-adjusted to `now`. - fn for_each_topic( - &self, - prefix: &[u8], - now: Tick, - f: &mut dyn FnMut(&Observation), - ); + fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)); /// Apply [`TrimPolicy`] rules in order (per-topic, age, count) and return /// the number of observations removed. Trimmed ids are not tombstoned: diff --git a/tines-core/src/wire.rs b/tines-core/src/wire.rs index feac9b7..1da4d08 100644 --- a/tines-core/src/wire.rs +++ b/tines-core/src/wire.rs @@ -4,7 +4,7 @@ //! — `tines-core` is `#![no_std]` and never panics on untrusted bytes. use crate::consts; -use crate::{Observation, NodeId, ObservationId}; +use crate::{NodeId, Observation, ObservationId}; use heapless::Vec; use serde::{Deserialize, Serialize}; diff --git a/tines-core/tests/pair.rs b/tines-core/tests/pair.rs index 6e564ae..31a4cc1 100644 --- a/tines-core/tests/pair.rs +++ b/tines-core/tests/pair.rs @@ -1,10 +1,10 @@ mod common; use common::PairTransport; +use std::collections::BTreeSet; use tines_core::observation::Payload; use tines_core::transport::{Event, FrameBuf, Transport}; use tines_core::{consts, LatestWins, MemStore, Node, ObserveError, Store}; -use std::collections::BTreeSet; #[test] fn pair_transport_connect_then_frame() { @@ -111,15 +111,23 @@ fn belief_latest_wins_across_nodes() { // observe stores at `now` (b's stored_at=500, a's stored_at=0), so b's // is younger everywhere it is held; a re-received copy has a slightly // different age, so compare by identity, not byte-equality. - let belief = nb.belief(b"door", &LatestWins, now).expect("door belief exists"); - assert_eq!(belief.observer, [2; 16], "nb belief should be b's observation"); + let belief = nb + .belief(b"door", &LatestWins, now) + .expect("door belief exists"); + assert_eq!( + belief.observer, [2; 16], + "nb belief should be b's observation" + ); assert_eq!(belief.seq, 0); assert_eq!(belief.payload, Payload::from_slice(b"late").unwrap()); assert_eq!(belief.observed_at, Some(500)); // a's node, looking at the same topic, agrees on the same observation. let belief_a = na.belief(b"door", &LatestWins, now).unwrap(); - assert_eq!(belief_a.observer, [2; 16], "na belief should be b's observation"); + assert_eq!( + belief_a.observer, [2; 16], + "na belief should be b's observation" + ); assert_eq!(belief_a.payload, Payload::from_slice(b"late").unwrap()); assert_eq!(belief_a.observed_at, Some(500)); } @@ -192,6 +200,9 @@ fn third_party_observation_propagates_within_contact() { nb.poll(t); t += 10; } - assert!(nb.store().contains(&id), "fresh observation must propagate to b"); + assert!( + nb.store().contains(&id), + "fresh observation must propagate to b" + ); assert_eq!(nb.store().len(), 1); } diff --git a/tines-sim/src/lib.rs b/tines-sim/src/lib.rs index 336f220..b8832e6 100644 --- a/tines-sim/src/lib.rs +++ b/tines-sim/src/lib.rs @@ -9,10 +9,10 @@ #![deny(warnings)] -pub mod trace; -pub mod world; -pub mod transport; pub mod sim; +pub mod trace; +pub mod transport; +pub mod world; pub use sim::{Sim, SimNode}; pub use trace::{Contact, ContactTrace}; diff --git a/tines-sim/src/trace.rs b/tines-sim/src/trace.rs index dd6bf6a..9dc2446 100644 --- a/tines-sim/src/trace.rs +++ b/tines-sim/src/trace.rs @@ -53,7 +53,12 @@ mod tests { #[test] fn in_contact_matches_both_orders() { let mut trace = ContactTrace::default(); - trace.contacts.push(Contact { start_ms: 100, end_ms: 200, a: 3, b: 7 }); + trace.contacts.push(Contact { + start_ms: 100, + end_ms: 200, + a: 3, + b: 7, + }); assert!(!trace.in_contact(3, 7, 50)); // before the window assert!(trace.in_contact(3, 7, 100)); // at the start edge assert!(trace.in_contact(7, 3, 150)); // mid-window, either order diff --git a/tines-sim/src/transport.rs b/tines-sim/src/transport.rs index 6d4612a..fbde1ee 100644 --- a/tines-sim/src/transport.rs +++ b/tines-sim/src/transport.rs @@ -49,8 +49,7 @@ impl Transport for SimTransport { if !w.trace.in_contact(self.me, *to, w.prev_t) { return Err(SendError::Unreachable); } - if frame.len() > w.cfg.mtu - && !matches!(Frame::decode(frame), Ok(Frame::Summary { .. })) { + 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 b80b364..e081cea 100644 --- a/tines-sim/src/world.rs +++ b/tines-sim/src/world.rs @@ -150,9 +150,12 @@ mod tests { seed, }; let mut trace = ContactTrace::default(); - trace - .contacts - .push(Contact { start_ms: start, end_ms: end, a: 0, b: 1 }); + trace.contacts.push(Contact { + start_ms: start, + end_ms: end, + a: 0, + b: 1, + }); SimWorld::new(cfg, trace) } @@ -163,9 +166,7 @@ mod tests { /// The number of `Frame` events queued in `node`'s inbox. fn frame_count(world: &Rc>, node: usize) -> usize { - world - .borrow() - .inboxes[node] + world.borrow().inboxes[node] .iter() .filter(|e| matches!(e, Event::Frame(_, _))) .count() @@ -244,9 +245,12 @@ mod tests { }; let make_trace = || { let mut trace = ContactTrace::default(); - trace - .contacts - .push(Contact { start_ms: 0, end_ms: 1000, a: 0, b: 1 }); + trace.contacts.push(Contact { + start_ms: 0, + end_ms: 1000, + a: 0, + b: 1, + }); trace }; let w1 = SimWorld::new(cfg.clone(), make_trace()); diff --git a/tines-sim/tests/convergence.rs b/tines-sim/tests/convergence.rs index 8747cec..ff90c6f 100644 --- a/tines-sim/tests/convergence.rs +++ b/tines-sim/tests/convergence.rs @@ -27,14 +27,25 @@ fn two_nodes_one_contact_converge() { 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()); + 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.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!( + 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); } @@ -66,7 +77,11 @@ fn relay_through_middle_node() { 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.observe( + 0, + format!("a-{}", i).as_bytes(), + format!("a-p{}", i).as_bytes(), + ); } sim.run_until(7000); @@ -74,7 +89,10 @@ fn relay_through_middle_node() { 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!( + node_2.contains(id), + "node 2 is missing node 0's observation {id:?}" + ); } assert_eq!(from_0.len(), 5, "node 0 observed 5 things"); } @@ -114,7 +132,10 @@ fn disconnected_component_does_not_converge() { 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, 2, 3]), + "cross-component never converges" + ); assert!(sim.converged(&[0, 1]), "component A converges"); assert!(sim.converged(&[2, 3]), "component B converges"); } @@ -139,8 +160,16 @@ fn lossy_link_still_converges_given_time() { 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.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); diff --git a/tines-sim/tests/props.rs b/tines-sim/tests/props.rs index 0bebf42..84b0af6 100644 --- a/tines-sim/tests/props.rs +++ b/tines-sim/tests/props.rs @@ -19,14 +19,7 @@ use tines_sim::{Contact, ContactTrace, Sim, WorldConfig}; /// A fixed set of short observation topics the tests reuse so the generated /// observations round-trip cleanly through the core (`MAX_TOPIC` is 64 bytes). const TOPICS: [&[u8]; 8] = [ - b"temp", - b"humid", - b"door", - b"motion", - b"light", - b"batt", - b"press", - b"power", + b"temp", b"humid", b"door", b"motion", b"light", b"batt", b"press", b"power", ]; /// A single contact window over `nodes` nodes. `start` is bounded well below the @@ -44,28 +37,28 @@ fn contact(nodes: usize) -> impl Strategy { 0usize..nodes, 0usize..nodes.saturating_sub(1), ) - .prop_map(|(start, duration, a, b_prime)| { - let mut b = b_prime; - if b >= a { - b += 1; - } - Contact { - start_ms: start, - end_ms: start + duration, - a, - b, - } - }) + .prop_map(|(start, duration, a, b_prime)| { + let mut b = b_prime; + if b >= a { + b += 1; + } + Contact { + start_ms: start, + end_ms: start + duration, + a, + b, + } + }) } /// A single observation: a topic drawn from [`TOPICS`] and a random 1..=64 byte /// payload. fn observation() -> impl Strategy, Vec)> { -( - 0u32..8u32, - proptest::collection::vec(any::(), 1..=64usize), -) -.prop_map(|(topic_idx, payload)| (TOPICS[topic_idx as usize].to_vec(), payload)) + ( + 0u32..8u32, + proptest::collection::vec(any::(), 1..=64usize), + ) + .prop_map(|(topic_idx, payload)| (TOPICS[topic_idx as usize].to_vec(), payload)) } /// A generated scenario: a node count, a contact trace over those nodes, and a @@ -93,8 +86,13 @@ fn scenario() -> impl Strategy { /// A 3-node observation list, reused by the order-independence test. fn observations_3nodes() -> impl Strategy, Vec)>> { - proptest::collection::vec(observation(), 3..=3) - .prop_map(|per_node| per_node.into_iter().enumerate().map(|(n, o)| (n, o.0, o.1)).collect()) + proptest::collection::vec(observation(), 3..=3).prop_map(|per_node| { + per_node + .into_iter() + .enumerate() + .map(|(n, o)| (n, o.0, o.1)) + .collect() + }) } /// Maps a `u64` seed to one of the six permutations of three slots.