diff --git a/crates/engine/src/log/mod.rs b/crates/engine/src/log/mod.rs index 4a73c8d..c6adfb5 100644 --- a/crates/engine/src/log/mod.rs +++ b/crates/engine/src/log/mod.rs @@ -812,9 +812,20 @@ impl NamespaceLog { files.push((self.active.borrow().file_id, self.active())); let mut events = Vec::new(); - for (_, file) in &files { + let active_id = self.active.borrow().file_id; + for (id, file) in &files { let end = file.data_end_offset().await; - scan_file_records(file, end, filter, since_revision, &self.values, &mut events).await?; + let sealed = *id != active_id; + scan_file_records( + file, + end, + filter, + since_revision, + &self.values, + &mut events, + sealed, + ) + .await?; } // Sort by revision so callers see a clean chronological stream. events.sort_by_key(|e| match e { @@ -1241,6 +1252,7 @@ async fn scan_file_records( since_revision: u64, values: &ValueStore, events: &mut Vec, + sealed: bool, ) -> Result<()> { use crate::watch::WatchEvent; @@ -1280,6 +1292,13 @@ async fn scan_file_records( // this file at the first bad CRC — the record_len we'd skip forward by // is itself covered by the CRC and can't be trusted past a mismatch. if record::verify_crc(&hdr, &hdr_bytes, &body, offset).is_err() { + // Sealed files are immutable: a mid-file CRC is bitrot, not a + // torn tail. Returning Ok(prefix) makes watch catch-up look + // complete while later keys never arrive. The active file + // still stops — that *is* a torn write. + if sealed { + return Err(EngineError::CrcMismatch { offset }); + } warn!( offset, "bad CRC during watch replay; stopping scan of this file" @@ -2701,6 +2720,72 @@ mod watch_valuesep_tests { } }); } + + /// rebuild_from_records already fail-stops on a sealed mid-file CRC. + /// scan_since (watch catch-up) used to return Ok([keep]) and drop the + /// suffix. A bitrot mid-sealed-file is not a torn tail. + #[test] + fn scan_since_sealed_mid_crc_is_not_a_silent_prefix() { + use crate::error::EngineError; + use crate::log::file::data_filename; + use crate::log::record::{HEADER_LEN, parse_header}; + + run(async { + let dir = TempDir::new().unwrap(); + let cfg = LogConfig { + rotate_threshold: 1 << 40, + fanout: 8, + value_sep_threshold: 1 << 20, + }; + let log = NamespaceLog::open(dir.path().to_path_buf(), cfg) + .await + .unwrap(); + log.put_full(Bytes::from_static(b"keep"), b"one", &[], None) + .await + .unwrap(); + log.put_full(Bytes::from_static(b"drop-me"), b"two", &[], None) + .await + .unwrap(); + log.put_full(Bytes::from_static(b"also-keep"), b"three", &[], None) + .await + .unwrap(); + log.reclaim().await.unwrap(); + + let sealed = dir.path().join(data_filename(0)); + let mut buf = std::fs::read(&sealed).unwrap(); + let mut offset = 0usize; + let mut recs = Vec::new(); + while offset + HEADER_LEN <= buf.len() { + let Ok(hdr) = parse_header(&buf[offset..offset + HEADER_LEN], offset as u64) else { + break; + }; + let rec_len = hdr.record_len(); + if offset + rec_len > buf.len() { + break; + } + recs.push((offset, hdr)); + offset += rec_len; + } + assert!( + recs.len() >= 3, + "reclaim should have sealed all three records, got {}", + recs.len() + ); + let (second_at, second) = recs[1]; + let val = second_at + HEADER_LEN + second.key_size as usize; + buf[val] ^= 0x01; + std::fs::write(&sealed, &buf).unwrap(); + + let err = log + .scan_since(&KeyFilter::Prefix(b""), 0) + .await + .expect_err("sealed mid-file CRC must fail-stop watch catch-up"); + assert!( + matches!(err, EngineError::CrcMismatch { .. }), + "got {err:?}" + ); + }); + } } #[cfg(test)]