From 1752bd23ff8ebc512d78db215a50073813f4592a Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Sat, 17 Oct 2026 23:58:15 -0700 Subject: [PATCH] refactor(core): dedupe trim/belief comparators and emit_one; add coverage tests --- tines-core/src/belief.rs | 28 ++++++++++++++------ tines-core/src/exchange.rs | 28 +++++++++++++++++--- tines-core/src/mem_store.rs | 51 +++++++++++++++++++++++++++---------- 3 files changed, 81 insertions(+), 26 deletions(-) diff --git a/tines-core/src/belief.rs b/tines-core/src/belief.rs index d13f7c5..06be901 100644 --- a/tines-core/src/belief.rs +++ b/tines-core/src/belief.rs @@ -31,6 +31,13 @@ pub struct LatestWins; /// `me` beats any observation from another observer regardless of age. pub struct SelfFirst; +/// The [`LatestWins`] ordering: `obs` beats `cur` when it is younger, with a +/// deterministic `(observer, seq)` ascending tie-break on equal age. +fn latest_beats(obs: &Observation, cur: &Observation) -> bool { + obs.age_ms < cur.age_ms + || (obs.age_ms == cur.age_ms && (obs.observer, obs.seq) < (cur.observer, cur.seq)) +} + impl BeliefPolicy for LatestWins { type Belief = Observation; @@ -43,10 +50,7 @@ impl BeliefPolicy for LatestWins { 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)) - { + if latest_beats(obs, &cur) { Some(obs.clone()) } else { Some(cur) @@ -78,10 +82,7 @@ impl BeliefPolicy for SelfFirst { // Both from `me` or both from someone else: fall back to // the LatestWins rule (youngest, tie broken by // (observer, seq) ascending). - if obs.age_ms < cur.age_ms - || (obs.age_ms == cur.age_ms - && (obs.observer, obs.seq) < (cur.observer, cur.seq)) - { + if latest_beats(obs, &cur) { obs.clone() } else { cur @@ -159,6 +160,17 @@ mod tests { assert_eq!(fold_rev(&LatestWins, me, &obs).unwrap().observer, [2; 16]); } + #[test] + fn latest_wins_tie_same_observer_breaks_by_seq() { + let me = [7; 16]; + // Same observer, same age: the tie-break must fall to the smaller seq. + let hi_seq = obs([5; 16], 9, 100); + let lo_seq = obs([5; 16], 2, 100); + let obs = [hi_seq, lo_seq]; + assert_eq!(fold(&LatestWins, me, &obs).unwrap().seq, 2); + assert_eq!(fold_rev(&LatestWins, me, &obs).unwrap().seq, 2); + } + #[test] fn self_first_prefers_own_even_if_older() { let me = [7; 16]; diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 68ed157..a2bb712 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -263,10 +263,8 @@ fn emit_one( } while let Some(&front) = peer.send_queue.first() { - if !store.contains(&front) { - peer.send_queue.remove(0); - continue; - } + // A single `get` both checks presence and returns the age-adjusted + // observation; a missing id (trimmed elsewhere) is skipped. let Some(obs) = store.get(&front, now) else { peer.send_queue.remove(0); continue; @@ -721,4 +719,26 @@ mod tests { assert!(ex.peers()[0].in_flight.is_none()); assert!(!ex.peers()[0].send_queue.contains(&id)); } + + #[test] + fn inbound_frags_are_reassembled_and_stored() { + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + let big = obs([2; 16], 7, 0, None, b"t", &[0u8; 1024]); + let id = big.id(); + + // Fragment the observation exactly as the peer would, then feed each + // Frag frame through the exchange's inbound path (no prior Contact). + let mut frames: Vec> = Vec::new(); + crate::frag::fragment([2; 16], &big, 200, &mut |b| { + frames.push(heapless::Vec::from_slice(b).unwrap()); + }) + .expect("fragment large obs"); + + for f in frames { + let _ = ex.on_event(Event::Frame(9, f), &mut store, 0); + } + assert!(store.contains(&id), "reassembled obs must be stored"); + assert_eq!(store.len(), 1); + } } diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs index a39595c..522c242 100644 --- a/tines-core/src/mem_store.rs +++ b/tines-core/src/mem_store.rs @@ -52,6 +52,18 @@ impl Default for MemStore { } } +#[cfg(feature = "alloc")] +impl MemStore { + /// Order two stored ids by their current age at `now` (youngest first). + fn age_cmp(&self, a: &ObservationId, b: &ObservationId, now: Tick) -> core::cmp::Ordering { + 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, + )) + } +} + #[cfg(feature = "alloc")] impl Store for MemStore { fn insert(&mut self, obs: Observation, now: Tick) -> Result { @@ -119,13 +131,7 @@ impl Store for MemStore { continue; } 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, - )) - }); + ids.sort_by(|&a, &b| self.age_cmp(&a, &b, now)); for id in ids.into_iter().skip(keep) { removed.insert(id); } @@ -151,13 +157,7 @@ impl Store for MemStore { .filter(|id| !removed.contains(id)) .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, - )) - }); + remaining.sort_by(|&a, &b| self.age_cmp(&a, &b, now)); let over = remaining.len() - max_count; for id in remaining.into_iter().rev().take(over) { removed.insert(id); @@ -344,4 +344,27 @@ mod tests { assert!(s.contains(&id)); assert_eq!(s.len(), 1); } + + #[test] + fn trim_keep_per_topic_uses_elapsed_age() { + let mut s = MemStore::new(); + // All stored with age_ms=0, but at different ticks: seq 1 is oldest + // (stored first), seq 3 youngest (stored last). + s.insert(obs([1; 16], 1, 0, None, b"t"), 1000).unwrap(); + s.insert(obs([1; 16], 2, 0, None, b"t"), 2000).unwrap(); + s.insert(obs([1; 16], 3, 0, None, b"t"), 3000).unwrap(); + let policy = TrimPolicy { + keep_per_topic: Some(2), + max_age_ms: None, + max_count: None, + }; + // At now=4000 the current ages are 3000/2000/1000 for seq 1/2/3, so + // the two youngest (seq 2 and 3) survive and seq 1 (oldest) drops. + assert_eq!(s.trim(&policy, 4000), 1); + let mut seqs = std::collections::BTreeSet::new(); + s.for_each_topic(b"t", 4000, &mut |o| { + seqs.insert(o.seq); + }); + assert_eq!(seqs, std::collections::BTreeSet::from_iter([2u64, 3])); + } }