From f85cdbd5b7007f557aaec22206d21338faaa76dd Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 15:12:07 +1100 Subject: [PATCH 1/3] fix: release blob integrity checks in 0.8.8 Signed-off-by: wchwawa --- CHANGELOG.md | 12 + Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 7 + src/api/stats.rs | 2 +- src/store/blob_store/file/integrity.rs | 206 ++++++++++++++ src/store/blob_store/file/mod.rs | 373 ++++++++++++++++++++++--- 7 files changed, 567 insertions(+), 37 deletions(-) create mode 100644 src/store/blob_store/file/integrity.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a6ddbb..8014d04 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,18 @@ fine-grained per-commit history is in `git log`. ## [Unreleased] +## [0.8.8] — 2026-10-26 + +### Fixed + +- File-backed blob reads now verify per-page checksums for the persisted part + of each blob before returning bytes. Damaged active blobs fail with + `NodeCorrupt` instead of silently returning altered values. The checksum + sidecar is synced before the manifest commits a new slot, and existing + version 6 stores migrate on their first writable open. A read-only open + requires migration to have completed first. This detects damage that occurs + after migration; existing damage cannot be reconstructed or authenticated. + ## [0.8.7] — 2026-10-16 ### Changed diff --git a/Cargo.lock b/Cargo.lock index 35212fa..4ac6165 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -316,7 +316,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "holt" -version = "0.8.7" +version = "0.8.8" dependencies = [ "crc32fast", "crossbeam-channel", diff --git a/Cargo.toml b/Cargo.toml index 6ee6f7a..fced75a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "holt" -version = "0.8.7" +version = "0.8.8" edition = "2021" rust-version = "1.82" autobenches = false diff --git a/README.md b/README.md index e871852..f3d05c5 100644 --- a/README.md +++ b/README.md @@ -175,6 +175,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/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/integrity.rs b/src/store/blob_store/file/integrity.rs new file mode 100644 index 0000000..0f2ea91 --- /dev/null +++ b/src/store/blob_store/file/integrity.rs @@ -0,0 +1,206 @@ +//! 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")) +} + +#[cfg(test)] +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. +#[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()); + } +} diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index a89947c..c1909e4 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; @@ -100,7 +102,6 @@ use crate::api::errors::{Error, Result}; use crate::api::stats::{StoreStats, VacuumStats}; use crate::layout::{BlobGuid, PAGE_SIZE}; use crate::store::blob_frame::clear_unpersisted_tail; -#[cfg(not(all(target_os = "linux", feature = "io-uring")))] use crate::store::blob_frame::persisted_extent; #[cfg(all(target_os = "linux", feature = "io-uring"))] @@ -112,6 +113,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"; @@ -169,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; @@ -309,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, @@ -316,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 @@ -353,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 @@ -532,6 +546,86 @@ fn open_read_accelerator_file( } } +/// 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 extent = persisted_extent(frame.as_slice()); + let record = integrity::encode_prefix(*guid, entry.slot, frame.as_slice(), extent); + 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`. /// @@ -626,8 +720,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); @@ -651,10 +746,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(()), @@ -709,8 +806,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); @@ -722,10 +820,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(()), @@ -760,6 +860,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)) } @@ -827,6 +928,37 @@ 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 frame = write.src.as_slice(); + let record = + integrity::encode_prefix(write.guid, write.slot, frame, persisted_extent(frame)); + self.integrity_file + .write_all_at(&record, integrity::record_offset(write.slot)?)?; + } + Ok(()) + } + fn reserve_write_entries( &self, guids: impl IntoIterator, @@ -1153,6 +1285,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) { @@ -1188,6 +1324,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); @@ -1208,6 +1348,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, @@ -1223,6 +1369,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) @@ -1501,8 +1648,11 @@ 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())?; clear_unpersisted_tail(dst.as_mut_slice()); Ok(()) } @@ -1539,20 +1689,20 @@ impl BlobStore for FileBlobStore { ring.pread_many_at(&mut batch) }; drop(batch); - if batch_result.is_ok() { - for (off, dst) in offsets.iter().zip(dsts.iter_mut()) { - if off.is_ok() { - clear_unpersisted_tail(dst.as_mut_slice()); - } - } - } - 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())?; + clear_unpersisted_tail(dst.as_mut_slice()); + Ok(()) + } Err(e) => Err(Error::BlobStoreIo(io::Error::other(format!( "batched uring read failed: {e}" )))), @@ -1617,7 +1767,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"))] { @@ -1630,7 +1781,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 { @@ -1764,7 +1916,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); } @@ -1781,7 +1936,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); } @@ -1801,7 +1959,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); } @@ -1815,7 +1976,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); } @@ -1871,7 +2035,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, @@ -1907,7 +2078,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 { @@ -2029,9 +2202,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)), }; @@ -2046,6 +2219,7 @@ impl Manifest { let reusable_slots = ReusableSlots::reconstruct(next_slot, &used_slots)?; Ok(Self { + version, entries, next_slot, reusable_slots, @@ -2057,7 +2231,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)?; @@ -2065,7 +2239,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)", )); @@ -2090,7 +2264,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) { @@ -3601,11 +3775,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); @@ -3905,7 +4085,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]; { @@ -3915,7 +4095,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 { @@ -3926,6 +4106,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 dd7732b01e15852312a334c40cc14a80f2027e1b Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 16:19:46 +1100 Subject: [PATCH 2/3] perf: bound reads and parallelize blob checksums Compact records only authenticate a prefix of a reused slot. Read that prefix, clear the rest, and verify independent blobs in parallel for large Linux batches. Keep the per-blob result order and reject damaged pages before callers receive data. Signed-off-by: wchwawa --- src/store/blob_store/file/integrity.rs | 84 +++++++--- src/store/blob_store/file/mod.rs | 217 +++++++++++++++++++------ src/store/blob_store/file/uring.rs | 83 +++++++--- 3 files changed, 296 insertions(+), 88 deletions(-) diff --git a/src/store/blob_store/file/integrity.rs b/src/store/blob_store/file/integrity.rs index 0f2ea91..7744cae 100644 --- a/src/store/blob_store/file/integrity.rs +++ b/src/store/blob_store/file/integrity.rs @@ -72,17 +72,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, @@ -90,31 +149,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; diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index c1909e4..4793d2e 100644 --- a/src/store/blob_store/file/mod.rs +++ b/src/store/blob_store/file/mod.rs @@ -860,7 +860,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)) } @@ -932,7 +932,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)?) @@ -944,7 +944,7 @@ impl FileBlobStore { Error::BlobStoreIo(error) } })?; - Ok(record) + integrity::ValidatedRecord::new(record, guid, slot) } fn write_integrity_records(&self, writes: &[PreparedBlobWrite<'_>]) -> Result<()> { @@ -1060,15 +1060,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(()) } @@ -1349,8 +1349,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); @@ -1650,9 +1650,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())?; + let covered = record.covered_bytes(); + self.pread_at(offset, covered, dst)?; + dst.as_mut_slice()[covered..].fill(0); + record.verify_full(dst.as_slice())?; clear_unpersisted_tail(dst.as_mut_slice()); Ok(()) } @@ -1666,15 +1668,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)); } } @@ -1689,26 +1697,62 @@ 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())?; - clear_unpersisted_tail(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())?; + clear_unpersisted_tail(dst.as_mut_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")))] @@ -1768,21 +1812,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 { @@ -3386,13 +3446,24 @@ mod tests { } assert!(batch.iter().all(|frame| frame.as_slice() == dst.as_slice())); - // Only the extent was written: the slot still holds the old - // frame's bytes past it. + // Only the extent was written. A range read beyond it must not + // expose bytes left by the old frame in the reused 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 == 0xab)); + 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] @@ -4142,6 +4213,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 a9b9797..a56da63 100644 --- a/src/store/blob_store/file/uring.rs +++ b/src/store/blob_store/file/uring.rs @@ -179,8 +179,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`. @@ -201,11 +206,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(()) @@ -266,13 +269,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; @@ -333,12 +336,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 @@ -451,20 +457,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() } @@ -525,7 +532,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; @@ -560,7 +567,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, @@ -748,8 +755,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)); @@ -784,10 +795,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); @@ -800,6 +811,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(); From 2ab0e4de104906f100d1136d261a6771e9e4eea9 Mon Sep 17 00:00:00 2001 From: wchwawa Date: Mon, 26 Oct 2026 16:24:01 +1100 Subject: [PATCH 3/3] chore: align release harness locks with 0.8.8 The soak and benchmark harnesses use locked path dependencies. Update only their Holt package versions so release validation can run against this candidate without changing third-party packages. Signed-off-by: wchwawa --- benches/Cargo.lock | 2 +- tools/soak/Cargo.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/benches/Cargo.lock b/benches/Cargo.lock index 51ffaf9..568c4ef 100644 --- a/benches/Cargo.lock +++ b/benches/Cargo.lock @@ -567,7 +567,7 @@ dependencies = [ [[package]] name = "holt" -version = "0.8.5" +version = "0.8.8" dependencies = [ "crc32fast", "crossbeam-channel", diff --git a/tools/soak/Cargo.lock b/tools/soak/Cargo.lock index d871e2a..b50dddc 100644 --- a/tools/soak/Cargo.lock +++ b/tools/soak/Cargo.lock @@ -60,7 +60,7 @@ checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" [[package]] name = "holt" -version = "0.8.7" +version = "0.8.8" dependencies = [ "crc32fast", "crossbeam-channel",