kyle/tines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: ci script and README

Browse Source
This commit is contained in:
kyle
2026-10-17 21:48:37 -07:00
parent 4706004f47
commit ab9b892d94
19 changed files with 468 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tines-core/src/belief.rs
+20 -4
View File
@@ -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<Self::Belief>, obs: &Observation) -> Option<Self::Belief>;
fn step(
&self,
me: NodeId,
acc: Option<Self::Belief>,
obs: &Observation,
) -> Option<Self::Belief>;
}
/// 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<Self::Belief>, obs: &Observation) -> Option<Self::Belief> {
fn step(
&self,
_me: NodeId,
acc: Option<Self::Belief>,
obs: &Observation,
) -> Option<Self::Belief> {
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<Self::Belief>, obs: &Observation) -> Option<Self::Belief> {
fn step(
&self,
me: NodeId,
acc: Option<Self::Belief>,
obs: &Observation,
) -> Option<Self::Belief> {
match acc {
None => Some(obs.clone()),
Some(cur) => {
tines-core/src/bloom.rs
+8 -2
View File
@@ -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"
);
}
}
tines-core/src/exchange.rs
+99 -21
View File
@@ -80,7 +80,12 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
///
/// Returns `Some(obs)` when an inbound observation is newly stored (the
/// caller may react); otherwise `None`.
pub fn on_event<S: Store>(&mut self, ev: Event<A>, store: &mut S, now: Tick) -> Option<Observation> {
pub fn on_event<S: Store>(
&mut self,
ev: Event<A>,
store: &mut S,
now: Tick,
) -> Option<Observation> {
match ev {
Event::Contact(addr) => {
self.add_peer(addr);
@@ -99,7 +104,13 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
///
/// 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<S: Store>(&mut self, store: &S, mtu: usize, now: Tick, out: &mut [u8]) -> Option<(A, usize)> {
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() {
@@ -173,7 +184,13 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
None
}
fn on_frame<S: Store>(&mut self, addr: A, bytes: &[u8], store: &mut S, now: Tick) -> Option<Observation> {
fn on_frame<S: Store>(
&mut self,
addr: A,
bytes: &[u8],
store: &mut S,
now: Tick,
) -> Option<Observation> {
let frame = match Frame::decode(bytes) {
Ok(frame) => frame,
Err(_) => return None,
@@ -194,7 +211,13 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
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<A>(peer: &Peer<A>, 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<A: Clone, S: Store>(me: NodeId, peer: &mut Peer<A>, store: &S, mtu: usize, now: Tick, out: &mut [u8]) -> Option<(A, usize)> {
fn emit_one<A: Clone, S: Store>(
me: NodeId,
peer: &mut Peer<A>,
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<A, S: Store>(peer: &mut Peer<A>, 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<u64>, topic: &[u8], payload: &[u8]) -> Observation {
fn obs(
observer: [u8; 16],
seq: u64,
age: u64,
at: Option<u64>,
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<u8>, 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());
tines-core/src/frag.rs
+77 -13
View File
@@ -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<Observation> {
pub fn feed(
&mut self,
id: ObservationId,
index: u8,
count: u8,
data: &[u8],
now: Tick,
) -> Option<Observation> {
// 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<Observation> = 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<Observation> = 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<u8>)> = 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<u8>)> = 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<u8>)> = 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<Observation> = None;
@@ -361,7 +418,14 @@ mod tests {
let mut frags: Vec<(ObservationId, u8, u8, Vec<u8>)> = 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");
tines-core/src/lib.rs
+4 -4
View File
@@ -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};
tines-core/src/mem_store.rs
+35 -22
View File
@@ -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<ObservationId>> = 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(&current_age(&self.map[&b].obs, self.map[&b].stored_at, now))
current_age(&self.map[&a].obs, self.map[&a].stored_at, now).cmp(&current_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(&current_age(&self.map[&b].obs, self.map[&b].stored_at, now))
current_age(&self.map[&a].obs, self.map[&a].stored_at, now).cmp(&current_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]
tines-core/src/node.rs
+10 -8
View File
@@ -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<S: Store, T: Transport> Node<S, T> {
// 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<S: Store, T: Transport> Node<S, T> {
now: Tick,
) -> Option<P::Belief> {
let mut acc: Option<P::Belief> = 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
}
tines-core/src/observation.rs
+2 -1
View File
@@ -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
]
);
}
tines-core/src/store.rs
+1 -6
View File
@@ -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:
tines-core/src/wire.rs
+1 -1
View File
@@ -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};
tines-core/tests/pair.rs
+16 -5
View File
@@ -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);
}