fix(store): crash-safe reopen, bounded record decode, durability docs

tines-store/src/log.rs

@@ -24,15 +24,45 @@
 //! and for observations inserted before the last flush — ages continue across a
 //! reboot instead of freezing at (and under-counting from) their pre-shutdown
 //! value.
+//!
+//! Crash recovery & robustness
+//!
+//! * Torn trailing records. The log is append-only, so a crash mid-append can
+//!   leave a partial trailing record — a length prefix whose body ran past
+//!   end-of-file — or a bit-roted/corrupt record. [`LogStore::new`] recovers by
+//!   truncating the file back to the last *complete* record boundary and
+//!   loading what survived: the partial tail is dropped, complete records are
+//!   preserved. A corrupt length prefix that claims an absurd body size is
+//!   rejected before any allocation (see [`MAX_RECORD_BYTES`]) and dropped the
+//!   same way, so a malformed prefix cannot exhaust memory or make the store
+//!   unloadable.
+//!
+//! * Durability contract. `flush()` appends a `Record::Meta` and
+//!   `sync_all`s, making the released set, `next_seq`, and the appended
+//!   observation records durable across a power loss. `insert`, `release`, and
+//!   `trim` update the in-memory index and (for `insert`/`trim`) append
+//!   records, but those records reach disk only on the next `flush()`. Hosts
+//!   that need crash durability must `flush()` before shutdown.
 
 use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::fs::{File, OpenOptions};
-use std::io::{Read, Write};
+use std::io::Write;
 use std::path::Path;
 
 use serde::{Deserialize, Serialize};
 use tines_core::{Observation, ObservationId, Store, StoreError, Tick, TrimPolicy};
 
+/// Maximum accepted body size, in bytes, for a single log record.
+///
+/// A record is `[u32 little-endian length][postcard bytes]`; this caps the
+/// postcard body so a corrupt or malicious length prefix (e.g. `0xFFFFFFFF`,
+/// which would otherwise try to reserve ~4 GiB) is rejected *before* any
+/// allocation. The bound is comfortably larger than the largest legitimate
+/// record — a fully populated `Record::Obs` is about 1.3 KiB of postcard plus
+/// 4 bytes of framing — so valid inserts are never dropped, while still
+/// bounding the allocation against a malformed prefix.
+pub const MAX_RECORD_BYTES: usize = 8 * 1024;
+
 /// Current (age-adjusted) age of a stored observation at tick `now`.
 fn current_age(obs: &Observation, stored_at: Tick, now: Tick) -> u64 {
     obs.age_ms.saturating_add(now.saturating_sub(stored_at))
@@ -80,6 +110,9 @@ impl LogStore {
     /// number to hand out; the persisted `next_seq` (if any) is honored when it
     /// is larger, so a resumed run never reuses a sequence.
     pub fn new(path: &Path, now: Tick, first_seq: u64) -> Result<Self, StoreError> {
+        // Open for reads (to load) and appends (to write later). Append mode is
+        // correct because we load by reading the whole file below, and every
+        // mutation appends to the end of the log.
         let file = OpenOptions::new()
             .read(true)
             .create(true)
@@ -87,19 +120,36 @@ impl LogStore {
             .open(path)
             .map_err(|_| StoreError::Backend)?;
 
+        // Read the whole log into memory to rebuild the index. Loading from a
+        // slice (rather than seeking the open file) lets us detect a torn or
+        // corrupt trailing record by byte offsets and truncate the file to the
+        // last complete boundary — see the crash-recovery note in the module docs.
+        let data = std::fs::read(path).map_err(|_| StoreError::Backend)?;
+
         let mut index: BTreeMap<ObservationId, Entry> = BTreeMap::new();
         let mut dead: HashSet<ObservationId> = HashSet::new();
         let mut next_seq = first_seq;
+        let mut records_loaded: usize = 0;
 
-        let mut file = file;
+        // Byte offset just past the last *complete* record replayed so far. A
-        while let Some(rec) = read_record(&mut file)? {
+        // torn or corrupt trailing record (from a crash mid-append, or a
+        // bit-roted/corrupt record) stops the loop and is truncated back to this
+        // boundary, so the store loads with whatever complete records survived
+        // instead of failing outright. See the crash-recovery note in the module docs.
+        let mut pos = 0usize;
+        let mut good_end = 0usize;
+
+        loop {
+            match parse_record(&data, pos) {
+                Ok(Some((new_pos, rec))) => {
                     match rec {
                         Record::Obs { id, obs, stored_at } => {
                             // A preceding Trim for this id is cancelled by the
                             // re-insert, so revive it; otherwise dedupe to the
-                    // first (oldest) occurrence. Fold in the time elapsed
+                            // first (oldest) occurrence. Fold in the time
-                    // since the record's own insert tick so the age is exact
+                            // elapsed since the record's own insert tick so the
-                    // regardless of when the last flush happened.
+                            // age is exact regardless of when the last flush
+                            // happened.
                             dead.remove(&id);
                             index.entry(id).or_insert_with(|| Entry {
                                 obs: obs.aged(now.saturating_sub(stored_at)),
@@ -123,12 +173,44 @@ impl LogStore {
                             }
                         }
                     }
+                    records_loaded += 1;
+                    // Advance the last-complete-record boundary now that this
+                    // record is fully read.
+                    good_end = new_pos;
+                    pos = new_pos;
+                }
+                // Clean end-of-file: nothing left to read.
+                Ok(None) => break,
+                // Torn or corrupt trailing record. `parse_record` only errors on
+                // a bad trailing record (a partial length prefix, a body that
+                // runs past end-of-file, a body length over MAX_RECORD_BYTES, or
+                // a decode failure). Chop the file back to the last complete
+                // boundary and load what survived — only the partial tail of a
+                // crashed append is lost, never a complete record.
+                Err(_) => break,
+            }
         }
 
         for id in dead {
             index.remove(&id);
         }
 
+        // We stopped before EOF: the trailing bytes are a torn or corrupt record.
+        // Truncate the file to the last complete boundary so it has a clean tail.
+        let dropped = data.len().saturating_sub(good_end);
+        if dropped > 0 {
+            file.set_len(good_end as u64)
+                .map_err(|_| StoreError::Backend)?;
+            eprintln!(
+                "tines-store: dropped {} torn/corrupt byte(s) at the tail of {} \
+                 while loading ({} complete record(s) loaded); call flush() \
+                 after the next mutation to rewrite a clean tail.",
+                dropped,
+                path.display(),
+                records_loaded,
+            );
+        }
+
         Ok(LogStore {
             file,
             index,
@@ -144,6 +226,13 @@ impl LogStore {
 
     /// Persist the current index state (released flags, `next_seq`, and the
     /// observation log) to disk.
+    ///
+    /// This is the durability boundary: `flush()` appends a `Record::Meta`
+    /// and calls `sync_all`, after which the released set, `next_seq`, and the
+    /// appended observation records are durable across a power loss. Because
+    /// `insert`, `release`, and `trim` only reach 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
         // `entry.obs.age_ms` is the true current age and `stored_at` marks when
@@ -191,21 +280,54 @@ fn write_record(file: &mut File, record: &Record) -> Result<(), StoreError> {
     Ok(())
 }
 
-/// Read one length-prefixed record, returning `Ok(None)` at end of file.
+/// Parse one length-prefixed record from `data` at byte offset `pos`.
-fn read_record(file: &mut File) -> Result<Option<Record>, StoreError> {
+///
-    let mut len_buf = [0u8; 4];
+/// This is a pure slice parser (no file I/O), so a torn or corrupt trailing
-    if file.read_exact(&mut len_buf).is_err() {
+/// record is detectable purely by byte offsets, which is exactly what
-        // Clean end-of-file: no length prefix left to read.
+/// [`LogStore::new`] needs to truncate to the last complete boundary.
+///
+/// * `Ok(None)` — clean end-of-file: `pos` is exactly at the end of `data`.
+/// * `Ok(Some((new_pos, rec)))` — a complete record was parsed; `new_pos` is
+///   the byte offset just past it (`pos + 4 + body_len`).
+/// * `Err(StoreError::Backend)` — a torn or corrupt *trailing* record: fewer
+///   than 4 prefix bytes remain (a partial prefix from a crash mid-append), the
+///   declared body length exceeds [`MAX_RECORD_BYTES`] (rejected *before* any
+///   allocation, so a corrupt `0xFFFFFFFF` prefix cannot exhaust memory), the
+///   body runs past `data.len()`, or the body fails to decode.
+///   [`LogStore::new`] treats this as a torn tail and truncates the file.
+fn parse_record(data: &[u8], pos: usize) -> Result<Option<(usize, Record)>, StoreError> {
+    // Clean EOF: nothing left.
+    if pos == data.len() {
         return Ok(None);
     }
-    let len = u32::from_le_bytes(len_buf) as usize;
+    // A partial length prefix (1..3 leftover bytes) from a crash mid-append.
-    let mut buf = vec![0u8; len];
+    if pos + 4 > data.len() {
-    file.read_exact(&mut buf).map_err(|_| StoreError::Backend)?;
+        return Err(StoreError::Backend);
-    let record = postcard::from_bytes(&buf).map_err(|_| StoreError::Backend)?;
+    }
-    Ok(Some(record))
+    let len = u32::from_le_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]) as usize;
+    // Reject an over-length declared body *before* allocating, so a corrupt or
+    // malicious prefix (e.g. `0xFFFFFFFF`, ~4 GiB) cannot exhaust memory.
+    // See [`MAX_RECORD_BYTES`].
+    if len > MAX_RECORD_BYTES {
+        return Err(StoreError::Backend);
+    }
+    let start = pos + 4;
+    let end = start + len;
+    // A body that runs past end-of-file is a torn trailing record.
+    if end > data.len() {
+        return Err(StoreError::Backend);
+    }
+    let rec = postcard::from_bytes(&data[start..end]).map_err(|_| StoreError::Backend)?;
+    Ok(Some((end, rec)))
 }
 
 impl Store for LogStore {
+    /// Insert an observation.
+    ///
+    /// 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
+    /// must flush before shutdown (see the durability contract in the module
+    /// docs).
     fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, StoreError> {
         let id = obs.id();
         if self.index.contains_key(&id) {
@@ -259,6 +381,13 @@ impl Store for LogStore {
         }
     }
 
+    /// Visit the ids of observations addressed to someone (`to == Some`),
+    /// excluding those already released.
+    ///
+    /// This is stricter than the [`Store`] trait's documented contract
+    /// (`addressed == to == 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.
     fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
         for (id, entry) in self.index.iter() {
             if entry.obs.to.is_some() && !entry.released {
@@ -267,6 +396,12 @@ impl Store for LogStore {
         }
     }
 
+    /// Mark the observation with `id` as released: it has been sent and must
+    /// not be offered for exchange again.
+    ///
+    /// The mark is set in memory immediately, but is only persisted to disk on
+    /// the next [`Self::flush`]; without a flush the release is lost across a
+    /// power loss (see the durability contract in the module docs).
     fn release(&mut self, id: &ObservationId) {
         if let Some(entry) = self.index.get_mut(id) {
             entry.released = true;
@@ -289,6 +424,13 @@ impl Store for LogStore {
         }
     }
 
+    /// Apply [`TrimPolicy`] rules in order (per-topic, age, count) and return
+    /// the number of observations removed. Trimmed ids are not tombstoned:
+    /// re-inserting them later is accepted.
+    ///
+    /// Each removed id appends a `Record::Trim` record, but that record is
+    /// durable across a power loss only after a [`Self::flush`]; hosts that
+    /// need crash durability must flush before shutdown.
     fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize {
         let mut removed: BTreeSet<ObservationId> = BTreeSet::new();
 

tines-store/tests/persistence.rs

@@ -3,6 +3,7 @@
 //! Each test runs against its own unique file under the temp directory so the
 //! parallel test runner cannot collide on disk state.
 
+use std::io::Write;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -145,3 +146,73 @@ fn next_seq_persists() {
         assert_eq!(s.next_seq(), 4);
     }
 }
+
+#[test]
+fn reopen_recovers_from_torn_trailing_record() {
+    // A crash mid-append can leave a partial trailing record — a length prefix
+    // whose body never reached disk. Reopen must truncate the torn tail and
+    // load the complete records, not fail the whole store.
+    let path = store_path("torn");
+
+    {
+        let mut s = LogStore::new(&path, 0, 1).expect("open fresh store");
+        let a = obs([1; 16], 1, 0, Some(10));
+        let b = obs([2; 16], 2, 0, Some(20));
+        assert!(s.insert(a, 1000).unwrap());
+        assert!(s.insert(b, 1001).unwrap());
+        s.flush().expect("flush before drop");
+    }
+
+    // Simulate a torn append: write a length prefix claiming 64 body bytes but
+    // only write 20 of them. The body runs past end-of-file.
+    {
+        let mut f = std::fs::OpenOptions::new()
+            .append(true)
+            .open(&path)
+            .expect("open log for torn write");
+        f.write_all(&64u32.to_le_bytes())
+            .expect("write torn prefix");
+        f.write_all(&[0u8; 20]).expect("write partial body");
+        f.flush().expect("flush torn tail");
+    }
+
+    // Reopen: the torn tail is truncated, the two complete records survive.
+    {
+        let s = LogStore::new(&path, 3000, 1).expect("reopen after torn tail");
+        assert!(s.contains(&obs([1; 16], 1, 0, Some(10)).id()));
+        assert!(s.contains(&obs([2; 16], 2, 0, Some(20)).id()));
+        assert_eq!(s.next_seq(), 3);
+    }
+}
+
+#[test]
+fn reopen_recovers_from_oversized_record_length() {
+    // A corrupt length prefix (e.g. u32::MAX) must neither trigger a giant
+    // allocation nor make the store unloadable; it should be dropped like a
+    // torn record, leaving the earlier complete records intact.
+    let path = store_path("oversize");
+
+    {
+        let mut s = LogStore::new(&path, 0, 1).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");
+    }
+
+    {
+        // Declare a body of u32::MAX bytes with no actual body. read_record
+        // must reject it before allocating any buffer.
+        let mut f = std::fs::OpenOptions::new()
+            .append(true)
+            .open(&path)
+            .expect("open log for oversized write");
+        f.write_all(&u32::MAX.to_le_bytes())
+            .expect("write oversized length prefix");
+        f.flush().expect("flush oversized tail");
+    }
+
+    {
+        let s = LogStore::new(&path, 3000, 1).expect("reopen after oversized prefix");
+        assert!(s.contains(&obs([1; 16], 1, 0, Some(10)).id()));
+    }
+}