From 0ac8f457ec13db2b8232f079b937229360bc7303 Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 15:06:10 +1100 Subject: [PATCH 1/3] fix: reject corrupt blob reads Signed-off-by: wchwawa --- README.md | 7 + src/store/blob_store/file/integrity.rs | 204 +++++++++++++++ src/store/blob_store/file/mod.rs | 345 +++++++++++++++++++++++-- 3 files changed, 533 insertions(+), 23 deletions(-) create mode 100644 src/store/blob_store/file/integrity.rs diff --git a/README.md b/README.md index 0484a16..37b5271 100644 --- a/README.md +++ b/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. diff --git a/src/store/blob_store/file/integrity.rs b/src/store/blob_store/file/integrity.rs new file mode 100644 index 0000000..2709e6c --- /dev/null +++ b/src/store/blob_store/file/integrity.rs @@ -0,0 +1,204 @@ +//! 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 { + 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(()) +} + +pub(super) fn verify_full( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + slot: u64, + frame: &[u8], +) -> Result<()> { + validate_record(record, guid, slot)?; + debug_assert_eq!(frame.len(), PAGE_SIZE as usize); + verify_pages(record, guid, 0, frame) +} + +pub(super) fn verify_range( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + slot: u64, + byte_offset: u64, + bytes: &[u8], +) -> Result<()> { + validate_record(record, guid, slot)?; + let start = usize::try_from(byte_offset) + .map_err(|_| Error::node_corrupt("FileBlobStore::Integrity::range"))?; + 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(guid)); + } + verify_pages(record, guid, start / PAGE_BYTES, bytes) +} + +fn verify_pages( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + first_page: usize, + bytes: &[u8], +) -> Result<()> { + let covered_pages = u32::from_le_bytes( + record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] + .try_into() + .unwrap(), + ) as usize; + 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. +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()); + } +} diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index ae65975..d8a5af3 100644 --- a/src/store/blob_store/file/mod.rs +++ b/src/store/blob_store/file/mod.rs @@ -12,6 +12,8 @@ //! / //! blobs.dat — single packed file, blob N lives at byte //! offset N * PAGE_SIZE +//! integrity.dat — slot-aligned per-page checksums for blobs.dat; +//! synced before a new manifest mapping //! manifest.bin — small file mapping BlobGuid → slot number //! plus `next_slot`; //! rewritten only when the manifest delta log @@ -80,13 +82,13 @@ //! Switching between them is an internal performance toggle; no //! caller-visible behaviour changes. +mod integrity; #[cfg(all(target_os = "linux", feature = "io-uring"))] mod uring; use std::collections::HashMap; use std::fs::{File, OpenOptions}; use std::io::{self, Read, Write}; -#[cfg(not(all(target_os = "linux", feature = "io-uring")))] use std::os::unix::fs::FileExt; use std::os::unix::fs::OpenOptionsExt; use std::os::unix::io::AsRawFd; @@ -109,6 +111,10 @@ use self::uring::UringContext; /// Filename of the packed blob data file inside `data_dir`. const DATA_FILENAME: &str = "blobs.dat"; +/// Authoritative per-page checksums, addressed by the same shadow slot. +const INTEGRITY_FILENAME: &str = "integrity.dat"; +/// Distinguishes an interrupted v6 log-only migration from a lost v7 snapshot. +const INTEGRITY_MIGRATING_FILENAME: &str = "integrity.migrating"; /// Advisory lock file inside `data_dir`, flock'd for the lifetime of /// an open instance. Readers hold shared locks; writers hold exclusive locks. const LOCK_FILENAME: &str = "store.lock"; @@ -168,8 +174,11 @@ const MANIFEST_MAGIC: [u8; 8] = *b"ARTSNMNF"; /// an older read-accelerator design. v6 drops it: the current packed /// read index validates against the blob header stamp, so the /// manifest generation field was dead weight. Older manifests (incl. -/// v5) are refused on load, not migrated. -const MANIFEST_VERSION: u16 = 6; +/// v5) are refused on load. v7 requires durable per-page checksum +/// records before a manifest mapping can be trusted; writable v6 +/// stores are migrated once without rewriting blob images. +const MANIFEST_VERSION: u16 = 7; +const LEGACY_MANIFEST_VERSION: u16 = 6; /// Per-record magic for `manifest.log`. const MANIFEST_LOG_MAGIC: [u8; 4] = *b"MLG1"; const MANIFEST_LOG_TY_SET: u8 = 1; @@ -308,6 +317,7 @@ pub struct FileBlobStore { flush_gap_hook: FlushGapHook, access: FileAccess, data_file: File, + integrity_file: File, read_index_file: File, value_segment_file: File, manifest: RwLock, @@ -315,6 +325,8 @@ pub struct FileBlobStore { /// Every blob rewrite uses a fresh shadow slot, so both inserts and /// replacements leave manifest work until [`Self::flush_locked`]. manifest_dirty: AtomicBool, + /// Checksum records must reach stable storage before a manifest delta. + integrity_dirty: AtomicBool, /// Monotonic counter bumped before each data-file write. /// `flush` syncs up to the observed epoch instead of clearing /// a single bool, so a racing writer cannot be hidden by a @@ -352,6 +364,9 @@ pub struct FileBlobStore { #[derive(Debug)] struct Manifest { + /// Format read from disk; a writable v6 open performs a one-time + /// checksum migration before exposing the store. + version: u16, /// guid → packed-file slot. entries: HashMap, /// Next never-used slot to hand out when no reusable slot is @@ -542,6 +557,85 @@ fn preflight_writable_journal(data_dir: &Path, access: FileAccess) -> Result<()> Ok(()) } +/// Migrate v6 by checksumming each currently referenced frame without +/// rewriting it. The v7 manifest snapshot is the final publication step: +/// a crash before its rename leaves a v6 store that repeats this pass; a +/// crash after it leaves every referenced slot with a durable checksum. +fn open_integrity_file( + data_dir: &Path, + data_file: &File, + manifest: &mut Manifest, + access: FileAccess, +) -> Result { + if manifest.version != MANIFEST_VERSION && access.is_read_only() { + return Err(Error::BlobStoreIo(io::Error::other( + "legacy store requires a writable integrity migration", + ))); + } + let path = data_dir.join(INTEGRITY_FILENAME); + let migrating = data_dir.join(INTEGRITY_MIGRATING_FILENAME); + if manifest.version == MANIFEST_VERSION && !path.exists() { + return Err(Error::node_corrupt( + "FileBlobStore::Integrity::missing file", + )); + } + if manifest.version == 0 + && !manifest.entries.is_empty() + && path.metadata().is_ok_and(|metadata| metadata.len() != 0) + && !migrating.exists() + { + return Err(Error::node_corrupt( + "FileBlobStore::Integrity::missing protected manifest", + )); + } + let mut options = OpenOptions::new(); + options + .read(true) + .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW | libc::O_NONBLOCK); + if !access.is_read_only() { + options.write(true).create(true).mode(0o600); + } + let file = options.open(path)?; + if !file.metadata()?.is_file() { + return Err(Error::node_corrupt( + "FileBlobStore::Integrity::not a regular file", + )); + } + if manifest.version == MANIFEST_VERSION { + return Ok(file); + } + + if manifest.version == 0 && !manifest.entries.is_empty() && !migrating.exists() { + let mut marker = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&migrating)?; + marker.write_all(b"HOLT-INTEGRITY-MIGRATION-V1\n")?; + marker.sync_all()?; + sync_dir(data_dir)?; + } + let mut frame = AlignedBlobBuf::zeroed(); + for (guid, entry) in &manifest.entries { + let data_offset = entry + .slot + .checked_mul(u64::from(PAGE_SIZE)) + .ok_or_else(|| Error::node_corrupt("FileBlobStore::Integrity::data offset"))?; + data_file.read_exact_at(frame.as_mut_slice(), data_offset)?; + let record = integrity::encode(*guid, entry.slot, frame.as_slice()); + file.write_all_at(&record, integrity::record_offset(entry.slot)?)?; + } + file.sync_all()?; + manifest.persist_snapshot(data_dir)?; + manifest.truncate_log()?; + manifest.version = MANIFEST_VERSION; + if migrating.exists() { + std::fs::remove_file(migrating)?; + sync_dir(data_dir)?; + } + Ok(file) +} + impl FileBlobStore { /// Open or create a persistent store at `data_dir`. /// @@ -637,8 +731,9 @@ impl FileBlobStore { let _ = libc::fcntl(data_file.as_raw_fd(), libc::F_NOCACHE, 1); } - let manifest = + let mut manifest = Manifest::load_or_create(&manifest_path, &manifest_log_path, !access.is_read_only())?; + let integrity_file = open_integrity_file(&data_dir, &data_file, &mut manifest, access)?; let file_slots = slots_for_len(data_file.metadata()?.len()); let preallocated_slots = file_slots.max(manifest.next_slot); @@ -662,10 +757,12 @@ impl FileBlobStore { flush_gap_hook: FlushGapHook::default(), access, data_file, + integrity_file, read_index_file, value_segment_file, manifest: RwLock::new(manifest), manifest_dirty: AtomicBool::new(false), + integrity_dirty: AtomicBool::new(false), data_write_epoch: AtomicU64::new(0), data_sync_epoch: AtomicU64::new(0), data_io_lock: Mutex::new(()), @@ -721,8 +818,9 @@ impl FileBlobStore { let _ = libc::fcntl(data_file.as_raw_fd(), libc::F_NOCACHE, 1); } - let manifest = + let mut manifest = Manifest::load_or_create(&manifest_path, &manifest_log_path, !access.is_read_only())?; + let integrity_file = open_integrity_file(&data_dir, &data_file, &mut manifest, access)?; let file_slots = slots_for_len(data_file.metadata()?.len()); let preallocated_slots = file_slots.max(manifest.next_slot); @@ -734,10 +832,12 @@ impl FileBlobStore { flush_gap_hook: FlushGapHook::default(), access, data_file, + integrity_file, read_index_file, value_segment_file, manifest: RwLock::new(manifest), manifest_dirty: AtomicBool::new(false), + integrity_dirty: AtomicBool::new(false), data_write_epoch: AtomicU64::new(0), data_sync_epoch: AtomicU64::new(0), data_io_lock: Mutex::new(()), @@ -772,6 +872,7 @@ impl FileBlobStore { } } + #[cfg(any(test, all(target_os = "linux", feature = "io-uring")))] fn offset_of(&self, guid: BlobGuid) -> Result { Ok(self.entry_of(guid)?.slot * u64::from(PAGE_SIZE)) } @@ -839,6 +940,35 @@ impl FileBlobStore { }) } + fn read_integrity_record( + &self, + guid: BlobGuid, + slot: u64, + ) -> Result<[u8; integrity::RECORD_BYTES]> { + let mut record = [0u8; integrity::RECORD_BYTES]; + self.integrity_file + .read_exact_at(&mut record, integrity::record_offset(slot)?) + .map_err(|error| { + if error.kind() == io::ErrorKind::UnexpectedEof { + Error::node_corrupt("FileBlobStore::Integrity::missing record") + .with_blob_guid(guid) + } else { + Error::BlobStoreIo(error) + } + })?; + Ok(record) + } + + fn write_integrity_records(&self, writes: &[PreparedBlobWrite<'_>]) -> Result<()> { + self.integrity_dirty.store(true, Ordering::Release); + for write in writes { + let record = integrity::encode(write.guid, write.slot, write.src.as_slice()); + self.integrity_file + .write_all_at(&record, integrity::record_offset(write.slot)?)?; + } + Ok(()) + } + fn reserve_write_entries( &self, guids: impl IntoIterator, @@ -1160,6 +1290,10 @@ impl FileBlobStore { self.sync_data_file()?; self.mark_data_synced(epoch); } + if self.integrity_dirty.load(Ordering::Acquire) { + self.integrity_file.sync_data()?; + self.integrity_dirty.store(false, Ordering::Release); + } let mut publish_free_slots = false; if self.manifest_dirty.swap(false, Ordering::AcqRel) { @@ -1195,6 +1329,10 @@ impl FileBlobStore { let len = slots.saturating_mul(u64::from(PAGE_SIZE)); let mut bytes = 0; bytes += shrink_file_to_len(&self.data_file, len)?; + bytes += shrink_file_to_len( + &self.integrity_file, + slots.saturating_mul(integrity::RECORD_BYTES as u64), + )?; bytes += shrink_file_to_len(&self.read_index_file, len)?; bytes += shrink_file_to_len(&self.value_segment_file, len)?; self.preallocated_slots.store(slots, Ordering::Release); @@ -1215,6 +1353,12 @@ impl FileBlobStore { item.to_slot, &mut data, )?); + let previous = self.read_integrity_record(item.guid, item.from_slot)?; + integrity::verify_full(&previous, item.guid, item.from_slot, data.as_slice())?; + let record = integrity::relocate_verified(&previous, item.to_slot); + self.integrity_file + .write_all_at(&record, integrity::record_offset(item.to_slot)?)?; + bytes = bytes.saturating_add(integrity::RECORD_BYTES as u64); bytes = bytes.saturating_add(copy_advisory_slot( &self.read_index_file, item.from_slot, @@ -1230,6 +1374,7 @@ impl FileBlobStore { } self.data_file.sync_all()?; + self.integrity_file.sync_all()?; self.read_index_file.sync_all()?; self.value_segment_file.sync_all()?; Ok(bytes) @@ -1508,8 +1653,12 @@ impl BlobStore for FileBlobStore { fn read_blob(&self, guid: BlobGuid, dst: &mut AlignedBlobBuf) -> Result<()> { let _slot = self.enter_slot_read(); - let offset = self.offset_of(guid)?; + let entry = self.entry_of(guid)?; + let offset = entry.slot * u64::from(PAGE_SIZE); self.pread_at(offset, dst)?; + let record = self.read_integrity_record(guid, entry.slot)?; + integrity::verify_full(&record, guid, entry.slot, dst.as_slice())?; + integrity::clear_uncovered_tail(&record, dst.as_mut_slice()); Ok(()) } @@ -1548,10 +1697,18 @@ impl BlobStore for FileBlobStore { offsets .into_iter() - .map(|off| match off { + .zip(guids) + .zip(dsts.iter_mut()) + .map(|((off, guid), dst)| match off { Err(e) => Err(e), - Ok(_) => match &batch_result { - Ok(()) => Ok(()), + Ok(offset) => match &batch_result { + Ok(()) => { + let slot = offset / u64::from(PAGE_SIZE); + let record = self.read_integrity_record(*guid, slot)?; + integrity::verify_full(&record, *guid, slot, dst.as_slice())?; + integrity::clear_uncovered_tail(&record, dst.as_mut_slice()); + Ok(()) + } Err(e) => Err(Error::BlobStoreIo(io::Error::other(format!( "batched uring read failed: {e}" )))), @@ -1616,7 +1773,8 @@ impl BlobStore for FileBlobStore { "ranged read length must be a 4 KB multiple" ); let _slot = self.enter_slot_read(); - let offset = self.offset_of(guid)? + byte_offset; + let entry = self.entry_of(guid)?; + let offset = entry.slot * u64::from(PAGE_SIZE) + byte_offset; #[cfg(all(target_os = "linux", feature = "io-uring"))] { @@ -1629,7 +1787,8 @@ impl BlobStore for FileBlobStore { use std::os::unix::fs::FileExt; self.data_file.read_exact_at(dst, offset)?; } - Ok(()) + let record = self.read_integrity_record(guid, entry.slot)?; + integrity::verify_range(&record, guid, entry.slot, byte_offset, dst) } fn read_index_range(&self, guid: BlobGuid, byte_offset: u64, dst: &mut [u8]) -> Result { @@ -1763,7 +1922,10 @@ impl BlobStore for FileBlobStore { let prepared = self.prepare_blob_writes(&writes)?; self.clear_reserved_read_accelerators(&prepared); self.mark_data_write_started(); - if let Err(error) = self.pwrite_at(prepared[0].offset, src) { + if let Err(error) = self + .pwrite_at(prepared[0].offset, src) + .and_then(|()| self.write_integrity_records(&prepared)) + { self.release_reserved_slots(prepared.iter().map(|write| write.slot)); return Err(error); } @@ -1780,7 +1942,10 @@ impl BlobStore for FileBlobStore { } self.clear_reserved_read_accelerators(&prepared); self.mark_data_write_started(); - if let Err(error) = self.pwrite_many_at(&prepared) { + if let Err(error) = self + .pwrite_many_at(&prepared) + .and_then(|()| self.write_integrity_records(&prepared)) + { self.release_reserved_slots(prepared.iter().map(|write| write.slot)); return Err(error); } @@ -1800,7 +1965,10 @@ impl BlobStore for FileBlobStore { let prepared = self.prepare_blob_writes(writes)?; self.clear_reserved_read_accelerators(&prepared); let epoch = self.mark_data_write_started(); - if let Err(error) = self.pwrite_many_and_sync_at(&prepared) { + if let Err(error) = self + .pwrite_many_and_sync_at(&prepared) + .and_then(|()| self.write_integrity_records(&prepared)) + { self.release_reserved_slots(prepared.iter().map(|write| write.slot)); return Err(error); } @@ -1814,7 +1982,10 @@ impl BlobStore for FileBlobStore { let prepared = self.prepare_blob_writes(writes)?; self.clear_reserved_read_accelerators(&prepared); self.mark_data_write_started(); - if let Err(error) = self.pwrite_many_at(&prepared) { + if let Err(error) = self + .pwrite_many_at(&prepared) + .and_then(|()| self.write_integrity_records(&prepared)) + { self.release_reserved_slots(prepared.iter().map(|write| write.slot)); return Err(error); } @@ -1906,7 +2077,9 @@ impl BlobStore for FileBlobStore { fn needs_flush(&self) -> bool { !self.access.is_read_only() - && (self.data_needs_sync().is_some() || self.manifest_dirty.load(Ordering::Acquire)) + && (self.data_needs_sync().is_some() + || self.integrity_dirty.load(Ordering::Acquire) + || self.manifest_dirty.load(Ordering::Acquire)) } fn vacuum(&self) -> Result { @@ -2028,9 +2201,9 @@ impl Manifest { None }; - let (mut entries, mut next_slot) = match File::open(path) { + let (mut entries, mut next_slot, version) = match File::open(path) { Ok(mut f) => Self::parse_snapshot(&mut f)?, - Err(e) if e.kind() == io::ErrorKind::NotFound => (HashMap::new(), 0), + Err(e) if e.kind() == io::ErrorKind::NotFound => (HashMap::new(), 0, 0), Err(e) => return Err(Error::BlobStoreIo(e)), }; @@ -2045,6 +2218,7 @@ impl Manifest { let reusable_slots = ReusableSlots::reconstruct(next_slot, &used_slots)?; Ok(Self { + version, entries, next_slot, reusable_slots, @@ -2056,7 +2230,7 @@ impl Manifest { }) } - fn parse_snapshot(f: &mut File) -> Result<(HashMap, u64)> { + fn parse_snapshot(f: &mut File) -> Result<(HashMap, u64, u16)> { // Header: magic 8 + version 2 + count 4 + reserved 2 + next_slot 8 = 24 B. let mut hdr = [0u8; 24]; f.read_exact(&mut hdr)?; @@ -2064,7 +2238,7 @@ impl Manifest { return Err(Error::node_corrupt("FileBlobStore::Manifest::magic")); } let version = u16::from_le_bytes([hdr[8], hdr[9]]); - if version != MANIFEST_VERSION { + if version != MANIFEST_VERSION && version != LEGACY_MANIFEST_VERSION { return Err(Error::node_corrupt( "FileBlobStore::Manifest::version (older manifests are not migrated)", )); @@ -2089,7 +2263,7 @@ impl Manifest { used_slots.push(s); } ReusableSlots::reconstruct(next_slot, &used_slots)?; - Ok((entries, next_slot)) + Ok((entries, next_slot, version)) } fn publish_write_entry(&mut self, guid: BlobGuid, slot: u64) { @@ -3850,7 +4024,7 @@ mod tests { } #[test] - fn manifest_delta_log_replays_without_snapshot() { + fn manifest_delta_log_replays_after_integrity_snapshot() { let dir = tempfile::tempdir().unwrap(); let g: BlobGuid = [0xC1; 16]; { @@ -3860,7 +4034,7 @@ mod tests { b.write_blob(g, &buf_with(9)).unwrap(); b.flush().unwrap(); assert!(dir.path().join(MANIFEST_LOG_FILENAME).exists()); - assert!(!dir.path().join(MANIFEST_FILENAME).exists()); + assert!(dir.path().join(MANIFEST_FILENAME).exists()); } let Some(b) = try_open(dir.path()) else { @@ -3871,6 +4045,131 @@ mod tests { assert_eq!(dst.as_slice()[100], 9); } + #[test] + fn corrupted_active_value_is_rejected_by_full_and_page_reads() { + let dir = tempfile::tempdir().unwrap(); + let guid: BlobGuid = [0xC7; 16]; + let slot = { + let Some(store) = try_open(dir.path()) else { + return; + }; + store.write_blob(guid, &buf_with(9)).unwrap(); + store.flush().unwrap(); + store.entry_of(guid).unwrap().slot + }; + let data = OpenOptions::new() + .read(true) + .write(true) + .open(dir.path().join(DATA_FILENAME)) + .unwrap(); + let offset = slot * u64::from(PAGE_SIZE) + 100; + let mut byte = [0u8; 1]; + data.read_exact_at(&mut byte, offset).unwrap(); + byte[0] ^= 1; + data.write_all_at(&byte, offset).unwrap(); + data.sync_all().unwrap(); + + let store = FileBlobStore::open(dir.path()).unwrap(); + let mut frame = AlignedBlobBuf::zeroed(); + assert!(matches!( + store.read_blob(guid, &mut frame), + Err(Error::NodeCorrupt { .. }) + )); + assert!(matches!( + store.read_blob_range(guid, 0, &mut frame.as_mut_slice()[..4096]), + Err(Error::NodeCorrupt { .. }) + )); + } + + #[test] + fn missing_protected_manifest_cannot_rebaseline_blob_checksums() { + let dir = tempfile::tempdir().unwrap(); + let guid: BlobGuid = [0xC8; 16]; + { + let Some(store) = try_open(dir.path()) else { + return; + }; + store.write_blob(guid, &buf_with(9)).unwrap(); + store.flush().unwrap(); + } + std::fs::remove_file(dir.path().join(MANIFEST_FILENAME)).unwrap(); + assert!(matches!( + FileBlobStore::open(dir.path()), + Err(Error::NodeCorrupt { .. }) + )); + } + + #[test] + fn integrity_sidecar_rejects_symlink_replacement() { + use std::os::unix::fs::symlink; + + let dir = tempfile::tempdir().unwrap(); + { + let Some(store) = try_open(dir.path()) else { + return; + }; + store.write_blob([0xCB; 16], &buf_with(7)).unwrap(); + store.flush().unwrap(); + } + let sidecar = dir.path().join(INTEGRITY_FILENAME); + let saved = dir.path().join("saved-integrity.dat"); + std::fs::rename(&sidecar, &saved).unwrap(); + symlink(&saved, &sidecar).unwrap(); + assert!(FileBlobStore::open(dir.path()).is_err()); + } + + #[test] + fn legacy_snapshot_requires_writable_migration_before_reads() { + let dir = tempfile::tempdir().unwrap(); + let guid: BlobGuid = [0xC9; 16]; + { + let Some(store) = try_open(dir.path()) else { + return; + }; + store.write_blob(guid, &buf_with(9)).unwrap(); + store.flush().unwrap(); + } + let path = dir.path().join(MANIFEST_FILENAME); + let manifest = OpenOptions::new().write(true).open(path).unwrap(); + manifest + .write_all_at(&LEGACY_MANIFEST_VERSION.to_le_bytes(), 8) + .unwrap(); + manifest.sync_all().unwrap(); + std::fs::remove_file(dir.path().join(INTEGRITY_FILENAME)).unwrap(); + + assert!(FileBlobStore::open_read_only(dir.path()).is_err()); + let store = FileBlobStore::open(dir.path()).unwrap(); + let mut frame = AlignedBlobBuf::zeroed(); + store.read_blob(guid, &mut frame).unwrap(); + assert_eq!(frame.as_slice()[100], 9); + drop(store); + assert!(dir.path().join(INTEGRITY_FILENAME).exists()); + assert!(FileBlobStore::open_read_only(dir.path()).is_ok()); + } + + #[test] + fn legacy_log_only_store_is_migrated_without_losing_mapping() { + let dir = tempfile::tempdir().unwrap(); + let guid: BlobGuid = [0xCA; 16]; + { + let Some(store) = try_open(dir.path()) else { + return; + }; + store.write_blob(guid, &buf_with(12)).unwrap(); + store.flush().unwrap(); + } + std::fs::remove_file(dir.path().join(MANIFEST_FILENAME)).unwrap(); + std::fs::remove_file(dir.path().join(INTEGRITY_FILENAME)).unwrap(); + + let store = FileBlobStore::open(dir.path()).unwrap(); + let mut frame = AlignedBlobBuf::zeroed(); + store.read_blob(guid, &mut frame).unwrap(); + assert_eq!(frame.as_slice()[100], 12); + drop(store); + assert!(dir.path().join(MANIFEST_FILENAME).exists()); + assert!(!dir.path().join(INTEGRITY_MIGRATING_FILENAME).exists()); + } + #[test] fn manifest_delta_log_ignores_torn_tail() { let dir = tempfile::tempdir().unwrap(); From 1308fd133e0976c31b581e196d5c8c6c1718606b Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 15:35:36 +1100 Subject: [PATCH 2/3] fix: account for integrity file reclamation Signed-off-by: wchwawa --- src/api/stats.rs | 2 +- src/store/blob_store/file/mod.rs | 17 +++++++++++++++-- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/src/api/stats.rs b/src/api/stats.rs index e49237f..a949758 100644 --- a/src/api/stats.rs +++ b/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. /// diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index d8a5af3..869c6f2 100644 --- a/src/store/blob_store/file/mod.rs +++ b/src/store/blob_store/file/mod.rs @@ -2041,7 +2041,14 @@ impl BlobStore for FileBlobStore { let tail_target = next_slot .saturating_sub(tail_reclaimable_slots) .saturating_mul(u64::from(PAGE_SIZE)); + let integrity_tail_target = next_slot + .saturating_sub(tail_reclaimable_slots) + .saturating_mul(integrity::RECORD_BYTES as u64); let tail_reclaimable_bytes = reclaimable_tail_bytes(&self.data_file, tail_target) + .saturating_add(reclaimable_tail_bytes( + &self.integrity_file, + integrity_tail_target, + )) .saturating_add(reclaimable_tail_bytes(&self.read_index_file, tail_target)) .saturating_add(reclaimable_tail_bytes( &self.value_segment_file, @@ -3720,11 +3727,17 @@ mod tests { let free = b.store_stats(); assert_eq!(free.tail_reclaimable_slots, 2); assert_eq!(free.middle_reusable_slots, 0); - assert!(free.tail_reclaimable_bytes >= 2 * u64::from(PAGE_SIZE)); + assert!( + free.tail_reclaimable_bytes + >= 2 * (u64::from(PAGE_SIZE) + integrity::RECORD_BYTES as u64) + ); let vacuum = b.vacuum().unwrap(); assert_eq!(vacuum.slots_trimmed, 2); - assert!(vacuum.bytes_truncated >= 2 * u64::from(PAGE_SIZE)); + assert!( + vacuum.bytes_truncated + >= 2 * (u64::from(PAGE_SIZE) + integrity::RECORD_BYTES as u64) + ); let after = b.store_stats(); assert_eq!(after.next_slot, 1); From 15302d04c82fec16f116c2c350b3061456558d6e Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 16:19:50 +1100 Subject: [PATCH 3/3] perf: bound reads and parallelize blob checksums Compact records from the 0.8 line must never expose bytes beyond their authenticated extent. Read that prefix, clear the rest, and verify independent blobs in parallel for large Linux batches while preserving per-blob results. Signed-off-by: wchwawa --- src/store/blob_store/file/integrity.rs | 85 ++++++-- src/store/blob_store/file/mod.rs | 273 ++++++++++++++++++++----- src/store/blob_store/file/uring.rs | 83 ++++++-- 3 files changed, 354 insertions(+), 87 deletions(-) diff --git a/src/store/blob_store/file/integrity.rs b/src/store/blob_store/file/integrity.rs index 2709e6c..555b7d4 100644 --- a/src/store/blob_store/file/integrity.rs +++ b/src/store/blob_store/file/integrity.rs @@ -71,17 +71,76 @@ fn validate_record(record: &[u8; RECORD_BYTES], guid: BlobGuid, slot: u64) -> Re 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 { + 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<()> { - validate_record(record, guid, slot)?; - debug_assert_eq!(frame.len(), PAGE_SIZE as usize); - verify_pages(record, guid, 0, frame) + ValidatedRecord::new(*record, guid, slot)?.verify_full(frame) } +#[cfg(test)] pub(super) fn verify_range( record: &[u8; RECORD_BYTES], guid: BlobGuid, @@ -89,31 +148,16 @@ pub(super) fn verify_range( byte_offset: u64, bytes: &[u8], ) -> Result<()> { - validate_record(record, guid, slot)?; - let start = usize::try_from(byte_offset) - .map_err(|_| Error::node_corrupt("FileBlobStore::Integrity::range"))?; - 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(guid)); - } - verify_pages(record, guid, start / PAGE_BYTES, bytes) + 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<()> { - let covered_pages = u32::from_le_bytes( - record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] - .try_into() - .unwrap(), - ) as usize; for (index, page) in bytes.chunks_exact(PAGE_BYTES).enumerate() { if first_page + index >= covered_pages { continue; @@ -129,6 +173,7 @@ fn verify_pages( /// 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] diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index 869c6f2..876025a 100644 --- a/src/store/blob_store/file/mod.rs +++ b/src/store/blob_store/file/mod.rs @@ -872,7 +872,7 @@ impl FileBlobStore { } } - #[cfg(any(test, all(target_os = "linux", feature = "io-uring")))] + #[cfg(test)] fn offset_of(&self, guid: BlobGuid) -> Result { Ok(self.entry_of(guid)?.slot * u64::from(PAGE_SIZE)) } @@ -944,7 +944,7 @@ impl FileBlobStore { &self, guid: BlobGuid, slot: u64, - ) -> Result<[u8; integrity::RECORD_BYTES]> { + ) -> Result { let mut record = [0u8; integrity::RECORD_BYTES]; self.integrity_file .read_exact_at(&mut record, integrity::record_offset(slot)?) @@ -956,7 +956,7 @@ impl FileBlobStore { Error::BlobStoreIo(error) } })?; - Ok(record) + integrity::ValidatedRecord::new(record, guid, slot) } fn write_integrity_records(&self, writes: &[PreparedBlobWrite<'_>]) -> Result<()> { @@ -1070,15 +1070,15 @@ impl FileBlobStore { // `write_blob` clean of any conditional plumbing. #[cfg(all(target_os = "linux", feature = "io-uring"))] - fn pread_at(&self, offset: u64, dst: &mut AlignedBlobBuf) -> Result<()> { + fn pread_at(&self, offset: u64, len: usize, dst: &mut AlignedBlobBuf) -> Result<()> { let mut ring = self.uring.lock().unwrap(); - ring.pread_at(offset, dst)?; + ring.pread_at(offset, len, dst)?; Ok(()) } #[cfg(not(all(target_os = "linux", feature = "io-uring")))] - fn pread_at(&self, offset: u64, dst: &mut AlignedBlobBuf) -> Result<()> { - let dst = dst.as_mut_slice(); + fn pread_at(&self, offset: u64, len: usize, dst: &mut AlignedBlobBuf) -> Result<()> { + let dst = &mut dst.as_mut_slice()[..len]; self.data_file.read_exact_at(dst, offset)?; Ok(()) } @@ -1354,8 +1354,8 @@ impl FileBlobStore { &mut data, )?); let previous = self.read_integrity_record(item.guid, item.from_slot)?; - integrity::verify_full(&previous, item.guid, item.from_slot, data.as_slice())?; - let record = integrity::relocate_verified(&previous, item.to_slot); + previous.verify_full(data.as_slice())?; + let record = previous.relocate_verified(item.to_slot); self.integrity_file .write_all_at(&record, integrity::record_offset(item.to_slot)?)?; bytes = bytes.saturating_add(integrity::RECORD_BYTES as u64); @@ -1655,10 +1655,11 @@ impl BlobStore for FileBlobStore { let _slot = self.enter_slot_read(); let entry = self.entry_of(guid)?; let offset = entry.slot * u64::from(PAGE_SIZE); - self.pread_at(offset, dst)?; let record = self.read_integrity_record(guid, entry.slot)?; - integrity::verify_full(&record, guid, entry.slot, dst.as_slice())?; - integrity::clear_uncovered_tail(&record, dst.as_mut_slice()); + let covered = record.covered_bytes(); + self.pread_at(offset, covered, dst)?; + dst.as_mut_slice()[covered..].fill(0); + record.verify_full(dst.as_slice())?; Ok(()) } @@ -1671,15 +1672,21 @@ impl BlobStore for FileBlobStore { fn read_blobs(&self, guids: &[BlobGuid], dsts: &mut [AlignedBlobBuf]) -> Vec> { debug_assert_eq!(guids.len(), dsts.len()); let _slot = self.enter_slot_read(); - // Resolve offsets up front (lock-free manifest read). A guid - // that doesn't resolve is reported in its own slot and left - // out of the ring batch, so one bad guid can't sink the rest. - let offsets: Vec> = guids.iter().map(|g| self.offset_of(*g)).collect(); - - let mut batch: Vec<(u64, &mut AlignedBlobBuf)> = Vec::with_capacity(dsts.len()); - for (off, dst) in offsets.iter().zip(dsts.iter_mut()) { - if let Ok(off) = off { - batch.push((*off, dst)); + // Resolve each slot and its durable covered extent before submitting + // data reads. Invalid records fail only their own guid. + let reads: Vec> = guids + .iter() + .map(|guid| { + let entry = self.entry_of(*guid)?; + let record = self.read_integrity_record(*guid, entry.slot)?; + Ok((entry.slot * u64::from(PAGE_SIZE), record)) + }) + .collect(); + + let mut batch: Vec<(u64, usize, &mut AlignedBlobBuf)> = Vec::with_capacity(dsts.len()); + for (read, dst) in reads.iter().zip(dsts.iter_mut()) { + if let Ok((offset, record)) = read { + batch.push((*offset, record.covered_bytes(), dst)); } } @@ -1694,27 +1701,61 @@ impl BlobStore for FileBlobStore { ring.pread_many_at(&mut batch) }; drop(batch); - - offsets - .into_iter() - .zip(guids) - .zip(dsts.iter_mut()) - .map(|((off, guid), dst)| match off { - Err(e) => Err(e), - Ok(offset) => match &batch_result { - Ok(()) => { - let slot = offset / u64::from(PAGE_SIZE); - let record = self.read_integrity_record(*guid, slot)?; - integrity::verify_full(&record, *guid, slot, dst.as_slice())?; - integrity::clear_uncovered_tail(&record, dst.as_mut_slice()); - Ok(()) - } - Err(e) => Err(Error::BlobStoreIo(io::Error::other(format!( - "batched uring read failed: {e}" + if let Err(error) = batch_result { + return reads + .into_iter() + .map(|read| match read { + Err(error) => Err(error), + Ok(_) => Err(Error::BlobStoreIo(io::Error::other(format!( + "batched uring read failed: {error}" )))), - }, - }) - .collect() + }) + .collect(); + } + + fn verify_read( + read: Result<(u64, integrity::ValidatedRecord)>, + dst: &mut AlignedBlobBuf, + ) -> Result<()> { + let (_, record) = read?; + dst.as_mut_slice()[record.covered_bytes()..].fill(0); + record.verify_full(dst.as_slice())?; + Ok(()) + } + + // The ring completes data I/O before verification. For large batches, + // independent page checksums can use multiple cores without changing + // read ordering or sharing a destination buffer. + const VERIFY_FANOUT: usize = 8; + if reads.len() < 32 { + return reads + .into_iter() + .zip(dsts.iter_mut()) + .map(|(read, dst)| verify_read(read, dst)) + .collect(); + } + let chunk_len = reads.len().div_ceil(VERIFY_FANOUT); + std::thread::scope(|scope| { + let mut items = reads.into_iter().zip(dsts.iter_mut()); + let mut handles = Vec::with_capacity(VERIFY_FANOUT); + loop { + let chunk: Vec<_> = items.by_ref().take(chunk_len).collect(); + if chunk.is_empty() { + break; + } + handles.push(scope.spawn(move || { + chunk + .into_iter() + .map(|(read, dst)| verify_read(read, dst)) + .collect::>() + })); + } + let mut results = Vec::with_capacity(guids.len()); + for handle in handles { + results.extend(handle.join().expect("blob checksum worker panicked")); + } + results + }) } #[cfg(not(all(target_os = "linux", feature = "io-uring")))] @@ -1774,21 +1815,37 @@ impl BlobStore for FileBlobStore { ); let _slot = self.enter_slot_read(); let entry = self.entry_of(guid)?; + let start = usize::try_from(byte_offset) + .map_err(|_| Error::node_corrupt("FileBlobStore::Integrity::range"))?; + if start % 4096 != 0 + || dst.len() % 4096 != 0 + || start + .checked_add(dst.len()) + .is_none_or(|end| end > PAGE_SIZE as usize) + { + return Err(Error::node_corrupt("FileBlobStore::Integrity::range").with_blob_guid(guid)); + } + let record = self.read_integrity_record(guid, entry.slot)?; + let covered = record.covered_bytes(); + let read_len = covered.saturating_sub(start).min(dst.len()); let offset = entry.slot * u64::from(PAGE_SIZE) + byte_offset; #[cfg(all(target_os = "linux", feature = "io-uring"))] { - let mut ring = self.uring.lock().unwrap(); - ring.pread_slice_at(offset, dst)?; + if read_len != 0 { + let mut ring = self.uring.lock().unwrap(); + ring.pread_slice_at(offset, &mut dst[..read_len])?; + } } #[cfg(not(all(target_os = "linux", feature = "io-uring")))] { - use std::os::unix::fs::FileExt; - self.data_file.read_exact_at(dst, offset)?; + if read_len != 0 { + self.data_file.read_exact_at(&mut dst[..read_len], offset)?; + } } - let record = self.read_integrity_record(guid, entry.slot)?; - integrity::verify_range(&record, guid, entry.slot, byte_offset, dst) + dst[read_len..].fill(0); + record.verify_range(byte_offset, dst) } fn read_index_range(&self, guid: BlobGuid, byte_offset: u64, dst: &mut [u8]) -> Result { @@ -3347,6 +3404,78 @@ mod tests { assert_eq!(dst.as_slice()[100], 7); } + /// A frame image whose data area ends at `space_used`, with every other + /// byte set to `fill` so persisted and cleared bytes are distinguishable. + fn frame_with_extent(space_used: u32, fill: u8) -> AlignedBlobBuf { + let mut frame = AlignedBlobBuf::zeroed(); + frame.as_mut_slice().fill(fill); + crate::store::blob_frame::BlobFrame::wrap(frame.as_mut_slice()) + .header_mut() + .space_used = space_used; + frame + } + + #[test] + fn compact_record_hides_bytes_beyond_its_authenticated_extent() { + use crate::layout::DATA_AREA_START; + + let dir = tempfile::tempdir().unwrap(); + let Some(b) = try_open(dir.path()) else { + return; + }; + let old: BlobGuid = [0x41; 16]; + let new: BlobGuid = [0x42; 16]; + // A full frame takes a slot, then moves to a shadow slot and + // releases the first one. + b.write_blob(old, &frame_with_extent(PAGE_SIZE, 0xab)) + .unwrap(); + let released = b.entry_of(old).unwrap().slot; + b.write_blob(old, &frame_with_extent(PAGE_SIZE, 0xab)) + .unwrap(); + b.flush().unwrap(); + + let small = frame_with_extent(DATA_AREA_START + 5_000, 0xcd); + b.write_blob(new, &small).unwrap(); + let new_slot = b.entry_of(new).unwrap().slot; + assert_eq!(new_slot, released); + let extent = (DATA_AREA_START + 2 * 4096) as usize; + // A release/0.8 store can carry a compact integrity record. Leave + // non-authoritative bytes physically present after its extent. + let record = integrity::encode_prefix(new, new_slot, small.as_slice(), extent); + b.integrity_file + .write_all_at(&record, integrity::record_offset(new_slot).unwrap()) + .unwrap(); + + let mut dst = AlignedBlobBuf::zeroed(); + b.read_blob(new, &mut dst).unwrap(); + assert_eq!(&dst.as_slice()[..extent], &small.as_slice()[..extent]); + assert!(dst.as_slice()[extent..].iter().all(|&byte| byte == 0)); + let mut batch = [AlignedBlobBuf::zeroed(), AlignedBlobBuf::zeroed()]; + for result in b.read_blobs(&[new, new], &mut batch) { + result.unwrap(); + } + assert!(batch.iter().all(|frame| frame.as_slice() == dst.as_slice())); + + // A range read beyond the authenticated extent must not expose + // non-authoritative bytes in the slot. + // A ranged read needs a 4 KB-aligned buffer for O_DIRECT. + let mut tail = AlignedBlobBuf::zeroed(); + b.read_blob_range(new, extent as u64, &mut tail.as_mut_slice()[..4096]) + .unwrap(); + assert!(tail.as_slice()[..4096].iter().all(|&byte| byte == 0)); + b.read_blob_range( + new, + (extent - 4096) as u64, + &mut tail.as_mut_slice()[..8192], + ) + .unwrap(); + assert_eq!( + &tail.as_slice()[..4096], + &small.as_slice()[extent - 4096..extent] + ); + assert!(tail.as_slice()[4096..8192].iter().all(|&byte| byte == 0)); + } + #[test] fn write_replaces_existing_through_shadow_slot() { let dir = tempfile::tempdir().unwrap(); @@ -4094,6 +4223,56 @@ mod tests { )); } + #[cfg(all(target_os = "linux", feature = "io-uring"))] + #[test] + fn large_batch_checks_each_blob_and_preserves_result_order() { + const COUNT: usize = 40; + const DAMAGED: usize = 17; + let dir = tempfile::tempdir().unwrap(); + let mut guids = Vec::with_capacity(COUNT); + let slot = { + let Some(store) = try_open(dir.path()) else { + return; + }; + for index in 0..COUNT { + let mut guid = [0u8; 16]; + guid[0] = index as u8; + store + .write_blob(guid, &frame_with_extent(PAGE_SIZE, index as u8)) + .unwrap(); + guids.push(guid); + } + store.flush().unwrap(); + store.entry_of(guids[DAMAGED]).unwrap().slot + }; + let data = OpenOptions::new() + .read(true) + .write(true) + .open(dir.path().join(DATA_FILENAME)) + .unwrap(); + let offset = slot * u64::from(PAGE_SIZE) + 8192; + let mut byte = [0u8; 1]; + data.read_exact_at(&mut byte, offset).unwrap(); + byte[0] ^= 1; + data.write_all_at(&byte, offset).unwrap(); + data.sync_all().unwrap(); + + let store = FileBlobStore::open(dir.path()).unwrap(); + let mut frames = (0..COUNT) + .map(|_| AlignedBlobBuf::zeroed()) + .collect::>(); + let results = store.read_blobs(&guids, &mut frames); + assert_eq!(results.len(), COUNT); + for (index, (result, frame)) in results.into_iter().zip(frames).enumerate() { + if index == DAMAGED { + assert!(matches!(result, Err(Error::NodeCorrupt { .. }))); + } else { + result.unwrap(); + assert_eq!(frame.as_slice()[100], index as u8); + } + } + } + #[test] fn missing_protected_manifest_cannot_rebaseline_blob_checksums() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/store/blob_store/file/uring.rs b/src/store/blob_store/file/uring.rs index f2008ba..cd865a6 100644 --- a/src/store/blob_store/file/uring.rs +++ b/src/store/blob_store/file/uring.rs @@ -175,8 +175,13 @@ impl UringContext { /// Synchronous `pread` via `io_uring`: same shape as /// [`Self::pwrite_at`]. - pub(super) fn pread_at(&mut self, offset: u64, buf: &mut AlignedBlobBuf) -> io::Result<()> { - let entry = self.read_entry(offset, buf).user_data(0); + pub(super) fn pread_at( + &mut self, + offset: u64, + len: usize, + buf: &mut AlignedBlobBuf, + ) -> io::Result<()> { + let entry = self.read_entry(offset, len, buf).user_data(0); // SAFETY: same argument as `pwrite_at` — `buf` outlives the // synchronous `submit_and_wait`. @@ -197,11 +202,9 @@ impl UringContext { if n < 0 { return Err(io::Error::from_raw_os_error(-n)); } - if (n as usize) != buf.len() { + if (n as usize) != len { return Err(io::Error::other(format!( - "short uring read: read {} of {}", - n, - buf.len() + "short uring read: read {n} of {len}" ))); } Ok(()) @@ -262,13 +265,13 @@ impl UringContext { /// `pread` backend the same queue depth comes for free from /// concurrent lock-free `read_exact_at` calls). /// - /// Each `reads[i]` is `(byte_offset, dst)`; every `dst` must be a + /// Each `reads[i]` is `(byte_offset, covered_bytes, dst)`; every `dst` must be a /// distinct buffer (the kernel writes each independently). Reads /// are submitted in slice order — callers that care about device /// LBA sequencing pre-sort by offset. pub(super) fn pread_many_at( &mut self, - reads: &mut [(u64, &mut AlignedBlobBuf)], + reads: &mut [(u64, usize, &mut AlignedBlobBuf)], ) -> io::Result<()> { let total = reads.len(); let mut base = 0usize; @@ -331,12 +334,15 @@ impl UringContext { self.drain_write_batch(chunk) } - fn submit_read_batch(&mut self, chunk: &mut [(u64, &mut AlignedBlobBuf)]) -> io::Result<()> { + fn submit_read_batch( + &mut self, + chunk: &mut [(u64, usize, &mut AlignedBlobBuf)], + ) -> io::Result<()> { debug_assert!(!chunk.is_empty()); debug_assert!(chunk.len() <= RING_DEPTH_USIZE); - for (idx, (offset, buf)) in chunk.iter_mut().enumerate() { - let entry = self.read_entry(*offset, buf).user_data(idx as u64); + for (idx, (offset, len, buf)) in chunk.iter_mut().enumerate() { + let entry = self.read_entry(*offset, *len, buf).user_data(idx as u64); // SAFETY: every SQE references a buffer borrowed from // `chunk`; this function synchronously waits for all @@ -450,20 +456,21 @@ impl UringContext { .build() } - fn read_entry(&self, offset: u64, buf: &mut AlignedBlobBuf) -> squeue::Entry { + fn read_entry(&self, offset: u64, len: usize, buf: &mut AlignedBlobBuf) -> squeue::Entry { + debug_assert!(len > 0 && len <= buf.len() && len % 4096 == 0); if self.fixed_buffers { if let Some(buffer_index) = buf.fixed_buffer_index() { return opcode::ReadFixed::new( self.fixed_fd, buf.as_mut_ptr(), - buf.len() as u32, + len as u32, buffer_index, ) .offset(offset) .build(); } } - opcode::Read::new(self.fixed_fd, buf.as_mut_ptr(), buf.len() as u32) + opcode::Read::new(self.fixed_fd, buf.as_mut_ptr(), len as u32) .offset(offset) .build() } @@ -524,7 +531,7 @@ impl UringContext { Ok(()) } - fn drain_read_batch(&mut self, chunk: &[(u64, &mut AlignedBlobBuf)]) -> io::Result<()> { + fn drain_read_batch(&mut self, chunk: &[(u64, usize, &mut AlignedBlobBuf)]) -> io::Result<()> { let mut seen = [0u64; CQ_BITMAP_WORDS]; let mut first_err: Option = None; @@ -559,7 +566,7 @@ impl UringContext { record_err(&mut first_err, io::Error::from_raw_os_error(-n)); continue; } - let expected = chunk[idx].1.len(); + let expected = chunk[idx].1; if (n as usize) != expected { record_err( &mut first_err, @@ -746,8 +753,12 @@ mod tests { let mut r0 = AlignedBlobBuf::zeroed(); let mut r1 = AlignedBlobBuf::zeroed(); let mut r2 = AlignedBlobBuf::zeroed(); - ring.pread_many_at(&mut [(2 * frame_len, &mut r2), (0, &mut r0), (frame_len, &mut r1)]) - .unwrap(); + ring.pread_many_at(&mut [ + (2 * frame_len, frame_len as usize, &mut r2), + (0, frame_len as usize, &mut r0), + (frame_len, frame_len as usize, &mut r1), + ]) + .unwrap(); assert!(r0.as_slice().iter().all(|&x| x == 0xAA)); assert!(r1.as_slice().iter().all(|&x| x == 0xBB)); @@ -782,10 +793,10 @@ mod tests { ring.sync_data().unwrap(); let mut reads: Vec = (0..count).map(|_| AlignedBlobBuf::zeroed()).collect(); - let mut read_refs: Vec<(u64, &mut AlignedBlobBuf)> = reads + let mut read_refs: Vec<(u64, usize, &mut AlignedBlobBuf)> = reads .iter_mut() .enumerate() - .map(|(i, b)| (i as u64 * frame_len, b)) + .map(|(i, b)| (i as u64 * frame_len, frame_len as usize, b)) .collect(); ring.pread_many_at(&mut read_refs).unwrap(); drop(read_refs); @@ -798,6 +809,38 @@ mod tests { } } + #[test] + fn pread_many_at_honors_each_covered_length() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("data"); + let file = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(true) + .open(&path) + .unwrap(); + let mut ring = UringContext::new(&file, None).unwrap(); + let first = frame(0xaa); + let second = frame(0xbb); + let frame_len = first.len() as u64; + ring.pwrite_many_at(&[(0, &first), (frame_len, &second)]) + .unwrap(); + ring.sync_data().unwrap(); + + let mut short = AlignedBlobBuf::zeroed(); + short.as_mut_slice().fill(0xee); + let mut full = AlignedBlobBuf::zeroed(); + ring.pread_many_at(&mut [ + (0, 4096, &mut short), + (frame_len, frame_len as usize, &mut full), + ]) + .unwrap(); + assert!(short.as_slice()[..4096].iter().all(|&byte| byte == 0xaa)); + assert!(short.as_slice()[4096..].iter().all(|&byte| byte == 0xee)); + assert!(full.as_slice().iter().all(|&byte| byte == 0xbb)); + } + #[test] fn pread_many_at_empty_is_ok() { let dir = tempfile::tempdir().unwrap();