kyle/tines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(store): snapshot the log after each fold; flush through the trait

Browse Source 
Review findings B1, B2, S2, S3, B3 (store sites), decision I, J and L.

- B1: the clock-marker flush is now the `Store::flush` impl, so
  `Node::flush` writes the marker and fsyncs. The inherent
  `LogStore::flush` is gone (calls resolve to the trait method).
- B2 / decision I: `open` (after the tick fold) and `set_wall_clock`
  (after the wall fold) rewrite the log as a snapshot: header, Boot,
  every live Obs at its current age with stored_at = now, a Release
  for every released id still held, and a closing Clock marker (no
  wall at open, wall_now after the fold). Temp file beside the log,
  fsync, rename, fsync dir. Every stored_at on disk is then in the
  current boot's ticks, so lineage survives any number of reboots, and
  an expiry or floor from the fold survives a later reboot. A leftover
  temp file from an interrupted snapshot is removed on open.
- Lock across the rename: the snapshot's temp file is created 0600 and
  locked before it is renamed over the log, so the file at the path is
  always locked by the live store. An opener that locked a file a
  snapshot just replaced detects it (dev/inode check) and retries.
- S2: the known-downtime fold adds downtime minus the ticks the open
  fold counted minus (now - open_tick), once, not per-entry
  now - stored_at, so a flush before trust no longer double-counts.
- S3: flush and set_wall_clock clamp `now` to the time anchor (the
  latest tick the store has seen); stored_at never moves backwards.
- B3: the derived marker wall and the expiry `lifetime + 1` saturate.
- Decision J: peer_time_estimate returns the latest estimate among
  non-quarantined observations with observed_at; rustdoc says peer
  time is unauthenticated.
- Decision L: trim rustdoc states quarantined entries are exempt from
  every rule; tested.
- A header torn while the log was first created (shorter than the
  header, starting with the magic) opens as a new log; any other short
  file is still refused and left alone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
kyle
2026-10-22 08:44:09 -07:00
co-authored by Claude Opus 5.5
parent 8c97963364
commit 382254faaf
3 changed files with 889 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tines-store/src/lib.rs
+2 -1
View File
@@ -1,4 +1,5 @@
//! tines-store — persistent append-only `LogStore` backend. //! tines-store — persistent `LogStore` backend: an append-only log, rewritten
//! as a snapshot at open and after the wall-clock fold.
//! //!
//! This crate is the host-side (std) persistence layer for tines. The `Store` //! This crate is the host-side (std) persistence layer for tines. The `Store`
//! trait, `TrimPolicy`, and `StoreError` live in `tines-core` (which stays //! trait, `TrimPolicy`, and `StoreError` live in `tines-core` (which stays
tines-store/src/log.rs
+512 -165
View File
@@ -30,9 +30,11 @@
//! incremented on every open. `first_seq() = boot_count << 32`, which is //! incremented on every open. `first_seq() = boot_count << 32`, which is
//! core's documented contract; a resumed node therefore never reuses a //! core's documented contract; a resumed node therefore never reuses a
//! sequence number. //! sequence number.
//! * `Record::Clock` — a tick marker written by [`LogStore::flush`]. On load //! * `Record::Clock` — a tick marker written by `flush` (the [`Store::flush`]
//! each live observation's age is folded up to the last clock marker so //! impl) and at the end of every snapshot, with a wall time once trusted
//! pre-reboot elapsed time is recovered (see [`LogStore::open`]). //! time is set. On load each live observation's age is folded up to the
//! last clock marker so pre-reboot elapsed time is recovered (see
//! [`LogStore::open`]).
//! //!
//! On load the index is rebuilt by replaying the log, then every age is //! On load the index is rebuilt by replaying the log, then every age is
//! rebased to load time: each entry's `age_ms` is folded by the time elapsed //! rebased to load time: each entry's `age_ms` is folded by the time elapsed
@@ -44,39 +46,64 @@
//! reboot instead of freezing at (and under-counting from) their pre-shutdown //! reboot instead of freezing at (and under-counting from) their pre-shutdown
//! value. //! value.
//! //!
//! Snapshots (spec §9, decision I)
//!
//! A `stored_at` is a tick from the boot that wrote it, so an append-only log
//! would mix tick domains across boots, and the fold's results would live only
//! in memory. So the log is rewritten as a snapshot at two points: at the end
//! of [`LogStore::open`] (after the tick fold) and at the end of
//! [`LogStore::set_wall_clock`]. A snapshot is the header, the current `Boot`
//! record, every live observation with its current age and `stored_at = now`,
//! a `Release` record for every released id still held, and a closing clock
//! marker at `now` (with the wall time after `set_wall_clock`, none at open).
//! It holds no trimmed records. Every `stored_at` on disk then belongs to the
//! current boot, and the fold's results (expiries, floors, known-downtime
//! advances) are durable.
//!
//! Crash recovery & robustness //! Crash recovery & robustness
//! //!
//! * Framing & checksums. Every record carries a [`SYNC_MARKER`] and a CRC-32C //! * Framing & checksums. Every record carries a [`SYNC_MARKER`] and a CRC-32C
//! over its body, so any corruption is detectable rather than silently //! over its body, so any corruption is detectable rather than silently
//! misparsed. //! misparsed.
//! * Torn trailing records. The log is append-only, so a crash mid-append can //! * Torn trailing records. Between snapshots the log is append-only, so a
//! leave a partial trailing record — a header whose body ran past //! crash mid-append can leave a partial trailing record — a header whose
//! end-of-file — or a bit-roted/corrupt record. [`LogStore::open`] recovers //! body ran past end-of-file — or a bit-roted/corrupt record.
//! by scanning for the next well-formed record: a record whose marker, //! [`LogStore::open`] recovers by scanning for the next well-formed record: a
//! length, CRC or body-decode is bad is *skipped* (scan forward for the next //! record whose marker, length, CRC or body-decode is bad is *skipped* (scan
//! `SYNC_MARKER` and verify its CRC before accepting it), and the bytes //! forward for the next `SYNC_MARKER` and verify its CRC before accepting
//! leading up to it are dropped. Only bad bytes at the very end, with no //! it), and the bytes leading up to it are dropped. Only bad bytes at the
//! valid record after them, count as a torn tail; a torn tail is truncated //! very end, with no valid record after them, count as a torn tail. The open
//! and the file and its directory are fsynced. //! snapshot rewrites the log without either.
//! * Snapshots are atomic. A snapshot is written to a temporary file in the
//! log's directory, fsynced, renamed over the log, and then the directory is
//! fsynced. A crash before the rename leaves the old log intact (the next
//! open repeats the same fold and removes the leftover temporary file);
//! after it, the snapshot is the log.
//! * Torn first header. A file shorter than the header that starts with the
//! magic was torn while the log was first created; it holds no records, so
//! it opens as a new log. Any other short file is refused.
//! * Write cap. `write_record` refuses a record over [`MAX_RECORD_BYTES`], so //! * Write cap. `write_record` refuses a record over [`MAX_RECORD_BYTES`], so
//! the writer can no longer produce a record the reader rejects. //! the writer can no longer produce a record the reader rejects.
//! //!
//! * Durability contract. `flush()` appends a `Record::Clock` and `sync_all`s, //! * Durability contract. `flush(now)` appends a `Record::Clock` and
//! making the appended observation and release records durable across a power //! `sync_all`s, making the appended observation and release records durable
//! loss. `insert`, `release`, and `trim` update the in-memory index and (for //! across a power loss. `insert`, `release`, and `trim` update the in-memory
//! `insert`/`trim`) append records, but those records reach disk only on the //! index and append records, but those records are durable across a power
//! next `flush()`. Hosts that need crash durability must `flush()` before //! loss only after the next flush (or snapshot). Hosts that need crash
//! shutdown (see the durability contract in the module docs). `release`, by //! durability must flush before shutdown. `release` appends its `Release`
//! contrast, appends its `Release` record immediately (best-effort: an append //! record immediately (best-effort: an append error is logged), so a release
//! error is logged), so a single release is durable across a kill-without-flush //! survives a process kill without waiting for a flush.
//! without waiting for a flush.
//! //!
//! Isolation & integrity //! Isolation & integrity
//! //!
//! * Exclusive lock. `open` takes an exclusive `File::try_lock`; a second //! * Exclusive lock. `open` takes an exclusive `File::try_lock` on the log
//! opener gets a clear error instead of interleaving on the same file. //! file; a second opener gets a clear error instead of interleaving on the
//! * Permissions. The log file is created mode `0600` and its parent directory //! same file. A snapshot locks its temporary file before renaming it over
//! (created if missing) mode `0700`. //! the log, so the file at the log's path stays locked while the store is
//! alive, and an opener that locked a file a snapshot just replaced notices
//! and retries on the new one.
//! * Permissions. The log file (and a snapshot's temporary file) is created
//! mode `0600` and its parent directory (created if missing) mode `0700`.
//! * Format version. The file starts with a `"TINESLOG"` header carrying a //! * Format version. The file starts with a `"TINESLOG"` header carrying a
//! `format_version`; a missing or unknown header is an error that names the //! `format_version`; a missing or unknown header is an error that names the
//! fix (delete the log). The 1.0 work bumps `format_version` for the new //! fix (delete the log). The 1.0 work bumps `format_version` for the new
@@ -244,18 +271,31 @@ struct Entry {
/// boot counter, durable releases, crash recovery, exclusive lock, and /// boot counter, durable releases, crash recovery, exclusive lock, and
/// permissions. /// permissions.
pub struct LogStore { pub struct LogStore {
/// Where the log lives; the snapshot's temporary file sits beside it.
path: PathBuf,
/// The live log, opened for appends and holding the exclusive lock.
file: File, file: File,
index: BTreeMap<ObservationId, Entry>, index: BTreeMap<ObservationId, Entry>,
boot_count: u64, boot_count: u64,
/// Number of corrupt records skipped while rebuilding the index on the last /// 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. /// [`Self::open`] (0 if the log was clean). Exposed for hosts to report.
skipped_records: usize, skipped_records: usize,
/// The last clock marker seen while rebuilding the index, if any: the /// The previous boot's last clock marker, `(tick, wall)`, as read from the
/// `(marker_tick, marker_wall)` pair. The reopen fold anchors to the later /// log at [`Self::open`]. The open snapshot replaces it on disk with a
/// of this and the boot record; the downtime fold anchors to it. /// marker that has no wall time, so it is kept here for the downtime fold
/// in [`Self::set_wall_clock`].
last_marker: Option<(Tick, Option<u64>)>, last_marker: Option<(Tick, Option<u64>)>,
/// Caller ticks of downtime already folded during [`Self::open`]. /// Caller ticks of downtime already folded during [`Self::open`]: `now`
/// at open minus the previous marker's tick, saturating.
counted_ticks: u64, counted_ticks: u64,
/// The caller's `now` at [`Self::open`]. The fold subtracts the ticks
/// elapsed since then exactly once (S2 of the 1.0 review).
open_tick: Tick,
/// The time anchor: the latest tick the store has been given (open,
/// insert, flush or fold). An earlier `now` passed to `flush` or
/// `set_wall_clock` is clamped up to it, so `stored_at` and the clock
/// markers never move backwards.
anchor: Tick,
/// The `unknown_downtime_age_ms` broadcast floor configured on open. /// The `unknown_downtime_age_ms` broadcast floor configured on open.
unknown_downtime_age_ms: u64, unknown_downtime_age_ms: u64,
/// Wall-clock value captured by [`Self::set_wall_clock`]; `Some` once the /// Wall-clock value captured by [`Self::set_wall_clock`]; `Some` once the
@@ -368,9 +408,16 @@ impl LogStore {
/// hand out up to 2³² sequence numbers before the high word would roll /// hand out up to 2³² sequence numbers before the high word would roll
/// over, which is more than any robot needs. /// over, which is more than any robot needs.
/// ///
/// If the log has a corrupt middle record it is skipped and only a torn /// If the log has a corrupt middle record it is skipped and a torn tail is
/// *tail* is truncated; the number of corrupt records skipped is reported /// dropped; the number of corrupt records skipped is reported once on
/// once on stderr and is available via [`Self::skipped_records`]. /// stderr and is available via [`Self::skipped_records`].
///
/// Before returning, `open` rewrites the log as a snapshot of what it
/// loaded (spec §9, decision I; see the module docs), with every age
/// folded to `now` and a closing clock marker at `now` that carries no
/// wall time. That makes the new boot counter durable, keeps every
/// `stored_at` on disk in this boot's ticks, and compacts the log. If the
/// snapshot can't be written, `open` fails and the old log is untouched.
/// ///
/// An exclusive lock is taken on the file, so a second `open` while this /// An exclusive lock is taken on the file, so a second `open` while this
/// one is alive fails with [`StoreError::Backend`]. /// one is alive fails with [`StoreError::Backend`].
@@ -382,44 +429,33 @@ impl LogStore {
// Ensure the parent directory exists (mode 0700 if we create it). // Ensure the parent directory exists (mode 0700 if we create it).
ensure_parent_dir(path)?; ensure_parent_dir(path)?;
// Open for reads (to load) and appends (to write later). Append mode is // Open the log and take its exclusive lock (see `open_locked`).
// correct because we load by reading the whole file below, and every let file = open_locked(path)?;
// mutation appends to the end of the log. Set mode 0600 via
// set_permissions below so it is enforced whether or not the file
// pre-existed.
let mut tmp = OpenOptions::new();
let opts = tmp.read(true).create(true).append(true);
let mut file = opts.open(path).map_err(|_| StoreError::Backend)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
file.set_permissions(std::fs::Permissions::from_mode(0o600))
.map_err(|_| StoreError::Backend)?;
}
// Exclusive lock: a second opener must not interleave on the same file. // We hold the lock, so nobody else is mid-snapshot: a temporary file
// (`try_lock` takes an exclusive lock on this toolchain's std; the // left beside the log is from a snapshot interrupted before its
// spec's `try_lock_exclusive` is not present here.) // rename. The log itself is intact, so the leftover is discarded.
file.try_lock().map_err(|_| { let tmp_path = snapshot_tmp_path(path);
if tmp_path.exists() {
let _ = std::fs::remove_file(&tmp_path);
eprintln!( eprintln!(
"tines-store: refusing to open {}: another process already \ "tines-store: removed {} left by an interrupted snapshot; \
holds the store lock (an earlier instance may still be \ the log itself is intact.",
running); delete nothing, just wait for it to release.", tmp_path.display()
path.display()
); );
StoreError::Backend }
})?;
let data = std::fs::read(path).map_err(|_| StoreError::Backend)?; let data = std::fs::read(path).map_err(|_| StoreError::Backend)?;
// Header: write a fresh one on an empty log, otherwise validate the // Header: an empty log (or one whose header was torn while it was
// existing one. A missing or unknown header is an error that names the // first created) is new; otherwise validate it. A missing or unknown
// fix (delete the log). // header is an error that names the fix (delete the log).
if data.is_empty() { let data: &[u8] = if is_torn_new_log(&data) {
write_header(&mut file)?; &[]
} else { } else {
validate_header(&data, path)?; validate_header(&data, path)?;
} &data
};
let mut index: BTreeMap<ObservationId, Entry> = BTreeMap::new(); let mut index: BTreeMap<ObservationId, Entry> = BTreeMap::new();
let mut dead: HashSet<ObservationId> = HashSet::new(); let mut dead: HashSet<ObservationId> = HashSet::new();
@@ -449,7 +485,7 @@ impl LogStore {
// at or after `pos`. If it starts after `pos`, the bytes in between // at or after `pos`. If it starts after `pos`, the bytes in between
// were a corrupt/torn record we skipped. If there is no well-formed // were a corrupt/torn record we skipped. If there is no well-formed
// record left, the remainder is a torn tail. // record left, the remainder is a torn tail.
match parse_valid_record(&data, pos) { match parse_valid_record(data, pos) {
Some((start, end, rec)) => { Some((start, end, rec)) => {
if start > pos { if start > pos {
skipped += 1; skipped += 1;
@@ -504,19 +540,13 @@ impl LogStore {
index.remove(&id); index.remove(&id);
} }
// A torn tail (bad bytes at EOF with no valid record after) is truncated // A torn tail (bad bytes at EOF with no valid record after) is dropped:
// back to the last complete boundary, then the file and its directory // the open snapshot below rewrites the log from the records loaded.
// are fsynced so the clean tail is durable.
let dropped = data.len().saturating_sub(good_end); let dropped = data.len().saturating_sub(good_end);
if dropped > 0 { if dropped > 0 {
file.set_len(good_end as u64)
.map_err(|_| StoreError::Backend)?;
file.sync_all().map_err(|_| StoreError::Backend)?;
let _ = sync_dir(path);
eprintln!( eprintln!(
"tines-store: truncated {} torn byte(s) at the tail of {} \ "tines-store: dropped {} torn byte(s) at the tail of {} \
while loading ({} complete record(s) loaded); fsynced file \ while loading ({} complete record(s) loaded).",
and directory.",
dropped, dropped,
path.display(), path.display(),
records_loaded, records_loaded,
@@ -526,20 +556,19 @@ impl LogStore {
if skipped > 0 { if skipped > 0 {
eprintln!( eprintln!(
"tines-store: skipped {} corrupt record(s) in {} while loading \ "tines-store: skipped {} corrupt record(s) in {} while loading \
({} complete record(s) loaded); the log has been repaired \ ({} complete record(s) loaded); the open snapshot rewrites \
and will rewrite a clean tail on the next mutation.", the log without them.",
skipped, skipped,
path.display(), path.display(),
records_loaded, records_loaded,
); );
} }
// Increment the boot counter and durably persist it before returning, so // Increment the boot counter. The open snapshot below makes it durable
// a later open reads a strictly higher value and first_seq() climbs. // before returning, so a later open reads a strictly higher value and
let boot_count = last_boot_count.map(|c| c + 1).unwrap_or(1); // first_seq() climbs. Saturating: a corrupt-but-valid huge count must
let record = Record::Boot { count: boot_count }; // not panic (first_seq then repeats, which core dedupes safely).
write_record(&mut file, &record)?; let boot_count = last_boot_count.map_or(1, |c| c.saturating_add(1));
file.sync_all().map_err(|_| StoreError::Backend)?;
// Fold each live observation's age up to the last clock marker and then // 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 // to load time. Anchoring to each entry's insert tick keeps the fold
@@ -562,18 +591,28 @@ impl LogStore {
entry.stored_at = now; entry.stored_at = now;
} }
Ok(LogStore { let mut store = LogStore {
path: path.to_path_buf(),
file, file,
index, index,
boot_count, boot_count,
skipped_records: skipped, skipped_records: skipped,
last_marker, last_marker,
counted_ticks: last_marker.map_or(0, |(tick, _)| now.saturating_sub(tick)), counted_ticks: last_marker.map_or(0, |(tick, _)| now.saturating_sub(tick)),
open_tick: now,
anchor: now,
unknown_downtime_age_ms, unknown_downtime_age_ms,
wall_at_set: None, wall_at_set: None,
tick_at_set: None, tick_at_set: None,
fold_report: None, fold_report: None,
}) };
// Decision I: rewrite the log as a snapshot, so every `stored_at` on
// disk belongs to this boot and the new boot count is durable. The
// store can't know the wall time at open, so the closing marker has
// none: a restart before trusted time takes the unknown path.
store.write_snapshot(now, None)?;
Ok(store)
} }
/// The sequence number the next `Node::observe` should use for this boot. /// The sequence number the next `Node::observe` should use for this boot.
@@ -595,50 +634,88 @@ impl LogStore {
self.skipped_records self.skipped_records
} }
/// Persist the current index state to disk. /// Clamp `now` to the time anchor and advance the anchor to it.
/// fn advance_anchor(&mut self, now: Tick) -> Tick {
/// This is the durability boundary: `flush(now)` advances each entry's age self.anchor = self.anchor.max(now);
/// to the caller's `now`, appends a `Record::Clock` tick marker stamped self.anchor
/// 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`]).
for entry in self.index.values_mut() {
entry.obs.age_ms = entry
.obs
.age_ms
.saturating_add(now.saturating_sub(entry.stored_at));
entry.stored_at = now;
}
// Record a clock marker stamped with the caller's `now` so the load-time /// Advance every entry's age to `now` in memory and set its `stored_at` to
// age fold anchors to when the host actually flushed — not the last /// `now`. `now` must not be behind any entry's `stored_at` (callers pass
// insert tick (that stale-marker bug was F18). /// the anchor), so no `stored_at` moves backwards.
// fn rebase_to(&mut self, now: Tick) {
// The wall value is *derived*: if a trusted wall clock has been for entry in self.index.values_mut() {
// established (see [`LogStore::set_wall_clock`]) the marker carries entry.obs.age_ms = current_age(&entry.obs, entry.stored_at, now);
// `wall_at_set + (now - tick_at_set)`; before that it is `None`, which entry.stored_at = entry.stored_at.max(now);
// is what makes the reopen fold treat this marker's downtime as unknown. }
let wall = match (self.wall_at_set, self.tick_at_set) { }
/// The wall time a clock marker written at tick `now` carries: derived as
/// `wall_at_set + (now - tick_at_set)` once trusted time is set, else
/// `None`. Saturating, so a peer's absurd wall time can't overflow it.
fn marker_wall(&self, now: Tick) -> Option<u64> {
match (self.wall_at_set, self.tick_at_set) {
(Some(wall_at_set), Some(tick_at_set)) => { (Some(wall_at_set), Some(tick_at_set)) => {
Some(wall_at_set + now.saturating_sub(tick_at_set)) Some(wall_at_set.saturating_add(now.saturating_sub(tick_at_set)))
} }
_ => None, _ => None,
}
}
/// Rewrite the log as a snapshot of the current state at tick `now`
/// (decision I): header, the current `Boot` record, every live observation
/// with its current age and `stored_at = now`, a `Release` record for every
/// released id still held, and a closing clock marker at `now` carrying
/// `wall`. Trimmed records are not carried over.
///
/// The snapshot goes to a temporary file beside the log (created mode
/// `0600` and locked before it becomes the log), is fsynced, renamed over
/// the log, and then the directory is fsynced. A crash before the rename
/// leaves the old log intact; after it, the snapshot is the log. Because
/// the temporary file is locked before the rename, the file at the log's
/// path is locked by this store at every instant (see `open_locked`).
///
/// `now` must already be clamped to the anchor. The in-memory entries are
/// rebased to `now` as well, so memory and disk agree.
fn write_snapshot(&mut self, now: Tick, wall: Option<u64>) -> Result<(), StoreError> {
self.rebase_to(now);
let mut buf = Vec::new();
buf.extend_from_slice(&header_bytes());
buf.extend(encode_record(&Record::Boot {
count: self.boot_count,
})?);
for entry in self.index.values() {
buf.extend(encode_record(&Record::Obs {
obs: Box::new(entry.obs.clone()),
stored_at: now,
})?);
}
for (id, entry) in self.index.iter() {
if entry.released {
buf.extend(encode_record(&Record::Release { id: *id })?);
}
}
buf.extend(encode_record(&Record::Clock { tick: now, wall })?);
let tmp_path = snapshot_tmp_path(&self.path);
// We hold the log's lock, so any file here is a leftover.
let _ = std::fs::remove_file(&tmp_path);
let tmp = match write_locked_tmp(&tmp_path, &buf) {
Ok(tmp) => tmp,
Err(e) => {
let _ = std::fs::remove_file(&tmp_path);
return Err(e);
}
}; };
let record = Record::Clock { tick: now, wall }; if std::fs::rename(&tmp_path, &self.path).is_err() {
write_record(&mut self.file, &record)?; let _ = std::fs::remove_file(&tmp_path);
self.file.sync_all().map_err(|_| StoreError::Backend)?; return Err(StoreError::Backend);
Ok(()) }
// The snapshot is the log now: append to it from here on. Dropping the
// old handle releases the lock on the old, now unlinked, file.
self.file = tmp;
sync_dir(&self.path)
} }
/// Append a single record to the log. /// Append a single record to the log.
@@ -654,16 +731,28 @@ impl LogStore {
/// observation is quarantined. This is the moment the host reports a /// observation is quarantined. This is the moment the host reports a
/// *trusted* wall clock. /// *trusted* wall clock.
/// ///
/// **Known downtime**: the last marker on disk carries a wall time and /// **Known downtime**: the previous boot's last marker (as read at open)
/// `wall_now` is not behind it. Every quarantined observation gains the wall /// carries a wall time and `wall_now` is not behind it. Every quarantined
/// elapsed since that marker (`wall_now - marker_wall`), minus the ticks /// observation gains the wall time elapsed since that marker
/// already counted during this open (saturating to zero). /// (`wall_now - marker_wall`), minus the ticks already counted for it:
/// those the open fold added past the marker tick, and this boot's ticks
/// since open (`now - open_tick`), counted once whether or not a flush
/// ran in between. Saturating to zero.
/// ///
/// **Unknown downtime**: no marker, a marker without a wall time, or a /// **Unknown downtime**: no marker, a marker without a wall time, or a
/// marker whose wall time is ahead of `wall_now`. Quarantined addressed /// marker whose wall time is ahead of `wall_now`. Quarantined addressed
/// observations are aged past their lifetime (never delivered/forwarded); /// observations are aged past their lifetime (never delivered/forwarded);
/// quarantined broadcasts are floored to `max(age, unknown_downtime_age_ms)`. /// quarantined broadcasts are floored to `max(age, unknown_downtime_age_ms)`.
/// ///
/// The fold then rewrites the log as a snapshot (decision I) whose closing
/// marker carries `wall_now`, so its results are durable: an addressed
/// observation it expired stays released after a later reboot, and a
/// floored broadcast keeps its floor. If the snapshot fails, the fold
/// still holds in memory, but clock markers carry no wall time until the
/// next open, so that open takes the unknown path again.
///
/// A `now` earlier than the store's time anchor is clamped to it.
///
/// This runs exactly once; a second call returns the first report unchanged /// This runs exactly once; a second call returns the first report unchanged
/// (see the `set_wall_clock_runs_once` test). /// (see the `set_wall_clock_runs_once` test).
/// ///
@@ -676,6 +765,7 @@ impl LogStore {
return report; return report;
} }
let now = self.advance_anchor(now);
let mut expired_addressed = 0usize; let mut expired_addressed = 0usize;
let mut floored_broadcasts = 0usize; let mut floored_broadcasts = 0usize;
let mut known_downtime_ms: Option<u64> = None; let mut known_downtime_ms: Option<u64> = None;
@@ -684,16 +774,18 @@ impl LogStore {
// There is a marker with a usable wall time. Downtime is known. // There is a marker with a usable wall time. Downtime is known.
Some((_marker_tick, Some(marker_wall))) if marker_wall <= wall_now => { Some((_marker_tick, Some(marker_wall))) if marker_wall <= wall_now => {
let downtime_ms = wall_now.saturating_sub(marker_wall); let downtime_ms = wall_now.saturating_sub(marker_wall);
// Ticks already folded in during the reopen (load) fold. // The wall time since the marker, minus the ticks the open fold
let counted = self.counted_ticks; // already counted (open tick past the marker tick), minus this
let extra_ticks = downtime_ms.saturating_sub(counted) as Tick; // boot's ticks since open. Every quarantined entry's age already
// includes the latter, whether or not a flush has rebased it in
// between, so it is subtracted once for all (S2).
let added = downtime_ms
.saturating_sub(self.counted_ticks)
.saturating_sub(now.saturating_sub(self.open_tick));
for entry in self.index.values_mut() { for entry in self.index.values_mut() {
if !entry.quarantined { if !entry.quarantined {
continue; 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.obs.age_ms = entry.obs.age_ms.saturating_add(added);
entry.quarantined = false; entry.quarantined = false;
} }
@@ -714,8 +806,10 @@ impl LogStore {
// lifetime and released (withheld from delivery), exactly // lifetime and released (withheld from delivery), exactly
// as if it had been delivered and acknowledged. // as if it had been delivered and acknowledged.
let lifetime = entry.obs.dest.as_ref().map(|d| d.lifetime_ms).unwrap_or(0); let lifetime = entry.obs.dest.as_ref().map(|d| d.lifetime_ms).unwrap_or(0);
// Saturating: at `lifetime_ms = u64::MAX` no age is
// past it, and `released` alone withholds it.
if entry.obs.age_ms <= lifetime { if entry.obs.age_ms <= lifetime {
entry.obs.age_ms = lifetime + 1; entry.obs.age_ms = lifetime.saturating_add(1);
expired_addressed += 1; expired_addressed += 1;
} }
entry.released = true; entry.released = true;
@@ -740,6 +834,24 @@ impl LogStore {
self.wall_at_set = Some(wall_now); self.wall_at_set = Some(wall_now);
self.tick_at_set = Some(now); self.tick_at_set = Some(now);
// Decision I: snapshot the folded state, so the fold is durable (an
// expired message stays released, a floored broadcast keeps its floor)
// and the closing marker carries the wall time.
if self.write_snapshot(now, Some(wall_now)).is_err() {
// The fold holds in memory, but not on disk. If later markers
// carried a wall time, the next open would see a *known* downtime
// and the unfolded records on disk would come back: an expired
// message resurrected. So markers stay without a wall time for
// this run, and the next open takes the unknown path again.
self.wall_at_set = None;
self.tick_at_set = None;
eprintln!(
"tines-store: failed to snapshot {} after the wall-clock fold; \
clock markers carry no wall time until the next open.",
self.path.display()
);
}
let report = FoldReport { let report = FoldReport {
downtime_ms: known_downtime_ms, downtime_ms: known_downtime_ms,
expired_addressed, expired_addressed,
@@ -759,40 +871,43 @@ impl LogStore {
report report
} }
/// Estimate the true wall time from the observations stored since this /// Estimate the current wall time from peers' observations: the latest
/// open — the peer-observation fallback in the trusted-time state machine /// estimate among the observations that aren't quarantined and carry an
/// (`HostClock::anchor_wall`). /// `observed_at` (decision J).
/// ///
/// Scans the entries stored *since this open*: the non-quarantined ones /// Each candidate's estimate is
/// (quarantined ones are invisible to the scan, being pre-open data whose
/// age cannot be verified). Among those carrying an `observed_at`, the
/// youngest (smallest `age_ms`) gives the best estimate, because a younger
/// observation's age has been trusted longest. The estimate is
/// ///
/// ```text /// ```text
/// observed_at + age_ms + (now - stored_at) /// observed_at + age_ms + (now - stored_at)
/// ``` /// ```
/// ///
/// i.e. the peer's wall time when it made the observation, aged by its /// i.e. the observer's wall time when it made the observation, plus its
/// current age plus the ticks elapsed since we stored it. `age_ms` is a /// age when it arrived, plus the ticks elapsed since. Every term saturates.
/// lower bound on the age at send time, so the estimate is never later than /// Because `age_ms` is a lower bound, an honest estimate is never later
/// the true current wall time — a peer cannot report a younger age than its /// than the true time, so the latest estimate is the closest one.
/// own clock implies.
/// ///
/// Returns `None` when no stored observation carries an `observed_at` yet: /// Only observations inserted since this open count: anything held from
/// there is nothing to derive time from. /// before it is quarantined, and its age isn't trusted yet.
///
/// **Peer time is unauthenticated** until signatures land in 1.1: a peer
/// can claim any `observed_at`, including one absurdly far ahead (this
/// returns `u64::MAX` rather than overflowing). The host bounds the
/// estimate and collects estimates for a short window before adopting
/// the latest one with [`Self::set_wall_clock`] (spec §9).
///
/// Returns `None` when no candidate carries an `observed_at` yet.
pub fn peer_time_estimate(&self, now: Tick) -> Option<u64> { pub fn peer_time_estimate(&self, now: Tick) -> Option<u64> {
self.index self.index
.values() .values()
.filter(|entry| !entry.quarantined && entry.obs.observed_at.is_some()) .filter(|entry| !entry.quarantined)
.min_by_key(|entry| entry.obs.age_ms) .filter_map(|entry| {
.and_then(|entry| {
entry.obs.observed_at.map(|observed_at| { entry.obs.observed_at.map(|observed_at| {
observed_at observed_at
.saturating_add(entry.obs.age_ms) .saturating_add(entry.obs.age_ms)
.saturating_add(now.saturating_sub(entry.stored_at)) .saturating_add(now.saturating_sub(entry.stored_at))
}) })
}) })
.max()
} }
/// The [`FoldReport`] from the (single) `set_wall_clock`, before it runs or /// The [`FoldReport`] from the (single) `set_wall_clock`, before it runs or
@@ -814,14 +929,127 @@ impl WallClockStore for LogStore {
} }
} }
/// Write the log header: `magic ‖ format_version ‖ reserved`. /// The log header: `magic ‖ format_version ‖ reserved`.
fn write_header(file: &mut File) -> Result<(), StoreError> { fn header_bytes() -> [u8; HEADER_LEN] {
file.write_all(&MAGIC).map_err(|_| StoreError::Backend)?; let mut out = [0u8; HEADER_LEN];
file.write_all(&FORMAT_VERSION.to_le_bytes()) out[0..8].copy_from_slice(&MAGIC);
.map_err(|_| StoreError::Backend)?; out[8..10].copy_from_slice(&FORMAT_VERSION.to_le_bytes());
file.write_all(&0u16.to_le_bytes()) // reserved for future use // out[10..12]: reserved for future use, zero.
.map_err(|_| StoreError::Backend)?; out
Ok(()) }
/// Whether `data` is a log with no records yet: empty, or a header torn by a
/// crash while the log was first created.
///
/// A torn header is shorter than [`HEADER_LEN`] and starts with (a prefix of)
/// the magic. Records only ever follow a complete header, so such a file holds
/// no data, and treating it as new loses nothing. Any other short file may not
/// be ours, so it is left to [`validate_header`] to refuse.
fn is_torn_new_log(data: &[u8]) -> bool {
if data.len() >= HEADER_LEN {
return false;
}
let n = data.len().min(MAGIC.len());
data[..n] == MAGIC[..n]
}
/// The temporary file a snapshot is written to before it is renamed over the
/// log at `path`: same directory, so the rename is atomic.
fn snapshot_tmp_path(path: &Path) -> PathBuf {
let mut name = path.file_name().unwrap_or_default().to_os_string();
name.push(".snapshot-tmp");
path.with_file_name(name)
}
/// Open the log at `path` (creating it, mode `0600`) and take its exclusive
/// lock.
///
/// The lock lives on the log file itself. A snapshot replaces that file by
/// rename, so the lock must follow it: the snapshot locks its temporary file
/// *before* renaming it over the log (see [`LogStore::write_snapshot`]), so
/// whatever file is at `path` is locked while the store is alive. A second
/// opener can still race a snapshot: it opens the old file, and the store
/// releases that file's lock right after the rename. So after locking, the
/// opener checks that its file is still the one at `path`. If not, it retries
/// on the new file, whose lock the live store holds.
fn open_locked(path: &Path) -> Result<File, StoreError> {
const ATTEMPTS: usize = 8;
for _ in 0..ATTEMPTS {
let mut opts = OpenOptions::new();
opts.read(true).create(true).append(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
opts.mode(0o600);
}
let file = opts.open(path).map_err(|_| StoreError::Backend)?;
// Enforce 0600 whether or not the file pre-existed.
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
file.set_permissions(std::fs::Permissions::from_mode(0o600))
.map_err(|_| StoreError::Backend)?;
}
// Exclusive lock: a second opener must not interleave on the same file.
file.try_lock().map_err(|_| {
eprintln!(
"tines-store: refusing to open {}: another process already \
holds the store lock (an earlier instance may still be \
running); delete nothing, just wait for it to release.",
path.display()
);
StoreError::Backend
})?;
if is_file_at(&file, path) {
return Ok(file);
}
}
eprintln!(
"tines-store: refusing to open {}: the log kept being replaced while \
taking its lock; another process may be using it.",
path.display()
);
Err(StoreError::Backend)
}
/// Whether `file` is the file currently at `path`, not one a rename replaced.
#[cfg(unix)]
fn is_file_at(file: &File, path: &Path) -> bool {
use std::os::unix::fs::MetadataExt;
match (file.metadata(), std::fs::metadata(path)) {
(Ok(a), Ok(b)) => a.dev() == b.dev() && a.ino() == b.ino(),
_ => false,
}
}
/// Without inode numbers there is nothing to compare; trust the open.
#[cfg(not(unix))]
fn is_file_at(_file: &File, _path: &Path) -> bool {
true
}
/// Create `tmp_path` (mode `0600`, which must not exist yet), lock it, write
/// `bytes` and fsync it. Returns the locked handle, ready to append to once it
/// is renamed over the log.
fn write_locked_tmp(tmp_path: &Path, bytes: &[u8]) -> Result<File, StoreError> {
let mut opts = OpenOptions::new();
opts.read(true).create_new(true).append(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
opts.mode(0o600);
}
let mut file = opts.open(tmp_path).map_err(|_| StoreError::Backend)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
file.set_permissions(std::fs::Permissions::from_mode(0o600))
.map_err(|_| StoreError::Backend)?;
}
file.try_lock().map_err(|_| StoreError::Backend)?;
file.write_all(bytes).map_err(|_| StoreError::Backend)?;
file.sync_all().map_err(|_| StoreError::Backend)?;
Ok(file)
} }
/// Reject a log whose header is missing, truncated, or an unknown version, /// Reject a log whose header is missing, truncated, or an unknown version,
@@ -1006,7 +1234,7 @@ impl Store for LogStore {
/// Insert an observation. /// Insert an observation.
/// ///
/// The record is appended to the log, but is durable across a power loss /// The record is appended to the log, but is durable across a power loss
/// only after a call to [`Self::flush`]; hosts that need crash durability /// only after a call to [`Store::flush`]; hosts that need crash durability
/// must flush before shutdown (see the durability contract in the module /// must flush before shutdown (see the durability contract in the module
/// docs). /// docs).
fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, StoreError> { fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, StoreError> {
@@ -1021,6 +1249,7 @@ impl Store for LogStore {
stored_at: now, stored_at: now,
}; };
self.append(&record)?; self.append(&record)?;
self.advance_anchor(now);
self.index.insert( self.index.insert(
id, id,
Entry { Entry {
@@ -1089,7 +1318,7 @@ impl Store for LogStore {
/// appended to the log at the same time (best-effort: an append error is /// appended to the log at the same time (best-effort: an append error is
/// logged, not raised, keeping [`Store::release`] infallible). The release /// logged, not raised, keeping [`Store::release`] infallible). The release
/// is therefore durable across a kill-without-flush — unlike a flush-only /// is therefore durable across a kill-without-flush — unlike a flush-only
/// release, it does not wait for a later [`Self::flush`]. /// release, it does not wait for a later [`Store::flush`].
fn release(&mut self, id: &ObservationId) { fn release(&mut self, id: &ObservationId) {
if let Some(entry) = self.index.get_mut(id) { if let Some(entry) = self.index.get_mut(id) {
entry.released = true; entry.released = true;
@@ -1132,8 +1361,15 @@ impl Store for LogStore {
/// the number of observations removed. Trimmed ids are not tombstoned: /// the number of observations removed. Trimmed ids are not tombstoned:
/// re-inserting them later is accepted. /// re-inserting them later is accepted.
/// ///
/// **Quarantined observations are exempt from every rule** (spec §9,
/// decision L): `keep_per_topic`, `max_age_ms` and `max_count` all rank by
/// age, and a quarantined observation's age isn't trusted until
/// [`LogStore::set_wall_clock`] folds the downtime in. They are neither
/// trimmed nor counted, so a store in quarantine can briefly hold more
/// than `max_count`, until the fold.
///
/// Each removed id appends a `Record::Trim` record, but that record is /// Each removed id appends a `Record::Trim` record, but that record is
/// durable across a power loss only after a [`Self::flush`]; hosts that /// durable across a power loss only after a [`Store::flush`]; hosts that
/// need crash durability must flush before shutdown. /// need crash durability must flush before shutdown.
fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize { fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize {
let mut removed: BTreeSet<ObservationId> = BTreeSet::new(); let mut removed: BTreeSet<ObservationId> = BTreeSet::new();
@@ -1201,6 +1437,34 @@ impl Store for LogStore {
} }
removed.len() removed.len()
} }
/// The durability point (spec §4).
///
/// Clamps `now` to the time anchor (an earlier `now` than any tick the
/// store has been given counts as that tick, so nothing moves backwards),
/// rebases every entry's age to it in memory, appends a clock marker at
/// `now` and fsyncs the log. Everything inserted, released or trimmed so
/// far is then durable, and the next [`LogStore::open`] folds ages from
/// that marker. After [`LogStore::set_wall_clock`] the marker carries the
/// derived wall time `wall_at_set + (now - tick_at_set)`; before it, none.
///
/// Flushing twice is harmless: it appends a second marker. A failure
/// returns [`StoreError::Backend`] and leaves the store usable; the
/// in-memory state is unchanged apart from the rebase, which doesn't
/// change the age `get` reports at `now` or later.
fn flush(&mut self, now: Tick) -> Result<(), StoreError> {
let now = self.advance_anchor(now);
self.rebase_to(now);
// The marker records the caller's `now`, not the last insert tick
// (that stale-marker bug was F18).
let record = Record::Clock {
tick: now,
wall: self.marker_wall(now),
};
write_record(&mut self.file, &record)?;
self.file.sync_all().map_err(|_| StoreError::Backend)?;
Ok(())
}
} }
impl LogStore { impl LogStore {
@@ -1255,7 +1519,7 @@ mod tests {
// error, so the writer can never emit a record the reader rejects. // error, so the writer can never emit a record the reader rejects.
let path = unique_path("oversize"); let path = unique_path("oversize");
let mut f = std::fs::File::create(&path).unwrap(); let mut f = std::fs::File::create(&path).unwrap();
write_header(&mut f).unwrap(); f.write_all(&header_bytes()).unwrap();
// A body exactly at the cap is accepted; one byte over is refused. The // A body exactly at the cap is accepted; one byte over is refused. The
// framing entry point exercises the same cap guard the private // framing entry point exercises the same cap guard the private
@@ -1281,8 +1545,9 @@ mod tests {
assert!(s.insert(c, 1002).unwrap()); assert!(s.insert(c, 1002).unwrap());
s.flush(1002).unwrap(); s.flush(1002).unwrap();
} }
// Flip a byte in the body of the second data record (Obs(b)). // Flip a byte in the body of the second data record (Obs(b)): the
corrupt_body(&path, 2); // open snapshot wrote Boot and Clock first, then Obs(a), Obs(b).
corrupt_body(&path, 3);
{ {
let s = LogStore::open(&path, 3000).unwrap(); let s = LogStore::open(&path, 3000).unwrap();
assert!( assert!(
@@ -1345,6 +1610,88 @@ mod tests {
); );
} }
#[test]
fn interrupted_snapshot_is_discarded() {
// A crash before the snapshot's rename leaves a temporary file beside
// an intact log. The next open ignores it (even a well-formed one) and
// removes it.
let path = unique_path("tmp_left");
let a = obs([1; 16], 1, 0, None);
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(a.clone(), 0).unwrap();
s.flush(0).unwrap();
}
// A complete, valid log holding something else, never renamed.
let tmp = snapshot_tmp_path(&path);
{
let other = unique_path("tmp_other");
let mut s = LogStore::open(&other, 0).unwrap();
s.insert(obs([2; 16], 2, 0, None), 0).unwrap();
s.flush(0).unwrap();
drop(s);
std::fs::rename(&other, &tmp).unwrap();
}
let s = LogStore::open(&path, 0).unwrap();
assert!(s.contains(&a.id()), "the log is what was renamed last");
assert!(!s.contains(&obs([2; 16], 2, 0, None).id()));
assert_eq!(s.len(), 1);
assert!(!tmp.exists(), "the leftover is removed");
}
#[test]
fn open_rewrites_the_log_as_a_snapshot() {
// Trimmed records and the previous boots' records don't survive the
// open snapshot: the log holds just the current state.
let path = unique_path("compact");
{
let mut s = LogStore::open(&path, 0).unwrap();
for seq in 0..20 {
s.insert(obs([1; 16], seq, 0, None), 0).unwrap();
}
let mut policy = TrimPolicy::default();
policy.max_count = Some(1);
assert_eq!(s.trim(&policy, 0), 19);
s.flush(0).unwrap();
}
let before = std::fs::metadata(&path).unwrap().len();
drop(LogStore::open(&path, 0).unwrap());
let after = std::fs::metadata(&path).unwrap().len();
assert!(after < before, "{after} < {before}");
let data = std::fs::read(&path).unwrap();
let mut kinds = Vec::new();
let mut pos = HEADER_LEN;
while let Some((_, end, rec)) = parse_valid_record(&data, pos) {
kinds.push(match rec {
Record::Obs { .. } => "obs",
Record::Trim { .. } => "trim",
Record::Release { .. } => "release",
Record::Boot { .. } => "boot",
Record::Clock { wall, .. } => {
assert_eq!(wall, None, "the open snapshot's marker has no wall");
"clock"
}
});
pos = end;
}
assert_eq!(kinds, ["boot", "obs", "clock"]);
}
#[cfg(unix)]
#[test]
fn a_lock_on_a_replaced_file_is_detected() {
// The race `open_locked` guards against: an opener's file is renamed
// over between its open and its lock.
let path = unique_path("replaced");
std::fs::write(&path, b"old").unwrap();
let stale = File::open(&path).unwrap();
assert!(is_file_at(&stale, &path));
let other = unique_path("replacement");
std::fs::write(&other, b"new").unwrap();
std::fs::rename(&other, &path).unwrap();
assert!(!is_file_at(&stale, &path));
}
fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option<u64>) -> Observation { fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option<u64>) -> Observation {
// `Observation` is `#[non_exhaustive]` outside `tines-core` (Task 5); // `Observation` is `#[non_exhaustive]` outside `tines-core` (Task 5);
// build it via the public constructor instead of a struct literal. // build it via the public constructor instead of a struct literal.
tines-store/tests/persistence.rs
+375 -2
View File
@@ -373,8 +373,9 @@ mod fleet_fixes_f5 {
assert!(s.insert(a_obs([3; 16], 3), 1002).unwrap()); assert!(s.insert(a_obs([3; 16], 3), 1002).unwrap());
s.flush(1002).expect("flush"); s.flush(1002).expect("flush");
} }
// Corrupt the middle observation (record index 2: Boot, obs-1, obs-2). // Corrupt the middle observation (record index 3: the open snapshot's
corrupt_record(&path, 2); // Boot and Clock, then obs-1, obs-2).
corrupt_record(&path, 3);
{ {
let s = LogStore::open(&path, 3000).expect("reopen"); let s = LogStore::open(&path, 3000).expect("reopen");
assert!( assert!(
@@ -1000,3 +1001,375 @@ mod quarantined_duplicate_exchange {
); );
} }
} }
// Review fixes B1, B2/S2 (decision I), S3, B3, decision J and decision L.
mod snapshots_and_flush {
use super::*;
use tines_core::{consts, Event, Node, SendError, Store, Transport, TrimPolicy};
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)
}
struct NullTransport;
impl Transport for NullTransport {
type Addr = u8;
fn poll(&mut self) -> Option<Event<u8>> {
None
}
fn send(&mut self, _to: &u8, _frame: &[u8]) -> Result<(), SendError> {
Ok(())
}
fn mtu(&self) -> usize {
consts::MAX_FRAME
}
}
/// Flush through the `Store` trait, never the backend's own method.
fn flush_generic<S: Store>(store: &mut S, now: u64) {
store.flush(now).expect("trait flush");
}
fn age_after_reopen(path: &std::path::Path, id: &tines_core::ObservationId) -> u64 {
let s = LogStore::open(path, 0).unwrap();
s.get(id, 0).unwrap().age_ms
}
/// B1: a flush through the `Store` trait (or `Node::flush`) writes the
/// clock marker, so a flushed store and an unflushed one give different
/// ages after a reboot.
#[test]
fn trait_flush_is_the_log_flush() {
let o = obs([1; 16], 1, 0, None);
let id = o.id();
let unflushed = store_path("b1_unflushed");
{
let mut s = LogStore::open(&unflushed, 0).unwrap();
s.insert(o.clone(), 1000).unwrap();
}
let generic = store_path("b1_generic");
{
let mut s = LogStore::open(&generic, 0).unwrap();
s.insert(o.clone(), 1000).unwrap();
flush_generic(&mut s, 5000);
}
let via_node = store_path("b1_node");
{
let mut s = LogStore::open(&via_node, 0).unwrap();
s.insert(o.clone(), 1000).unwrap();
let mut node = Node::new([9; 16], s.first_seq(), s, NullTransport);
node.flush(5000).expect("Node::flush");
}
assert_eq!(
age_after_reopen(&unflushed, &id),
0,
"no marker after the insert: the crash window isn't counted"
);
assert_eq!(age_after_reopen(&generic, &id), 4000);
assert_eq!(age_after_reopen(&via_node, &id), 4000);
}
/// B2 (lineage): ages survive any number of reboots, because the open
/// snapshot rewrites every `stored_at` into the current boot's ticks.
#[test]
fn age_lineage_across_three_opens() {
let path = store_path("b2_lineage");
let o = obs([1; 16], 1, 0, None);
let id = o.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(o, 1000).unwrap();
s.flush(50_000).unwrap();
}
{
// Every boot's ticks restart at 0.
let mut s = LogStore::open(&path, 0).unwrap();
assert_eq!(s.get(&id, 0).unwrap().age_ms, 49_000);
s.flush(500).unwrap();
}
{
let s = LogStore::open(&path, 0).unwrap();
assert_eq!(
s.get(&id, 0).unwrap().age_ms,
49_500,
"the third open keeps the lineage instead of resetting it"
);
}
}
/// B2 (resurrection): an addressed message expired by an unknown-downtime
/// fold stays released after a later reboot whose downtime is known.
#[test]
fn fold_expiry_survives_known_downtime_reboot() {
let path = store_path("b2_resurrect");
let a = addressed([2; 16], 1, [3; 16], 10_000);
let id = a.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(a, 0).unwrap();
s.flush(0).unwrap(); // no wall time yet
}
{
let mut s = LogStore::open(&path, 0).unwrap();
let report = s.set_wall_clock(1_000_000, 0);
assert_eq!(report.downtime_ms, None);
assert_eq!(report.expired_addressed, 1);
s.flush(100).unwrap(); // marker wall 1_000_100
}
{
let mut s = LogStore::open(&path, 0).unwrap();
let report = s.set_wall_clock(1_000_200, 0);
assert_eq!(report.downtime_ms, Some(100), "downtime is known now");
assert!(s.is_released(&id), "the expired message stays released");
let mut offered = 0usize;
s.for_each_addressed_id(&mut |_| offered += 1);
assert_eq!(offered, 0, "never offered for custody again");
assert!(
s.get(&id, 0).unwrap().age_ms > 10_000,
"still past lifetime"
);
}
}
/// B2: a broadcast floored by an unknown-downtime fold keeps its floor.
#[test]
fn fold_floor_survives_reboot() {
let path = store_path("b2_floor");
let b = obs([1; 16], 1, 0, None);
let id = b.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(b, 0).unwrap();
s.flush(0).unwrap();
}
{
let mut s = LogStore::open_with(&path, 0, 30_000).unwrap();
assert_eq!(s.set_wall_clock(1_000_000, 0).floored_broadcasts, 1);
}
{
// Killed without a flush after the fold: the fold's snapshot is
// the durable state.
let mut s = LogStore::open_with(&path, 0, 30_000).unwrap();
assert_eq!(s.get(&id, 0).unwrap().age_ms, 30_000);
let report = s.set_wall_clock(1_000_000, 0);
assert_eq!(
report.downtime_ms,
Some(0),
"the fold snapshot's marker has a wall"
);
assert_eq!(s.get(&id, 0).unwrap().age_ms, 30_000);
}
}
/// S2: a flush between open and `set_wall_clock` has already counted its
/// share of this boot's ticks; the fold must not count them again.
#[test]
fn flush_before_fold_does_not_double_count() {
let path = store_path("s2_double");
let b = obs([1; 16], 1, 0, None);
let id = b.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(b, 0).unwrap();
s.set_wall_clock(1_000_000, 0);
s.flush(0).unwrap(); // marker: tick 0, wall 1_000_000
}
{
// Off for 11 s, then up for 4 s before trusted time arrives.
let mut s = LogStore::open(&path, 0).unwrap();
s.flush(4000).unwrap();
let report = s.set_wall_clock(1_015_000, 4000);
assert_eq!(report.downtime_ms, Some(15_000));
assert_eq!(
s.get(&id, 4000).unwrap().age_ms,
15_000,
"true age is 15 s, not 19 s"
);
}
}
/// The open snapshot's marker has no wall time: a restart before trusted
/// time takes the unknown path, even if the previous boot had trust.
#[test]
fn restart_before_trust_takes_unknown_path() {
let path = store_path("open_marker_no_wall");
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(addressed([2; 16], 1, [3; 16], 60_000), 0).unwrap();
s.set_wall_clock(1_000_000, 0);
s.flush(10).unwrap();
}
{
// Opened, never trusted, never flushed.
let _s = LogStore::open(&path, 0).unwrap();
}
{
let mut s = LogStore::open(&path, 0).unwrap();
let report = s.set_wall_clock(1_000_500, 0);
assert_eq!(report.downtime_ms, None);
assert_eq!(report.expired_addressed, 1);
}
}
/// S3: a `now` earlier than the last anchor is clamped, and `stored_at`
/// never moves backwards.
#[test]
fn flush_clamps_an_earlier_now() {
let path = store_path("s3_clamp");
let b = obs([1; 16], 1, 0, None);
let id = b.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(b, 5000).unwrap();
s.flush(3000).unwrap();
assert_eq!(s.get(&id, 5000).unwrap().age_ms, 0);
s.flush(8000).unwrap();
s.flush(6000).unwrap(); // earlier than the last anchor
assert_eq!(s.get(&id, 8000).unwrap().age_ms, 3000);
}
// The clamped marker is at 8000, not 6000.
assert_eq!(age_after_reopen(&path, &id), 3000);
}
/// B3: wall times near `u64::MAX` never overflow, in the marker or in the
/// unknown-downtime expiry.
#[test]
fn huge_wall_and_lifetime_saturate() {
let path = store_path("b3_saturate");
let a = addressed([2; 16], 1, [3; 16], u64::MAX);
let id = a.id();
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(a, 0).unwrap();
s.set_wall_clock(u64::MAX - 5, 0);
s.flush(1_000).unwrap(); // derived wall saturates
}
{
let mut s = LogStore::open(&path, 0).unwrap();
let report = s.set_wall_clock(u64::MAX, 0);
assert_eq!(report.downtime_ms, Some(0));
}
let path = store_path("b3_lifetime");
{
let mut s = LogStore::open(&path, 0).unwrap();
s.insert(addressed([2; 16], 1, [3; 16], u64::MAX), 0)
.unwrap();
}
{
let mut s = LogStore::open(&path, 0).unwrap();
let report = s.set_wall_clock(1000, 0); // unknown downtime
assert_eq!(report.downtime_ms, None);
assert!(
s.is_released(&id),
"never delivered after an unknown downtime"
);
}
}
/// Decision J: the latest estimate wins, quarantined entries never count,
/// and a huge `observed_at` saturates instead of overflowing.
#[test]
fn peer_time_estimate_is_the_latest() {
let path = store_path("j_latest");
{
let mut s = LogStore::open(&path, 0).unwrap();
// Held across the reboot below: quarantined, so never a candidate.
s.insert(obs([7; 16], 1, 0, Some(9_000_000)), 0).unwrap();
s.flush(0).unwrap();
}
let mut s = LogStore::open(&path, 0).unwrap();
assert_eq!(
s.peer_time_estimate(0),
None,
"quarantined data is no source"
);
// Younger, but further behind: estimate 1_000_000 + 10 + 100.
s.insert(obs([1; 16], 1, 10, Some(1_000_000)), 100).unwrap();
// Older, but later: estimate 1_000_500 + 5_000 + 100.
s.insert(obs([2; 16], 1, 5_000, Some(1_000_500)), 100)
.unwrap();
assert_eq!(s.peer_time_estimate(200), Some(1_005_600));
s.insert(obs([3; 16], 1, 0, Some(u64::MAX - 5)), 100)
.unwrap();
assert_eq!(s.peer_time_estimate(200), Some(u64::MAX));
}
/// Decision L: quarantined entries are exempt from every trim rule, not
/// just the age trim.
#[test]
fn trim_keeps_quarantined_entries() {
let path = store_path("l_trim");
{
let mut s = LogStore::open(&path, 0).unwrap();
for seq in 1..=3 {
s.insert(obs([1; 16], seq, 1_000_000, None), 0).unwrap();
}
s.flush(0).unwrap();
}
let mut s = LogStore::open(&path, 0).unwrap();
let mut policy = TrimPolicy::default();
policy.keep_per_topic = Some(1);
policy.max_age_ms = Some(1);
policy.max_count = Some(1);
assert_eq!(s.trim(&policy, 0), 0, "nothing quarantined is trimmed");
assert_eq!(s.len(), 3, "the store may exceed max_count in quarantine");
// Fresh inserts are trimmed normally alongside them.
s.insert(obs([2; 16], 1, 0, None), 0).unwrap();
s.insert(obs([2; 16], 2, 5, None), 0).unwrap();
assert_eq!(s.trim(&policy, 10), 2);
assert_eq!(s.len(), 3);
}
/// The exclusive lock survives the snapshot's rename, and the new file
/// keeps mode 0600.
#[cfg(unix)]
#[test]
fn lock_and_mode_survive_snapshot() {
use std::os::unix::fs::MetadataExt;
let path = store_path("lock_snapshot");
{
let mut s = LogStore::open(&path, 0).unwrap();
assert!(
LogStore::open(&path, 0).is_err(),
"locked after the open snapshot"
);
s.insert(obs([1; 16], 1, 0, None), 0).unwrap();
s.set_wall_clock(1_000_000, 0);
assert!(
LogStore::open(&path, 0).is_err(),
"locked after the fold snapshot"
);
let mode = std::fs::metadata(&path).unwrap().mode() & 0o777;
assert_eq!(mode, 0o600);
// Writes after the rename land in the live log.
s.insert(obs([1; 16], 2, 0, None), 0).unwrap();
s.flush(10).unwrap();
}
let s = LogStore::open(&path, 0).unwrap();
assert!(s.contains(&obs([1; 16], 2, 0, None).id()));
assert_eq!(s.len(), 2);
}
/// A header torn by a crash while the log was first created holds no
/// records, so it opens as a new log. Anything else that short is refused.
#[test]
fn torn_first_header_opens_as_new() {
let path = store_path("torn_header");
std::fs::write(&path, b"TINESL").unwrap();
let s = LogStore::open(&path, 0).expect("a torn header is an empty log");
assert_eq!(s.len(), 0);
drop(s);
let path = store_path("short_foreign");
std::fs::write(&path, b"HELLO").unwrap();
assert!(LogStore::open(&path, 0).is_err(), "not our file: refuse it");
assert_eq!(
std::fs::read(&path).unwrap(),
b"HELLO",
"and leave it alone"
);
}
}