refactor(core): scan only addressed ids for custody; share frag plan

- Add Store::for_each_addressed_id (default: every id; MemStore filters)
  so custody selection no longer clones every broadcast observation on
  each poll.
- Factor fragment sizing into frag::plan and MAX_FRAGMENTS, used by both
  frag::fragment and the exchange, and by the reassembler's bounds.
- Document `to` in the Observation::id hash input, matching the
  addressed-delivery plan.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tines-core/src/exchange.rs

@@ -17,12 +17,6 @@ use crate::{
 };
 use heapless::Vec;
 
-/// The maximum number of fragments an observation may be split into.
-///
-/// Matches [`crate::fragment`]: the reassembler tracks up to 64 indices via a
-/// bitset, so `count` is capped at 64 on both ends.
-const MAX_FRAGMENTS: usize = 64;
-
 /// Worst-case bytes a [`Frame::Summary`] adds around its bloom bits: the
 /// variant tag (1), `from` (16), `salt` as a varint (up to 5), and the bits
 /// length as a varint (up to 2).
@@ -590,7 +584,7 @@ fn select_addressed<S: Store, A: PartialEq>(
 ) -> Option<(ObservationId, Observation)> {
     let mut ahead: Option<(ObservationId, Observation)> = None;
     let mut wrapped: Option<(ObservationId, Observation)> = None;
-    store.for_each_id(&mut |id| {
+    store.for_each_addressed_id(&mut |id| {
         let is_ahead = after.is_none_or(|a| *id > a);
         let beats = |pick: &Option<(ObservationId, Observation)>| {
             pick.as_ref().is_none_or(|(best, _)| id < best)
@@ -683,14 +677,7 @@ fn encode_frame(
     };
     let mut obs_buf = [0u8; consts::MAX_FRAME * 2];
     let obs_bytes = postcard::to_slice(&frozen, &mut obs_buf).ok()?;
-    let chunk = mtu.saturating_sub(crate::frag::FRAG_OVERHEAD);
-    if chunk == 0 {
-        return None;
-    }
-    let count = obs_bytes.len().div_ceil(chunk);
-    if count == 0 || count > MAX_FRAGMENTS {
-        return None;
-    }
+    let (chunk, count) = crate::frag::plan(obs_bytes.len(), mtu)?;
     let index = match resume {
         Some(f) if f.len != obs_bytes.len() || usize::from(f.count) != count => return None,
         Some(f) => f.next,
@@ -1803,4 +1790,77 @@ mod tests {
             }
         }
     }
+
+    /// A `MemStore` that counts `get` calls.
+    struct CountingStore {
+        inner: MemStore,
+        gets: core::cell::Cell<usize>,
+    }
+
+    impl Store for CountingStore {
+        fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, crate::StoreError> {
+            self.inner.insert(obs, now)
+        }
+        fn contains(&self, id: &ObservationId) -> bool {
+            self.inner.contains(id)
+        }
+        fn get(&self, id: &ObservationId, now: Tick) -> Option<Observation> {
+            self.gets.set(self.gets.get() + 1);
+            self.inner.get(id, now)
+        }
+        fn len(&self) -> usize {
+            self.inner.len()
+        }
+        fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_id(f)
+        }
+        fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_broadcast_id(f)
+        }
+        fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_addressed_id(f)
+        }
+        fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) {
+            self.inner.for_each_topic(prefix, now, f)
+        }
+        fn trim(&mut self, policy: &crate::TrimPolicy, now: Tick) -> usize {
+            self.inner.trim(policy, now)
+        }
+    }
+
+    #[test]
+    fn custody_scan_skips_broadcast_observations() {
+        let mut ex = Exchange::new(ME);
+        let mut store = CountingStore {
+            inner: MemStore::new(),
+            gets: core::cell::Cell::new(0),
+        };
+        let mut theirs = Bloom::new(0);
+        for seq in 0..200 {
+            let o = obs([3; 16], seq, 0, None, b"t", b"x");
+            theirs.insert(&o.id());
+            store.insert(o, 0).unwrap();
+        }
+        ex.on_event(Event::Contact(1), &mut store, 0);
+        // The peer already holds every broadcast, so there is nothing to send.
+        let sum = Frame::Summary {
+            from: [2; 16],
+            salt: 0,
+            bits: heapless::Vec::from_slice(theirs.bits()).unwrap(),
+        };
+        ex.on_event(Event::Frame(1, frame_bytes(&sum)), &mut store, 0);
+        let mut buf = [0u8; consts::MAX_FRAME];
+        ex.next_frame(&store, consts::MAX_FRAME, 0, &mut buf); // our summary
+        store.gets.set(0);
+        for _ in 0..10 {
+            assert!(ex
+                .next_frame(&store, consts::MAX_FRAME, 0, &mut buf)
+                .is_none());
+        }
+        assert_eq!(
+            store.gets.get(),
+            0,
+            "custody must not read broadcast observations"
+        );
+    }
 }

tines-core/src/frag.rs

@@ -7,6 +7,26 @@ use heapless::Vec;
 /// Header allowance per fragment (bytes), pinned.
 pub const FRAG_OVERHEAD: usize = 64;
 
+/// The most fragments one observation may be split into: the reassembler
+/// tracks received indices in a `u64` bitset.
+pub(crate) const MAX_FRAGMENTS: usize = 64;
+
+/// Splits an encoded observation of `len` bytes for a link of `mtu` bytes,
+/// returning `(chunk, count)`: the data bytes per fragment and the number of
+/// fragments. `None` when `mtu` leaves no room for data or `count` would
+/// exceed [`MAX_FRAGMENTS`].
+pub(crate) fn plan(len: usize, mtu: usize) -> Option<(usize, usize)> {
+    let chunk = mtu.saturating_sub(FRAG_OVERHEAD);
+    if chunk == 0 {
+        return None;
+    }
+    let count = len.div_ceil(chunk);
+    if count == 0 || count > MAX_FRAGMENTS {
+        return None;
+    }
+    Some((chunk, count))
+}
+
 /// Encodes `obs` as one Obs frame if it fits in `mtu`, else as N Frag frames.
 /// Returns the number of frames; writes each via `emit(frame_bytes)`.
 pub fn fragment(
@@ -32,15 +52,7 @@ pub fn fragment(
     let mut obs_buf = [0u8; consts::MAX_FRAME * 2];
     let obs_bytes = postcard::to_slice(obs, &mut obs_buf).map_err(|_| WireError::TooLarge)?;
 
-    let chunk = mtu.saturating_sub(FRAG_OVERHEAD);
-    if chunk == 0 {
-        return Err(WireError::TooLarge);
-    }
-    #[allow(clippy::manual_div_ceil)]
-    let count = (obs_bytes.len() + chunk - 1) / chunk;
-    if count > 64 {
-        return Err(WireError::TooLarge);
-    }
+    let (chunk, count) = plan(obs_bytes.len(), mtu).ok_or(WireError::TooLarge)?;
 
     for i in 0..count {
         let start = i * chunk;
@@ -68,8 +80,8 @@ pub struct Reassembler {
     received: u64,
     started_at: Tick,
     buf: Vec<u8, { consts::MAX_FRAME * 2 }>,
-    offs: [u16; 64],
-    lens: [u16; 64],
+    offs: [u16; MAX_FRAGMENTS],
+    lens: [u16; MAX_FRAGMENTS],
 }
 
 impl Reassembler {
@@ -80,8 +92,8 @@ impl Reassembler {
             received: 0,
             started_at: 0,
             buf: heapless::Vec::new(),
-            offs: [u16::MAX; 64],
-            lens: [0u16; 64],
+            offs: [u16::MAX; MAX_FRAGMENTS],
+            lens: [0u16; MAX_FRAGMENTS],
         }
     }
 
@@ -96,7 +108,11 @@ impl Reassembler {
         // 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.
-        if count == 0 || count > 64 || (index as usize) >= 64 || index >= count {
+        if count == 0
+            || usize::from(count) > MAX_FRAGMENTS
+            || usize::from(index) >= MAX_FRAGMENTS
+            || index >= count
+        {
             return None;
         }
 
@@ -112,8 +128,8 @@ impl Reassembler {
             self.received = 0;
             self.started_at = now;
             self.buf.clear();
-            self.offs = [u16::MAX; 64];
-            self.lens = [0u16; 64];
+            self.offs = [u16::MAX; MAX_FRAGMENTS];
+            self.lens = [0u16; MAX_FRAGMENTS];
         }
 
         // 3. Append the fragment data to the flat buffer in arrival order and
@@ -181,8 +197,8 @@ impl Reassembler {
         self.received = 0;
         self.started_at = 0;
         self.buf.clear();
-        self.offs = [u16::MAX; 64];
-        self.lens = [0u16; 64];
+        self.offs = [u16::MAX; MAX_FRAGMENTS];
+        self.lens = [0u16; MAX_FRAGMENTS];
     }
 }
 

tines-core/src/mem_store.rs

@@ -110,6 +110,14 @@ impl Store for MemStore {
         }
     }
 
+    fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+        for (id, held) in self.map.iter() {
+            if held.obs.to.is_some() {
+                f(id);
+            }
+        }
+    }
+
     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) {

tines-core/src/observation.rs

@@ -32,11 +32,14 @@ pub struct Observation {
 }
 
 impl Observation {
-    /// Content-derived identity: Blake2s-256 over
-    /// `(observer, seq, topic_len, topic, payload_len, payload)`, first 16 bytes.
+    /// Content-derived identity: the first 16 bytes of Blake2s-256 over
+    /// `observer ‖ seq_le ‖ topic_len_le(u16) ‖ topic ‖ payload_len_le(u16) ‖
+    /// payload ‖ to_marker`, where `to_marker` is `0x00` for broadcast
+    /// (`to = None`) and `0x01 ‖ node_id` for `to = Some(node_id)`.
     ///
-    /// The hash ignores `age_ms` and `observed_at`, so two observations that
-    /// differ only in age/time are still treated as the same key.
+    /// The hash ignores `age_ms`, `observed_at` and `lifetime_ms`, so two
+    /// observations that differ only in age, time or lifetime are still
+    /// treated as the same key.
     pub fn id(&self) -> ObservationId {
         let mut hasher = Blake2s256::new();
         hasher.update(self.observer);

tines-core/src/store.rs

@@ -64,6 +64,16 @@ pub trait Store {
     /// (`to == None`), i.e. broadcasts. Order is unspecified.
     fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId));
 
+    /// Visit the ids of observations addressed to someone (`to == Some`).
+    /// Order is unspecified.
+    ///
+    /// The default visits every id, which is correct since callers still
+    /// check `to`; backends that can filter cheaply should override it so the
+    /// exchange does not read every broadcast on each poll.
+    fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+        self.for_each_id(f)
+    }
+
     /// 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));