refactor(core): dedupe trim/belief comparators and emit_one; add coverage tests

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
+                if latest_beats(obs, &cur) {
-                    || (obs.age_ms == cur.age_ms
-                        && (obs.observer, obs.seq) < (cur.observer, cur.seq))
-                {
                     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
+                    if latest_beats(obs, &cur) {
-                        || (obs.age_ms == cur.age_ms
-                            && (obs.observer, obs.seq) < (cur.observer, cur.seq))
-                    {
                         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];

tines-core/src/exchange.rs

@@ -263,10 +263,8 @@ fn emit_one<A: Clone, S: Store>(
     }
 
     while let Some(&front) = peer.send_queue.first() {
-        if !store.contains(&front) {
+        // A single `get` both checks presence and returns the age-adjusted
-            peer.send_queue.remove(0);
+        // observation; a missing id (trimmed elsewhere) is skipped.
-            continue;
-        }
         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<heapless::Vec<u8, { consts::MAX_FRAME }>> = 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);
+    }
 }

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(&current_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<bool, StoreError> {
@@ -119,13 +131,7 @@ impl Store for MemStore {
                     continue;
                 }
                 let mut ids = member_ids.clone();
-                ids.sort_by(|&a, &b| {
+                ids.sort_by(|&a, &b| self.age_cmp(&a, &b, 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);
                 }
@@ -151,13 +157,7 @@ impl Store for MemStore {
                 .filter(|id| !removed.contains(id))
                 .collect();
             if remaining.len() > max_count {
-                remaining.sort_by(|&a, &b| {
+                remaining.sort_by(|&a, &b| self.age_cmp(&a, &b, 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) {
                     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]));
+    }
 }