[storage] invalidate rewinded pages in page cache (#3653)

Author: roberto-bayardo

runtime/src/utils/buffer/paged/append.rs

@@ -857,12 +857,6 @@ impl<B: Blob> Append<B> {
             return Ok(());
         }
 
-        // Implementation note: rewinding the blob across a page boundary potentially results in
-        // stale data remaining in the page cache. We don't proactively purge the data
-        // within this function since it would be inaccessible anyway. Instead we ensure it is
-        // always updated should the blob grow back to the point where we have new data for the same
-        // page, if any old data hasn't expired naturally by then.
-
         let logical_page_size = self.cache_ref.page_size();
         let physical_page_size = logical_page_size + CHECKSUM_SIZE;
 
@@ -889,6 +883,12 @@ impl<B: Blob> Append<B> {
         blob_guard.blob.resize(new_physical_size).await?;
         blob_guard.partial_page_state = None;
 
+        // Evict cached pages at or beyond the new full-page boundary. The page at `full_pages` (if
+        // partial) is now owned by the tip buffer, and anything above is beyond the new logical
+        // size. Leaving their pre-resize contents in the cache lets `try_read_sync` (which bypasses
+        // the tip buffer) observe stale bytes once the tip is repopulated.
+        self.cache_ref.invalidate_from(self.id, full_pages);
+
         // Update blob state and buffer based on the desired logical size. The partial page data is
         // read with CRC validation; the validated length may exceed partial_bytes (reflecting the
         // old data length), but we only load the prefix we need. The next sync will write the
@@ -2456,6 +2456,54 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_resize_invalidates_cache() {
+        // Regression: shrinking a blob across a page boundary must drop cached pages for the
+        // truncated region. Before the fix, `try_read_sync` (which bypasses the tip buffer)
+        // would observe pre-resize bytes at offsets later reclaimed by new appends.
+        let executor = deterministic::Runner::default();
+        executor.start(|context: deterministic::Context| async move {
+            let cache_ref = CacheRef::from_pooler(
+                &context.with_label("cache"),
+                PAGE_SIZE,
+                NZUsize!(BUFFER_SIZE),
+            );
+            let (blob, blob_size) = context
+                .open("test_partition", b"resize_invalidates_cache")
+                .await
+                .unwrap();
+            let append = Append::new(blob, blob_size, BUFFER_SIZE, cache_ref)
+                .await
+                .unwrap();
+
+            // Write + sync a full page so it lands in the page cache. Use a distinct byte
+            // pattern so a stale cache read would be obvious.
+            let page_size = PAGE_SIZE.get() as usize;
+            let old_bytes = vec![0xAAu8; page_size];
+            append.append(&old_bytes).await.unwrap();
+            append.sync().await.unwrap();
+
+            // Confirm page 0 is reachable via the cache-only fast path.
+            let mut probe = vec![0u8; 16];
+            assert!(append.try_read_sync(0, &mut probe));
+            assert_eq!(probe, vec![0xAAu8; 16]);
+
+            // Rewind to 0 (crossing the page boundary) and append a new, distinct pattern.
+            append.resize(0).await.unwrap();
+            let new_bytes = vec![0xBBu8; 16];
+            append.append(&new_bytes).await.unwrap();
+
+            // The cache must not serve pre-resize bytes. Either try_read_sync misses (cache
+            // was invalidated) or it returns the new pattern; it must never return 0xAA.
+            let mut probe = vec![0u8; 16];
+            let hit = append.try_read_sync(0, &mut probe);
+            assert!(
+                !hit || probe == new_bytes,
+                "try_read_sync served stale pre-resize bytes: {probe:?}"
+            );
+        });
+    }
+
     #[test]
     fn test_reopen_partial_tail_append_and_resize() {
         let executor = deterministic::Runner::default();

runtime/src/utils/buffer/paged/cache.rs

@@ -107,12 +107,15 @@ struct Cache {
     /// # Invariants
     ///
     /// Each `index` entry maps to exactly one `entries` slot, and that entry always has a
-    /// matching key.
+    /// matching key. (The converse is not true: after [Self::invalidate_from] a slot may retain
+    /// a stale key that is no longer present in `index`.)
     index: HashMap<(u64, u64), usize>,
 
     /// Metadata for each cache slot.
     ///
-    /// Each `entries` slot has exactly one corresponding `index` entry.
+    /// Every entry reachable via `index` has a matching key here. Slots that were invalidated by
+    /// [Self::invalidate_from] retain their stale key but are unreachable from `index` and will
+    /// be reclaimed by the Clock evictor on the next sweep.
     entries: Vec<CacheEntry>,
 
     /// Per-slot page buffers allocated from the pool.
@@ -137,6 +140,12 @@ struct Cache {
 /// Metadata for a single cache entry (page data stored in per-slot buffers).
 struct CacheEntry {
     /// The cache key which is composed of the blob id and page number of the page.
+    ///
+    /// # Invariant
+    ///
+    /// Every live cache slot has a matching entry in `index`. Slots that have been invalidated (see
+    /// [Cache::invalidate_from]) retain their stale key here but are no longer reachable via
+    /// `index` and will be reclaimed first by the Clock evictor.
     key: (u64, u64),
 
     /// A bit indicating whether this page was recently referenced.
@@ -425,6 +434,13 @@ impl CacheRef {
 
         buf.len()
     }
+
+    /// 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 {
@@ -525,18 +541,24 @@ impl Cache {
             return;
         }
 
-        // Cache full: find slot to evict using Clock algorithm
+        // Cache full: find slot to evict using Clock algorithm. Invalidated slots (`referenced =
+        // false`, stale `entry.key` no longer in `index`) are reclaimed on the first sweep.
         while self.entries[self.clock].referenced.load(Ordering::Relaxed) {
             self.entries[self.clock]
                 .referenced
                 .store(false, Ordering::Relaxed);
             self.clock = (self.clock + 1) % self.entries.len();
         }
 
-        // Evict and replace
+        // Evict and replace. Only drop the old `entry.key` from `index` when it still points
+        // to this slot: after `invalidate_from` a slot may hold a stale key that has since
+        // been re-cached at a different slot, and an unconditional `remove` would orphan
+        // that live entry.
         let slot = self.clock;
         let entry = &mut self.entries[slot];
-        assert!(self.index.remove(&entry.key).is_some());
+        if self.index.get(&entry.key) == Some(&slot) {
+            self.index.remove(&entry.key);
+        }
         self.index.insert(key, slot);
         entry.key = key;
         entry.referenced.store(true, Ordering::Relaxed);
@@ -545,6 +567,21 @@ impl Cache {
         // Move the clock forward.
         self.clock = (self.clock + 1) % self.entries.len();
     }
+
+    /// Drop any cached pages for `blob_id` at `page_num >= start_page`. The slots keep their
+    /// (now stale) `entry.key` so the Clock evictor can reclaim them; `read_at` and the
+    /// duplicate-update path never reach them because `index` no longer maps to them.
+    fn invalidate_from(&mut self, blob_id: u64, start_page: u64) {
+        self.index.retain(|&(bid, page_num), &mut slot| {
+            if bid != blob_id || page_num < start_page {
+                return true;
+            }
+            self.entries[slot]
+                .referenced
+                .store(false, Ordering::Relaxed);
+            false
+        });
+    }
 }
 
 /// Fetch one logical page for insertion into the page cache, rejecting partial pages because cache
@@ -755,6 +792,56 @@ mod tests {
         );
     }
 
+    #[test_traced]
+    fn test_invalidate_from_does_not_orphan_re_cached_page() {
+        // Regression: when the Clock evictor lands on an invalidated slot whose stale key has
+        // since been re-cached at a different slot, the old index entry (pointing to the
+        // live slot) must not be removed.
+        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 carry stale keys with referenced=false.
+        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. Clock sits at slot 0, which is referenced=false, so the insert
+        // lands at slot 0 (slot 1 still holds its stale (blob, 1) key).
+        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. Clock now advances to slot 1 (still referenced=false), evicts it.
+        // With the buggy unconditional `index.remove(entry.key)` this would remove the live
+        // (blob, 1) -> slot 0 mapping, orphaning slot 0.
+        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