diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index 01ac8ee6..c4f2d278 100644 --- a/tines-store/src/log.rs +++ b/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 { + // 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,41 +120,74 @@ 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 = BTreeMap::new(); let mut dead: HashSet = HashSet::new(); let mut next_seq = first_seq; + let mut records_loaded: usize = 0; - let mut file = file; - while let Some(rec) = read_record(&mut file)? { - 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 - // since the record's own insert tick so the age is exact - // regardless of when the last flush happened. - dead.remove(&id); - index.entry(id).or_insert_with(|| Entry { - obs: obs.aged(now.saturating_sub(stored_at)), - stored_at: now, - released: false, - }); - } - Record::Trim { id } => { - dead.insert(id); - index.remove(&id); - } - Record::Meta { - next_seq: ns, - released, - } => { - next_seq = ns.max(next_seq); - for rid in released { - if let Some(entry) = index.get_mut(&rid) { - entry.released = true; + // Byte offset just past the last *complete* record replayed so far. A + // 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 since the record's own insert tick so the + // age is exact regardless of when the last flush + // happened. + dead.remove(&id); + index.entry(id).or_insert_with(|| Entry { + obs: obs.aged(now.saturating_sub(stored_at)), + stored_at: now, + released: false, + }); + } + Record::Trim { id } => { + dead.insert(id); + index.remove(&id); + } + Record::Meta { + next_seq: ns, + released, + } => { + next_seq = ns.max(next_seq); + for rid in released { + if let Some(entry) = index.get_mut(&rid) { + entry.released = true; + } + } } } + 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, } } @@ -129,6 +195,22 @@ impl LogStore { 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. -fn read_record(file: &mut File) -> Result, StoreError> { - let mut len_buf = [0u8; 4]; - if file.read_exact(&mut len_buf).is_err() { - // Clean end-of-file: no length prefix left to read. +/// Parse one length-prefixed record from `data` at byte offset `pos`. +/// +/// This is a pure slice parser (no file I/O), so a torn or corrupt trailing +/// record is detectable purely by byte offsets, which is exactly what +/// [`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, StoreError> { + // Clean EOF: nothing left. + if pos == data.len() { return Ok(None); } - let len = u32::from_le_bytes(len_buf) as usize; - let mut buf = vec![0u8; len]; - file.read_exact(&mut buf).map_err(|_| StoreError::Backend)?; - let record = postcard::from_bytes(&buf).map_err(|_| StoreError::Backend)?; - Ok(Some(record)) + // A partial length prefix (1..3 leftover bytes) from a crash mid-append. + if pos + 4 > data.len() { + return Err(StoreError::Backend); + } + 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 { 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 = BTreeSet::new(); diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index e4adccdd..c537891a 100644 --- a/tines-store/tests/persistence.rs +++ b/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())); + } +}