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/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..555b7d4 --- /dev/null +++ b/src/store/blob_store/file/integrity.rs @@ -0,0 +1,249 @@ +//! Per-page checksums for authoritative packed blob slots. +//! +//! A record is stored at the same slot number as its blob. The manifest may +//! reference a new slot only after both files have been synced. Reusable slots +//! are fenced by the manifest's existing reader drain before either record is +//! overwritten. + +use crate::api::errors::{Error, Result}; +use crate::layout::{BlobGuid, PAGE_SIZE}; + +const MAGIC: [u8; 8] = *b"HOLTCKS1"; +const PAGE_BYTES: usize = 4096; +const PAGE_COUNT: usize = PAGE_SIZE as usize / PAGE_BYTES; +const PAGE_CHECKSUMS_OFFSET: usize = 32; +const COVERED_PAGES_OFFSET: usize = 544; +const RECORD_CHECKSUM_OFFSET: usize = 572; +pub(super) const RECORD_BYTES: usize = 576; + +const _: () = assert!(PAGE_COUNT == 128); +const _: () = assert!(PAGE_CHECKSUMS_OFFSET + PAGE_COUNT * 4 <= RECORD_CHECKSUM_OFFSET); + +pub(super) fn record_offset(slot: u64) -> Result { + slot.checked_mul(RECORD_BYTES as u64) + .ok_or(Error::node_corrupt("FileBlobStore::Integrity::slot offset")) +} + +pub(super) fn encode(guid: BlobGuid, slot: u64, frame: &[u8]) -> [u8; RECORD_BYTES] { + encode_prefix(guid, slot, frame, frame.len()) +} + +pub(super) fn encode_prefix( + guid: BlobGuid, + slot: u64, + frame: &[u8], + covered_bytes: usize, +) -> [u8; RECORD_BYTES] { + debug_assert_eq!(frame.len(), PAGE_SIZE as usize); + debug_assert!( + covered_bytes > 0 && covered_bytes <= frame.len() && covered_bytes % PAGE_BYTES == 0 + ); + let mut record = [0u8; RECORD_BYTES]; + record[..8].copy_from_slice(&MAGIC); + record[8..24].copy_from_slice(&guid); + record[24..32].copy_from_slice(&slot.to_le_bytes()); + for (page, bytes) in frame[..covered_bytes].chunks_exact(PAGE_BYTES).enumerate() { + let start = PAGE_CHECKSUMS_OFFSET + page * 4; + record[start..start + 4].copy_from_slice(&crc32fast::hash(bytes).to_le_bytes()); + } + record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] + .copy_from_slice(&((covered_bytes / PAGE_BYTES) as u32).to_le_bytes()); + let checksum = crc32fast::hash(&record[..RECORD_CHECKSUM_OFFSET]); + record[RECORD_CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes()); + record +} + +fn validate_record(record: &[u8; RECORD_BYTES], guid: BlobGuid, slot: u64) -> Result<()> { + let stored = u32::from_le_bytes(record[RECORD_CHECKSUM_OFFSET..].try_into().unwrap()); + let covered_pages = u32::from_le_bytes( + record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] + .try_into() + .unwrap(), + ); + if record[..8] != MAGIC + || record[8..24] != guid + || record[24..32] != slot.to_le_bytes() + || !(1..=PAGE_COUNT as u32).contains(&covered_pages) + || stored != crc32fast::hash(&record[..RECORD_CHECKSUM_OFFSET]) + { + return Err(Error::node_corrupt("FileBlobStore::Integrity::record").with_blob_guid(guid)); + } + Ok(()) +} + +/// The record's extent is trusted only after its binding and checksum pass. +pub(super) struct ValidatedRecord { + bytes: [u8; RECORD_BYTES], + guid: BlobGuid, + covered_pages: usize, +} + +impl ValidatedRecord { + pub(super) fn new(bytes: [u8; RECORD_BYTES], guid: BlobGuid, slot: u64) -> Result { + validate_record(&bytes, guid, slot)?; + let covered_pages = u32::from_le_bytes( + bytes[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] + .try_into() + .unwrap(), + ) as usize; + Ok(Self { + bytes, + guid, + covered_pages, + }) + } + + pub(super) fn covered_bytes(&self) -> usize { + self.covered_pages * PAGE_BYTES + } + + pub(super) fn verify_full(&self, frame: &[u8]) -> Result<()> { + debug_assert_eq!(frame.len(), PAGE_SIZE as usize); + verify_pages(&self.bytes, self.guid, self.covered_pages, 0, frame) + } + + pub(super) fn verify_range(&self, byte_offset: u64, bytes: &[u8]) -> Result<()> { + let start = usize::try_from(byte_offset).map_err(|_| { + Error::node_corrupt("FileBlobStore::Integrity::range").with_blob_guid(self.guid) + })?; + if start % PAGE_BYTES != 0 + || bytes.len() % PAGE_BYTES != 0 + || start + .checked_add(bytes.len()) + .is_none_or(|end| end > PAGE_SIZE as usize) + { + return Err( + Error::node_corrupt("FileBlobStore::Integrity::range").with_blob_guid(self.guid) + ); + } + verify_pages( + &self.bytes, + self.guid, + self.covered_pages, + start / PAGE_BYTES, + bytes, + ) + } + + pub(super) fn relocate_verified(&self, new_slot: u64) -> [u8; RECORD_BYTES] { + relocate_verified(&self.bytes, new_slot) + } +} + +#[cfg(test)] +pub(super) fn verify_full( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + slot: u64, + frame: &[u8], +) -> Result<()> { + ValidatedRecord::new(*record, guid, slot)?.verify_full(frame) +} + +#[cfg(test)] +pub(super) fn verify_range( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + slot: u64, + byte_offset: u64, + bytes: &[u8], +) -> Result<()> { + ValidatedRecord::new(*record, guid, slot)?.verify_range(byte_offset, bytes) +} + +fn verify_pages( + record: &[u8; RECORD_BYTES], + guid: BlobGuid, + covered_pages: usize, + first_page: usize, + bytes: &[u8], +) -> Result<()> { + for (index, page) in bytes.chunks_exact(PAGE_BYTES).enumerate() { + if first_page + index >= covered_pages { + continue; + } + let start = PAGE_CHECKSUMS_OFFSET + (first_page + index) * 4; + let expected = u32::from_le_bytes(record[start..start + 4].try_into().unwrap()); + if crc32fast::hash(page) != expected { + return Err(Error::node_corrupt("FileBlobStore::Integrity::page").with_blob_guid(guid)); + } + } + Ok(()) +} + +/// Legacy compact frames may leave stale, non-authoritative bytes beyond +/// their persisted prefix. A verified record makes that boundary explicit. +#[cfg(test)] +pub(super) fn clear_uncovered_tail(record: &[u8; RECORD_BYTES], frame: &mut [u8]) { + let covered_pages = u32::from_le_bytes( + record[COVERED_PAGES_OFFSET..COVERED_PAGES_OFFSET + 4] + .try_into() + .unwrap(), + ) as usize; + frame[covered_pages * PAGE_BYTES..].fill(0); +} + +/// Retarget a record after the caller has verified its source frame. +pub(super) fn relocate_verified(record: &[u8; RECORD_BYTES], new_slot: u64) -> [u8; RECORD_BYTES] { + let mut relocated = *record; + relocated[24..32].copy_from_slice(&new_slot.to_le_bytes()); + let checksum = crc32fast::hash(&relocated[..RECORD_CHECKSUM_OFFSET]); + relocated[RECORD_CHECKSUM_OFFSET..].copy_from_slice(&checksum.to_le_bytes()); + relocated +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn page_fault_is_rejected_in_full_and_range_reads() { + let guid = [7; 16]; + let mut frame = vec![0u8; PAGE_SIZE as usize]; + frame[PAGE_BYTES * 19 + 31] = 0x42; + let record = encode(guid, 3, &frame); + verify_full(&record, guid, 3, &frame).unwrap(); + verify_range( + &record, + guid, + 3, + (19 * PAGE_BYTES) as u64, + &frame[19 * PAGE_BYTES..20 * PAGE_BYTES], + ) + .unwrap(); + frame[PAGE_BYTES * 19 + 31] ^= 1; + assert!(verify_full(&record, guid, 3, &frame).is_err()); + assert!(verify_range( + &record, + guid, + 3, + (19 * PAGE_BYTES) as u64, + &frame[19 * PAGE_BYTES..20 * PAGE_BYTES] + ) + .is_err()); + } + + #[test] + fn record_is_bound_to_guid_slot_and_checksum() { + let frame = vec![0u8; PAGE_SIZE as usize]; + let guid = [7; 16]; + let mut record = encode(guid, 3, &frame); + assert!(verify_full(&record, [8; 16], 3, &frame).is_err()); + assert!(verify_full(&record, guid, 4, &frame).is_err()); + record[100] ^= 1; + assert!(verify_full(&record, guid, 3, &frame).is_err()); + } + + #[test] + fn compact_prefix_ignores_stale_tail_but_clears_it_for_full_reads() { + let guid = [9; 16]; + let mut frame = vec![3u8; PAGE_SIZE as usize]; + let record = encode_prefix(guid, 2, &frame, PAGE_BYTES * 2); + frame[PAGE_BYTES * 3] ^= 1; + verify_full(&record, guid, 2, &frame).unwrap(); + clear_uncovered_tail(&record, &mut frame); + assert_eq!(frame[PAGE_BYTES * 2], 0); + frame[PAGE_BYTES] ^= 1; + assert!(verify_full(&record, guid, 2, &frame).is_err()); + } +} diff --git a/src/store/blob_store/file/mod.rs b/src/store/blob_store/file/mod.rs index ae65975..876025a 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(test)] 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 { + 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) + } + })?; + integrity::ValidatedRecord::new(record, guid, slot) + } + + 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, @@ -940,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(()) } @@ -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)?; + 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); 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,13 @@ 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)?; - self.pread_at(offset, dst)?; + let entry = self.entry_of(guid)?; + let offset = entry.slot * u64::from(PAGE_SIZE); + let record = self.read_integrity_record(guid, entry.slot)?; + 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(()) } @@ -1522,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)); } } @@ -1545,19 +1701,61 @@ impl BlobStore for FileBlobStore { ring.pread_many_at(&mut batch) }; drop(batch); - - offsets - .into_iter() - .map(|off| match off { - Err(e) => Err(e), - Ok(_) => match &batch_result { - Ok(()) => 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")))] @@ -1616,20 +1814,38 @@ 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 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)?; + } } - Ok(()) + 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 { @@ -1763,7 +1979,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 +1999,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 +2022,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 +2039,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); } @@ -1870,7 +2098,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, @@ -1906,7 +2141,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 +2265,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 +2282,7 @@ impl Manifest { let reusable_slots = ReusableSlots::reconstruct(next_slot, &used_slots)?; Ok(Self { + version, entries, next_slot, reusable_slots, @@ -2056,7 +2294,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 +2302,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 +2327,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) { @@ -3166,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(); @@ -3546,11 +3856,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); @@ -3850,7 +4166,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 +4176,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 +4187,181 @@ 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 { .. }) + )); + } + + #[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(); + 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(); 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();