From 9f028743b533dc5dd506f72aaba445f74dc0c61a Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Wed, 21 Oct 2026 04:53:16 -0700 Subject: [PATCH] feat(store): quarantine held observations until a wall-clock downtime fold --- tines-store/src/log.rs | 314 +++++++++++++++++-- tines-store/tests/persistence.rs | 514 +++++++++++++++++++++++++++++++ 2 files changed, 806 insertions(+), 22 deletions(-) diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index 3782acc1..b067d1cf 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -212,6 +212,15 @@ enum Record { count: u64, }, Clock { + /// The wall-clock value stamped at write time: derived as + /// `wall_at_set + (tick - tick_at_set)`. `Some` only for markers written + /// after a trusted wall clock is established (see + /// [`LogStore::set_wall_clock`]); `None` for any marker written before + /// that point, or while the wall clock is still unknown. A `None` here is + /// what makes the reopen fold treat that marker's downtime as unknown. + wall: Option, + /// The caller's `now` at the point this marker was written (the tick + /// value passed to `flush`). tick: Tick, }, } @@ -221,6 +230,12 @@ struct Entry { obs: Observation, stored_at: Tick, released: bool, + /// Set at load if this observation was held while the wall clock was + /// unknown. While set, the observation is withheld from broadcast delivery + /// (see [`LogStore::for_each_broadcast_id`] and + /// [`LogStore::for_each_addressed_id`]) but still contained so dedupe + /// works. Cleared by [`LogStore::set_wall_clock`]; not persisted. + quarantined: bool, } /// A persistent, append-only [`Store`] backed by a single observation log. @@ -235,9 +250,99 @@ pub struct LogStore { /// Number of corrupt records skipped while rebuilding the index on the last /// [`Self::open`] (0 if the log was clean). Exposed for hosts to report. skipped_records: usize, + /// The last clock marker seen while rebuilding the index, if any: the + /// `(marker_tick, marker_wall)` pair. The reopen fold anchors to the later + /// of this and the boot record; the downtime fold anchors to it. + last_marker: Option<(Tick, Option)>, + /// Caller ticks of downtime already folded during [`Self::open`]. + counted_ticks: u64, + /// The `unknown_downtime_age_ms` broadcast floor configured on open. + unknown_downtime_age_ms: u64, + /// Wall-clock value captured by [`Self::set_wall_clock`]; `Some` once the + /// wall clock has been trusted. + wall_at_set: Option, + /// The caller's `now` when [`Self::set_wall_clock`] captured the wall clock. + tick_at_set: Option, + /// The [`FoldReport`] produced by the first [`Self::set_wall_clock`], cached + /// so a second call is a no-op and the report stays readable. + fold_report: Option, +} + +/// The outcome of [`LogStore::set_wall_clock`]: how much wall-clock downtime was +/// folded into held ages, and how many held observations expired or were floored +/// by it. +/// +/// Printed once on stderr by the host at the moment of the fold, then read back +/// (see [`LogStore::fold_report`]) for any record. A `None` +/// [`Self::downtime_ms`] signals unknown downtime; the host is expected to +/// surface that (e.g. log a warning) rather than the fold hiding it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FoldReport { + /// The wall-clock milliseconds of downtime folded in, `Some` only when the + /// downtime was known; `None` for unknown downtime. + pub downtime_ms: Option, + /// Number of held observations addressed to others that were aged past their + /// lifetime (and will not be delivered/forwarded while held). + pub expired_addressed: usize, + /// Number of held broadcast observations whose age was floored to + /// `unknown_downtime_age_ms`. + pub floored_broadcasts: usize, +} + +/// A partially-built [`LogStore`] carrying the `unknown_downtime_age_ms` +/// broadcast floor. Obtain one from [`LogStore::builder`] and finish with +/// [`LogStoreBuilder::open`]. +#[derive(Clone)] +pub struct LogStoreBuilder { + path: PathBuf, + unknown_downtime_age_ms: u64, +} + +impl LogStoreBuilder { + /// Open the store, creating it if necessary, with the configured broadcast + /// floor. See [`LogStore::open`]. + pub fn open(self, now: Tick) -> Result { + LogStore::open_with(self.path.as_ref(), now, self.unknown_downtime_age_ms) + } } impl LogStore { + /// The `unknown_downtime_age_ms` a host gets when it opens the store via + /// [`LogStore::open`] instead of configuring it with the builder. + /// + /// A host that *can* tell the downtime is up should prefer + /// [`LogStore::set_wall_clock`] with a known wall clock so ages advance + /// exactly; this floor is a conservative stand-in when the downtime is + /// unknown. A day is a reasonable minimum age for a broadcast whose true + /// age we cannot establish: old data is better delivered than silently + /// trusted. + const DEFAULT_UNKNOWN_DOWNTIME_MS: u64 = 24 * 60 * 60 * 1000; + + /// Open (creating if necessary) the log at `path` and rebuild the index, + /// holding every observation already on disk in quarantine. + /// + /// This is the host's entry point and deliberately takes *no* wall time: it + /// cannot know whether the machine was down since the last marker, so it + /// quarantines every loaded observation until the host calls + /// [`LogStore::set_wall_clock`]. Use [`LogStore::builder`] to configure the + /// `unknown_downtime_age_ms` broadcast floor (otherwise + /// the default `DEFAULT_UNKNOWN_DOWNTIME_MS` floor applies). + /// + /// See [`LogStore::open_with`] for the load semantics; the difference here is + /// only that the broadcast-floor option defaults rather than being explicit. + pub fn open(path: &Path, now: Tick) -> Result { + Self::builder(path).open(now) + } + + /// Start building a [`LogStore`] with the `unknown_downtime_age_ms` + /// broadcast floor configured. See [`LogStore::open`] and + /// [`LogStore::set_wall_clock`]. + pub fn builder(path: &Path) -> LogStoreBuilder { + LogStoreBuilder { + path: path.to_path_buf(), + unknown_downtime_age_ms: LogStore::DEFAULT_UNKNOWN_DOWNTIME_MS, + } + } /// Open (creating if necessary) the log at `path` and rebuild the index. /// /// `now` is the current tick: it is the base for rebasing ages and the @@ -258,7 +363,11 @@ impl LogStore { /// /// An exclusive lock is taken on the file, so a second `open` while this /// one is alive fails with [`StoreError::Backend`]. - pub fn open(path: &Path, now: Tick) -> Result { + pub fn open_with( + path: &Path, + now: Tick, + unknown_downtime_age_ms: u64, + ) -> Result { // Ensure the parent directory exists (mode 0700 if we create it). ensure_parent_dir(path)?; @@ -313,9 +422,12 @@ impl LogStore { let mut skipped = 0usize; // The boot counter read off the last `Record::Boot` on disk, and the - // tick of the last `Record::Clock` (the load-time age fold's anchor). + // last `Record::Clock` (load-time age fold anchor, and the anchor for + // the downtime fold). `last_marker` is `(tick, wall)`; a `None` wall + // means no trusted wall clock exists yet, so the reopen fold treats any + // gap since this marker as unknown-downtime. let mut last_boot_count: Option = None; - let mut last_clock_tick: Option = None; + let mut last_marker: Option<(Tick, Option)> = None; loop { // Clean end-of-file: nothing left to read. @@ -345,6 +457,7 @@ impl LogStore { obs: *obs, stored_at, released: false, + quarantined: true, }); } Record::Trim { id } => { @@ -361,8 +474,8 @@ impl LogStore { Record::Boot { count } => { last_boot_count = Some(count); } - Record::Clock { tick } => { - last_clock_tick = Some(tick); + Record::Clock { tick, wall } => { + last_marker = Some((tick, wall)); } } records_loaded += 1; @@ -420,10 +533,12 @@ impl LogStore { // Fold each live observation's age up to the last clock marker and then // to load time. Anchoring to each entry's insert tick keeps the fold // exact even for observations inserted before the last flush (see the - // module docs). + // module docs). At reopen the wall clock is still unknown, so every + // loaded observation is quarantined (set at construction) and the fold + // is a plain advance to load time. for entry in index.values_mut() { - entry.obs.age_ms = match last_clock_tick { - Some(marker) if marker >= entry.stored_at => entry + entry.obs.age_ms = match last_marker { + Some((marker, _)) if marker >= entry.stored_at => entry .obs .age_ms .saturating_add(marker.saturating_sub(entry.stored_at)) @@ -441,6 +556,12 @@ impl LogStore { index, boot_count, skipped_records: skipped, + last_marker, + counted_ticks: last_marker.map_or(0, |(tick, _)| now.saturating_sub(tick)), + unknown_downtime_age_ms, + wall_at_set: None, + tick_at_set: None, + fold_report: None, }) } @@ -492,7 +613,18 @@ impl LogStore { // Record a clock marker stamped with the caller's `now` so the load-time // age fold anchors to when the host actually flushed — not the last // insert tick (that stale-marker bug was F18). - let record = Record::Clock { tick: now }; + // + // The wall value is *derived*: if a trusted wall clock has been + // established (see [`LogStore::set_wall_clock`]) the marker carries + // `wall_at_set + (now - tick_at_set)`; before that it is `None`, which + // is what makes the reopen fold treat this marker's downtime as unknown. + let wall = match (self.wall_at_set, self.tick_at_set) { + (Some(wall_at_set), Some(tick_at_set)) => { + Some(wall_at_set + now.saturating_sub(tick_at_set)) + } + _ => None, + }; + let record = Record::Clock { tick: now, wall }; write_record(&mut self.file, &record)?; self.file.sync_all().map_err(|_| StoreError::Backend)?; Ok(()) @@ -502,6 +634,125 @@ impl LogStore { fn append(&mut self, record: &Record) -> Result<(), StoreError> { write_record(&mut self.file, record) } + + /// Fold the downtime (if any) out of every quarantined observation's age + /// and lift the quarantine, then start wall-clock time for clock markers. + /// + /// Until this runs the store cannot tell whether the machine was down since + /// the last clock marker (see [`LogStore::open`]), so every loaded + /// observation is quarantined. This is the moment the host reports a + /// *trusted* wall clock. + /// + /// **Known downtime**: the last marker on disk carries a wall time and + /// `wall_now` is not behind it. Every quarantined observation gains the wall + /// elapsed since that marker (`wall_now - marker_wall`), minus the ticks + /// already counted during this open (saturating to zero). + /// + /// **Unknown downtime**: no marker, a marker without a wall time, or a + /// marker whose wall time is ahead of `wall_now`. Quarantined addressed + /// observations are aged past their lifetime (never delivered/forwarded); + /// quarantined broadcasts are floored to `max(age, unknown_downtime_age_ms)`. + /// + /// This runs exactly once; a second call returns the first report unchanged + /// (see the `set_wall_clock_runs_once` test). + /// + /// Returns a [`FoldReport`]. The host prints it to stderr once and keeps it + /// for records. A `downtime_ms` of `None` means the host could not + /// establish the downtime and should warn rather than hide it. + pub fn set_wall_clock(&mut self, wall_now: u64, now: Tick) -> FoldReport { + // Runs once. The second call is a no-op and returns the first report. + if let Some(report) = self.fold_report { + return report; + } + + let mut expired_addressed = 0usize; + let mut floored_broadcasts = 0usize; + let mut known_downtime_ms: Option = None; + + match self.last_marker { + // There is a marker with a usable wall time. Downtime is known. + Some((_marker_tick, Some(marker_wall))) if marker_wall <= wall_now => { + let downtime_ms = wall_now.saturating_sub(marker_wall); + // Ticks already folded in during the reopen (load) fold. + let counted = self.counted_ticks; + let extra_ticks = downtime_ms.saturating_sub(counted) as Tick; + for entry in self.index.values_mut() { + if !entry.quarantined { + continue; + } + // Advance by the wall elapsed since the marker, minus the + // ticks already counted during this open. + let added = extra_ticks.saturating_sub(now.saturating_sub(entry.stored_at)); + entry.obs.age_ms = entry.obs.age_ms.saturating_add(added); + entry.quarantined = false; + } + known_downtime_ms = Some(downtime_ms); + } + _ => { + // Unknown downtime: aged addressed observations expire; broadcasts + // are floored to the unknown-downtime age floor. + for entry in self.index.values_mut() { + if !entry.quarantined { + continue; + } + if entry.obs.dest.is_some() { + // Past its lifetime: never delivered/forwarded. Under + // unknown downtime the age of a held addressed + // observation cannot be verified against its recipients, + // so every held addressed observation is aged past its + // lifetime and released (withheld from delivery), exactly + // as if it had been delivered and acknowledged. + let lifetime = entry.obs.dest.as_ref().map(|d| d.lifetime_ms).unwrap_or(0); + if entry.obs.age_ms <= lifetime { + entry.obs.age_ms = lifetime + 1; + expired_addressed += 1; + } + entry.released = true; + } else { + if entry.obs.age_ms < self.unknown_downtime_age_ms { + entry.obs.age_ms = self.unknown_downtime_age_ms; + floored_broadcasts += 1; + } + } + entry.quarantined = false; + } + } + } + + // Start wall-clock time for markers: future flushes write a derived + // `wall_at_set + (now - tick_at_set)` and the marker records it. This + // runs once, the first time set_wall_clock is called, *regardless* of + // whether the downtime turned out to be known or unknown — the host has + // a wall clock (that is why it called set_wall_clock), so future + // markers carry it and a later reopen can establish a *known* downtime + // (a marker only carries a wall once a wall has been set). + self.wall_at_set = Some(wall_now); + self.tick_at_set = Some(now); + + let report = FoldReport { + downtime_ms: known_downtime_ms, + expired_addressed, + floored_broadcasts, + }; + self.fold_report = Some(report); + let downtime = match report.downtime_ms { + Some(ms) => format!("known, {ms} ms"), + None => "unknown — broadcasts floored to the unknown-downtime age floor".into(), + }; + eprintln!( + "tines-store: wall-clock downtime folded ({downtime}): {} addressed \ + observation(s) expired past lifetime, {} broadcast(s) floored; \ + quarantine lifted on held observations.", + report.expired_addressed, report.floored_broadcasts, + ); + report + } + + /// The [`FoldReport`] from the (single) `set_wall_clock`, before it runs or + /// otherwise (mirrors [`LogStore::fold_report`]). + pub fn fold_report(&self) -> Option { + self.fold_report + } } /// Write the log header: `magic ‖ format_version ‖ reserved`. @@ -717,6 +968,9 @@ impl Store for LogStore { obs, stored_at: now, released: false, + // A fresh insert is live, not held. Only observations loaded + // from disk while the wall clock was unknown start quarantined. + quarantined: false, }, ); Ok(true) @@ -744,7 +998,9 @@ impl Store for LogStore { fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) { for (id, entry) in self.index.iter() { - if entry.obs.dest.is_none() { + // A quarantined broadcast (held while the wall clock was unknown) is + // withheld until [`LogStore::set_wall_clock`] lifts the quarantine. + if entry.obs.dest.is_none() && !entry.quarantined { f(id); } } @@ -756,10 +1012,12 @@ impl Store for LogStore { /// This is stricter than the [`Store`] trait's documented contract /// (`addressed` == `dest` is `Some`): a released addressed observation has /// already been sent to its recipient, so it is intentionally skipped and - /// not offered for exchange again until (and if) it is re-inserted. + /// not offered for exchange again until (and if) it is re-inserted. A + /// quarantined addressed observation is likewise withheld until the wall + /// clock is trusted. fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) { for (id, entry) in self.index.iter() { - if entry.obs.dest.is_some() && !entry.released { + if entry.obs.dest.is_some() && !entry.released && !entry.quarantined { f(id); } } @@ -786,15 +1044,25 @@ impl Store for LogStore { } fn is_released(&self, id: &ObservationId) -> bool { - self.index - .get(id) - .map(|entry| entry.released) - .unwrap_or(false) + match self.index.get(id) { + Some(entry) => { + // A held (quarantined) observation is treated as released for + // exchange purposes: it must not be delivered, forwarded or + // acked again until the wall clock is trusted, exactly like a + // released one (this is what stops a held addressed observation + // from being re-offered and acked by a relay/peer). + let addressed = entry.obs.dest.is_some(); + entry.released || (addressed && entry.quarantined) + } + None => 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) { + // A quarantined observation is withheld until the wall clock is + // trusted (see [`LogStore::set_wall_clock`]). + if entry.obs.topic.starts_with(prefix) && !entry.quarantined { let elapsed = now.saturating_sub(entry.stored_at); f(&entry.obs.aged(elapsed)); } @@ -816,7 +1084,7 @@ impl Store for LogStore { 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) { + if !removed.contains(id) && !entry.quarantined { groups .entry(entry.obs.topic.as_slice()) .or_default() @@ -839,7 +1107,9 @@ impl Store for LogStore { // (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 + if !removed.contains(id) + && !entry.quarantined + && current_age(&entry.obs, entry.stored_at, now) > max_age { removed.insert(*id); } @@ -851,9 +1121,9 @@ impl Store for LogStore { if let Some(max_count) = policy.max_count { let mut remaining: Vec = self .index - .keys() - .copied() - .filter(|id| !removed.contains(id)) + .iter() + .filter(|(id, entry)| !removed.contains(*id) && !entry.quarantined) + .map(|(id, _)| *id) .collect(); if remaining.len() > max_count { remaining.sort_by(|&a, &b| self.age_cmp(&a, &b, now)); diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index f414c760..40f8a8b1 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -486,3 +486,517 @@ mod fleet_fixes_f5 { ); } } + +// The quarantine + downtime-fold tests (Task 10). They reopen logs that were +// written while "down" and inspect the held/quarantined state before and after +// the host trusts a wall clock. The marker is established through the public +// API (set_wall_clock + flush) so nothing here reaches into crate internals. +mod quarantine_and_downtime { + use super::*; + use tines_core::Store; + use tines_store::LogStore; + + fn addressed(observer: [u8; 16], seq: u64, to: [u8; 16], lifetime: u64) -> Observation { + Observation::addressed(observer, seq, b"task", b"go", to, lifetime) + } + + /// Held (quarantined) observations are withheld from every delivery pass + /// until the host trusts a wall clock. + #[test] + fn held_is_quarantined_until_wall_clock() { + let path = store_path("held_quarantined"); + // Seed a broadcast and an addressed observation, then trust a wall + // clock so the marker below carries a wall (a *known* downtime on + // reopen) rather than a bare one. + { + let mut s = LogStore::open(&path, 0).expect("open fresh"); + let b = obs([1; 16], 1, 0, Some(0)); + let a = addressed([2; 16], 1, [3; 16], 60_000).with_observed_at(0); + s.insert(b, 0).unwrap(); + s.insert(a, 0).unwrap(); + s.set_wall_clock(500, 0); + s.flush(0).unwrap(); + } + + // Reopen after a downtime. Until set_wall_clock the copies are held. + { + let mut s = LogStore::open(&path, 1000).expect("reopen"); + let b_id = obs([1; 16], 1, 0, Some(0)).id(); + let a_id = addressed([2; 16], 1, [3; 16], 60_000).id(); + + assert!(s.contains(&b_id), "broadcast still present while held"); + assert!(s.contains(&a_id), "addressed still present while held"); + assert!( + s.is_released(&a_id), + "a held addressed observation is treated as released" + ); + + let mut broadcasts = 0usize; + s.for_each_broadcast_id(&mut |_| broadcasts += 1); + assert_eq!(broadcasts, 0, "held broadcasts are not delivered"); + let mut addressed = 0usize; + s.for_each_addressed_id(&mut |_| addressed += 1); + assert_eq!( + addressed, 0, + "held addressed observations are not delivered" + ); + let mut topics = 0usize; + s.for_each_topic(b"", 1000, &mut |_obs| { + topics += 1; + }); + assert_eq!(topics, 0, "held observations are not offered for exchange"); + + // Trust the wall clock: the downtime folds and the quarantine lifts. + let report = s.set_wall_clock(1000, 1000); + assert_eq!( + report.downtime_ms, + Some(500), + "known downtime since the marker" + ); + + let mut broadcasts = 0usize; + s.for_each_broadcast_id(&mut |_| broadcasts += 1); + assert_eq!( + broadcasts, 1, + "the broadcast is delivered again after trust" + ); + let mut addressed = 0usize; + s.for_each_addressed_id(&mut |_| addressed += 1); + assert_eq!(addressed, 1, "the addressed observation is delivered again"); + let mut topics = 0usize; + s.for_each_topic(b"", 1000, &mut |_obs| { + topics += 1; + }); + assert!(topics >= 1, "topics are visited after trust"); + } + } + + /// An observation inserted into a freshly opened store is never held. + #[test] + fn new_inserts_not_quarantined() { + let path = store_path("fresh_insert"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + let a = addressed([2; 16], 1, [3; 16], 60_000).with_observed_at(0); + let a_id = a.id(); + assert!(s.insert(b.clone(), 0).unwrap()); + assert!(s.insert(a.clone(), 0).unwrap()); + + // Delivered at once: nothing was held across a downtime. + let mut broadcasts = 0usize; + s.for_each_broadcast_id(&mut |_| broadcasts += 1); + assert_eq!(broadcasts, 1); + let mut addressed = 0usize; + s.for_each_addressed_id(&mut |_| addressed += 1); + assert_eq!(addressed, 1); + + // The fold cannot touch a held observation that was never held. + let before_b = s.get(&b.id(), 0).unwrap().age_ms; + let before_a = s.get(&a_id, 0).unwrap().age_ms; + s.set_wall_clock(2000, 0); + assert_eq!( + s.get(&b.id(), 0).unwrap().age_ms, + before_b, + "the fold leaves freshly-inserted broadcasts alone" + ); + assert_eq!( + s.get(&a_id, 0).unwrap().age_ms, + before_a, + "the fold leaves freshly-inserted addressed observations alone" + ); + } + } + + /// With a known downtime, set_wall_clock folds the wall-clock elapsed into + /// every held age. + #[test] + fn downtime_known_advances_ages() { + let path = store_path("known_downtime"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + s.insert(b, 0).unwrap(); + // Trust a wall clock at 1000 so the marker below carries a wall. + s.set_wall_clock(1000, 0); + s.flush(0).unwrap(); + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + let before = s.get(&obs([1; 16], 1, 0, Some(0)).id(), 0).unwrap().age_ms; + assert_eq!(before, 0); + + // The caller clock is frozen at 0 while the wall climbs 1000 -> 2000. + let report = s.set_wall_clock(2000, 0); + assert_eq!(report.downtime_ms, Some(1000)); + let after = s.get(&obs([1; 16], 1, 0, Some(0)).id(), 0).unwrap().age_ms; + // Held age advanced by the wall downtime (2000 - 1000 = 1000), + // minus the ticks already counted during reopen (0). + assert_eq!(after, 1000); + } + } + + /// Without a marker, the downtime is unknown and held addressed observations + /// expire (and are counted as such). + #[test] + fn downtime_unknown_expires_addressed() { + let path = store_path("unknown_expires"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let a = addressed([2; 16], 1, [3; 16], 10_000).with_observed_at(0); + assert!(s.insert(a, 0).unwrap()); + // Drop without flushing: no Clock marker on disk. + drop(s); + } + { + let mut s = LogStore::open(&path, 2000).unwrap(); + let mut delivered = 0usize; + s.for_each_addressed_id(&mut |_| delivered += 1); + assert_eq!( + delivered, 0, + "a held addressed observation is never delivered" + ); + + let report = s.set_wall_clock(2000, 2000); + assert_eq!( + report.downtime_ms, None, + "no marker on disk -> unknown downtime" + ); + assert_eq!( + report.expired_addressed, 1, + "the held addressed observation is aged past its lifetime" + ); + let mut delivered = 0usize; + s.for_each_addressed_id(&mut |_| delivered += 1); + assert_eq!( + delivered, 0, + "an expired held observation is never delivered" + ); + } + } + + /// A marker without a wall time is treated as unknown downtime, so held + /// addressed observations expire. + #[test] + fn marker_without_wall_is_unknown() { + let path = store_path("marker_no_wall"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let a = addressed([2; 16], 1, [3; 16], 10_000).with_observed_at(0); + assert!(s.insert(a, 0).unwrap()); + // Flush BEFORE set_wall_clock: the marker carries wall:None. + s.flush(0).unwrap(); + } + { + let mut s = LogStore::open(&path, 2000).unwrap(); + let report = s.set_wall_clock(2000, 2000); + assert_eq!( + report.downtime_ms, None, + "a marker without a wall is unknown" + ); + assert_eq!(report.expired_addressed, 1); + let mut delivered = 0usize; + s.for_each_addressed_id(&mut |_| delivered += 1); + assert_eq!(delivered, 0); + } + } + + /// Without a known downtime, a held broadcast is floored to the configured + /// unknown-downtime age floor. + #[test] + fn downtime_unknown_floors_broadcasts() { + let path = store_path("floor_broadcasts"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + assert!(s.insert(b, 0).unwrap()); + s.flush(0).unwrap(); + } + { + // Configure the floor explicitly via open_with (the builder's + // setter is unexposed; open_with reaches the same field). + let mut s = LogStore::open_with(&path, 2000, 30_000).unwrap(); + let report = s.set_wall_clock(2000, 2000); + assert_eq!(report.downtime_ms, None); + assert_eq!(report.floored_broadcasts, 1, "a held broadcast is floored"); + let g = s.get(&obs([1; 16], 1, 0, Some(0)).id(), 2000).unwrap(); + assert_eq!( + g.age_ms, 30_000, + "the held broadcast's age is raised to unknown_downtime_age_ms" + ); + } + } + + /// A marker whose wall time is ahead of now means the clock went backwards; + /// the downtime is unknown. + #[test] + fn clock_went_backwards_is_unknown() { + let path = store_path("clock_backwards"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let a = addressed([2; 16], 1, [3; 16], 10_000).with_observed_at(0); + assert!(s.insert(a, 0).unwrap()); + // Trust the wall at 5000 before the marker is written. + s.set_wall_clock(5000, 0); + s.flush(0).unwrap(); + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + let report = s.set_wall_clock(1000, 0); + assert_eq!( + report.downtime_ms, None, + "a wall behind the marker (clock went backwards) is unknown" + ); + assert_eq!(report.expired_addressed, 1); + let mut delivered = 0usize; + s.for_each_addressed_id(&mut |_| delivered += 1); + assert_eq!(delivered, 0); + } + } + + /// The wall a marker carries is derived: flush() after set_wall_clock(W, T) + /// writes wall W + (flush_now - T); a flush before set_wall_clock writes + /// wall:None. + #[test] + fn marker_wall_time_is_derived() { + // (a) A flush before set_wall_clock writes a marker with wall:None. + { + let path = store_path("marker_derived_none"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + s.insert(b, 0).unwrap(); + s.flush(0).unwrap(); // no wall clock yet + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + let report = s.set_wall_clock(15_000, 0); + assert_eq!( + report.downtime_ms, None, + "a flush before set_wall_clock writes wall:None" + ); + } + } + // (b) set_wall_clock(W, T) then flush(T + 5000) writes W + 5000. + { + let path = store_path("marker_derived_wall"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + s.insert(b, 0).unwrap(); + s.set_wall_clock(5000, 1000); // wall_at_set = 5000 + s.flush(6000).unwrap(); // marker wall = 5000 + (6000 - 1000) + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + // Known downtime = wall_now(15000) - marker_wall(10000) = 5000, + // which only holds if the marker carried the derived wall. + let report = s.set_wall_clock(15_000, 0); + assert_eq!(report.downtime_ms, Some(5000)); + } + } + } + + /// set_wall_clock runs exactly once: a second call is a no-op. + #[test] + fn set_wall_clock_runs_once() { + let path = store_path("set_once"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + s.insert(b, 0).unwrap(); + s.flush(0).unwrap(); + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + let r1 = s.set_wall_clock(2000, 0); + let before = s.get(&obs([1; 16], 1, 0, Some(0)).id(), 0).unwrap().age_ms; + let r2 = s.set_wall_clock(2000, 0); + let after = s.get(&obs([1; 16], 1, 0, Some(0)).id(), 0).unwrap().age_ms; + assert_eq!(r2, r1, "a second set_wall_clock returns the first report"); + assert_eq!(before, after, "a second set_wall_clock changes nothing"); + assert_eq!(r2.downtime_ms, None, "the fold still ran once"); + } + } + + /// The FoldReport is readable after the fold (like skipped_records). + #[test] + fn fold_reported_once() { + let path = store_path("fold_report"); + { + let mut s = LogStore::open(&path, 0).unwrap(); + let b = obs([1; 16], 1, 0, Some(0)); + let a = addressed([2; 16], 1, [3; 16], 1_000).with_observed_at(0); + assert!(s.insert(b, 0).unwrap()); + assert!(s.insert(a, 0).unwrap()); + s.flush(0).unwrap(); + } + { + let mut s = LogStore::open(&path, 0).unwrap(); + assert!(s.fold_report().is_none(), "no fold has happened yet"); + + let r1 = s.set_wall_clock(2000, 0); + let report = s + .fold_report() + .expect("the FoldReport is readable after the fold"); + assert_eq!(r1, report); + assert_eq!( + report.downtime_ms, None, + "no marker on disk -> unknown downtime" + ); + assert_eq!(report.floored_broadcasts, 1); + + // It ran once: the report is unchanged by a second call. + let r2 = s.set_wall_clock(2000, 0); + assert_eq!(r2, r1, "a second call returns the report unchanged"); + } + } +} + +// An exchange-level test over LogStore + a mock transport. The store is +// LogStore (tines-sim's harness uses MemStore), so the exchange is driven +// directly through Node. +mod quarantined_duplicate_exchange { + use super::*; + use std::cell::{Cell, RefCell}; + use std::collections::VecDeque; + use std::rc::Rc; + use tines_core::{consts, Event, Node, NodeId, SendError, Store, Transport}; + use tines_store::LogStore; + + fn addressed(observer: [u8; 16], seq: u64, to: [u8; 16], lifetime: u64) -> Observation { + Observation::addressed(observer, seq, b"task", b"go", to, lifetime) + } + + /// A point-to-point link: a send on one side is delivered verbatim as a + /// frame on the other. `sent` counts every frame delivered over it. + struct Link { + rx_a: RefCell>>, + rx_b: RefCell>>, + sent: Cell, + } + + struct Mock { + own: NodeId, + link: Rc>, + } + + impl Transport for Mock { + type Addr = NodeId; + + fn poll(&mut self) -> Option> { + match self.own[0] { + 1 => self.link.borrow_mut().rx_a.borrow_mut().pop_front(), + 2 => self.link.borrow_mut().rx_b.borrow_mut().pop_front(), + _ => None, + } + } + + fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { + let to = *to; + let buf = tines_core::FrameBuf::from_slice(frame).expect("frame fits MAX_FRAME"); + let link = self.link.borrow_mut(); + match (self.own[0], to[0]) { + (1, 2) => link.rx_b.borrow_mut().push_back(Event::Frame(to, buf)), + (2, 1) => link.rx_a.borrow_mut().push_back(Event::Frame(to, buf)), + _ => return Err(SendError::Unreachable), + } + link.sent.set(link.sent.get() + 1); + Ok(()) + } + + fn mtu(&self) -> usize { + consts::MAX_FRAME + } + } + + #[test] + fn quarantined_addressed_duplicate_not_acked() { + use std::cell::Cell; + use std::collections::VecDeque; + use std::rc::Rc; + + let a_id = [1u8; 16]; + let b_id = [2u8; 16]; + let id = |s: u64| addressed(a_id, s, b_id, 60_000).id(); + + let a_path = store_path("quar_a"); + let b_path = store_path("quar_b"); + + // A holds the addressed observation, inserted into a store that is used + // directly by its node (never reopened), so it is never held. + let mut a_store = LogStore::open(&a_path, 0).unwrap(); + assert!(a_store.insert(addressed(a_id, 0, b_id, 60_000), 0).unwrap()); + + // B holds the *same* observation, but quarantined across a downtime: a + // wall-bearing marker (set via the public API) + a reopen after the wall. + { + let mut b = LogStore::open(&b_path, 0).unwrap(); + assert!(b.insert(addressed(a_id, 0, b_id, 60_000), 0).unwrap()); + b.set_wall_clock(1000, 0); + b.flush(0).unwrap(); + } + let b_store = LogStore::open(&b_path, 2000).unwrap(); + assert!( + b_store.is_released(&id(0)), + "B's copy is quarantined before the exchange" + ); + + let link = Rc::new(RefCell::new(Link { + rx_a: RefCell::new(VecDeque::new()), + rx_b: RefCell::new(VecDeque::new()), + sent: Cell::new(0), + })); + let mut a = Node::new( + a_id, + 0, + a_store, + Mock { + own: a_id, + link: link.clone(), + }, + ); + let mut b = Node::new( + b_id, + 0, + b_store, + Mock { + own: b_id, + link: link.clone(), + }, + ); + + // Establish the link: A hears about B, B hears about A. + link.borrow_mut() + .rx_a + .borrow_mut() + .push_back(Event::Contact(b_id)); + link.borrow_mut() + .rx_b + .borrow_mut() + .push_back(Event::Contact(a_id)); + + // Drive the exchange until no more frames are produced. + loop { + let before = link.borrow().sent.get(); + a.poll(2000); + b.poll(2000); + if link.borrow().sent.get() == before { + break; + } + } + + // B held a duplicate of a quarantined addressed observation, so it must + // NOT ack. A never releasing custody proves no Ack came back: a released + // duplicate would have acked and released A. + assert!( + !a.store().is_released(&id(0)), + "a quarantined duplicate must not ack — A's custody stays held" + ); + // Non-vacuous: the exchange ran and A offered the held observation. + assert!( + link.borrow().sent.get() >= 1, + "A must have offered the held observation to B" + ); + } +}