diff --git a/scripts/ci.sh b/scripts/ci.sh index f2a34441..177b05a1 100755 --- a/scripts/ci.sh +++ b/scripts/ci.sh @@ -3,8 +3,16 @@ set -e cargo fmt --all -- --check cargo clippy --locked --workspace --all-targets -- -D warnings cargo clippy --locked -p tines-core --no-default-features --all-targets -- -D warnings +# The backend conformance suite lives behind the `conformance` feature so it +# stays out of the default/std builds, but every backend crate still runs it. +cargo clippy --locked -p tines-core --features conformance --all-targets -- -D warnings +cargo clippy --locked -p tines-store --features conformance --all-targets -- -D warnings +cargo clippy --locked -p tines-sim --features conformance --all-targets -- -D warnings RUSTDOCFLAGS="-D warnings" cargo doc --locked --workspace --no-deps RUSTDOCFLAGS="-D warnings" cargo doc --locked -p tines-core --no-deps --no-default-features cargo test --locked --workspace +cargo test --locked -p tines-core --features conformance +cargo test --locked -p tines-store --features conformance +cargo test --locked -p tines-sim --features conformance cargo test --locked -p tines-core --no-default-features sh scripts/check-no-std.sh diff --git a/tines-core/Cargo.toml b/tines-core/Cargo.toml index e91483a3..7f0ea735 100644 --- a/tines-core/Cargo.toml +++ b/tines-core/Cargo.toml @@ -14,6 +14,11 @@ categories = ["no-std", "embedded", "network-programming", "science::robotics"] [features] default = ["alloc"] alloc = [] +# A `no_std`-clean conformance suite ([`store_conformance`]) that concrete +# backends run against themselves to prove they honor the `Store` contract. +# Std-free so it builds with `--no-default-features`; enabled only by the +# backend crates that exercise the suite (MemStore, LogStore, CappedStore). +conformance = [] [dependencies] heapless = { version = "0.8", features = ["serde"] } diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index 7717c56a..a31eff08 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -14,6 +14,8 @@ pub mod mem_store; pub mod node; pub mod observation; pub mod store; +#[cfg(feature = "conformance")] +pub mod store_conformance; pub mod transport; mod wire; diff --git a/tines-core/src/mem_store.rs b/tines-core/src/mem_store.rs index dc9bfd1a..d1f4e427 100644 --- a/tines-core/src/mem_store.rs +++ b/tines-core/src/mem_store.rs @@ -479,3 +479,24 @@ mod tests { assert!(!s.is_released(&[0u8; 16])); } } + +#[cfg(all(test, feature = "conformance"))] +mod conformance_tests { + use crate::store_conformance; + + use super::MemStore; + + // The whole suite runs against `MemStore` from its own crate. + #[test] + fn store_conforms() { + store_conformance::insert_get(MemStore::new); + store_conformance::dedupe(MemStore::new); + store_conformance::age_lineage(MemStore::new); + store_conformance::saturating_ticks(MemStore::new); + store_conformance::release_surviving_get(MemStore::new); + store_conformance::sig_round_trip(MemStore::new); + store_conformance::trim_order(MemStore::new); + store_conformance::no_tombstones(MemStore::new); + store_conformance::idempotent_flush(MemStore::new); + } +} diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 927a9874..6348a80a 100644 --- a/tines-core/src/node.rs +++ b/tines-core/src/node.rs @@ -219,14 +219,15 @@ impl Node { &self.store } - /// Mutable access to the backing store. + /// Best-effort durability flush, advancing each stored observation's age + /// up to `now`. /// - /// Lets a host call backend-specific methods that aren't on the [`Store`] - /// trait — e.g. `tines_store::LogStore::flush`, the durability boundary a - /// host reaches on its flush schedule. [`Self::trim`] takes `&mut self`; - /// this exposes the concrete backend the same way. - pub fn store_mut(&mut self) -> &mut S { - &mut self.store + /// This is the durability boundary a host reaches on its flush schedule — + /// delegates to [`Store::flush`], so it is infallible by contract: a failed + /// flush is, at worst, a duplicate custodian on the next exchange, never a + /// lost message. + pub fn flush(&mut self, now: Tick) -> Result<(), StoreError> { + self.store.flush(now) } /// Access the transport (e.g. to `connect`/`disconnect` a peer in tests). diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs index a87fe463..6ea9e91e 100644 --- a/tines-core/src/store.rs +++ b/tines-core/src/store.rs @@ -43,7 +43,6 @@ pub enum StoreError { /// 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. /// @@ -62,6 +61,11 @@ pub trait Store { /// Number of observations currently stored. fn len(&self) -> usize; + /// `true` if the store holds no observations. + fn is_empty(&self) -> bool { + self.len() == 0 + } + /// Visit every stored observation id. fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)); @@ -85,6 +89,10 @@ pub trait Store { /// Releasing an id that is not currently stored is a no-op. The mark /// persists for the lifetime of the entry (until it is trimmed); a /// re-inserted observation starts un-released. + /// + /// Release is best-effort and infallible: it never returns an error, and a + /// failed release is, at worst, a duplicate custodian on the next + /// exchange — never a lost message. fn release(&mut self, id: &ObservationId); /// Returns `true` if the observation with `id` has been released. @@ -98,6 +106,19 @@ pub trait Store { /// 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; + + /// Best-effort durability flush, advancing each stored observation's age + /// up to the caller's `now` and recording `now` as the flush clock. + /// + /// Flush must not lose messages, must never resurrect a released message, + /// and is infallible by contract: a failed flush is, at worst, a + /// duplicate custodian on the next exchange — never a lost message. + /// Backends that cannot persist more cheaply default to a no-op that + /// simply returns `Ok(())`. + fn flush(&mut self, now: Tick) -> Result<(), StoreError> { + let _ = now; + Ok(()) + } } /// Current (age-adjusted) age of a stored observation at tick `now`. diff --git a/tines-core/src/store_conformance.rs b/tines-core/src/store_conformance.rs new file mode 100644 index 00000000..40fdd81a --- /dev/null +++ b/tines-core/src/store_conformance.rs @@ -0,0 +1,223 @@ +//! Backend conformance suite. +//! +//! A `no_std`-clean set of property tests that any [`Store`] backend can run to +//! prove it honors the contract in [`Store`](crate::Store). Each public function +//! exercises exactly one property against a backend produced by a +//! caller-supplied factory, so the same body runs for [`MemStore`], a +//! [`LogStore`](crate::store::LogStore) opened on a fresh file per case and +//! [`CappedStore`] — see the `#[test]` shims in each backend crate. +//! +//! The functions build their own [`Observation`]s, including one carrying a +//! `sig`, which only the crate that owns the `#[non_exhaustive]` type may +//! construct by struct literal — this is why the suite lives *inside* +//! `tines-core` rather than as a `tines-core/tests/` file. +//! +//! The suite is deliberately free of `std`/`alloc` so it can build under +//! `--no-default-features` (with this feature on): every call is `heapless` +//! plus `assert!`. + +use crate::observation::Observation; +use crate::{Payload, SigAlgo, Signature, Store, Topic, TrimPolicy}; + +/// Build a signed broadcast observation for the suite. +/// +/// `signature` is a byte string the backend need not understand; the test only +/// checks it round-trips verbatim, so it is any length `<= MAX_SIG`. +fn signed_obs(seq: u64, topic: &[u8], signature: &[u8]) -> Observation { + let payload = Payload::from_slice(b"pay").expect("payload fits capacity"); + Observation { + observer: [1u8; 16], + seq, + age_ms: 0, + observed_at: None, + topic: Topic::from_slice(topic).expect("topic fits capacity"), + payload, + dest: None, + sig: Some(Signature { + version: 1, + algo: SigAlgo::Ed25519, + bytes: heapless::Vec::from_slice(signature).expect("signature fits capacity"), + }), + } +} + +/// Two observations that carry the *same* id: same content, different age. +/// The signature is ignored by identity, so an aged copy reuses the id. +fn same_id(seq: u64, topic: &[u8]) -> (Observation, Observation) { + let fresh = signed_obs(seq, topic, &[0u8; 8]); + let mut aged = signed_obs(seq, topic, &[0u8; 8]); + aged.age_ms = 500; + (fresh, aged) +} + +/// A freshly inserted observation must be retrievable by id at `now`. +pub fn insert_get(mut make: impl FnMut() -> S) { + let mut store = make(); + let (obs, _) = same_id(1, b"t"); + let id = obs.id(); + assert!(store.insert(obs, 1000).unwrap(), "insert succeeds"); + let got = store.get(&id, 1000).expect("get after insert"); + assert_eq!(got.seq, 1); + assert_eq!(got.topic.as_slice(), b"t"); + assert_eq!(got.payload.as_slice(), b"pay"); +} + +/// A second insert of an already-held id is a no-op: the entry, its stored age +/// lineage and its release state are left untouched. +pub fn dedupe(mut make: impl FnMut() -> S) { + let mut store = make(); + assert!(store.is_empty()); + let (fresh, aged) = same_id(1, b"t"); + let id = fresh.id(); + assert_eq!(id, aged.id(), "an aged copy shares the id"); + assert!(store.insert(fresh, 1000).unwrap(), "first insert"); + assert_eq!(store.len(), 1); + assert!( + !store.insert(aged, 1500).unwrap(), + "duplicate is not reinserted" + ); + assert_eq!(store.len(), 1); + assert!(store.contains(&id)); +} + +/// `get` advances age by elapsed ticks; a deduped insert must not reset the +/// stored age lineage. +pub fn age_lineage(mut make: impl FnMut() -> S) { + let mut store = make(); + let (fresh, _) = same_id(1, b"t"); + let id = fresh.id(); + store.insert(fresh, 1000).unwrap(); + let first = store.get(&id, 1500).unwrap(); + assert_eq!(first.age_ms, 500, "age advances to elapsed ticks"); + // A later insert of the same id must not reset stored_at. + store.insert(same_id(1, b"t").0, 3000).unwrap(); + let second = store.get(&id, 3500).unwrap(); + assert_eq!( + second.age_ms, 2500, + "age keeps climbing from the first insert, not the duplicate" + ); +} + +/// `get` with `now` before the stored tick saturates to zero age — it must not +/// underflow. +pub fn saturating_ticks(mut make: impl FnMut() -> S) { + let mut store = make(); + let (fresh, _) = same_id(1, b"t"); + let id = fresh.id(); + store.insert(fresh, 2000).unwrap(); + let got = store.get(&id, 1000).unwrap(); + assert_eq!(got.age_ms, 0, "elapsed underflow saturates at zero"); +} + +/// Releasing marks an id but leaves it stored, contained and gettable; it must +/// not evict the entry. +pub fn release_surviving_get(mut make: impl FnMut() -> S) { + let mut store = make(); + let (fresh, _) = same_id(1, b"t"); + let id = fresh.id(); + store.insert(fresh, 1000).unwrap(); + store.release(&id); + assert!(store.is_released(&id)); + assert!(store.contains(&id), "release does not evict"); + assert!( + store.get(&id, 1000).is_some(), + "released id is still gettable" + ); + assert_eq!(store.len(), 1); +} + +/// A signature carried by an observation round-trips through the store intact, +/// and a different signature over the same content shares the id. +pub fn sig_round_trip(mut make: impl FnMut() -> S) { + let mut store = make(); + let sig_bytes = [7u8; 32]; + let fresh = signed_obs(1, b"sig", &sig_bytes); + let id = fresh.id(); + store.insert(fresh, 1000).unwrap(); + let got = store.get(&id, 1000).unwrap(); + let sig = got + .sig + .expect("the signature round-trips through the store"); + assert_eq!(sig.version, 1); + assert_eq!(sig.algo, SigAlgo::Ed25519); + assert_eq!(&sig.bytes[..], &sig_bytes[..]); + // A different signature over identical content must not change identity. + let other = signed_obs(1, b"sig", &[9u8; 40]); + assert_eq!(other.id(), id, "the signature must not affect identity"); +} + +/// With a count cap, `trim` drops the oldest (highest current age) first and +/// reports how many it removed. +pub fn trim_order(mut make: impl FnMut() -> S) { + let mut store = make(); + for i in 0..5u64 { + let fresh = signed_obs(i, b"t", &[i as u8]); + assert!(store.insert(fresh, i * 10).unwrap(), "insert {}", i); + } + let policy = TrimPolicy { + keep_per_topic: None, + max_age_ms: None, + max_count: Some(2), + }; + assert_eq!( + store.trim(&policy, 50), + 3, + "trim returns the number removed" + ); + // The two youngest (seq 3 and 4, lowest current age) survive. + for dropped in 0..3u64 { + let fresh = signed_obs(dropped, b"t", &[dropped as u8]); + assert!(!store.contains(&fresh.id()), "seq {} is trimmed", dropped); + } + for kept in 3..5u64 { + let fresh = signed_obs(kept, b"t", &[kept as u8]); + assert!(store.contains(&fresh.id()), "seq {} survives", kept); + } +} + +/// A trimmed id is not tombstoned: it can be reinserted and counted again. +pub fn no_tombstones(mut make: impl FnMut() -> S) { + let mut store = make(); + let fresh = signed_obs(1, b"t", &[1u8; 8]); + let id = fresh.id(); + store.insert(fresh, 0).unwrap(); + let policy = TrimPolicy { + keep_per_topic: None, + max_age_ms: Some(1), + max_count: None, + }; + assert_eq!(store.trim(&policy, 100), 1); + assert!(store.is_empty()); + // The trimmed id is not tombstoned: re-insert it (aged, still same id). + let reinserted = signed_obs(1, b"t", &[1u8; 8]); + let mut reinserted = reinserted; + reinserted.age_ms = 5; + assert_eq!(reinserted.id(), id, "an aged re-insert keeps the id"); + assert!( + store.insert(reinserted, 0).unwrap(), + "a trimmed id can be reinserted" + ); + assert!(store.contains(&id)); + assert_eq!(store.len(), 1); +} + +/// Flushing twice is safe: it must not error and must not evict, and age must +/// keep advancing afterwards. +pub fn idempotent_flush(mut make: impl FnMut() -> S) { + let mut store = make(); + let fresh = signed_obs(1, b"t", &[1u8; 8]); + let id = fresh.id(); + store.insert(fresh, 1000).unwrap(); + assert!( + store.flush(2000).is_ok(), + "the first flush is a best-effort success" + ); + assert!( + store.flush(2000).is_ok(), + "the second flush is a no-op that still succeeds" + ); + assert!(store.contains(&id), "flush does not evict"); + assert_eq!(store.len(), 1); + let got = store.get(&id, 2500).unwrap(); + assert_eq!(got.age_ms, 1500, "age keeps advancing past a flush"); +} diff --git a/tines-sim/Cargo.toml b/tines-sim/Cargo.toml index 48cf75ee..e17e146a 100644 --- a/tines-sim/Cargo.toml +++ b/tines-sim/Cargo.toml @@ -17,3 +17,8 @@ rand = "0.8" [dev-dependencies] proptest = "1" + +[features] +# Re-export tines-core's std-free conformance suite so a backend can exercise +# CappedStore against it: `cargo test --features conformance`. +conformance = ["tines-core/conformance"] diff --git a/tines-sim/src/capped.rs b/tines-sim/src/capped.rs index 249ef76d..3720ccf6 100644 --- a/tines-sim/src/capped.rs +++ b/tines-sim/src/capped.rs @@ -80,4 +80,41 @@ impl Store for CappedStore { fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize { self.inner.trim(policy, now) } + + fn is_empty(&self) -> bool { + self.inner.is_empty() + } + + fn flush(&mut self, now: Tick) -> Result<(), StoreError> { + self.inner.flush(now) + } +} + +/// Conformance-suite shim for [`CappedStore`]. +/// +/// `CappedStore` is a capacity-bounded wrapper over [`MemStore`], so it +/// inherits the memory backend's conformance by delegating; this shim proves +/// the delegation is complete. Run with: +/// `cargo test -p tines-sim --features conformance`. +#[cfg(all(test, feature = "conformance"))] +mod conformance_tests { + use super::CappedStore; + use tines_core::store_conformance; + + fn make() -> impl FnMut() -> CappedStore { + || CappedStore::default() + } + + #[test] + fn capped_conforms() { + store_conformance::insert_get(make()); + store_conformance::dedupe(make()); + store_conformance::age_lineage(make()); + store_conformance::saturating_ticks(make()); + store_conformance::release_surviving_get(make()); + store_conformance::sig_round_trip(make()); + store_conformance::trim_order(make()); + store_conformance::no_tombstones(make()); + store_conformance::idempotent_flush(make()); + } } diff --git a/tines-store/Cargo.toml b/tines-store/Cargo.toml index 2b972be9..487d05d6 100644 --- a/tines-store/Cargo.toml +++ b/tines-store/Cargo.toml @@ -15,3 +15,8 @@ categories = ["science::robotics"] tines-core = { path = "../tines-core", version = "0.3.0" } postcard = { version = "1", features = ["use-std"] } serde = { version = "1", features = ["derive"] } + +[features] +# Re-export tines-core's std-free conformance suite so a backend can exercise +# LogStore against it: `cargo test --features conformance`. +conformance = ["tines-core/conformance"] diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index 0858c9d4..3782acc1 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -232,7 +232,6 @@ pub struct LogStore { file: File, index: BTreeMap, boot_count: u64, - now: Tick, /// 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, @@ -441,7 +440,6 @@ impl LogStore { file, index, boot_count, - now, skipped_records: skipped, }) } @@ -467,30 +465,34 @@ impl LogStore { /// Persist the current index state to disk. /// - /// This is the durability boundary: `flush()` advances each entry's age to - /// the current tick, appends a `Record::Clock` tick marker, and calls - /// `sync_all`, after which the appended observation and release records are - /// durable across a power loss. Because `insert` only reaches disk via a - /// subsequent `flush()`, hosts that need crash durability must call - /// `flush()` before shutdown (see the durability contract in the module - /// docs). - pub fn flush(&mut self) -> Result<(), StoreError> { - // Advance every entry's age up to this tick in memory, so each + /// This is the durability boundary: `flush(now)` advances each entry's age + /// to the caller's `now`, appends a `Record::Clock` tick marker stamped + /// with `now`, and calls `sync_all`, after which the appended observation + /// and release records are durable across a power loss. Because `insert` + /// only reaches disk via a subsequent `flush()`, hosts that need crash + /// durability must call `flush()` before shutdown (see the durability + /// contract in the module docs). + /// + /// `now` is the caller's current tick, which the marker records: anchoring + /// the reopen fold to when the host actually flushed — rather than the last + /// insert tick — keeps ages exact across a reboot. + pub fn flush(&mut self, now: Tick) -> Result<(), StoreError> { + // Advance every entry's age up to the caller's `now` 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::open`]), so this keeps - // the in-memory representation coherent within this run. + // `stored_at` on the next load (see [`LogStore::open`]). 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; + .saturating_add(now.saturating_sub(entry.stored_at)); + entry.stored_at = now; } - // Record a clock marker so the load-time age fold has a durable anchor - // for the time elapsed up to this flush. - let record = Record::Clock { tick: self.now }; + // 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 }; write_record(&mut self.file, &record)?; self.file.sync_all().map_err(|_| StoreError::Backend)?; Ok(()) @@ -703,7 +705,6 @@ impl Store for LogStore { // Dedupe: keep the first-stored entry and its age lineage. return Ok(false); } - self.now = now; // The id is recomputed from `obs` on load, so it is not stored here. let record = Record::Obs { obs: Box::new(obs.clone()), @@ -949,7 +950,7 @@ mod tests { assert!(s.insert(a, 1000).unwrap()); assert!(s.insert(b, 1001).unwrap()); assert!(s.insert(c, 1002).unwrap()); - s.flush().unwrap(); + s.flush(1002).unwrap(); } // Flip a byte in the body of the second data record (Obs(b)). corrupt_body(&path, 2); @@ -980,7 +981,7 @@ mod tests { let mut s = LogStore::open(&path, 0).unwrap(); let a = obs([1; 16], 1, 0, Some(10)); assert!(s.insert(a, 1000).unwrap()); - s.flush().unwrap(); + s.flush(1000).unwrap(); } { let mut f = std::fs::OpenOptions::new() @@ -1044,3 +1045,50 @@ mod tests { std::fs::write(path, data).unwrap(); } } + +/// Conformance-suite shim for [`LogStore`]. +/// +/// Each suite case opens a brand-new temp log, so a case never inherits +/// records (or an age-fold state) from a previous one. Run with: +/// `cargo test -p tines-store --features conformance`. +#[cfg(all(test, feature = "conformance"))] +mod conformance_tests { + use crate::log::LogStore; + use std::path::PathBuf; + use tines_core::store_conformance; + + fn unique_path() -> PathBuf { + static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let n = COUNTER.fetch_add(1, std::sync::atomic::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-log-conformance-{}-{}", + std::process::id(), + key + )); + p + } + + fn make() -> impl FnMut() -> LogStore { + // A fresh file (and thus a fresh age-fold state) on every call. + move || LogStore::open(&unique_path(), 0).expect("open fresh log for conformance") + } + + #[test] + fn log_conforms() { + store_conformance::insert_get(make()); + store_conformance::dedupe(make()); + store_conformance::age_lineage(make()); + store_conformance::saturating_ticks(make()); + store_conformance::release_surviving_get(make()); + store_conformance::sig_round_trip(make()); + store_conformance::trim_order(make()); + store_conformance::no_tombstones(make()); + store_conformance::idempotent_flush(make()); + } +} diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index 4921aae3..f414c760 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -77,7 +77,7 @@ fn reopen_preserves_store() { assert!(!s.is_released(&ib)); s.release(&ib); assert!(s.is_released(&ib)); - s.flush().expect("flush before drop"); + s.flush(1001).expect("flush before drop"); } { @@ -99,7 +99,7 @@ fn ages_rebase_on_load() { let mut s = LogStore::open(&path, 1000).unwrap(); let o = obs([1; 16], 1, 500, Some(10)); assert!(s.insert(o, 1000).unwrap()); - s.flush().unwrap(); + s.flush(1000).unwrap(); } { @@ -130,7 +130,7 @@ fn ages_rebase_after_nonidle_flush() { // non-idle. let b = obs([2; 16], 2, 0, Some(20)); assert!(s.insert(b, 3000).unwrap()); - s.flush().unwrap(); + s.flush(3000).unwrap(); } { @@ -162,7 +162,7 @@ fn ages_include_time_to_clock_marker() { // earlier (at 1000) and sits idle until the store is dropped. let d = obs([2; 16], 2, 0, Some(20)); assert!(s.insert(d, 3000).unwrap()); - s.flush().unwrap(); + s.flush(3000).unwrap(); } { @@ -268,7 +268,7 @@ fn reopen_recovers_from_torn_trailing_record() { let b = obs([2; 16], 2, 0, Some(20)); assert!(s.insert(a, 1000).unwrap()); assert!(s.insert(b, 1001).unwrap()); - s.flush().expect("flush before drop"); + s.flush(1001).expect("flush before drop"); } // Simulate a torn append: write a length prefix claiming 64 body bytes but @@ -305,7 +305,7 @@ fn reopen_recovers_from_oversized_record_length() { let mut s = LogStore::open(&path, 0).expect("open fresh store"); let a = obs([1; 16], 1, 0, Some(10)); assert!(s.insert(a, 1000).unwrap()); - s.flush().expect("flush before drop"); + s.flush(1000).expect("flush before drop"); } { @@ -371,7 +371,7 @@ mod fleet_fixes_f5 { assert!(s.insert(a_obs([1; 16], 1), 1000).unwrap()); assert!(s.insert(a_obs([2; 16], 2), 1001).unwrap()); assert!(s.insert(a_obs([3; 16], 3), 1002).unwrap()); - s.flush().expect("flush"); + s.flush(1002).expect("flush"); } // Corrupt the middle observation (record index 2: Boot, obs-1, obs-2). corrupt_record(&path, 2); @@ -399,7 +399,7 @@ mod fleet_fixes_f5 { let mut s = LogStore::open(&path, 0).expect("open fresh"); assert!(s.insert(a_obs([1; 16], 1), 1000).unwrap()); assert!(s.insert(a_obs([2; 16], 2), 1001).unwrap()); - s.flush().expect("flush"); + s.flush(1001).expect("flush"); } // Append a record header (sync + len = 64) with only 20 of the body bytes. { diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index 7c38847e..befbe02e 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -221,12 +221,12 @@ where } // On a schedule: trim stale observations and flush the store. Flush is - // `LogStore::flush` (not on the `Store` trait), so it's called on the - // concrete store through `Node::store_mut` — tines-core keeps no flush hook. + // `LogStore::flush`, reached through `Node::flush(now)`, which delegates to + // `Store::flush` — the infallible durability boundary. if *rounds >= POLL_CAP { *rounds = 0; node.trim(&TrimPolicy::default(), clock.now()); - match node.store_mut().flush() { + match node.flush(clock.now()) { Ok(()) => {} Err(e) => eprintln!("tines: store flush failed: {e:?}"), }