From 9a055b5b5722948546fea0a4a294dfbdab553b07 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Sun, 18 Oct 2026 05:09:07 -0700 Subject: [PATCH] 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 --- tines-core/src/exchange.rs | 90 +++++++++++++++++++++++++++++------ tines-core/src/frag.rs | 52 +++++++++++++------- tines-core/src/mem_store.rs | 8 ++++ tines-core/src/observation.rs | 11 +++-- tines-core/src/store.rs | 10 ++++ 5 files changed, 134 insertions(+), 37 deletions(-) diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 47113b5..fde0840 100644 --- a/tines-core/src/exchange.rs +++ b/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( ) -> 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, + } + + impl Store for CountingStore { + fn insert(&mut self, obs: Observation, now: Tick) -> Result { + self.inner.insert(obs, now) + } + fn contains(&self, id: &ObservationId) -> bool { + self.inner.contains(id) + } + fn get(&self, id: &ObservationId, now: Tick) -> Option { + 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" + ); + } } diff --git a/tines-core/src/frag.rs b/tines-core/src/frag.rs index 93f327b..aca0e4b 100644 --- a/tines-core/src/frag.rs +++ b/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, - 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]; } } diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs index a26bbd8..0034558 100644 --- a/tines-core/src/mem_store.rs +++ b/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) { diff --git a/tines-core/src/observation.rs b/tines-core/src/observation.rs index 20d5d95..180ae80 100644 --- a/tines-core/src/observation.rs +++ b/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); diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs index ab77ea6..f93adce 100644 --- a/tines-core/src/store.rs +++ b/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));