diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index 55a485cd..01ac8ee6 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -8,17 +8,22 @@ //! //! Record kinds: //! -//! * `Record::Obs` — a stored [`Observation`] (deduped on `id` at load). +//! * `Record::Obs` — a stored [`Observation`] together with the tick it was +//! inserted at (`stored_at`). 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. +//! * `Record::Meta` — the current `next_seq` 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. +//! rebased to load time: each entry's `age_ms` is folded by the time elapsed +//! since its own `stored_at` (the tick it was inserted, carried on the `Obs` +//! record). Anchoring the fold to each entry's insert tick rather than to a +//! single persisted clock marker makes the reopen fold exact for idle stores +//! and for observations inserted before the last flush — ages continue across a +//! reboot instead of freezing at (and under-counting from) their pre-shutdown +//! value. use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::fs::{File, OpenOptions}; @@ -38,13 +43,13 @@ enum Record { Obs { id: ObservationId, obs: Box, + stored_at: Tick, }, Trim { id: ObservationId, }, Meta { next_seq: u64, - last_clock: Tick, released: Vec, }, } @@ -65,7 +70,6 @@ pub struct LogStore { index: BTreeMap, next_seq: u64, now: Tick, - last_clock: Option, } impl LogStore { @@ -86,18 +90,19 @@ impl LogStore { 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 } => { + Record::Obs { id, obs, stored_at } => { // A preceding Trim for this id is cancelled by the // re-insert, so revive it; otherwise dedupe to the - // first (oldest) occurrence. + // first (oldest) occurrence. Fold in the time elapsed + // since the record's own insert tick so the age is exact + // regardless of when the last flush happened. dead.remove(&id); - index.entry(id).or_insert(Entry { - obs: *obs, + index.entry(id).or_insert_with(|| Entry { + obs: obs.aged(now.saturating_sub(stored_at)), stored_at: now, released: false, }); @@ -108,11 +113,9 @@ impl LogStore { } 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; @@ -126,21 +129,11 @@ impl LogStore { 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, }) } @@ -150,8 +143,21 @@ impl LogStore { } /// Persist the current index state (released flags, `next_seq`, and the - /// clock marker) and the observation log to disk. + /// observation log) to disk. pub fn flush(&mut self) -> Result<(), StoreError> { + // Advance every entry's age up to this tick in memory, so each + // `entry.obs.age_ms` is the true current age and `stored_at` marks when + // that age was computed. Ages are recomputed from each entry's own + // `stored_at` on the next load (see [`LogStore::new`]), so this keeps + // the in-memory representation coherent within this run. + for entry in self.index.values_mut() { + entry.obs.age_ms = entry + .obs + .age_ms + .saturating_add(self.now.saturating_sub(entry.stored_at)); + entry.stored_at = self.now; + } + let mut released: Vec = self .index .iter() @@ -162,12 +168,10 @@ impl LogStore { 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(()) } @@ -212,6 +216,7 @@ impl Store for LogStore { let record = Record::Obs { id, obs: Box::new(obs.clone()), + stored_at: now, }; self.append(&record)?; self.index.insert( diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index b049ebcb..e4adccdd 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -95,6 +95,37 @@ fn ages_rebase_on_load() { } } +#[test] +fn ages_rebase_after_nonidle_flush() { + // An observation inserted *before* the last flush must lose no time when + // ages are rebased on load: the reopen fold has to be exact, not folded + // against the flush clock as if the entry had been inserted then. + let path = store_path("ages_nonidle"); + + { + let mut s = LogStore::new(&path, 1000, 1).unwrap(); + let a = obs([1; 16], 1, 100, Some(10)); + assert!(s.insert(a, 1000).unwrap()); + // Advance the store clock via a second, distinct insert so the flush + // below runs at a tick later than `a`'s insert — the store is + // non-idle. + let b = obs([2; 16], 2, 0, Some(20)); + assert!(s.insert(b, 3000).unwrap()); + s.flush().unwrap(); + } + + { + // Reopen at 9000. `a`'s true age is 100 (insert-time) + (9000 - 1000) + // = 8100. A flush that fails to catch `age_ms` up folds the reopen + // against the flush clock (3000) and under-counts by the 2000ms + // between insert and flush, reporting only 6100. + let s = LogStore::new(&path, 9000, 1).unwrap(); + let a = obs([1; 16], 1, 100, Some(10)); + let g = s.get(&a.id(), 9000).unwrap(); + assert_eq!(g.age_ms, 8100); + } +} + #[test] fn next_seq_persists() { let path = store_path("seq");