Sealed watch catch-up must not drop the tail on mid-file CRC

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<crate::watch::WatchEvent>,
+    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)]