diff --git a/Cargo.lock b/Cargo.lock index dc69b524..fd48dcb1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,15 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "atomic-polyfill" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4" +dependencies = [ + "critical-section", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -134,6 +143,12 @@ dependencies = [ "libc", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + [[package]] name = "crypto-common" version = "0.1.7" @@ -155,6 +170,18 @@ dependencies = [ "subtle", ] +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "equivalent" version = "1.0.3" @@ -216,6 +243,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "hash32" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67" +dependencies = [ + "byteorder", +] + [[package]] name = "hash32" version = "0.3.1" @@ -231,13 +267,27 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "heapless" +version = "0.7.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" +dependencies = [ + "atomic-polyfill", + "hash32 0.2.1", + "rustc_version", + "serde", + "spin", + "stable_deref_trait", +] + [[package]] name = "heapless" version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bfb9eb618601c89945a70e254898da93b13be0388091d42117462b265bb3fad" dependencies = [ - "hash32", + "hash32 0.3.1", "serde", "stable_deref_trait", ] @@ -264,6 +314,15 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "memchr" version = "2.8.3" @@ -292,6 +351,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" dependencies = [ "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "heapless 0.7.17", "serde", ] @@ -424,6 +486,15 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.5" @@ -449,6 +520,18 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + [[package]] name = "serde" version = "1.0.229" @@ -479,6 +562,15 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -551,7 +643,7 @@ name = "tines-core" version = "0.3.0" dependencies = [ "blake2", - "heapless", + "heapless 0.8.0", "postcard", "serde", ] @@ -565,6 +657,15 @@ dependencies = [ "tines-core", ] +[[package]] +name = "tines-store" +version = "0.3.0" +dependencies = [ + "postcard", + "serde", + "tines-core", +] + [[package]] name = "toml_datetime" version = "1.1.2+spec-1.1.0" diff --git a/Cargo.toml b/Cargo.toml index 1af11d4d..f46241d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["tines-core", "tines-sim"] +members = ["tines-core", "tines-sim", "tines-store"] resolver = "2" [workspace.package] diff --git a/tines-store/Cargo.toml b/tines-store/Cargo.toml new file mode 100644 index 00000000..2b972be9 --- /dev/null +++ b/tines-store/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "tines-store" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Persistent append-only LogStore backend for tines: memory survives reboot with rebased ages and persisted release/sequence state." +authors.workspace = true +repository.workspace = true +readme.workspace = true +documentation = "https://docs.rs/tines-store" +keywords = ["robotics", "gossip", "delay-tolerant", "store", "persistent"] +categories = ["science::robotics"] + +[dependencies] +tines-core = { path = "../tines-core", version = "0.3.0" } +postcard = { version = "1", features = ["use-std"] } +serde = { version = "1", features = ["derive"] } diff --git a/tines-store/src/lib.rs b/tines-store/src/lib.rs new file mode 100644 index 00000000..de6d0771 --- /dev/null +++ b/tines-store/src/lib.rs @@ -0,0 +1,11 @@ +//! tines-store — persistent append-only `LogStore` backend. +//! +//! This crate is the host-side (std) persistence layer for tines. The `Store` +//! trait, `TrimPolicy`, and `StoreError` live in `tines-core` (which stays +//! `no_std` and I/O-free); everything durable lives here. + +#![forbid(unsafe_code)] + +pub mod log; + +pub use log::LogStore; diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs new file mode 100644 index 00000000..55a485cd --- /dev/null +++ b/tines-store/src/log.rs @@ -0,0 +1,361 @@ +//! Append-only observation log backing [`LogStore`]. +//! +//! The log is a single file of self-describing, length-prefixed records: +//! +//! ```text +//! [u32 little-endian length][postcard bytes of the record] +//! ``` +//! +//! Record kinds: +//! +//! * `Record::Obs` — a stored [`Observation`] (deduped on `id` at load). +//! * `Record::Trim` — the id of an observation removed by `trim`; the last +//! event for an id wins, so a re-inserted (new `Obs` record) id survives. +//! * `Record::Meta` — the current `next_seq`, the last `now` seen (the clock +//! marker), and the full set of released ids. The last `Meta` on disk wins, +//! so it always reflects the latest flushed state. +//! +//! On load the index is rebuilt by replaying the log, then every age is +//! rebased to load time: the time elapsed since the last persisted clock marker +//! is folded into `age_ms` and `stored_at` is set to the load `now`, so ages +//! continue across a reboot instead of freezing at their pre-shutdown value. + +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::Path; + +use serde::{Deserialize, Serialize}; +use tines_core::{Observation, ObservationId, Store, StoreError, Tick, TrimPolicy}; + +/// Current (age-adjusted) age of a stored observation at tick `now`. +fn current_age(obs: &Observation, stored_at: Tick, now: Tick) -> u64 { + obs.age_ms.saturating_add(now.saturating_sub(stored_at)) +} + +#[derive(Serialize, Deserialize)] +enum Record { + Obs { + id: ObservationId, + obs: Box, + }, + Trim { + id: ObservationId, + }, + Meta { + next_seq: u64, + last_clock: Tick, + released: Vec, + }, +} + +#[derive(Clone)] +struct Entry { + obs: Observation, + stored_at: Tick, + released: bool, +} + +/// A persistent, append-only [`Store`] backed by a single observation log. +/// +/// See the [module documentation](crate) for the on-disk record format and the +/// rebase-on-load behavior. +pub struct LogStore { + file: File, + index: BTreeMap, + next_seq: u64, + now: Tick, + last_clock: Option, +} + +impl LogStore { + /// Open (creating if necessary) the log at `path` and rebuild the index. + /// + /// `now` is the current tick and is used both as the base for rebasing ages + /// and as the store's current clock. `first_seq` is the next sequence + /// number to hand out; the persisted `next_seq` (if any) is honored when it + /// is larger, so a resumed run never reuses a sequence. + pub fn new(path: &Path, now: Tick, first_seq: u64) -> Result { + let file = OpenOptions::new() + .read(true) + .create(true) + .append(true) + .open(path) + .map_err(|_| StoreError::Backend)?; + + let mut index: BTreeMap = BTreeMap::new(); + let mut dead: HashSet = HashSet::new(); + let mut next_seq = first_seq; + let mut last_clock: Option = None; + + let mut file = file; + while let Some(rec) = read_record(&mut file)? { + match rec { + Record::Obs { id, obs } => { + // A preceding Trim for this id is cancelled by the + // re-insert, so revive it; otherwise dedupe to the + // first (oldest) occurrence. + dead.remove(&id); + index.entry(id).or_insert(Entry { + obs: *obs, + stored_at: now, + released: false, + }); + } + Record::Trim { id } => { + dead.insert(id); + index.remove(&id); + } + Record::Meta { + next_seq: ns, + last_clock: lc, + released, + } => { + next_seq = ns.max(next_seq); + last_clock = Some(lc); + for rid in released { + if let Some(entry) = index.get_mut(&rid) { + entry.released = true; + } + } + } + } + } + + for id in dead { + index.remove(&id); + } + + // Fold the pre-reboot elapsed time into every age and reset the stored + // base to load time. When there is no clock marker yet (a fresh store) + // or the clock moved backwards, nothing is folded (saturating). + let elapsed = now.saturating_sub(last_clock.unwrap_or(now)); + for entry in index.values_mut() { + entry.obs.age_ms = entry.obs.age_ms.saturating_add(elapsed); + entry.stored_at = now; + } + + Ok(LogStore { + file, + index, + next_seq, + now, + last_clock, + }) + } + + /// The next sequence number this store will hand out. + pub fn next_seq(&self) -> u64 { + self.next_seq + } + + /// Persist the current index state (released flags, `next_seq`, and the + /// clock marker) and the observation log to disk. + pub fn flush(&mut self) -> Result<(), StoreError> { + let mut released: Vec = self + .index + .iter() + .filter(|(_, entry)| entry.released) + .map(|(id, _)| *id) + .collect(); + released.sort_unstable(); + + let record = Record::Meta { + next_seq: self.next_seq, + last_clock: self.now, + released, + }; + write_record(&mut self.file, &record)?; + self.file.sync_all().map_err(|_| StoreError::Backend)?; + self.last_clock = Some(self.now); + Ok(()) + } + + /// Append a single length-prefixed record to the log. + fn append(&mut self, record: &Record) -> Result<(), StoreError> { + write_record(&mut self.file, record) + } +} + +/// Write a record as `[u32 length][postcard bytes]`. +fn write_record(file: &mut File, record: &Record) -> Result<(), StoreError> { + let bytes = postcard::to_stdvec(record).map_err(|_| StoreError::Backend)?; + let len = bytes.len() as u32; + file.write_all(&len.to_le_bytes()) + .map_err(|_| StoreError::Backend)?; + file.write_all(&bytes).map_err(|_| StoreError::Backend)?; + Ok(()) +} + +/// Read one length-prefixed record, returning `Ok(None)` at end of file. +fn read_record(file: &mut File) -> Result, StoreError> { + let mut len_buf = [0u8; 4]; + if file.read_exact(&mut len_buf).is_err() { + // Clean end-of-file: no length prefix left to read. + return Ok(None); + } + let len = u32::from_le_bytes(len_buf) as usize; + let mut buf = vec![0u8; len]; + file.read_exact(&mut buf).map_err(|_| StoreError::Backend)?; + let record = postcard::from_bytes(&buf).map_err(|_| StoreError::Backend)?; + Ok(Some(record)) +} + +impl Store for LogStore { + fn insert(&mut self, obs: Observation, now: Tick) -> Result { + let id = obs.id(); + if self.index.contains_key(&id) { + // Dedupe: keep the first-stored entry and its age lineage. + return Ok(false); + } + self.now = now; + let record = Record::Obs { + id, + obs: Box::new(obs.clone()), + }; + self.append(&record)?; + self.index.insert( + id, + Entry { + obs, + stored_at: now, + released: false, + }, + ); + self.next_seq += 1; + Ok(true) + } + + fn contains(&self, id: &ObservationId) -> bool { + self.index.contains_key(id) + } + + fn get(&self, id: &ObservationId, now: Tick) -> Option { + let entry = self.index.get(id)?; + let elapsed = now.saturating_sub(entry.stored_at); + Some(entry.obs.aged(elapsed)) + } + + fn len(&self) -> usize { + self.index.len() + } + + fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)) { + for id in self.index.keys() { + f(id); + } + } + + fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) { + for (id, entry) in self.index.iter() { + if entry.obs.to.is_none() { + f(id); + } + } + } + + fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) { + for (id, entry) in self.index.iter() { + if entry.obs.to.is_some() && !entry.released { + f(id); + } + } + } + + fn release(&mut self, id: &ObservationId) { + if let Some(entry) = self.index.get_mut(id) { + entry.released = true; + } + } + + fn is_released(&self, id: &ObservationId) -> bool { + self.index + .get(id) + .map(|entry| entry.released) + .unwrap_or(false) + } + + fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) { + for entry in self.index.values() { + if entry.obs.topic.starts_with(prefix) { + let elapsed = now.saturating_sub(entry.stored_at); + f(&entry.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, entry) in self.index.iter() { + if !removed.contains(id) { + groups + .entry(entry.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| self.age_cmp(&a, &b, 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, entry) in self.index.iter() { + if !removed.contains(id) && current_age(&entry.obs, entry.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 + .index + .keys() + .copied() + .filter(|id| !removed.contains(id)) + .collect(); + if remaining.len() > max_count { + 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); + } + } + } + + for id in &removed { + // `Store::trim` cannot surface I/O errors, so best-effort: the + // in-memory index is updated regardless of the append result. + let _ = self.append(&Record::Trim { id: *id }); + self.index.remove(id); + } + removed.len() + } +} + +impl LogStore { + /// 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.index[a].obs, self.index[a].stored_at, now).cmp(¤t_age( + &self.index[b].obs, + self.index[b].stored_at, + now, + )) + } +} diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs new file mode 100644 index 00000000..b049ebcb --- /dev/null +++ b/tines-store/tests/persistence.rs @@ -0,0 +1,116 @@ +//! Persistence tests for the append-only [`LogStore`] backend. +//! +//! Each test runs against its own unique file under the temp directory so the +//! parallel test runner cannot collide on disk state. + +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; + +use tines_core::{Observation, Payload, Store, Topic}; +use tines_store::LogStore; + +static COUNTER: AtomicU64 = AtomicU64::new(0); + +fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option) -> Observation { + Observation { + observer, + seq, + age_ms: age, + observed_at: at, + topic: Topic::from_slice(b"door/3").unwrap(), + payload: Payload::from_slice(b"open").unwrap(), + to: None, + lifetime_ms: None, + } +} + +fn store_path(tag: &str) -> PathBuf { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let key = n as u128 * 1_000_000 + nanos; + let mut p = std::env::temp_dir(); + p.push(format!( + "tines-store-{}-{}-{}.log", + tag, + std::process::id(), + key + )); + p +} + +#[test] +fn reopen_preserves_store() { + let path = store_path("reopen"); + + { + let mut s = LogStore::new(&path, 0, 1).expect("open fresh store"); + let a = obs([1; 16], 1, 0, Some(10)); + let b = obs([2; 16], 2, 0, Some(20)); + let ib = b.id(); + + assert!(s.insert(a, 1000).unwrap()); + assert!(s.insert(b, 1001).unwrap()); + assert!(!s.is_released(&ib)); + s.release(&ib); + assert!(s.is_released(&ib)); + assert_eq!(s.next_seq(), 3); + s.flush().expect("flush before drop"); + } + + { + let s = LogStore::new(&path, 2000, 3).expect("reopen store"); + let a = obs([1; 16], 1, 0, Some(10)); + let b = obs([2; 16], 2, 0, Some(20)); + assert!(s.contains(&a.id())); + assert!(s.contains(&b.id())); + assert!(!s.is_released(&a.id())); + assert!(s.is_released(&b.id())); + assert_eq!(s.next_seq(), 3); + } +} + +#[test] +fn ages_rebase_on_load() { + let path = store_path("ages"); + + { + let mut s = LogStore::new(&path, 1000, 1).unwrap(); + let o = obs([1; 16], 1, 500, Some(10)); + assert!(s.insert(o, 1000).unwrap()); + s.flush().unwrap(); + } + + { + let s = LogStore::new(&path, 5000, 1).unwrap(); + let o = obs([1; 16], 1, 500, Some(10)); + let id = o.id(); + // 500ms original age plus the 4000ms that elapsed while the store was + // closed (5000 - 1000). A store that forgot to fold the downtime in + // would report only the frozen 500ms. + let g = s.get(&id, 5000).unwrap(); + assert_eq!(g.age_ms, 4500); + } +} + +#[test] +fn next_seq_persists() { + let path = store_path("seq"); + + { + let mut s = LogStore::new(&path, 0, 1).unwrap(); + for i in 0..3u64 { + let o = obs([1; 16], 1 + i, 0, Some(10)); + assert!(s.insert(o, 1 + i).unwrap()); + } + assert_eq!(s.next_seq(), 4); + s.flush().unwrap(); + } + + { + let s = LogStore::new(&path, 100, 1).unwrap(); + assert_eq!(s.next_seq(), 4); + } +}