//! A page cache for caching _logical_ pages of [Blob] data in memory. The cache is unaware of the //! physical page format used by the blob, which is left to the blob implementation. use super::get_page_from_blob; use crate::{Blob, BufferPool, BufferPooler, Error, IoBuf, IoBufMut}; use ahash::AHashMap; use commonware_utils::{cache::Clock, sync::RwLock}; use futures::{future::Shared, FutureExt}; use std::{ collections::hash_map::Entry, future::Future, num::{NonZeroU16, NonZeroUsize}, pin::Pin, sync::{ atomic::{AtomicU64, Ordering}, Arc, }, }; use tracing::{error, trace}; /// Shared future for one logical page fetch. The output uses `Arc` because `Shared` /// requires cloneable results. The `IoBuf` contains only the logical, validated page bytes. type PageFetchFuture = Shared>> + Send>>>; /// Shared handle to one in-flight fetch generation. The cache keeps one copy in `page_fetches`, /// and each waiter clones the `Arc` while it is still interested in the result. type PageFetch = Arc; /// One in-flight fetch generation for a single `(blob_id, page_num)`. /// /// `fetch` is shared by every waiter that joined this generation. `waiters` counts the still /// armed waiters whose drop path may need to remove this entry if they become the last /// unresolved waiter. If `page_fetches[key]` is later replaced by a newer generation, stale /// waiters from the old generation must ignore it and rely on `Arc::ptr_eq` against their saved /// `fetch`. struct PageFetchEntry { /// Shared page fetch future that reads and validates the logical page exactly once. fetch: PageFetch, /// Count of waiters that still need cancellation cleanup for this fetch generation. waiters: usize, } /// Removes a stale in-flight page fetch when the last unresolved waiter is dropped. struct PageFetchGuard { cache: Arc>, key: (u64, u64), fetch: PageFetch, armed: bool, } impl PageFetchGuard { const fn new(cache: Arc>, key: (u64, u64), fetch: PageFetch) -> Self { Self { cache, key, fetch, armed: true, } } const fn disarm(&mut self) { self.armed = false; } } impl Drop for PageFetchGuard { fn drop(&mut self) { if !self.armed { return; } // A resolved fetch removes `page_fetches[key]` before waiters resume and disarm their // guards. If that fetch failed, the page remains uncached, so a new reader can install a // new fetch for the same key before an old waiter is cancelled. Ignore drops from stale // waiters so they cannot decrement or remove a newer generation. A surviving waiter keeps // the current generation installed, which lets the shared future finish and cache the page // on success. let mut cache = self.cache.write(); let Entry::Occupied(mut current) = cache.page_fetches.entry(self.key) else { return; }; if !Arc::ptr_eq(¤t.get().fetch, &self.fetch) { return; } if current.get().waiters == 1 { current.remove(); } else { current.get_mut().waiters -= 1; } } } /// A [Cache] caches pages of [Blob] data in memory after verifying the integrity of each. /// /// A single page cache can be used to cache data from multiple blobs by assigning a unique id to /// each. /// /// Eviction is delegated to a [Clock], which uses the Clock (second-chance) replacement /// policy, a lightweight approximation of LRU. All page buffers are pre-allocated from `pool` at /// construction (via [Clock::prefill]) and reused in place, so caching never allocates after /// construction. /// /// Reads first resolve pages through `hints`, a fixed-size direct-mapped array from /// [Self::hint_index] to the [Clock] slot the page was last cached in: a lookup is one array /// load instead of a hash-table probe chain, which the out-of-order core cannot overlap across /// items. Hints are best-effort, never truth: [Clock::get_at] only resolves a slot that still /// holds the page's key live, so entries staled by eviction, invalidation, or hint collisions /// read as misses and fall back to the [Clock]'s own lookup. Hints need no maintenance on /// eviction or invalidation, and their memory is fixed at construction, so no blob offset can /// grow them. struct Cache { /// Maps each (blob id, page number) to its logical page buffer. cache: Clock<(u64, u64), IoBufMut>, /// Direct-mapped [Clock] slot hints, indexed by [Self::hint_index]. Initialized /// out-of-range so untouched entries read as misses. The length is a power of two so /// [Self::hint_index] can wrap with a mask instead of a division, and at least twice the /// cache capacity: a full cache has one live page per `capacity`, so sizing at capacity /// makes hint collisions (and their slower fallback lookups) common. hints: Vec, /// Size of each page in bytes. page_size: usize, /// Pool the page buffers were allocated from. pool: BufferPool, /// A map of currently executing page fetches to ensure only one task at a time is trying to /// fetch a specific page. page_fetches: AHashMap<(u64, u64), PageFetchEntry>, } /// A reference to a page cache that can be shared across threads via cloning, along with the page /// size that will be used with it. Provides the API for interacting with the page cache in a /// thread-safe manner. #[derive(Clone)] pub struct CacheRef { /// The size of each page in the underlying blobs managed by this page cache. /// /// # Warning /// /// You cannot change the page size once data has been written without invalidating it. (Reads /// on blobs that were written with a different page size will fail their integrity check.) page_size: u64, /// The next id to assign to a blob that will be managed by this cache. next_id: Arc, /// Shareable reference to the page cache. cache: Arc>, /// Pool used for page-cache and associated buffer allocations. pool: BufferPool, } impl CacheRef { /// Create a shared page-cache handle backed by `pool`. /// /// The cache stores at most `capacity` pages, each exactly `page_size` bytes. /// Initialization eagerly allocates and zeroes all cache slots from `pool`. pub fn new(pool: BufferPool, page_size: NonZeroU16, capacity: NonZeroUsize) -> Self { let page_size_u64 = page_size.get() as u64; Self { page_size: page_size_u64, next_id: Arc::new(AtomicU64::new(0)), cache: Arc::new(RwLock::new(Cache::new(pool.clone(), page_size, capacity))), pool, } } /// Create a shared page-cache handle, extracting the storage [BufferPool] from a /// [BufferPooler]. pub fn from_pooler( pooler: &impl BufferPooler, page_size: NonZeroU16, capacity: NonZeroUsize, ) -> Self { Self::new(pooler.storage_buffer_pool().clone(), page_size, capacity) } /// The page size used by this page cache. #[inline] pub const fn page_size(&self) -> u64 { self.page_size } /// Returns the storage buffer pool associated with this cache. #[inline] pub const fn pool(&self) -> &BufferPool { &self.pool } /// Returns a unique id for the next blob that will use this page cache. pub fn next_id(&self) -> u64 { self.next_id.fetch_add(1, Ordering::Relaxed) } /// Convert a logical offset into the number of the page it belongs to and the offset within /// that page. pub const fn offset_to_page(&self, offset: u64) -> (u64, u64) { Cache::offset_to_page(self.page_size, offset) } /// Try to read the specified bytes from the page cache only. Returns the number of bytes /// successfully read from cache and copied to `buf` before a page fault, if any. pub(super) fn read_cached( &self, blob_id: u64, mut buf: &mut [u8], mut logical_offset: u64, ) -> usize { let original_len = buf.len(); let page_cache = self.cache.read(); while !buf.is_empty() { let count = page_cache.read_at(blob_id, buf, logical_offset); if count == 0 { // Cache miss - return how many bytes we successfully read break; } logical_offset += count as u64; buf = &mut buf[count..]; } original_len - buf.len() } /// Read multiple disjoint byte ranges from the page cache in a single lock acquisition. /// /// Each element of `ranges` is `(dest_slice, logical_offset)`. Fully-cached ranges have /// their data written to the destination slice and are removed from `ranges`. Entries left /// in `ranges` correspond to cache misses that the caller must read from the underlying /// blob. pub(super) fn read_cached_many(&self, blob_id: u64, ranges: &mut Vec<(&mut [u8], u64)>) { let page_cache = self.cache.read(); let page_size = page_cache.page_size; // Resolve every range's first page before copying any data. The lookups are // independent, so batching them lets the core overlap their memory latency instead of // stalling each lookup behind the previous range's copy. let mut srcs: Vec> = Vec::with_capacity(ranges.len()); for (buf, offset) in ranges.iter() { let (page_num, offset_in_page) = Cache::offset_to_page(page_size as u64, *offset); let offset_in_page = offset_in_page as usize; let seg = std::cmp::min(buf.len(), page_size - offset_in_page); srcs.push( page_cache .get_page(blob_id, page_num) .map(|page| &page.as_ref()[offset_in_page..offset_in_page + seg]), ); } // Copy resolved pages, dropping fully-cached ranges and keeping misses. A range whose // first page missed is kept untouched, and one that continues past its first page reads // the rest page by page, staying a miss if any later page faults. let mut next = 0; ranges.retain_mut(|(buf, offset)| { let src = srcs[next]; next += 1; if buf.is_empty() { return false; } let Some(src) = src else { return true; }; buf[..src.len()].copy_from_slice(src); let mut done = src.len(); while done < buf.len() { let count = page_cache.read_at(blob_id, &mut buf[done..], *offset + done as u64); if count == 0 { return true; } done += count; } false }); } /// Read the specified bytes, preferentially from the page cache. Bytes not found in the cache /// will be read from the provided `blob` and cached for future reads. pub(super) async fn read( &self, blob: &B, blob_id: u64, mut buf: &mut [u8], mut offset: u64, ) -> Result<(), Error> { // Read up to a page worth of data at a time from either the page cache or the `blob`, // until the requested data is fully read. while !buf.is_empty() { // Read lock the page cache and see if we can get (some of) the data from it. { let page_cache = self.cache.read(); let count = page_cache.read_at(blob_id, buf, offset); if count != 0 { offset += count as u64; buf = &mut buf[count..]; continue; } } // Handle page fault. let count = self .read_after_page_fault(blob, blob_id, buf, offset) .await?; offset += count as u64; buf = &mut buf[count..]; } Ok(()) } /// Fetch the requested page after encountering a page fault, which may involve retrieving it /// from `blob` & caching the result in the page cache. Returns the number of bytes read, which /// should always be non-zero. pub(super) async fn read_after_page_fault( &self, blob: &B, blob_id: u64, buf: &mut [u8], offset: u64, ) -> Result { assert!(!buf.is_empty()); let (page_num, offset_in_page) = Cache::offset_to_page(self.page_size, offset); let offset_in_page = offset_in_page as usize; trace!(page_num, blob_id, "page fault"); // Create or clone a future that retrieves the desired page from the underlying blob. This // requires a write lock on the page cache since we may need to modify `page_fetches` if // this task is the first fetcher. let (fetch_future, mut fetch_guard) = { let mut cache = self.cache.write(); // There's a (small) chance the page was fetched & buffered by another task before we // were able to acquire the write lock, so check the cache before doing anything else. let count = cache.read_at(blob_id, buf, offset); if count != 0 { return Ok(count); } let key = (blob_id, page_num); match cache.page_fetches.entry(key) { Entry::Occupied(o) => { // Another thread is already fetching this page, so clone its existing future. let entry = o.into_mut(); entry.waiters += 1; let fetch_future = entry.fetch.as_ref().clone(); let fetch = Arc::clone(&entry.fetch); ( fetch_future, PageFetchGuard::new(Arc::clone(&self.cache), key, fetch), ) } Entry::Vacant(v) => { // Nobody is currently fetching this page, so create a future that will do the // work. get_page_from_blob handles CRC validation and returns only logical bytes. let blob = blob.clone(); let cache = Arc::clone(&self.cache); let page_size = self.page_size; let future = async move { let result = fetch_cacheable_page(&blob, page_num, page_size).await; if let Err(err) = &result { error!(page_num, ?err, "Page fetch failed"); } // This shared future still owns `page_fetches[key]`. As long as at least // one waiter remains armed, that entry pins this generation in place, so a // replacement fetch for the same page cannot be inserted before we cache // the successful result below. Only when every waiter cancels can the last // guard remove the entry and let a later reader start a new generation. let mut cache = cache.write(); if let Ok(page) = &result { cache.cache(blob_id, page.as_ref(), page_num); } let _ = cache.page_fetches.remove(&key); result }; // Make the future shareable and insert it into the map. let fetch_future = future.boxed().shared(); let fetch = Arc::new(fetch_future.clone()); v.insert(PageFetchEntry { fetch: Arc::clone(&fetch), waiters: 1, }); ( fetch_future, PageFetchGuard::new(Arc::clone(&self.cache), key, fetch), ) } } }; // Await the shared fetch. The future itself logs failures, caches the resolved page, and // removes the in-flight marker before it returns, so waiters only need cancellation // cleanup while the fetch is still unresolved. let fetch_result = fetch_future.await; fetch_guard.disarm(); let page_buf = match fetch_result { Ok(page_buf) => page_buf, Err(_) => return Err(Error::ReadFailed), }; // Copy the requested portion of the page into the buffer. let bytes_to_copy = std::cmp::min(buf.len(), page_buf.len() - offset_in_page); buf[..bytes_to_copy] .copy_from_slice(&page_buf.as_ref()[offset_in_page..offset_in_page + bytes_to_copy]); Ok(bytes_to_copy) } /// Cache the provided pages of data in the page cache, returning the remaining bytes that /// didn't fill a whole page. `offset` must be page aligned. /// /// # Panics /// /// - Panics if `offset` is not page aligned. /// - If the buffer is not the size of a page. pub fn cache(&self, blob_id: u64, mut buf: &[u8], offset: u64) -> usize { let (mut page_num, offset_in_page) = self.offset_to_page(offset); assert_eq!(offset_in_page, 0); { // Write lock the page cache. let page_size = self.page_size as usize; let mut page_cache = self.cache.write(); while buf.len() >= page_size { page_cache.cache(blob_id, &buf[..page_size], page_num); buf = &buf[page_size..]; page_num = match page_num.checked_add(1) { Some(next) => next, None => break, }; } } buf.len() } /// Drop all cached pages while retaining the backing page buffers for reuse. /// /// Call only when no reads are in flight for this cache. #[cfg(any(test, feature = "test-utils"))] pub fn clear(&self) { self.cache.write().clear(); } /// Drop any cached pages for `blob_id` at `page_num >= start_page`. Used after a blob is /// truncated so subsequent reads can't observe pre-truncation bytes in a page that the tip /// buffer (or future writes) now owns. pub(super) fn invalidate_from(&self, blob_id: u64, start_page: u64) { self.cache.write().invalidate_from(blob_id, start_page); } } impl Cache { /// Return a new empty page cache with a max cache capacity of `capacity` pages, each of size /// `page_size` bytes. pub fn new(pool: BufferPool, page_size: NonZeroU16, capacity: NonZeroUsize) -> Self { let page_size = page_size.get() as usize; let mut cache = Clock::new(capacity); cache.prefill(|| pool.alloc_zeroed(page_size)); let hints = capacity.get().saturating_mul(2).next_power_of_two(); Self { cache, hints: vec![usize::MAX; hints], page_size, pool, page_fetches: AHashMap::new(), } } /// Convert an offset into the number of the page it belongs to and the offset within that page. const fn offset_to_page(page_size: u64, offset: u64) -> (u64, u64) { (offset / page_size, offset % page_size) } /// Attempt to fetch blob data starting at `offset` from the page cache. Returns the number of /// bytes read, which could be 0 if the first page in the requested range isn't buffered, and is /// never more than `self.page_size` or the length of `buf`. The returned bytes won't cross a /// page boundary, so multiple reads may be required even if all data in the desired range is /// buffered. fn read_at(&self, blob_id: u64, buf: &mut [u8], logical_offset: u64) -> usize { let (page_num, offset_in_page) = Self::offset_to_page(self.page_size as u64, logical_offset); let Some(page) = self.get_page(blob_id, page_num) else { return 0; }; let page = page.as_ref(); let offset_in_page = offset_in_page as usize; let bytes_to_copy = std::cmp::min(buf.len(), self.page_size - offset_in_page); buf[..bytes_to_copy].copy_from_slice(&page[offset_in_page..offset_in_page + bytes_to_copy]); bytes_to_copy } /// Put the given `page` into the page cache and record its slot hint. fn cache(&mut self, blob_id: u64, page: &[u8], page_num: u64) { assert_eq!(page.len(), self.page_size); let pool = &self.pool; let page_size = self.page_size; let (slot, buf) = self .cache .get_or_insert_mut((blob_id, page_num), || pool.alloc_zeroed(page_size)); buf.as_mut().copy_from_slice(page); let hint = self.hint_index(blob_id, page_num); self.hints[hint] = slot; } /// The hint slot for `(blob_id, page_num)`: the page number offset by a per-blob salt, /// wrapped to the array. /// /// Adding (rather than hashing in) the page number keeps consecutive pages in consecutive /// hint entries, so the sorted batches issued by [CacheRef::read_cached_many] walk the /// array sequentially instead of taking a cache miss per lookup. The salt spreads blobs' /// ranges apart; two blobs whose ranges still overlap only evict each other's hints, which /// [Self::get_page] repairs through the fallback lookup. #[inline] const fn hint_index(&self, blob_id: u64, page_num: u64) -> usize { let salted = page_num.wrapping_add(blob_id.wrapping_mul(commonware_utils::GOLDEN_RATIO)); (salted & (self.hints.len() as u64 - 1)) as usize } /// Look up a page, preferring its direct-mapped slot hint over the [Clock]'s own lookup. #[inline] fn get_page(&self, blob_id: u64, page_num: u64) -> Option<&IoBufMut> { let key = (blob_id, page_num); let slot = self.hints[self.hint_index(blob_id, page_num)]; if let Some(page) = self.cache.get_at(slot, &key) { return Some(page); } self.cache.get(&key) } /// Drop any cached pages for `blob_id` at `page_num >= start_page`. fn invalidate_from(&mut self, blob_id: u64, start_page: u64) { self.cache .retain(|&(bid, page_num), _| bid != blob_id || page_num < start_page); } /// Drop all cached pages while retaining backing page buffers for reuse. #[cfg(any(test, feature = "test-utils"))] fn clear(&mut self) { self.cache.retain(|_, _| false); self.page_fetches.clear(); } } /// Fetch one logical page for insertion into the page cache, rejecting partial pages because cache /// entries must always contain a full logical page. async fn fetch_cacheable_page( blob: &impl Blob, page_num: u64, page_size: u64, ) -> Result> { let page = get_page_from_blob(blob, page_num, page_size) .await .map_err(Arc::new)?; // We should never be fetching partial pages through the page cache. This can happen if a // non-last page is corrupted and falls back to a partial CRC. let len = page.len(); if len != page_size as usize { error!( page_num, expected = page_size, actual = len, "attempted to fetch partial page from blob" ); return Err(Arc::new(Error::InvalidChecksum)); } Ok(page) } #[cfg(test)] mod tests { use super::{super::Checksum, *}; use crate::{ buffer::paged::CHECKSUM_SIZE, deterministic, telemetry::metrics::Registry, Buf, BufferPool, BufferPoolConfig, Clock as _, Handle, IoBufs, IoBufsMut, Runner as _, Spawner as _, Storage as _, Supervisor as _, }; use commonware_cryptography::Crc32; use commonware_macros::test_traced; use commonware_utils::{channel::oneshot, sync::Mutex, NZUsize, NZU16}; use futures::future::pending; use rstest::rstest; use std::{ num::NonZeroU16, sync::{ atomic::{AtomicUsize, Ordering}, Arc, }, time::Duration, }; fn test_pool() -> BufferPool { let mut registry = Registry::default(); BufferPool::new(BufferPoolConfig::for_storage(), &mut registry) } // Logical page size (what CacheRef uses and what gets cached). const PAGE_SIZE: NonZeroU16 = NZU16!(1024); const PAGE_SIZE_U64: u64 = PAGE_SIZE.get() as u64; fn expected_cached_bytes(logical_offset: u64, len: usize) -> Vec { (0..len) .map(|i| { let page = (logical_offset + i as u64) / PAGE_SIZE_U64; page as u8 + 1 }) .collect() } /// A blob that signals once a read starts and then never returns. #[derive(Clone)] struct BlockingBlob { started: Arc>>>, } impl Blob for BlockingBlob { async fn read_at(&self, offset: u64, len: usize) -> Result { self.read_at_buf(offset, len, IoBufsMut::default()).await } async fn read_at_buf( &self, _offset: u64, _len: usize, _bufs: impl Into + Send, ) -> Result { let sender = self .started .lock() .take() .expect("blocking blob read started more than once"); let _ = sender.send(()); pending::<()>().await; unreachable!() } async fn write_at( &self, _offset: u64, _bufs: impl Into + Send, ) -> Result<(), Error> { Ok(()) } async fn write_at_sync( &self, offset: u64, bufs: impl Into + Send, ) -> Result<(), Error> { let bufs = bufs.into(); if !bufs.has_remaining() { return Ok(()); } self.write_at(offset, bufs).await?; self.sync().await } async fn resize(&self, _len: u64) -> Result<(), Error> { Ok(()) } async fn sync(&self) -> Result<(), Error> { Ok(()) } async fn start_sync(&self) -> Handle<()> { Handle::ready(self.sync().await) } } #[derive(Clone)] enum ControlledBlobResult { Success(Arc>), Error, } /// A blob that blocks its first physical page read until released and counts total reads. #[derive(Clone)] struct ControlledBlob { started: Arc>>>, release: Arc>>>, reads: Arc, result: ControlledBlobResult, } impl Blob for ControlledBlob { async fn read_at(&self, offset: u64, len: usize) -> Result { self.read_at_buf(offset, len, IoBufsMut::default()).await } async fn read_at_buf( &self, _offset: u64, _len: usize, _bufs: impl Into + Send, ) -> Result { if self.reads.fetch_add(1, Ordering::Relaxed) == 0 { let sender = self .started .lock() .take() .expect("controlled blob start signal consumed more than once"); let _ = sender.send(()); let release = self .release .lock() .take() .expect("controlled blob release receiver consumed more than once"); release.await.expect("release signal dropped"); } match &self.result { ControlledBlobResult::Success(page) => Ok(IoBufsMut::from(page.as_ref().clone())), ControlledBlobResult::Error => Err(Error::ReadFailed), } } async fn write_at( &self, _offset: u64, _bufs: impl Into + Send, ) -> Result<(), Error> { Ok(()) } async fn write_at_sync( &self, offset: u64, bufs: impl Into + Send, ) -> Result<(), Error> { let bufs = bufs.into(); if !bufs.has_remaining() { return Ok(()); } self.write_at(offset, bufs).await?; self.sync().await } async fn resize(&self, _len: u64) -> Result<(), Error> { Ok(()) } async fn sync(&self) -> Result<(), Error> { Ok(()) } async fn start_sync(&self) -> Handle<()> { Handle::ready(self.sync().await) } } #[test_traced] fn test_cache_basic() { let pool = test_pool(); let mut cache: Cache = Cache::new(pool, PAGE_SIZE, NZUsize!(10)); // Cache stores logical-sized pages. let mut buf = vec![0; PAGE_SIZE.get() as usize]; let bytes_read = cache.read_at(0, &mut buf, 0); assert_eq!(bytes_read, 0); cache.cache(0, &[1; PAGE_SIZE.get() as usize], 0); let bytes_read = cache.read_at(0, &mut buf, 0); assert_eq!(bytes_read, PAGE_SIZE.get() as usize); assert_eq!(buf, [1; PAGE_SIZE.get() as usize]); // Test replacement -- re-caching the same page overwrites it in place. cache.cache(0, &[2; PAGE_SIZE.get() as usize], 0); let bytes_read = cache.read_at(0, &mut buf, 0); assert_eq!(bytes_read, PAGE_SIZE.get() as usize); assert_eq!(buf, [2; PAGE_SIZE.get() as usize]); // Test exceeding the cache capacity. for i in 0u64..11 { cache.cache(0, &[i as u8; PAGE_SIZE.get() as usize], i); } // Page 0 should have been evicted. let bytes_read = cache.read_at(0, &mut buf, 0); assert_eq!(bytes_read, 0); // Page 1-10 should be in the cache. for i in 1u64..11 { let bytes_read = cache.read_at(0, &mut buf, i * PAGE_SIZE_U64); assert_eq!(bytes_read, PAGE_SIZE.get() as usize); assert_eq!(buf, [i as u8; PAGE_SIZE.get() as usize]); } // Test reading from an unaligned offset by adding 2 to an aligned offset. The read // should be 2 bytes short of a full logical page. let mut buf = vec![0; PAGE_SIZE.get() as usize]; let bytes_read = cache.read_at(0, &mut buf, PAGE_SIZE_U64 + 2); assert_eq!(bytes_read, PAGE_SIZE.get() as usize - 2); assert_eq!( &buf[..PAGE_SIZE.get() as usize - 2], [1; PAGE_SIZE.get() as usize - 2] ); } #[test_traced] fn test_invalidate_from_does_not_orphan_re_cached_page() { // Invalidating pages, re-caching one, then forcing an eviction must keep every live page // readable. Freed slots are reused cleanly, so an invalidated-then-re-cached page is never // orphaned by a later eviction. let mut registry = Registry::default(); let pool = BufferPool::new(BufferPoolConfig::for_storage(), &mut registry); let mut cache: Cache = Cache::new(pool, PAGE_SIZE, NZUsize!(2)); let blob_id = 0u64; let page_size = PAGE_SIZE.get() as usize; // Fill both slots, then invalidate them so both slots are freed for reuse. cache.cache(blob_id, &vec![0xAA; page_size], 0); cache.cache(blob_id, &vec![0xBB; page_size], 1); cache.invalidate_from(blob_id, 0); // Re-cache page 1 into a reused slot. cache.cache(blob_id, &vec![0xCC; page_size], 1); let mut buf = vec![0u8; page_size]; assert_eq!( cache.read_at(blob_id, &mut buf, PAGE_SIZE_U64), page_size, "page 1 should be readable after re-cache" ); assert_eq!(buf, vec![0xCC; page_size]); // Cache a new page, which reuses the other freed slot rather than evicting live page 1. cache.cache(blob_id, &vec![0xDD; page_size], 2); // Slot 0 must still be reachable via its live index entry. let mut buf = vec![0u8; page_size]; assert_eq!( cache.read_at(blob_id, &mut buf, PAGE_SIZE_U64), page_size, "live page 1 was orphaned by stale-slot eviction" ); assert_eq!(buf, vec![0xCC; page_size]); // And the newly cached page 2 is also reachable. let mut buf = vec![0u8; page_size]; assert_eq!( cache.read_at(blob_id, &mut buf, PAGE_SIZE_U64 * 2), page_size ); assert_eq!(buf, vec![0xDD; page_size]); } #[test_traced] fn test_cache_read_with_blob() { // Initialize the deterministic context let executor = deterministic::Runner::default(); // Start the test within the executor executor.start(|context| async move { // Physical page size = logical + CRC record. let physical_page_size = PAGE_SIZE_U64 + CHECKSUM_SIZE; // Populate a blob with 11 consecutive pages of CRC-protected data. let (blob, size) = context .open("test", "blob".as_bytes()) .await .expect("Failed to open blob"); assert_eq!(size, 0); for i in 0..11 { // Write logical data followed by Checksum. let logical_data = vec![i as u8; PAGE_SIZE.get() as usize]; let crc = Crc32::checksum(&logical_data); let record = Checksum::new(PAGE_SIZE.get(), crc); let mut page_data = logical_data; page_data.extend_from_slice(&record.to_bytes()); blob.write_at(i * physical_page_size, page_data) .await .unwrap(); } // Fill the page cache with the blob's data via CacheRef::read. let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(10)); assert_eq!(cache_ref.next_id(), 0); assert_eq!(cache_ref.next_id(), 1); for i in 0..11 { // Read expects logical bytes only (CRCs are stripped). let mut buf = vec![0; PAGE_SIZE.get() as usize]; cache_ref .read(&blob, 0, &mut buf, i * PAGE_SIZE_U64) .await .unwrap(); assert_eq!(buf, [i as u8; PAGE_SIZE.get() as usize]); } // Repeat the read to exercise reading from the page cache. Must start at 1 because // page 0 should be evicted. for i in 1..11 { let mut buf = vec![0; PAGE_SIZE.get() as usize]; cache_ref .read(&blob, 0, &mut buf, i * PAGE_SIZE_U64) .await .unwrap(); assert_eq!(buf, [i as u8; PAGE_SIZE.get() as usize]); } // Cleanup. blob.sync().await.unwrap(); }); } #[test_traced] fn test_cache_clear_forces_blob_read() { #[derive(Clone)] struct CountingBlob { reads: Arc, page: Arc>, } impl Blob for CountingBlob { async fn read_at(&self, offset: u64, len: usize) -> Result { self.read_at_buf(offset, len, IoBufsMut::default()).await } async fn read_at_buf( &self, _offset: u64, _len: usize, _bufs: impl Into + Send, ) -> Result { self.reads.fetch_add(1, Ordering::Relaxed); Ok(IoBufsMut::from(self.page.as_ref().clone())) } async fn write_at( &self, _offset: u64, _bufs: impl Into + Send, ) -> Result<(), Error> { Ok(()) } async fn write_at_sync( &self, offset: u64, bufs: impl Into + Send, ) -> Result<(), Error> { self.write_at(offset, bufs).await } async fn resize(&self, _len: u64) -> Result<(), Error> { Ok(()) } async fn sync(&self) -> Result<(), Error> { Ok(()) } async fn start_sync(&self) -> Handle<()> { Handle::ready(self.sync().await) } } let executor = deterministic::Runner::default(); executor.start(|context| async move { let page = vec![7u8; PAGE_SIZE.get() as usize]; let crc = Crc32::checksum(&page); let record = Checksum::new(PAGE_SIZE.get(), crc); let mut physical_page = page.clone(); physical_page.extend_from_slice(&record.to_bytes()); let blob = CountingBlob { reads: Arc::new(AtomicUsize::new(0)), page: Arc::new(physical_page), }; let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(2)); let mut buf = vec![0u8; page.len()]; cache_ref.read(&blob, 0, &mut buf, 0).await.unwrap(); assert_eq!(buf, page); assert_eq!(blob.reads.load(Ordering::Relaxed), 1); let mut buf = vec![0u8; page.len()]; cache_ref.read(&blob, 0, &mut buf, 0).await.unwrap(); assert_eq!(buf, page); assert_eq!(blob.reads.load(Ordering::Relaxed), 1); cache_ref.clear(); let mut buf = vec![0u8; page.len()]; cache_ref.read(&blob, 0, &mut buf, 0).await.unwrap(); assert_eq!(buf, page); assert_eq!(blob.reads.load(Ordering::Relaxed), 2); }); } #[test_traced] fn test_cache_max_page() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(2)); // Use the largest page-aligned offset representable for the configured PAGE_SIZE. let aligned_max_offset = u64::MAX - (u64::MAX % PAGE_SIZE_U64); // CacheRef::cache expects only logical bytes (no CRC). let logical_data = vec![42u8; PAGE_SIZE.get() as usize]; // Caching exactly one page at the maximum offset should succeed. let remaining = cache_ref.cache(0, logical_data.as_slice(), aligned_max_offset); assert_eq!(remaining, 0); // Reading from the cache should return the logical bytes. let mut buf = vec![0u8; PAGE_SIZE.get() as usize]; let page_cache = cache_ref.cache.read(); let bytes_read = page_cache.read_at(0, &mut buf, aligned_max_offset); assert_eq!(bytes_read, PAGE_SIZE.get() as usize); assert!(buf.iter().all(|b| *b == 42)); }); } #[test_traced] fn test_cache_at_high_offset() { let executor = deterministic::Runner::default(); executor.start(|context| async move { // Use the minimum page size (CHECKSUM_SIZE + 1 = 13) with high offset. const MIN_PAGE_SIZE: u64 = CHECKSUM_SIZE + 1; let cache_ref = CacheRef::from_pooler(&context, NZU16!(MIN_PAGE_SIZE as u16), NZUsize!(2)); // Create two pages worth of logical data (no CRCs - CacheRef::cache expects logical // only). let data = vec![1u8; MIN_PAGE_SIZE as usize * 2]; // Cache pages at a high (but not max) aligned offset so we can verify both pages. // Use an offset that's a few pages below max to avoid overflow when verifying. let aligned_max_offset = u64::MAX - (u64::MAX % MIN_PAGE_SIZE); let high_offset = aligned_max_offset - (MIN_PAGE_SIZE * 2); let remaining = cache_ref.cache(0, &data, high_offset); // Both pages should be cached. assert_eq!(remaining, 0); // Verify the first page was cached correctly. let mut buf = vec![0u8; MIN_PAGE_SIZE as usize]; let page_cache = cache_ref.cache.read(); assert_eq!( page_cache.read_at(0, &mut buf, high_offset), MIN_PAGE_SIZE as usize ); assert!(buf.iter().all(|b| *b == 1)); // Verify the second page was cached correctly. assert_eq!( page_cache.read_at(0, &mut buf, high_offset + MIN_PAGE_SIZE), MIN_PAGE_SIZE as usize ); assert!(buf.iter().all(|b| *b == 1)); }); } #[test_traced] fn test_page_fetches_entry_removed_when_first_fetcher_cancelled() { let executor = deterministic::Runner::default(); executor.start(|context| async move { // Set up a small cache and a blob whose read never completes once started. let blob_id = 0; let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(10)); let (started_tx, started_rx) = oneshot::channel(); let blob = BlockingBlob { started: Arc::new(Mutex::new(Some(started_tx))), }; let mut read_buf = vec![0u8; PAGE_SIZE.get() as usize]; // Spawn the first fetcher. It will insert into `page_fetches` and then block forever. let cache_ref_for_task = cache_ref.clone(); let blob_for_task = blob.clone(); let handle = context.spawn(move |_| async move { let _ = cache_ref_for_task .read(&blob_for_task, blob_id, &mut read_buf, 0) .await; }); // Wait until the underlying read has started, ensuring the in-flight marker exists. started_rx.await.expect("blocking read never started"); { let page_cache = cache_ref.cache.read(); assert!(page_cache.page_fetches.contains_key(&(blob_id, 0))); } // Cancel the first fetcher before it reaches explicit cleanup. handle.abort(); assert!(matches!(handle.await, Err(Error::Closed))); // The guard drop path should have removed the stale in-flight entry. let page_cache = cache_ref.cache.read(); assert!( !page_cache.page_fetches.contains_key(&(blob_id, 0)), "cancelled first fetcher should not leave stale page_fetches entry" ); }); } #[test_traced] fn test_followers_keep_single_flight_after_first_fetcher_cancellation() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let blob_id = 0; let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(10)); // Return one valid full page, but hold the underlying read until the test releases it. let logical_page = vec![7u8; PAGE_SIZE.get() as usize]; let crc = Crc32::checksum(&logical_page); let mut physical_page = logical_page.clone(); physical_page.extend_from_slice(&Checksum::new(PAGE_SIZE.get(), crc).to_bytes()); let (started_tx, started_rx) = oneshot::channel(); let (release_tx, release_rx) = oneshot::channel(); let reads = Arc::new(AtomicUsize::new(0)); let blob = ControlledBlob { started: Arc::new(Mutex::new(Some(started_tx))), release: Arc::new(Mutex::new(Some(release_rx))), reads: reads.clone(), result: ControlledBlobResult::Success(Arc::new(physical_page)), }; // Start the fetch that installs the shared in-flight entry. let mut first_buf = vec![0u8; PAGE_SIZE.get() as usize]; let cache_ref_for_first = cache_ref.clone(); let blob_for_first = blob.clone(); let first = context.child("first").spawn(move |_| async move { let _ = cache_ref_for_first .read(&blob_for_first, blob_id, &mut first_buf, 0) .await; }); started_rx.await.expect("first read never started"); // Join as a follower while the first fetch is still blocked in the blob. let mut second_buf = vec![0u8; PAGE_SIZE.get() as usize]; let cache_ref_for_second = cache_ref.clone(); let blob_for_second = blob.clone(); let second = context.child("second").spawn(move |_| async move { cache_ref_for_second .read(&blob_for_second, blob_id, &mut second_buf, 0) .await .expect("second read failed"); second_buf }); // Wait until both tasks are registered against the same in-flight fetch. loop { let joined = { let page_cache = cache_ref.cache.read(); page_cache .page_fetches .get(&(blob_id, 0)) .map(|fetch| fetch.waiters == 2) .unwrap_or(false) }; if joined { break; } context.sleep(Duration::from_millis(1)).await; } // Cancel the original fetcher; the follower should keep the generation alive. first.abort(); assert!(matches!(first.await, Err(Error::Closed))); // A later reader should still join the existing in-flight fetch instead of starting a // second blob read. let mut third_buf = vec![0u8; PAGE_SIZE.get() as usize]; let cache_ref_for_third = cache_ref.clone(); let blob_for_third = blob.clone(); let third = context.child("third").spawn(move |_| async move { cache_ref_for_third .read(&blob_for_third, blob_id, &mut third_buf, 0) .await .expect("third read failed"); third_buf }); // Either the third reader bumps the waiter count back to 2, or a bug starts a second // blob read. loop { let third_entered = { let page_cache = cache_ref.cache.read(); reads.load(Ordering::Relaxed) > 1 || page_cache .page_fetches .get(&(blob_id, 0)) .map(|fetch| fetch.waiters == 2) .unwrap_or(false) }; if third_entered { break; } context.sleep(Duration::from_millis(1)).await; } // Let the single underlying fetch complete and satisfy both surviving waiters. let _ = release_tx.send(()); let second_buf = second.await.expect("second task failed"); let third_buf = third.await.expect("third task failed"); assert_eq!(second_buf, logical_page); assert_eq!(third_buf, logical_page); // All waiters should have shared the same blob read. assert_eq!(reads.load(Ordering::Relaxed), 1); // The successful fetch should populate the cache for later readers. let mut cached = vec![0u8; PAGE_SIZE.get() as usize]; assert_eq!( cache_ref.read_cached(blob_id, &mut cached, 0), PAGE_SIZE.get() as usize ); assert_eq!(cached, logical_page); // A later read should hit the cached page and avoid touching the blob again. let mut fourth_buf = vec![0u8; PAGE_SIZE.get() as usize]; cache_ref .read(&blob, blob_id, &mut fourth_buf, 0) .await .unwrap(); assert_eq!(fourth_buf, logical_page); assert_eq!(reads.load(Ordering::Relaxed), 1); let page_cache = cache_ref.cache.read(); assert!( !page_cache.page_fetches.contains_key(&(blob_id, 0)), "completed fetch should leave no stale page_fetches entry" ); }); } #[test_traced] fn test_page_fetch_error_removes_entry_for_all_waiters() { let executor = deterministic::Runner::default(); executor.start(|context| async move { let blob_id = 0; let cache_ref = CacheRef::from_pooler(&context, PAGE_SIZE, NZUsize!(10)); // Hold one shared fetch in flight, then make the underlying read fail. let (started_tx, started_rx) = oneshot::channel(); let (release_tx, release_rx) = oneshot::channel(); let reads = Arc::new(AtomicUsize::new(0)); let blob = ControlledBlob { started: Arc::new(Mutex::new(Some(started_tx))), release: Arc::new(Mutex::new(Some(release_rx))), reads: reads.clone(), result: ControlledBlobResult::Error, }; // Start the fetch that creates the in-flight entry. let mut first_buf = vec![0u8; PAGE_SIZE.get() as usize]; let cache_ref_for_first = cache_ref.clone(); let blob_for_first = blob.clone(); let first = context.child("first").spawn(move |_| async move { cache_ref_for_first .read(&blob_for_first, blob_id, &mut first_buf, 0) .await }); started_rx.await.expect("first erroring read never started"); // Join with a second waiter that should observe the same failure. let mut second_buf = vec![0u8; PAGE_SIZE.get() as usize]; let cache_ref_for_second = cache_ref.clone(); let blob_for_second = blob.clone(); let second = context.child("second").spawn(move |_| async move { cache_ref_for_second .read(&blob_for_second, blob_id, &mut second_buf, 0) .await }); // Wait until both tasks share the same in-flight fetch entry. loop { let joined = { let page_cache = cache_ref.cache.read(); page_cache .page_fetches .get(&(blob_id, 0)) .map(|fetch| fetch.waiters == 2) .unwrap_or(false) }; if joined { break; } context.sleep(Duration::from_millis(1)).await; } // Release the blocked read so the shared fetch resolves with an error. let _ = release_tx.send(()); assert!(matches!(first.await, Ok(Err(Error::ReadFailed)))); assert!(matches!(second.await, Ok(Err(Error::ReadFailed)))); // Both waiters should still have shared a single blob read. assert_eq!(reads.load(Ordering::Relaxed), 1); // The failed generation must remove its in-flight entry and avoid caching data. { let page_cache = cache_ref.cache.read(); assert!( !page_cache.page_fetches.contains_key(&(blob_id, 0)), "erroring fetch should leave no stale page_fetches entry" ); } let mut cached = vec![0u8; PAGE_SIZE.get() as usize]; assert_eq!(cache_ref.read_cached(blob_id, &mut cached, 0), 0); // A later read should start a fresh fetch rather than reusing stale error state. let mut third_buf = vec![0u8; PAGE_SIZE.get() as usize]; assert!(matches!( cache_ref.read(&blob, blob_id, &mut third_buf, 0).await, Err(Error::ReadFailed) )); assert_eq!(reads.load(Ordering::Relaxed), 2); }); } #[test_traced] fn test_read_cached_many_all_cached() { let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(10)); let blob_id = cache_ref.next_id(); let page0 = vec![0xAA; PAGE_SIZE.get() as usize]; let page1 = vec![0xBB; PAGE_SIZE.get() as usize]; // Populate two pages with distinct data. { let mut cache = cache_ref.cache.write(); cache.cache(blob_id, &page0, 0); cache.cache(blob_id, &page1, 1); } let mut buf0 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut buf1 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut ranges: Vec<(&mut [u8], u64)> = vec![(&mut buf0, 0), (&mut buf1, PAGE_SIZE_U64)]; cache_ref.read_cached_many(blob_id, &mut ranges); // All ranges served from cache, so the vec is now empty. assert!(ranges.is_empty()); drop(ranges); // Buffers should contain the cached page data. assert!(buf0 == page0); assert!(buf1 == page1); } #[test_traced] fn test_read_cached_many_none_cached() { let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(10)); let blob_id = cache_ref.next_id(); let mut buf0 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut buf1 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut ranges: Vec<(&mut [u8], u64)> = vec![(&mut buf0, 0), (&mut buf1, PAGE_SIZE_U64)]; // Empty cache: both ranges should miss and remain in the vec unchanged. cache_ref.read_cached_many(blob_id, &mut ranges); assert_eq!(ranges.len(), 2); assert_eq!(ranges[0].1, 0); assert_eq!(ranges[1].1, PAGE_SIZE_U64); } #[test_traced] fn test_read_cached_many_scattered_misses() { // Verify that read_cached_many checks ALL ranges, not just up to the // first miss. Pages 0 and 2 are cached, page 1 is not. let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(10)); let blob_id = cache_ref.next_id(); let page0 = vec![0x11; PAGE_SIZE.get() as usize]; let page2 = vec![0x33; PAGE_SIZE.get() as usize]; { let mut cache = cache_ref.cache.write(); cache.cache(blob_id, &page0, 0); // page 1 deliberately not cached cache.cache(blob_id, &page2, 2); } let mut buf0 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut buf1 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut buf2 = vec![0u8; PAGE_SIZE_U64 as usize]; let mut ranges: Vec<(&mut [u8], u64)> = vec![ (&mut buf0, 0), (&mut buf1, PAGE_SIZE_U64), (&mut buf2, PAGE_SIZE_U64 * 2), ]; cache_ref.read_cached_many(blob_id, &mut ranges); // Only the page 1 miss should remain (page 2 is still processed despite // the earlier miss). assert_eq!(ranges.len(), 1); assert_eq!(ranges[0].1, PAGE_SIZE_U64); drop(ranges); // Cached pages should have their data written to the buffers. assert!(buf0 == page0); assert!(buf2 == page2); // Missed page's buffer should be untouched (still zeroed). assert!(buf1.iter().all(|b| *b == 0)); } #[test_traced] fn test_read_cached_many_stale_hint_after_eviction() { // Insert one page past capacity so the CLOCK evicts page 0 and reuses its slot for // page 2. Page 0's hint now points at a slot holding page 2's key, so the batched // read must report page 0 as a miss (never page 2's bytes) while still serving the // live pages. let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(2)); let blob_id = cache_ref.next_id(); let page_size = PAGE_SIZE.get() as usize; { let mut cache = cache_ref.cache.write(); for page in 0u64..3 { cache.cache(blob_id, &vec![page as u8 + 1; page_size], page); } } let mut bufs: Vec> = (0..3).map(|_| vec![0u8; page_size]).collect(); let mut iter = bufs.iter_mut(); let mut ranges: Vec<(&mut [u8], u64)> = (0..3u64) .map(|page| (iter.next().unwrap().as_mut_slice(), page * PAGE_SIZE_U64)) .collect(); cache_ref.read_cached_many(blob_id, &mut ranges); // Page 0 was evicted: it must be the one remaining miss, untouched. assert_eq!(ranges.len(), 1); assert_eq!(ranges[0].1, 0); drop(ranges); assert!(bufs[0].iter().all(|b| *b == 0)); assert_eq!(bufs[1], vec![2u8; page_size]); assert_eq!(bufs[2], vec![3u8; page_size]); } #[test_traced] fn test_read_cached_many_cross_blob_hint_collision() { // Two blobs whose salted ranges overlap share a hint entry, and the later insert // overwrites the earlier blob's hint. The hint only proposes a slot: [Clock::get_at] // validates the full (blob, page) key, so each blob reads back its own bytes (the // clobbered one through the fallback lookup), never the other's. let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(4)); let blob_a = cache_ref.next_id(); let blob_b = cache_ref.next_id(); let page_size = PAGE_SIZE.get() as usize; let page_a = 5u64; let page_b = { let mut cache = cache_ref.cache.write(); // Solve hint_index(blob_b, page_b) == hint_index(blob_a, page_a) for page_b. let mask = cache.hints.len() as u64 - 1; let page_b = page_a .wrapping_add(blob_a.wrapping_mul(commonware_utils::GOLDEN_RATIO)) .wrapping_sub(blob_b.wrapping_mul(commonware_utils::GOLDEN_RATIO)) & mask; assert_eq!( cache.hint_index(blob_a, page_a), cache.hint_index(blob_b, page_b) ); cache.cache(blob_a, &vec![0xAA; page_size], page_a); cache.cache(blob_b, &vec![0xBB; page_size], page_b); page_b }; for (blob, page, byte) in [(blob_a, page_a, 0xAAu8), (blob_b, page_b, 0xBB)] { let mut buf = vec![0u8; page_size]; let mut ranges: Vec<(&mut [u8], u64)> = vec![(&mut buf, page * PAGE_SIZE_U64)]; cache_ref.read_cached_many(blob, &mut ranges); assert!( ranges.is_empty(), "blob {blob} page {page} should be cached" ); drop(ranges); assert_eq!(buf, vec![byte; page_size]); } } #[test_traced] fn test_read_cached_many_sparse_page_number_keeps_hints_fixed() { // Hint memory is fixed at construction: caching at an extreme page number must not // grow any structure, and the page is still served through the hint path. let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(2)); let blob_id = cache_ref.next_id(); let page_size = PAGE_SIZE.get() as usize; let page_num = u64::MAX / PAGE_SIZE_U64 - 1; { let mut cache = cache_ref.cache.write(); let hints = cache.hints.len(); cache.cache(blob_id, &vec![0x5A; page_size], page_num); assert_eq!(cache.hints.len(), hints); } let mut buf = vec![0u8; page_size]; let mut ranges: Vec<(&mut [u8], u64)> = vec![(&mut buf, page_num * PAGE_SIZE_U64)]; cache_ref.read_cached_many(blob_id, &mut ranges); assert!(ranges.is_empty()); drop(ranges); assert_eq!(buf, vec![0x5A; page_size]); } #[test_traced] fn test_read_cached_many_invalidated_page_is_a_miss() { // Invalidated pages free their slots but keep their keys. Their hints need no // cleanup: a freed slot is not live, so the dropped page reads as a miss until // re-cached. let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(4)); let blob_id = cache_ref.next_id(); let page_size = PAGE_SIZE.get() as usize; { let mut cache = cache_ref.cache.write(); for page in 0u64..4 { cache.cache(blob_id, &vec![page as u8 + 1; page_size], page); } } cache_ref.invalidate_from(blob_id, 2); let read_page = |page: u64| { let mut buf = vec![0u8; page_size]; let mut ranges: Vec<(&mut [u8], u64)> = vec![(&mut buf, page * PAGE_SIZE_U64)]; cache_ref.read_cached_many(blob_id, &mut ranges); let hit = ranges.is_empty(); drop(ranges); hit.then_some(buf) }; assert_eq!(read_page(0), Some(vec![1u8; page_size])); assert_eq!(read_page(1), Some(vec![2u8; page_size])); assert_eq!(read_page(2), None); assert_eq!(read_page(3), None); // Re-caching a dropped page restores it through the hint path. { let mut cache = cache_ref.cache.write(); cache.cache(blob_id, &vec![0xCC; page_size], 2); } assert_eq!(read_page(2), Some(vec![0xCC; page_size])); } #[rstest] #[case::empty_read(vec![], 0, 0, 0)] #[case::single_cached_page(vec![0], 3, 5, 5)] #[case::cached_range_can_cross_pages(vec![0, 1], PAGE_SIZE_U64 - 2, 4, 4)] #[case::missing_first_page_reads_nothing(vec![1], 0, 4, 0)] #[case::missing_later_page_truncates_read(vec![0], PAGE_SIZE_U64 - 2, 4, 2)] fn test_read_cached( #[case] cached_pages: Vec, #[case] logical_offset: u64, #[case] len: usize, #[case] expected_count: usize, ) { let pool = test_pool(); let cache_ref = CacheRef::new(pool, PAGE_SIZE, NZUsize!(10)); let blob_id = cache_ref.next_id(); let sentinel = 0xEE; let page_size = PAGE_SIZE.get() as usize; { let mut cache = cache_ref.cache.write(); for page in cached_pages { // Use a distinct byte per page so cross-page reads prove both halves were copied. cache.cache(blob_id, &vec![page as u8 + 1; page_size], page); } } let mut buf = vec![sentinel; len]; let count = cache_ref.read_cached(blob_id, &mut buf, logical_offset); assert_eq!(count, expected_count); // The satisfied prefix holds cached bytes; everything past the first fault is untouched. assert_eq!(buf[..count], expected_cached_bytes(logical_offset, count)); assert!(buf[count..].iter().all(|b| *b == sentinel)); } }