Sealed length inflate past EOF must not drop the tail

crates/engine/src/log/record.rs

@@ -181,13 +181,40 @@ pub fn encode(
     Ok(buf)
 }
 
+/// True if `bytes` contains at least one fully framed, CRC-valid record.
+///
+/// Distinguishes a torn last write (header landed, body did not; nothing
+/// after it verifies) from mid-file length bitrot (a later record still
+/// sits in the file, hidden behind an inflated `val_size`).
+pub fn has_later_valid_record(bytes: &[u8]) -> bool {
+    let mut i = 0;
+    while i + HEADER_LEN <= bytes.len() {
+        if let Ok(hdr) = parse_header(&bytes[i..i + HEADER_LEN], i as u64) {
+            let rec_len = hdr.record_len();
+            if rec_len >= HEADER_LEN && i + rec_len <= bytes.len() {
+                let header_buf = &bytes[i..i + HEADER_LEN];
+                let body = &bytes[i + HEADER_LEN..i + rec_len];
+                if verify_crc(&hdr, header_buf, body, i as u64).is_ok() {
+                    return true;
+                }
+            }
+        }
+        i += 1;
+    }
+    false
+}
+
 /// Walk a sealed log image.
 ///
 /// A short last record at EOF is a torn tail and is dropped (`Ok` with the
-/// prefix). A CRC mismatch on a *fully framed* mid-file record is bitrot:
-/// the rest of the sealed file is still there and must not be silently
-/// discarded. Callers treat that as `CrcMismatch` (fail-stop) rather than
-/// "this is the torn active tail".
+/// prefix). That includes a *complete* header whose body never landed —
+/// the last write flushed 37 bytes and then crashed. A CRC mismatch on a
+/// fully framed mid-file record is bitrot: the rest of the sealed file is
+/// still there and must not be silently discarded.
+///
+/// The same applies to a mid-file `val_size` inflate that does not fit:
+/// later records are still in the file, so this is `CrcMismatch`, not a
+/// torn tail. [`has_later_valid_record`] is the distinguisher.
 ///
 /// [`crate::log::recover`] uses the same rule when a sealed footer is
 /// missing and the file is rebuilt from records.
@@ -205,6 +232,11 @@ pub fn scan_sealed(bytes: &[u8]) -> Result<Vec<u64>> {
         };
         let rec_len = hdr.record_len();
         if offset + rec_len > bytes.len() {
+            if has_later_valid_record(&bytes[offset + HEADER_LEN..]) {
+                return Err(EngineError::CrcMismatch {
+                    offset: offset as u64,
+                });
+            }
             break;
         }
         let body = &bytes[offset + HEADER_LEN..offset + rec_len];
@@ -303,4 +335,45 @@ mod tests {
         let offs = scan_sealed(&buf).expect("torn tail is not corruption");
         assert_eq!(offs, vec![0]);
     }
+
+    /// Last write flushed a complete header and only part of the body.
+    /// Nothing after it verifies — this is a torn tail, not bitrot.
+    #[test]
+    fn sealed_incomplete_last_record_is_torn_tail() {
+        let mut buf = Vec::new();
+        encode_into(&mut buf, 1, flags::NO_EXPIRY, 0, b"keep", b"one", b"").unwrap();
+        encode_into(&mut buf, 2, flags::NO_EXPIRY, 0, b"tail", b"partial", b"").unwrap();
+        buf.truncate(buf.len() - 5);
+        let offs = scan_sealed(&buf).expect("complete header + short body at EOF is torn");
+        assert_eq!(offs, vec![0]);
+    }
+
+    /// A mid-file val_size inflate that does not fit in the file never
+    /// reaches CRC. Treating it as a torn tail drops every later record.
+    #[test]
+    fn sealed_inflated_mid_file_length_is_not_a_silent_tail() {
+        let mut buf = Vec::new();
+        encode_into(&mut buf, 1, flags::NO_EXPIRY, 0, b"keep", b"one", b"").unwrap();
+        let second_at = buf.len();
+        encode_into(&mut buf, 2, flags::NO_EXPIRY, 0, b"drop-me", b"two", b"").unwrap();
+        encode_into(
+            &mut buf,
+            3,
+            flags::NO_EXPIRY,
+            0,
+            b"also-keep",
+            b"three",
+            b"",
+        )
+        .unwrap();
+
+        buf[second_at + 29..second_at + 33].copy_from_slice(&0x00FF_FFFFu32.to_le_bytes());
+
+        let err = scan_sealed(&buf)
+            .expect_err("inflated mid-file length in a sealed image is corruption");
+        assert!(
+            matches!(err, EngineError::CrcMismatch { offset } if offset == second_at as u64),
+            "got {err:?}"
+        );
+    }
 }

crates/engine/src/log/recover.rs

@@ -6,7 +6,9 @@ use tracing::warn;
 use crate::error::{EngineError, Result};
 use crate::log::file::{FooterEntry, LogFile, list_data_files};
 use crate::log::index::{IndexEntry, NsIndex};
-use crate::log::record::{HEADER_LEN, flags as rflags, parse_header, verify_crc};
+use crate::log::record::{
+    HEADER_LEN, flags as rflags, has_later_valid_record, parse_header, verify_crc,
+};
 
 /// Result of opening a namespace directory.
 pub struct OpenedFiles {
@@ -143,8 +145,37 @@ async fn rebuild_from_records(file: &LogFile, file_id: u32, index: &mut NsIndex)
             Err(_) => break,
         };
         let body_size = hdr.body_len();
+        // Don't ask the pool for an inflated val_size (up to 4 GiB). A body
+        // that cannot fit in the remaining file is either a torn last write
+        // or mid-file length bitrot — distinguisher is a later CRC-ok record.
+        let remaining = total.saturating_sub(offset.saturating_add(HEADER_LEN as u64));
+        if (body_size as u64) > remaining {
+            let rest = if remaining == 0 {
+                Vec::new()
+            } else {
+                match file
+                    .read_at(offset + HEADER_LEN as u64, remaining as usize)
+                    .await
+                {
+                    Ok(b) => b.to_vec(),
+                    Err(e) => {
+                        warn!(file_id, offset, error = %e, "I/O error peeking sealed short body; stopping scan at this offset");
+                        break;
+                    }
+                }
+            };
+            if has_later_valid_record(&rest) {
+                return Err(EngineError::CrcMismatch { offset });
+            }
+            break;
+        }
         let body = match file.read_at(offset + HEADER_LEN as u64, body_size).await {
-            Ok(b) if b.len() < body_size => break,
+            Ok(b) if b.len() < body_size => {
+                if has_later_valid_record(&b) {
+                    return Err(EngineError::CrcMismatch { offset });
+                }
+                break;
+            }
             Ok(b) => b,
             Err(e) => {
                 warn!(file_id, offset, error = %e, "I/O error reading sealed file body; stopping scan at this offset");