fix: reject corrupt blob reads

README.md

@@ -183,6 +183,13 @@ checkpointing:
 
 - WAL records make acknowledged mutations replayable.
 - Checkpoints flush dirty blob frames and compact the manifest.
+- Cold blob reads verify per-page checksums from a separate, durably ordered
+  integrity file. A checksum failure is reported as corruption rather than
+  returning bytes. Hot cached reads do not add a checksum lookup.
+- The first writable open of an existing v6 store creates the integrity file
+  and a v7 manifest snapshot. Back up the store before this one-way upgrade;
+  older Holt binaries cannot reopen v7. The migration establishes checksums
+  for existing bytes and cannot identify corruption that predates it.
 - Indexed point reads are served page-granularly from the in-blob routing
   region (built at compaction); it is an accelerator, never the source
   of truth.

src/api/stats.rs

@@ -139,7 +139,7 @@ pub struct VacuumStats {
     /// Bytes copied while relocating live blob slots.
     pub bytes_relocated: u64,
     /// Bytes removed from store files. Counts `blobs.dat`,
-    /// `read.idx`, and `value.seg` truncation.
+    /// `integrity.dat`, `read.idx`, and `value.seg` truncation.
     pub bytes_truncated: u64,
     /// Reusable middle slots submitted to filesystem hole punching.
     ///

src/store/blob_store/file/integrity.rs

@@ -0,0 +1,249 @@
+//! Per-page checksums for authoritative packed blob slots.
+//!
+//! A record is stored at the same slot number as its blob. The manifest may
+//! reference a new slot only after both files have been synced. Reusable slots
+//! are fenced by the manifest's existing reader drain before either record is
+//! overwritten.
+
+use crate::api::errors::{Error, Result};
+use crate::layout::{BlobGuid, PAGE_SIZE};
+
+const MAGIC: [u8; 8] = *b"HOLTCKS1";
+const PAGE_BYTES: usize = 4096;
+const PAGE_COUNT: usize = PAGE_SIZE as usize / PAGE_BYTES;
+const PAGE_CHECKSUMS_OFFSET: usize = 32;
+const COVERED_PAGES_OFFSET: usize = 544;
+const RECORD_CHECKSUM_OFFSET: usize = 572;
+pub(super) const RECORD_BYTES: usize = 576;
+
+const _: () = assert!(PAGE_COUNT == 128);
+const _: () = assert!(PAGE_CHECKSUMS_OFFSET + PAGE_COUNT * 4 <= RECORD_CHECKSUM_OFFSET);
+
+pub(super) fn record_offset(slot: u64) -> Result<u64> {
+    slot.checked_mul(RECORD_BYTES as u64)
+        .ok_or(Error::node_corrupt("FileBlobStore::Integrity::slot offset"))
+}
+
+pub(super) fn encode(guid: BlobGuid, slot: u64, frame: &[u8]) -> [u8; RECORD_BYTES] {
+    encode_prefix(guid, slot, frame, frame.len())
+}
+
+pub(super) fn encode_prefix(
+    guid: BlobGuid,
+    slot: u64,
+    frame: &[u8],
+    covered_bytes: usize,
+) -> [u8; RECORD_BYTES] {
+    debug_assert_eq!(frame.len(), PAGE_SIZE as usize);
+    debug_assert!(
+        covered_bytes > 0 && covered_bytes <= frame.len() && covered_bytes % PAGE_BYTES == 0
+    );
+    let mut record = [0u8; RECORD_BYTES];
+    record[..8].copy_from_slice(&MAGIC);
+    record[8..24].copy_from_slice(&guid);
+    record[24..32].copy_from_slice(&slot.to_le_bytes());
+    for (page, bytes) in frame[..covered_bytes].chunks_exact(PAGE_BYTES).enumerate() {
+        let start = PAGE_CHECKSUMS_OFFSET + page * 4;
+        record[start..start + 4].copy_from_slice(&crc32fast::hash(bytes).to_le_bytes());
+    }
+    record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4]
+        .copy_from_slice(&((covered_bytes / PAGE_BYTES) as u32).to_le_bytes());
+    let checksum = crc32fast::hash(&record[..RECORD_CHECKSUM_OFFSET]);
+    record[RECORD_CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());
+    record
+}
+
+fn validate_record(record: &[u8; RECORD_BYTES], guid: BlobGuid, slot: u64) -> Result<()> {
+    let stored = u32::from_le_bytes(record[RECORD_CHECKSUM_OFFSET..].try_into().unwrap());
+    let covered_pages = u32::from_le_bytes(
+        record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4]
+            .try_into()
+            .unwrap(),
+    );
+    if record[..8] != MAGIC
+        || record[8..24] != guid
+        || record[24..32] != slot.to_le_bytes()
+        || !(1..=PAGE_COUNT as u32).contains(&covered_pages)
+        || stored != crc32fast::hash(&record[..RECORD_CHECKSUM_OFFSET])
+    {
+        return Err(Error::node_corrupt("FileBlobStore::Integrity::record").with_blob_guid(guid));
+    }
+    Ok(())
+}
+
+/// The record's extent is trusted only after its binding and checksum pass.
+pub(super) struct ValidatedRecord {
+    bytes: [u8; RECORD_BYTES],
+    guid: BlobGuid,
+    covered_pages: usize,
+}
+
+impl ValidatedRecord {
+    pub(super) fn new(bytes: [u8; RECORD_BYTES], guid: BlobGuid, slot: u64) -> Result<Self> {
+        validate_record(&bytes, guid, slot)?;
+        let covered_pages = u32::from_le_bytes(
+            bytes[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4]
+                .try_into()
+                .unwrap(),
+        ) as usize;
+        Ok(Self {
+            bytes,
+            guid,
+            covered_pages,
+        })
+    }
+
+    pub(super) fn covered_bytes(&self) -> usize {
+        self.covered_pages * PAGE_BYTES
+    }
+
+    pub(super) fn verify_full(&self, frame: &[u8]) -> Result<()> {
+        debug_assert_eq!(frame.len(), PAGE_SIZE as usize);
+        verify_pages(&self.bytes, self.guid, self.covered_pages, 0, frame)
+    }
+
+    pub(super) fn verify_range(&self, byte_offset: u64, bytes: &[u8]) -> Result<()> {
+        let start = usize::try_from(byte_offset).map_err(|_| {
+            Error::node_corrupt("FileBlobStore::Integrity::range").with_blob_guid(self.guid)
+        })?;
+        if start % PAGE_BYTES != 0
+            || bytes.len() % PAGE_BYTES != 0
+            || start
+                .checked_add(bytes.len())
+                .is_none_or(|end| end > PAGE_SIZE as usize)
+        {
+            return Err(
+                Error::node_corrupt("FileBlobStore::Integrity::range").with_blob_guid(self.guid)
+            );
+        }
+        verify_pages(
+            &self.bytes,
+            self.guid,
+            self.covered_pages,
+            start / PAGE_BYTES,
+            bytes,
+        )
+    }
+
+    pub(super) fn relocate_verified(&self, new_slot: u64) -> [u8; RECORD_BYTES] {
+        relocate_verified(&self.bytes, new_slot)
+    }
+}
+
+#[cfg(test)]
+pub(super) fn verify_full(
+    record: &[u8; RECORD_BYTES],
+    guid: BlobGuid,
+    slot: u64,
+    frame: &[u8],
+) -> Result<()> {
+    ValidatedRecord::new(*record, guid, slot)?.verify_full(frame)
+}
+
+#[cfg(test)]
+pub(super) fn verify_range(
+    record: &[u8; RECORD_BYTES],
+    guid: BlobGuid,
+    slot: u64,
+    byte_offset: u64,
+    bytes: &[u8],
+) -> Result<()> {
+    ValidatedRecord::new(*record, guid, slot)?.verify_range(byte_offset, bytes)
+}
+
+fn verify_pages(
+    record: &[u8; RECORD_BYTES],
+    guid: BlobGuid,
+    covered_pages: usize,
+    first_page: usize,
+    bytes: &[u8],
+) -> Result<()> {
+    for (index, page) in bytes.chunks_exact(PAGE_BYTES).enumerate() {
+        if first_page + index >= covered_pages {
+            continue;
+        }
+        let start = PAGE_CHECKSUMS_OFFSET + (first_page + index) * 4;
+        let expected = u32::from_le_bytes(record[start..start + 4].try_into().unwrap());
+        if crc32fast::hash(page) != expected {
+            return Err(Error::node_corrupt("FileBlobStore::Integrity::page").with_blob_guid(guid));
+        }
+    }
+    Ok(())
+}
+
+/// Legacy compact frames may leave stale, non-authoritative bytes beyond
+/// their persisted prefix. A verified record makes that boundary explicit.
+#[cfg(test)]
+pub(super) fn clear_uncovered_tail(record: &[u8; RECORD_BYTES], frame: &mut [u8]) {
+    let covered_pages = u32::from_le_bytes(
+        record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4]
+            .try_into()
+            .unwrap(),
+    ) as usize;
+    frame[covered_pages * PAGE_BYTES..].fill(0);
+}
+
+/// Retarget a record after the caller has verified its source frame.
+pub(super) fn relocate_verified(record: &[u8; RECORD_BYTES], new_slot: u64) -> [u8; RECORD_BYTES] {
+    let mut relocated = *record;
+    relocated[24..32].copy_from_slice(&new_slot.to_le_bytes());
+    let checksum = crc32fast::hash(&relocated[..RECORD_CHECKSUM_OFFSET]);
+    relocated[RECORD_CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes());
+    relocated
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn page_fault_is_rejected_in_full_and_range_reads() {
+        let guid = [7; 16];
+        let mut frame = vec![0u8; PAGE_SIZE as usize];
+        frame[PAGE_BYTES * 19 + 31] = 0x42;
+        let record = encode(guid, 3, &frame);
+        verify_full(&record, guid, 3, &frame).unwrap();
+        verify_range(
+            &record,
+            guid,
+            3,
+            (19 * PAGE_BYTES) as u64,
+            &frame[19 * PAGE_BYTES..20 * PAGE_BYTES],
+        )
+        .unwrap();
+        frame[PAGE_BYTES * 19 + 31] ^= 1;
+        assert!(verify_full(&record, guid, 3, &frame).is_err());
+        assert!(verify_range(
+            &record,
+            guid,
+            3,
+            (19 * PAGE_BYTES) as u64,
+            &frame[19 * PAGE_BYTES..20 * PAGE_BYTES]
+        )
+        .is_err());
+    }
+
+    #[test]
+    fn record_is_bound_to_guid_slot_and_checksum() {
+        let frame = vec![0u8; PAGE_SIZE as usize];
+        let guid = [7; 16];
+        let mut record = encode(guid, 3, &frame);
+        assert!(verify_full(&record, [8; 16], 3, &frame).is_err());
+        assert!(verify_full(&record, guid, 4, &frame).is_err());
+        record[100] ^= 1;
+        assert!(verify_full(&record, guid, 3, &frame).is_err());
+    }
+
+    #[test]
+    fn compact_prefix_ignores_stale_tail_but_clears_it_for_full_reads() {
+        let guid = [9; 16];
+        let mut frame = vec![3u8; PAGE_SIZE as usize];
+        let record = encode_prefix(guid, 2, &frame, PAGE_BYTES * 2);
+        frame[PAGE_BYTES * 3] ^= 1;
+        verify_full(&record, guid, 2, &frame).unwrap();
+        clear_uncovered_tail(&record, &mut frame);
+        assert_eq!(frame[PAGE_BYTES * 2], 0);
+        frame[PAGE_BYTES] ^= 1;
+        assert!(verify_full(&record, guid, 2, &frame).is_err());
+    }
+}