From 7ece265dbd16b77a8f61f10dd97800c680c622c1 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Fri, 16 Oct 2026 18:46:31 -0700 Subject: [PATCH] feat(core): Store trait, TrimPolicy, MemStore --- tines-core/src/lib.rs | 7 + tines-core/src/mem_store.rs | 334 ++++++++++++++++++++++++++++++++++++ tines-core/src/store.rs | 83 +++++++++ 3 files changed, 424 insertions(+) create mode 100644 tines-core/src/mem_store.rs create mode 100644 tines-core/src/store.rs diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index ae8ce2f..a3443f1 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -6,10 +6,17 @@ extern crate alloc; pub mod consts; pub mod observation; +pub mod store; +#[cfg(feature = "alloc")] +pub mod mem_store; pub mod wire; pub use observation::{Observation, NodeId, ObservationId, Payload, Tick, Topic}; +pub use store::{Store, TrimPolicy, StoreError}; pub use wire::{Frame, WireError}; +#[cfg(feature = "alloc")] +pub use mem_store::MemStore; + #[cfg(test)] extern crate std; diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs new file mode 100644 index 0000000..9f59a33 --- /dev/null +++ b/tines-core/src/mem_store.rs @@ -0,0 +1,334 @@ +//! In-memory observation store backed by a `BTreeMap` keyed by content-derived +//! `ObservationId`. +//! +//! `MemStore` implements [`Store`](crate::store::Store). It is unbounded, so it +//! never returns `StoreError::Full`; allocation is therefore required and the +//! module is gated behind the `alloc` feature. + +#[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")] +struct Held { + obs: Observation, + stored_at: Tick, +} + +/// An unbounded [`Store`] backed by a `BTreeMap` keyed by content-derived +/// [`ObservationId`]. +/// +/// Because a `BTreeMap` has no capacity limit, `MemStore` never returns +/// `StoreError::Full`. It is the only store type that requires the `alloc` +/// feature; the [`Store`] trait, [`TrimPolicy`], and [`StoreError`] are defined +/// in [`store`](crate::store) and compile without allocation. +#[cfg(feature = "alloc")] +pub struct MemStore { + map: BTreeMap, +} + +#[cfg(feature = "alloc")] +impl MemStore { + /// Create an empty store. + pub fn new() -> Self { + MemStore { + map: BTreeMap::new(), + } + } +} + +#[cfg(feature = "alloc")] +impl Default for MemStore { + fn default() -> Self { + Self::new() + } +} + +#[cfg(feature = "alloc")] +impl Store for MemStore { + fn insert(&mut self, obs: Observation, now: Tick) -> Result { + let id = obs.id(); + if self.map.contains_key(&id) { + // Dedupe: keep the first-stored entry and its age lineage. + return Ok(false); + } + self.map.insert(id, Held { obs, stored_at: now }); + Ok(true) + } + + fn contains(&self, id: &ObservationId) -> bool { + self.map.contains_key(id) + } + + fn get(&self, id: &ObservationId, now: Tick) -> Option { + let held = self.map.get(id)?; + let elapsed = now.saturating_sub(held.stored_at); + Some(held.obs.aged(elapsed)) + } + + fn len(&self) -> usize { + self.map.len() + } + + fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)) { + for id in self.map.keys() { + 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) { + let elapsed = now.saturating_sub(held.stored_at); + f(&held.obs.aged(elapsed)); + } + } + } + + fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize { + let mut removed: BTreeSet = BTreeSet::new(); + + // (a) keep_per_topic: per exact topic, drop everything but the `N` + // youngest (lowest current age). + if let Some(keep) = policy.keep_per_topic { + let mut groups: BTreeMap<&[u8], Vec> = BTreeMap::new(); + for (id, held) in self.map.iter() { + if !removed.contains(id) { + groups.entry(held.obs.topic.as_slice()).or_default().push(*id); + } + } + for member_ids in groups.values() { + if member_ids.len() <= keep { + 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)) + }); + for id in ids.into_iter().skip(keep) { + removed.insert(id); + } + } + } + + // (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 + { + removed.insert(*id); + } + } + } + + // (c) max_count: drop the oldest (highest current age) until the number + // of remaining (untrimmed) observations is at most `max_count`. + if let Some(max_count) = policy.max_count { + let mut remaining: Vec = self + .map + .keys() + .copied() + .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)) + }); + let over = remaining.len() - max_count; + for id in remaining.into_iter().rev().take(over) { + removed.insert(id); + } + } + } + + self.map.retain(|id, _| !removed.contains(id)); + removed.len() + } +} + +#[cfg(all(test, feature = "alloc"))] +mod tests { + use super::MemStore; + use crate::observation::{Observation, Payload, Topic}; + use crate::store::{Store, TrimPolicy}; + + fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option, topic: &[u8]) -> Observation { + Observation { + observer, + seq, + age_ms: age, + observed_at: at, + topic: Topic::from_slice(topic).unwrap(), + payload: Payload::from_slice(b"").unwrap(), + } + } + + #[test] + fn insert_then_contains_and_len() { + let mut s = MemStore::new(); + let o = obs([1; 16], 1, 0, None, b"door/1"); + let id = o.id(); + assert!(s.insert(o, 1000).unwrap()); + assert!(s.contains(&id)); + assert_eq!(s.len(), 1); + } + + #[test] + fn duplicate_id_is_not_reinserted() { + let mut s = MemStore::new(); + let o1 = obs([1; 16], 1, 100, None, b"door/1"); + let id = o1.id(); + assert!(s.insert(o1, 1000).unwrap()); + // Same (observer, seq, topic, payload), different age -> deduped. + let o2 = obs([1; 16], 1, 999, None, b"door/1"); + assert_eq!(id, o2.id()); + assert!(!s.insert(o2, 2000).unwrap()); + assert_eq!(s.len(), 1); + // get() still returns the first-stored age lineage. + let g = s.get(&id, 1500).unwrap(); + assert_eq!(g.age_ms, 600); + } + + #[test] + fn get_adjusts_age_by_elapsed() { + let mut s = MemStore::new(); + let o = obs([1; 16], 1, 100, None, b"door/1"); + let id = o.id(); + s.insert(o, 1000).unwrap(); + let g = s.get(&id, 1500).unwrap(); + assert_eq!(g.age_ms, 600); + } + + #[test] + fn get_with_backwards_tick_saturates() { + let mut s = MemStore::new(); + let o = obs([1; 16], 1, 100, None, b"door/1"); + let id = o.id(); + s.insert(o, 1000).unwrap(); + let g = s.get(&id, 500).unwrap(); + assert_eq!(g.age_ms, 100); + } + + #[test] + fn for_each_topic_prefix_matches() { + let mut s = MemStore::new(); + s.insert(obs([1; 16], 1, 0, None, b"door/1"), 0).unwrap(); + s.insert(obs([1; 16], 2, 0, None, b"door/2"), 0).unwrap(); + s.insert(obs([1; 16], 3, 0, None, b"wall/1"), 0).unwrap(); + let mut count = 0; + s.for_each_topic(b"door/", 0, &mut |_o| { + count += 1; + }); + assert_eq!(count, 2); + } + + #[test] + fn for_each_topic_exact_empty_prefix_is_all() { + let mut s = MemStore::new(); + s.insert(obs([1; 16], 1, 0, None, b"door/1"), 0).unwrap(); + s.insert(obs([1; 16], 2, 0, None, b"door/2"), 0).unwrap(); + s.insert(obs([1; 16], 3, 0, None, b"wall/1"), 0).unwrap(); + let mut count = 0; + s.for_each_topic(b"", 0, &mut |_o| { + count += 1; + }); + assert_eq!(count, s.len()); + } + + #[test] + 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(); + } + let policy = TrimPolicy { + keep_per_topic: Some(2), + max_age_ms: None, + max_count: None, + }; + assert_eq!(s.trim(&policy, 0), 3); + let mut ages = std::collections::BTreeSet::new(); + s.for_each_topic(b"t", 0, &mut |o| { + ages.insert(o.age_ms); + }); + assert_eq!(ages, std::collections::BTreeSet::from_iter([10u64, 20])); + } + + #[test] + fn trim_max_age_drops_old() { + let mut s = MemStore::new(); + s.insert(obs([1; 16], 1, 10, None, b"t"), 0).unwrap(); + s.insert(obs([1; 16], 2, 10_000, None, b"t"), 0).unwrap(); + let policy = TrimPolicy { + keep_per_topic: None, + max_age_ms: Some(1000), + max_count: None, + }; + assert_eq!(s.trim(&policy, 0), 1); + let mut found = false; + s.for_each_topic(b"t", 0, &mut |o| { + assert_eq!(o.age_ms, 10); + found = true; + }); + assert!(found); + } + + #[test] + 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(); + } + let policy = TrimPolicy { + keep_per_topic: None, + max_age_ms: None, + max_count: Some(4), + }; + assert_eq!(s.trim(&policy, 0), 6); + assert_eq!(s.len(), 4); + let mut ages = std::collections::BTreeSet::new(); + 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])); + } + + #[test] + fn trimmed_obs_can_be_reinserted() { + let mut s = MemStore::new(); + let o = obs([1; 16], 1, 10, None, b"t"); + let id = o.id(); + assert!(s.insert(o, 0).unwrap()); + assert_eq!(s.len(), 1); + let policy = TrimPolicy { + keep_per_topic: None, + max_age_ms: Some(1), + max_count: None, + }; + assert_eq!(s.trim(&policy, 0), 1); + assert_eq!(s.len(), 0); + assert!(!s.contains(&id)); + // Trimmed then re-received is accepted. + let o2 = obs([1; 16], 1, 20, None, b"t"); + assert_eq!(id, o2.id()); + assert!(s.insert(o2, 0).unwrap()); + assert!(s.contains(&id)); + assert_eq!(s.len(), 1); + } +} diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs new file mode 100644 index 0000000..5e6fc3d --- /dev/null +++ b/tines-core/src/store.rs @@ -0,0 +1,83 @@ +//! Shared-memory observation store. +//! +//! This module defines the [`Store`] trait plus its [`TrimPolicy`] and +//! [`StoreError`] types. Everything here is `#![no_std]`-clean and does **not** +//! require the `alloc` feature — only concrete backends (see +//! [`mem_store`](crate::mem_store)) need allocation, so the trait, policy, and +//! error types remain available on `no_std` targets as well. + +use crate::observation::{Observation, ObservationId, Tick}; + +/// Ejection-policy knobs consumed by [`Store::trim`]. +/// +/// Rules are applied in the order listed here: per-topic retention, then a +/// global age cutoff, then a global count cap. Any field left as `None` is +/// ignored. +#[derive(Clone, Debug, Default)] +pub struct TrimPolicy { + /// Keep the `N` youngest (lowest current age) observations per exact topic. + pub keep_per_topic: Option, + /// Drop any observation whose current age (in ms) exceeds this value. + pub max_age_ms: Option, + /// Drop the oldest observations until the total count is at most this value. + pub max_count: Option, +} + +/// Errors returned by [`Store`] operations. +#[derive(Debug, PartialEq, Eq)] +pub enum StoreError { + /// The store is full and cannot accept new observations. + Full, + /// A backend-internal failure. + Backend, +} + +/// A content-addressed, age-aware collection of [`Observation`]s. +/// +/// Identity is the content-derived [`ObservationId`]; two observations that +/// differ only in age or `observed_at` share an id and dedupe on insert. Ages +/// are reported as *current* ages, advanced by the elapsed ticks since each +/// observation was stored. +#[allow(clippy::len_without_is_empty)] +pub trait Store { + /// Insert an observation. + /// + /// Returns `Ok(true)` if it was newly inserted, or `Ok(false)` if an + /// observation with the same id already exists (in which case the existing + /// entry — and its stored age lineage — is left untouched). + fn insert(&mut self, obs: Observation, now: Tick) -> Result; + + /// Returns `true` if an observation with `id` is currently stored. + fn contains(&self, id: &ObservationId) -> bool; + + /// Returns the observation with `id`, with `age_ms` advanced by + /// `now - stored_at` (saturating at 0 elapsed when `now < stored_at`). + fn get(&self, id: &ObservationId, now: Tick) -> Option; + + /// Number of observations currently stored. + fn len(&self) -> usize; + + /// Visit every stored observation id. + fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)); + + /// 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), + ); + + /// Apply [`TrimPolicy`] rules in order (per-topic, age, count) and return + /// the number of observations removed. Trimmed ids are not tombstoned: + /// re-inserting them later is accepted. + fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize; +} + +/// Current (age-adjusted) age of a stored observation at tick `now`. +#[inline] +#[cfg(feature = "alloc")] +pub(crate) fn current_age(obs: &Observation, stored_at: Tick, now: Tick) -> u64 { + obs.age_ms.saturating_add(now.saturating_sub(stored_at)) +}