Fix caching & memory infrastructure review findings

native/src/disk_cache/background.rs

@@ -115,11 +115,24 @@ impl L2DiskCache {
                             }
                         }
 
-                        // For size-based mode: drain bytes and notify waiting senders
+                        // For size-based mode: drain bytes and notify waiting senders.
+                        //
+                        // The mutex MUST be held across the fetch_sub + notify_all. The producer
+                        // (WriteSender::send) holds this same mutex across its check-and-wait, so
+                        // acquiring it here closes the lost-wakeup race: either we decrement before
+                        // the producer's load (it sees room and never waits), or the producer is
+                        // already parked in cvar.wait (which released the mutex) and our notify wakes
+                        // it. Without the lock, a notify issued between the producer's load and its
+                        // wait() would be lost, hanging the producer forever.
                         if let Some((queued_bytes, backpressure)) = sb_state {
-                            let remaining = queued_bytes.fetch_sub(data_len, Ordering::Release) - data_len;
-                            let (_lock, cvar) = &*backpressure;
-                            cvar.notify_all();
+                            let (lock, cvar) = &*backpressure;
+                            let remaining = {
+                                let _guard = lock.lock().unwrap();
+                                let remaining =
+                                    queued_bytes.fetch_sub(data_len, Ordering::Release) - data_len;
+                                cvar.notify_all();
+                                remaining
+                            };
 
                             // When queue fully drains, release overflow memory back to pool
                             if remaining == 0 {

native/src/disk_cache/compression.rs

@@ -3,6 +3,7 @@
 
 #![allow(dead_code)]
 
+use std::borrow::Cow;
 use std::io;
 use lz4_flex::{compress_prepend_size, decompress_size_prepended};
 use super::types::{CompressionAlgorithm, DiskCacheConfig};
@@ -47,31 +48,31 @@ pub fn should_compress(config: &DiskCacheConfig, component: &str, data_size: usi
     }
 }
 
-/// Compress data if appropriate
-pub fn compress_data(config: &DiskCacheConfig, component: &str, data: &[u8]) -> (Vec<u8>, CompressionAlgorithm) {
+/// Compress data if appropriate.
+///
+/// Returns a `Cow` so the uncompressed path (the current default for every
+/// component — see `should_compress`) is zero-copy: the caller writes the
+/// borrowed slice straight to disk instead of allocating a throwaway `Vec`.
+pub fn compress_data<'a>(
+    config: &DiskCacheConfig,
+    component: &str,
+    data: &'a [u8],
+) -> (Cow<'a, [u8]>, CompressionAlgorithm) {
     if !should_compress(config, component, data.len()) {
-        return (data.to_vec(), CompressionAlgorithm::None);
+        return (Cow::Borrowed(data), CompressionAlgorithm::None);
     }
 
     match config.compression {
-        CompressionAlgorithm::None => (data.to_vec(), CompressionAlgorithm::None),
-        CompressionAlgorithm::Lz4 => {
+        CompressionAlgorithm::None => (Cow::Borrowed(data), CompressionAlgorithm::None),
+        // Zstd is not available as a direct dependency; it falls back to LZ4, which
+        // provides good compression at higher speed.
+        CompressionAlgorithm::Lz4 | CompressionAlgorithm::Zstd => {
             let compressed = compress_prepend_size(data);
-            // Only use compression if it actually saves space
+            // Only use compression if it actually saves space.
             if compressed.len() < data.len() {
-                (compressed, CompressionAlgorithm::Lz4)
+                (Cow::Owned(compressed), CompressionAlgorithm::Lz4)
             } else {
-                (data.to_vec(), CompressionAlgorithm::None)
-            }
-        }
-        CompressionAlgorithm::Zstd => {
-            // Zstd not currently available as direct dependency
-            // Fall back to LZ4 which provides good compression with better speed
-            let compressed = compress_prepend_size(data);
-            if compressed.len() < data.len() {
-                (compressed, CompressionAlgorithm::Lz4)
-            } else {
-                (data.to_vec(), CompressionAlgorithm::None)
+                (Cow::Borrowed(data), CompressionAlgorithm::None)
             }
         }
     }

native/src/disk_cache/get_ops.rs

@@ -277,7 +277,19 @@ pub fn get_coalesced(
         }
 
         // Overlap with this cached range
-        if let Some(overlap) = cached_range.overlap_with(requested_range.start, requested_range.end) {
+        if let Some(raw_overlap) = cached_range.overlap_with(requested_range.start, requested_range.end) {
+            // Clip the overlap to the current cursor. Cached ranges may nest/overlap
+            // (see RangeIndex::find_overlapping), so a later range can re-cover bytes
+            // an earlier segment already emitted. Clipping to `cursor` prevents emitting
+            // overlapping segments (which would make cached_bytes exceed the request and
+            // let combine_segments copy one segment over another).
+            let overlap = max(raw_overlap.start, cursor)..raw_overlap.end;
+            if overlap.start >= overlap.end {
+                // Fully behind the cursor — already covered by an earlier segment.
+                cursor = max(cursor, raw_overlap.end);
+                continue;
+            }
+
             // Use fast sub-range extraction via mmap (avoids loading entire file)
             if let Some(data) = get_subrange(
                 config,
@@ -294,6 +306,16 @@ pub fn get_coalesced(
                     data,
                 });
                 cached_bytes += (overlap.end - overlap.start) as u64;
+            } else {
+                // The cached file for this range could not be read — it was likely
+                // evicted between the range-index snapshot and this read (do_evict
+                // deletes the directory before updating the manifest). Treat the
+                // sub-range as a gap so it is refetched from L3, rather than advancing
+                // the cursor past it and silently reporting the request as fully
+                // cached (which combine_segments would then serve as zero-filled or
+                // truncated bytes).
+                gaps.push(overlap.clone());
+                gap_bytes += overlap.end - overlap.start;
             }
 
             cursor = max(cursor, overlap.end);

native/src/disk_cache/lru.rs

@@ -1,66 +1,115 @@
 // lru.rs - LRU tracking for split eviction
 // Extracted from mod.rs during refactoring
 
+use std::collections::HashMap;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-/// LRU tracking entry for a split
-pub(crate) struct SplitLruEntry {
-    pub key: String,
-    pub size_bytes: u64,
-    pub last_accessed: u64,
+fn now_secs() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
 }
 
-/// LRU table for tracking split access patterns
+/// Per-split LRU bookkeeping.
+struct LruEntry {
+    size_bytes: u64,
+    /// Wall-clock seconds of the last access (used as the primary LRU key).
+    last_accessed: u64,
+    /// Monotonic tiebreaker so accesses within the same second still order
+    /// deterministically (seconds granularity alone would be arbitrary).
+    seq: u64,
+}
+
+/// LRU table for tracking split access patterns.
+///
+/// Keyed by split_key (`storage_loc/split_id`) for O(1) touch/remove on the hot
+/// read path. Eviction candidates are produced by sorting on demand, which only
+/// happens when the cache is over its high-water mark.
 pub(crate) struct SplitLruTable {
-    entries: Vec<SplitLruEntry>,
+    entries: HashMap<String, LruEntry>,
+    next_seq: u64,
 }
 
 impl SplitLruTable {
     pub fn new() -> Self {
-        Self { entries: Vec::new() }
+        Self {
+            entries: HashMap::new(),
+            next_seq: 0,
+        }
     }
 
+    fn bump_seq(&mut self) -> u64 {
+        let seq = self.next_seq;
+        self.next_seq += 1;
+        seq
+    }
+
+    /// Record an access to a split, updating its recency and size. O(1).
     pub fn touch(&mut self, key: &str, size_bytes: u64) {
-        let now = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap_or_default()
-            .as_secs();
+        let now = now_secs();
+        let seq = self.bump_seq();
+        match self.entries.get_mut(key) {
+            Some(entry) => {
+                entry.last_accessed = now;
+                entry.size_bytes = size_bytes;
+                entry.seq = seq;
+            }
+            None => {
+                self.entries.insert(
+                    key.to_string(),
+                    LruEntry {
+                        size_bytes,
+                        last_accessed: now,
+                        seq,
+                    },
+                );
+            }
+        }
+    }
 
-        if let Some(entry) = self.entries.iter_mut().find(|e| e.key == key) {
-            entry.last_accessed = now;
-            entry.size_bytes = size_bytes;
-        } else {
-            self.entries.push(SplitLruEntry {
-                key: key.to_string(),
-                size_bytes,
-                last_accessed: now,
-            });
+    /// Seed an entry restored from the persisted manifest at startup, preserving
+    /// its recorded `last_accessed` timestamp so cold splits from prior runs remain
+    /// visible to eviction. Never overwrites a live entry.
+    pub fn seed(&mut self, key: &str, size_bytes: u64, last_accessed: u64) {
+        if self.entries.contains_key(key) {
+            return;
         }
+        let seq = self.bump_seq();
+        self.entries.insert(
+            key.to_string(),
+            LruEntry {
+                size_bytes,
+                last_accessed,
+                seq,
+            },
+        );
     }
 
     pub fn remove(&mut self, key: &str) {
-        self.entries.retain(|e| e.key != key);
+        self.entries.remove(key);
     }
 
-    /// Get splits to evict to reach target size, ordered by LRU
+    /// Get splits to evict to reach target size, ordered by LRU (oldest first).
     pub fn get_eviction_candidates(&self, current_bytes: u64, target_bytes: u64) -> Vec<String> {
         if current_bytes <= target_bytes {
             return Vec::new();
         }
 
-        // Sort by last_accessed (oldest first)
-        let mut sorted: Vec<_> = self.entries.iter().collect();
-        sorted.sort_by_key(|e| e.last_accessed);
+        // Sort by (last_accessed, seq) so oldest — and, within a second, least
+        // recently touched — come first.
+        let mut sorted: Vec<(&String, &LruEntry)> = self.entries.iter().collect();
+        sorted.sort_by_key(|(_, e)| (e.last_accessed, e.seq));
 
         let mut to_evict = Vec::new();
         let mut freed = 0u64;
         let need_to_free = current_bytes - target_bytes;
 
-        for entry in sorted {
+        for (key, entry) in sorted {
             if freed >= need_to_free {
                 break;
             }
-            to_evict.push(entry.key.clone());
+            to_evict.push(key.clone());
             freed += entry.size_bytes;
         }
 

native/src/disk_cache/mmap_cache.rs

@@ -134,15 +134,27 @@ impl MmapCache {
         self.lru_order.push_back(path.to_path_buf());
     }
 
-    /// Remove a specific path from the cache (e.g., when file is deleted)
-    #[allow(dead_code)]
+    /// Remove a specific path from the cache (e.g., when a file is deleted or
+    /// about to be overwritten). Dropping the `Arc<Mmap>` here (once all in-flight
+    /// readers release their clones) unmaps the region, so the underlying inode's
+    /// disk space can actually be reclaimed and a subsequent read re-maps fresh
+    /// content instead of a stale mapping of the old file.
     pub fn remove(&mut self, path: &Path) {
         self.maps.remove(path);
         if let Some(pos) = self.lru_order.iter().position(|p| p == path) {
             self.lru_order.remove(pos);
         }
     }
 
+    /// Remove every cached mapping whose file lives under `dir` (used when a split
+    /// directory is evicted). Without this, evicted splits' files stay mapped —
+    /// pinning the deleted inodes so disk space is not reclaimed, and risking stale
+    /// reads if the same component path is later re-written.
+    pub fn remove_under_dir(&mut self, dir: &Path) {
+        self.maps.retain(|p, _| !p.starts_with(dir));
+        self.lru_order.retain(|p| !p.starts_with(dir));
+    }
+
     /// Clear all mappings
     #[allow(dead_code)]
     pub fn clear(&mut self) {