diff --git a/native/src/disk_cache/background.rs b/native/src/disk_cache/background.rs
index 9a8290cb..d9d173fc 100644
--- a/native/src/disk_cache/background.rs
+++ b/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 {
diff --git a/native/src/disk_cache/compression.rs b/native/src/disk_cache/compression.rs
index 516ba429..e42e8e7e 100644
--- a/native/src/disk_cache/compression.rs
+++ b/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)
             }
         }
     }
diff --git a/native/src/disk_cache/get_ops.rs b/native/src/disk_cache/get_ops.rs
index acff4762..d41c8881 100644
--- a/native/src/disk_cache/get_ops.rs
+++ b/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);
diff --git a/native/src/disk_cache/lru.rs b/native/src/disk_cache/lru.rs
index 3668e793..27b18f83 100644
--- a/native/src/disk_cache/lru.rs
+++ b/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;
         }
 
diff --git a/native/src/disk_cache/mmap_cache.rs b/native/src/disk_cache/mmap_cache.rs
index 272aaa2a..90f4bde4 100644
--- a/native/src/disk_cache/mmap_cache.rs
+++ b/native/src/disk_cache/mmap_cache.rs
@@ -134,8 +134,11 @@ 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) {
@@ -143,6 +146,15 @@ impl MmapCache {
         }
     }
 
+    /// 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) {
diff --git a/native/src/disk_cache/mod.rs b/native/src/disk_cache/mod.rs
index b701ba4f..fc5f2cb8 100644
--- a/native/src/disk_cache/mod.rs
+++ b/native/src/disk_cache/mod.rs
@@ -49,7 +49,6 @@ use std::io::{self, Read};
 use std::ops::Range;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
-use std::time::Duration;
 
 use crate::debug_println;
 use crate::memory_pool::{self, DiskCacheMemoryBudget};
@@ -106,26 +105,12 @@ impl WriteSender {
         }
     }
 
-    /// Try to send without blocking (used in Drop/shutdown paths).
-    ///
-    /// In Fragment mode, uses the bounded channel's `try_send` (may fail if full).
-    /// In SizeBased mode, unconditionally enqueues — the unbounded channel only fails
-    /// if the receiver is disconnected (shutdown). Bytes are tracked optimistically;
-    /// if `tx.send()` fails during shutdown, the counter drifts but is harmless since
-    /// no further reads will occur.
-    fn try_send(&self, req: WriteRequest) -> Result<(), ()> {
+    /// Bytes currently enqueued but not yet written to disk (size-based mode only).
+    /// Fragment mode has no byte counter, so this returns 0 there.
+    fn pending_bytes(&self) -> u64 {
         match self {
-            WriteSender::Fragment(tx) => tx.try_send(req).map_err(|_| ()),
-            WriteSender::SizeBased { tx, queued_bytes, .. } => {
-                let data_len = match &req {
-                    WriteRequest::Put { data, .. } => data.len() as u64,
-                    _ => 0,
-                };
-                if data_len > 0 {
-                    queued_bytes.fetch_add(data_len, Ordering::Release);
-                }
-                tx.send(req).map_err(|_| ())
-            }
+            WriteSender::SizeBased { queued_bytes, .. } => queued_bytes.load(Ordering::Acquire),
+            WriteSender::Fragment(_) => 0,
         }
     }
 
@@ -180,8 +165,6 @@ pub struct L2DiskCache {
     max_bytes: u64,
     /// Shutdown flag for background threads
     shutdown_flag: Arc<std::sync::atomic::AtomicBool>,
-    /// Thread handles for cleanup
-    thread_handles: Mutex<Vec<std::thread::JoinHandle<()>>>,
     /// Dirty flag - set when manifest has uncommitted changes
     manifest_dirty: Arc<std::sync::atomic::AtomicBool>,
     /// Memory budget for write queue (staircase-up/cliff-down pattern)
@@ -213,6 +196,16 @@ impl L2DiskCache {
             split_states.insert(split_key.clone(), SplitState::from_entry(split_entry));
         }
 
+        // Seed the LRU table from the persisted manifest so splits restored from a
+        // previous process are immediately visible to eviction (using their recorded
+        // last_accessed time). Without this, a full-at-startup cache would keep its
+        // cold splits pinned above the eviction threshold forever, since the LRU only
+        // learned about splits that happened to be touched in the current process.
+        let mut seeded_lru = SplitLruTable::new();
+        for (split_key, split_entry) in &manifest.splits {
+            seeded_lru.seed(split_key, split_entry.total_size_bytes, split_entry.last_accessed);
+        }
+
         // Create background writer channel based on configured mode.
         let (write_tx, write_rx, size_based_state) = match &config.write_queue_mode {
             WriteQueueMode::Fragment { capacity } => {
@@ -280,40 +273,33 @@ impl L2DiskCache {
             config: config.clone(),
             manifest: RwLock::new(manifest),
             split_states: RwLock::new(split_states),
-            lru_table: Mutex::new(SplitLruTable::new()),
+            lru_table: Mutex::new(seeded_lru),
             mmap_cache: Mutex::new(MmapCache::new(mmap_size)),
             write_tx,
             total_bytes: AtomicU64::new(total_bytes),
             max_bytes,
             shutdown_flag: Arc::clone(&shutdown_flag),
-            thread_handles: Mutex::new(Vec::new()),
             manifest_dirty: Arc::clone(&manifest_dirty),
             memory_budget,
         });
 
-        // Start background writer (uses Weak reference - doesn't prevent Drop)
+        // Start background writer (uses Weak reference - doesn't prevent Drop).
+        // Detached on purpose: these threads transiently upgrade the Weak, so Drop
+        // can run on them; joining from Drop would deadlock (see Drop impl).
         let cache_weak = Arc::downgrade(&cache);
-        let writer_handle = std::thread::spawn(move || {
+        std::thread::spawn(move || {
             Self::background_writer_static(write_rx, cache_weak, size_based_state);
         });
 
-        if let Ok(mut handles) = cache.thread_handles.lock() {
-            handles.push(writer_handle);
-        }
-
         // Start manifest sync timer - checks every second, syncs if dirty
         {
             let shutdown_flag_clone = Arc::clone(&shutdown_flag);
             let cache_weak = Arc::downgrade(&cache);
             let dirty_flag = Arc::clone(&manifest_dirty);
 
-            let timer_handle = std::thread::spawn(move || {
+            std::thread::spawn(move || {
                 Self::manifest_sync_timer_static(cache_weak, shutdown_flag_clone, dirty_flag);
             });
-
-            if let Ok(mut handles) = cache.thread_handles.lock() {
-                handles.push(timer_handle);
-            }
         }
 
         Ok(cache)
@@ -472,8 +458,10 @@ impl L2DiskCache {
         byte_range: Option<Range<u64>>,
         data: &[u8],
     ) {
-        // Check if we need to evict before adding
-        let current = self.total_bytes.load(Ordering::Relaxed);
+        // Check if we need to evict before adding. Count bytes already queued but not
+        // yet written so a burst doesn't overshoot the high-water mark before the
+        // background writer has flushed (and accounted) the in-flight fragments.
+        let current = self.total_bytes.load(Ordering::Relaxed) + self.write_tx.pending_bytes();
         let new_size = data.len() as u64;
 
         // Trigger eviction if we'd exceed 95% capacity
@@ -514,7 +502,7 @@ impl L2DiskCache {
         byte_range: Option<Range<u64>>,
         data: &[u8],
     ) -> bool {
-        let current = self.total_bytes.load(Ordering::Relaxed);
+        let current = self.total_bytes.load(Ordering::Relaxed) + self.write_tx.pending_bytes();
         let new_size = data.len() as u64;
 
         if current + new_size > (self.max_bytes * 95) / 100 {
@@ -640,6 +628,7 @@ impl L2DiskCache {
             &self.manifest,
             &self.split_states,
             &self.lru_table,
+            &self.mmap_cache,
             &self.total_bytes,
             storage_loc,
             split_id,
@@ -656,6 +645,7 @@ impl L2DiskCache {
             &self.manifest,
             &self.split_states,
             &self.lru_table,
+            &self.mmap_cache,
             &self.total_bytes,
             storage_loc,
             split_id,
@@ -676,6 +666,7 @@ impl L2DiskCache {
             &self.manifest,
             &self.split_states,
             &self.lru_table,
+            &self.mmap_cache,
             &self.total_bytes,
             storage_loc,
             split_id,
@@ -693,6 +684,7 @@ impl L2DiskCache {
             &self.manifest,
             &self.split_states,
             &self.lru_table,
+            &self.mmap_cache,
             &self.total_bytes,
             storage_loc,
             split_id,
@@ -738,15 +730,29 @@ impl Drop for L2DiskCache {
     fn drop(&mut self) {
         debug_println!("🔄 L2DiskCache::drop() - Starting cleanup");
 
-        // Signal shutdown to timer thread
-        self.shutdown_flag.store(true, Ordering::SeqCst);
+        // NOTE: no flush here. Inside Drop the Arc<L2DiskCache> strong count is already
+        // zero, so the background writer's `cache_weak.upgrade()` returns None and it
+        // cannot process a flush. Durable manifest sync is done by the live JNI close
+        // path via `flush_blocking()` before it releases its Arc.
 
-        // Signal shutdown to writer thread and sync manifest first.
-        let _ = self.write_tx.try_send(WriteRequest::SyncManifest);
-        let _ = self.write_tx.try_send(WriteRequest::Shutdown);
+        // Signal the timer thread to stop (it polls the flag every 100ms).
+        self.shutdown_flag.store(true, Ordering::SeqCst);
 
-        // Wait for background threads to process shutdown
-        std::thread::sleep(Duration::from_millis(250));
+        // Best-effort, NON-BLOCKING wake for a writer idling in recv(), so it observes
+        // the dropped Arc (upgrade() == None) and exits promptly. Using send_or_drop
+        // (never blocks) is essential: this Drop can itself be running on the writer
+        // thread — both background threads transiently upgrade the Weak they hold — and
+        // a blocking send on a full queue with no other receiver would deadlock. If the
+        // wake is dropped, the writer still exits when the write channel's sender (a
+        // field of self) is dropped as this cache is freed.
+        let _ = self.write_tx.send_or_drop(WriteRequest::Shutdown);
+
+        // We deliberately do NOT join the background threads. They transiently upgrade
+        // the Weak<Self>, so this Drop can run *on* the writer/timer thread (or on a
+        // tokio worker owned by the writer's runtime); joining from there would
+        // self-deadlock, or panic by dropping the runtime from its own worker. The
+        // threads terminate on their own — the timer on `shutdown_flag`, the writer
+        // when the write channel closes as `self` is freed.
 
         debug_println!("🔄 L2DiskCache::drop() - Cleanup complete");
     }
diff --git a/native/src/disk_cache/range_index.rs b/native/src/disk_cache/range_index.rs
index 9973870c..1d778274 100644
--- a/native/src/disk_cache/range_index.rs
+++ b/native/src/disk_cache/range_index.rs
@@ -70,44 +70,29 @@ impl RangeIndex {
     }
 
     /// Find all ranges that overlap with [start, end)
-    /// Returns ranges in sorted order by start position
-    /// O(log n + k) where k is the number of overlapping ranges
+    /// Returns ranges in sorted order by start position.
+    ///
+    /// Ranges are sorted by `start`, so we can stop as soon as `range.start >= end`
+    /// (no later range can begin before `end`). We must, however, examine every range
+    /// with `range.start < end`, because cached ranges may nest/overlap arbitrarily
+    /// (e.g. a full-file prewarm `[0, len)` alongside small query sub-ranges). A
+    /// "stop at first predecessor whose end <= start" backward scan — as used
+    /// previously — silently skips an earlier low-`start`/high-`end` range that still
+    /// overlaps, producing false gaps and, worse, dropping cached data from the result.
+    /// This linear prefix scan is O(k) in the number of ranges beginning before `end`,
+    /// which is bounded by the (typically small) number of distinct cached ranges per
+    /// component; correctness with nesting is worth more than the log-n prefix skip.
     pub fn find_overlapping(&self, start: u64, end: u64) -> Vec<&CachedRange> {
-        if self.ranges.is_empty() {
-            return Vec::new();
-        }
-
-        // Binary search to find first range that might overlap
-        // A range overlaps if: range.start < end && start < range.end
-        // First candidate: last range where range.start < end
-        let first_idx = match self.ranges.binary_search_by(|r| {
-            if r.start >= end {
-                std::cmp::Ordering::Greater
-            } else {
-                std::cmp::Ordering::Less
-            }
-        }) {
-            Ok(i) | Err(i) => i.saturating_sub(1),
-        };
-
-        // Scan backwards to find first actually overlapping range
-        // (needed because binary search found last range with start < end)
-        let mut scan_start = first_idx;
-        while scan_start > 0 && self.ranges[scan_start - 1].end > start {
-            scan_start -= 1;
-        }
-
-        // Collect all overlapping ranges
         let mut result = Vec::new();
-        for range in &self.ranges[scan_start..] {
+        for range in &self.ranges {
             if range.start >= end {
-                break; // No more overlaps possible
+                break; // Sorted by start: no later range can overlap.
             }
-            if range.overlaps(start, end) {
+            // range.start < end already holds; overlap iff range.end > start.
+            if range.end > start {
                 result.push(range);
             }
         }
-
         result
     }
 
diff --git a/native/src/disk_cache/write_ops.rs b/native/src/disk_cache/write_ops.rs
index 438c9da6..090102b4 100644
--- a/native/src/disk_cache/write_ops.rs
+++ b/native/src/disk_cache/write_ops.rs
@@ -4,7 +4,7 @@
 #![allow(dead_code)]
 
 use std::collections::HashMap;
-use std::fs::{self, File, OpenOptions};
+use std::fs::{self, OpenOptions};
 use std::io::{self, Write};
 use std::ops::Range;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -14,6 +14,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use super::compression::compress_data;
 use super::lru::SplitLruTable;
 use super::manifest::{CacheManifest, SplitEntry, SplitState};
+use super::mmap_cache::MmapCache;
 use super::path_helpers::{cache_dir, component_path, split_dir};
 use super::range_index::CachedRange;
 use super::types::{ComponentEntry, DiskCacheConfig};
@@ -24,6 +25,7 @@ pub fn do_put(
     manifest: &RwLock<CacheManifest>,
     split_states: &RwLock<HashMap<String, SplitState>>,
     lru_table: &Mutex<SplitLruTable>,
+    mmap_cache: &Mutex<MmapCache>,
     total_bytes: &AtomicU64,
     storage_loc: &str,
     split_id: &str,
@@ -31,7 +33,7 @@ pub fn do_put(
     byte_range: Option<Range<u64>>,
     data: &[u8],
 ) -> io::Result<()> {
-    // Compress if appropriate
+    // Compress if appropriate (Cow — borrowed, no copy, on the uncompressed path)
     let (compressed, compression) = compress_data(config, component, data);
 
     // Create split directory
@@ -61,6 +63,14 @@ pub fn do_put(
 
     fs::rename(&temp_path, &final_path)?;
 
+    // Drop any cached mmap of the destination AFTER the rename, so a subsequent read
+    // re-maps the new content rather than a stale mapping of the replaced inode.
+    // (Doing it before the rename would leave a window where a concurrent reader
+    // re-maps the old inode between the eviction and the rename.)
+    if let Ok(mut mc) = mmap_cache.lock() {
+        mc.remove(&final_path);
+    }
+
     // Update manifest
     let split_key = CacheManifest::split_key(storage_loc, split_id);
     let comp_key = CacheManifest::component_key(component, byte_range.clone());
@@ -170,6 +180,7 @@ pub async fn do_put_async(
     manifest: &RwLock<CacheManifest>,
     split_states: &RwLock<HashMap<String, SplitState>>,
     lru_table: &Mutex<SplitLruTable>,
+    mmap_cache: &Mutex<MmapCache>,
     total_bytes: &AtomicU64,
     storage_loc: &str,
     split_id: &str,
@@ -179,7 +190,7 @@ pub async fn do_put_async(
 ) -> io::Result<()> {
     use tokio::io::AsyncWriteExt;
 
-    // Compress if appropriate (CPU-bound, but fast)
+    // Compress if appropriate (Cow — borrowed, no copy, on the uncompressed path)
     let (compressed, compression) = compress_data(config, component, data);
 
     // Create split directory
@@ -206,6 +217,12 @@ pub async fn do_put_async(
 
     tokio::fs::rename(&temp_path, &final_path).await?;
 
+    // Drop any cached mmap of the destination AFTER the rename (see do_put for why),
+    // so a subsequent read re-maps the new content rather than a stale mapping.
+    if let Ok(mut mc) = mmap_cache.lock() {
+        mc.remove(&final_path);
+    }
+
     // Update manifest (sync - needs locking, but fast)
     let split_key = CacheManifest::split_key(storage_loc, split_id);
     let comp_key = CacheManifest::component_key(component, byte_range.clone());
@@ -311,6 +328,7 @@ pub fn do_evict(
     manifest: &RwLock<CacheManifest>,
     split_states: &RwLock<HashMap<String, SplitState>>,
     lru_table: &Mutex<SplitLruTable>,
+    mmap_cache: &Mutex<MmapCache>,
     total_bytes: &AtomicU64,
     storage_loc: &str,
     split_id: &str,
@@ -328,6 +346,13 @@ pub fn do_evict(
             .unwrap_or(0)
     };
 
+    // Drop any cached mmaps of this split's files first, so removing the directory
+    // actually reclaims disk space (an open mapping pins the inode) and no stale
+    // mapping survives the eviction.
+    if let Ok(mut mc) = mmap_cache.lock() {
+        mc.remove_under_dir(&split_dir_path);
+    }
+
     // Remove directory
     if split_dir_path.exists() {
         fs::remove_dir_all(&split_dir_path)?;
@@ -361,6 +386,7 @@ pub async fn do_evict_async(
     manifest: &RwLock<CacheManifest>,
     split_states: &RwLock<HashMap<String, SplitState>>,
     lru_table: &Mutex<SplitLruTable>,
+    mmap_cache: &Mutex<MmapCache>,
     total_bytes: &AtomicU64,
     storage_loc: &str,
     split_id: &str,
@@ -377,6 +403,12 @@ pub async fn do_evict_async(
             .unwrap_or(0)
     };
 
+    // Drop any cached mmaps of this split's files first, so removing the directory
+    // actually reclaims disk space and no stale mapping survives the eviction.
+    if let Ok(mut mc) = mmap_cache.lock() {
+        mc.remove_under_dir(&split_dir_path);
+    }
+
     // Async directory removal
     if tokio::fs::try_exists(&split_dir_path).await.unwrap_or(false) {
         tokio::fs::remove_dir_all(&split_dir_path).await?;
@@ -412,16 +444,21 @@ pub fn do_sync_manifest(
     let backup_path = cache_dir_path.join("manifest.json.bak");
     let temp_path = cache_dir_path.join("manifest.json.tmp");
 
-    // Serialize manifest
-    let manifest_data = {
+    // Snapshot the manifest under a brief write lock (just to stamp last_sync and
+    // clone), then serialize outside the lock. Serializing while holding the write
+    // lock would block every cache read/write for the whole (potentially large)
+    // serialization, which runs once a second while dirty. Compact JSON keeps the
+    // on-disk manifest small — it can hold one component entry per cached byte range.
+    let snapshot = {
         let mut manifest_guard = manifest.write().unwrap();
         manifest_guard.last_sync = SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
-        serde_json::to_string_pretty(&*manifest_guard)
-            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?
+        manifest_guard.clone()
     };
+    let manifest_data = serde_json::to_string(&snapshot)
+        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
 
     // Write to temp file
     {
diff --git a/native/src/global_cache/components.rs b/native/src/global_cache/components.rs
index f6079204..f385363f 100644
--- a/native/src/global_cache/components.rs
+++ b/native/src/global_cache/components.rs
@@ -5,49 +5,46 @@ use std::sync::Arc;
 
 use bytesize::ByteSize;
 use quickwit_config::{CacheConfig, SearcherConfig};
-use quickwit_search::list_fields_cache::ListFieldsCache;
-use quickwit_search::leaf_cache::LeafSearchCache;
-use quickwit_search::search_permit_provider::SearchPermitProvider;
 use quickwit_search::SearcherContext;
-use quickwit_storage::{
-    MemorySizedCache, QuickwitCache, SplitCache, StorageCache, STORAGE_METRICS,
-};
-use tantivy::aggregation::AggregationLimitsGuard;
+use quickwit_storage::SplitCache;
 use tempfile::TempDir;
 
 use crate::memory_pool::{global_pool, MemoryReservation};
 
-use super::cache_debug::{debug_arc_string_cache_identity, debug_cache_summary};
 use crate::debug_println;
 use crate::disk_cache::L2DiskCache;
 
 use super::config::GlobalCacheConfig;
 use super::storage_resolver::GLOBAL_STORAGE_RESOLVER;
 
-/// Global SearcherContext components
-/// These are the shared caches that should be reused across all searcher instances
-/// We use Arc to share these non-clonable types across multiple SearcherContext instances
+/// Global SearcherContext components.
+///
+/// Holds the process-wide split cache, L2 disk cache, and the cache/limit sizing
+/// derived from `GlobalCacheConfig`. Every `SearcherContext` is built by
+/// `SearcherContext::new_without_invoker`, which constructs its *own* fast-field,
+/// footer, leaf-search, list-fields, permit-provider and aggregation-limit
+/// instances from the `SearcherConfig` we hand it — so the way `GlobalCacheConfig`
+/// knobs take effect is by being written into that `SearcherConfig` (see
+/// `build_searcher_config`), NOT by holding separate cache instances here. Sharing
+/// across searchers is achieved by there being a single cached `SearcherContext`
+/// (see `get_global_searcher_context`), plus one per credential set.
 pub struct GlobalSearcherComponents {
-    /// Fast fields cache - shared across all searchers
-    pub fast_fields_cache: Arc<dyn StorageCache>,
-    /// Split footer cache - shared across all searchers (wrapped in Arc for sharing)
-    pub split_footer_cache: Arc<MemorySizedCache<String>>,
-    /// Leaf search cache - shared across all searchers (wrapped in Arc for sharing)
-    pub leaf_search_cache: Arc<LeafSearchCache>,
-    /// List fields cache - shared across all searchers (wrapped in Arc for sharing)
-    pub list_fields_cache: Arc<ListFieldsCache>,
-    /// Search permit provider - manages concurrent searches (wrapped in Arc for sharing)
-    pub search_permit_provider: Arc<SearchPermitProvider>,
-    /// Aggregation limits guard - shared memory tracking
-    pub aggregation_limit: AggregationLimitsGuard,
     /// Split cache - caches entire split files on disk (optional)
     pub split_cache_opt: Option<Arc<SplitCache>>,
-    /// L2 disk cache - tiered persistent disk cache with compression (optional)
+    /// L2 disk cache - tiered persistent disk cache (optional)
     pub disk_cache: Option<Arc<L2DiskCache>>,
     /// Temp directory for split cache (kept alive to prevent cleanup)
     _temp_dir: Option<TempDir>,
-    /// Predicate cache capacity — stored so SearcherContext can be built with the right size
-    pub predicate_cache_capacity: ByteSize,
+    /// Cache/limit sizing written into every `SearcherConfig` we build, so the
+    /// Java-configured `GlobalCacheConfig` knobs actually size the live caches.
+    fast_field_cache_capacity: ByteSize,
+    split_footer_cache_capacity: ByteSize,
+    partial_request_cache_capacity: ByteSize,
+    predicate_cache_capacity: ByteSize,
+    max_concurrent_splits: usize,
+    aggregation_memory_limit: ByteSize,
+    aggregation_bucket_limit: u32,
+    warmup_memory_budget: ByteSize,
     /// Memory reservation for the predicate cache, held for the lifetime of the components
     _predicate_cache_reservation: Option<MemoryReservation>,
 }
@@ -57,37 +54,16 @@ impl GlobalSearcherComponents {
     pub fn new(config: GlobalCacheConfig) -> Self {
         debug_println!("RUST DEBUG: Creating new GlobalSearcherComponents");
 
-        // Create fast field cache
-        let fast_field_cache_config = CacheConfig::default_with_capacity(config.fast_field_cache_capacity);
-        let fast_fields_cache = Arc::new(QuickwitCache::new(&fast_field_cache_config));
-
-        // Create split footer cache (wrapped in Arc for sharing)
-        let split_footer_cache_config = CacheConfig::default_with_capacity(config.split_footer_cache_capacity);
-        let split_footer_cache = Arc::new(MemorySizedCache::from_config(
-            &split_footer_cache_config,
-            &STORAGE_METRICS.split_footer_cache,
-        ));
-        debug_arc_string_cache_identity(&split_footer_cache, "split_footer_cache");
-
-        // Create leaf search cache (wrapped in Arc for sharing)
-        let partial_cache_config = CacheConfig::default_with_capacity(config.partial_request_cache_capacity);
-        let leaf_search_cache = Arc::new(LeafSearchCache::new(&partial_cache_config));
-
-        // Create list fields cache (wrapped in Arc for sharing)
-        let list_fields_cache = Arc::new(ListFieldsCache::new(&partial_cache_config));
-
-        // Create sync search permit provider to avoid async channel conflicts
-        // Using new_sync() method that doesn't use async channels
-        let search_permit_provider = Arc::new(SearchPermitProvider::new_sync(
-            config.max_concurrent_splits,
-            config.warmup_memory_budget,
-        ));
-
-        // Create aggregation limits guard
-        let aggregation_limit = AggregationLimitsGuard::new(
-            Some(config.aggregation_memory_limit.as_u64()),
-            Some(config.aggregation_bucket_limit),
-        );
+        // Capture the cache/limit sizing so build_searcher_config can apply it. The
+        // actual cache instances are created inside each SearcherContext from the
+        // SearcherConfig these values produce.
+        let fast_field_cache_capacity = config.fast_field_cache_capacity;
+        let split_footer_cache_capacity = config.split_footer_cache_capacity;
+        let partial_request_cache_capacity = config.partial_request_cache_capacity;
+        let max_concurrent_splits = config.max_concurrent_splits;
+        let aggregation_memory_limit = config.aggregation_memory_limit;
+        let aggregation_bucket_limit = config.aggregation_bucket_limit;
+        let warmup_memory_budget = config.warmup_memory_budget;
 
         // Create SplitCache if configured
         let (split_cache_opt, temp_dir) = if let Some(limits) = config.split_cache_limits {
@@ -188,35 +164,53 @@ impl GlobalSearcherComponents {
         };
 
         Self {
-            fast_fields_cache,
-            split_footer_cache,
-            leaf_search_cache,
-            list_fields_cache,
-            search_permit_provider,
-            aggregation_limit,
             split_cache_opt,
             disk_cache,
             _temp_dir: temp_dir,
+            fast_field_cache_capacity,
+            split_footer_cache_capacity,
+            partial_request_cache_capacity,
             predicate_cache_capacity,
+            max_concurrent_splits,
+            aggregation_memory_limit,
+            aggregation_bucket_limit,
+            warmup_memory_budget,
             _predicate_cache_reservation: predicate_cache_reservation,
         }
     }
 
-    /// Build a SearcherConfig with the predicate cache capacity stored in this component set.
-    /// Use this for the default/credential contexts so they respect Java-configured cache sizes.
+    /// Configured aggregation limits `(memory_bytes, bucket_limit)` from the
+    /// Java-supplied `GlobalCacheConfig`, for code paths that build their own
+    /// `AggregationLimitsGuard` (which are not fed by the `SearcherContext`).
+    pub fn aggregation_limits(&self) -> (u64, u32) {
+        (self.aggregation_memory_limit.as_u64(), self.aggregation_bucket_limit)
+    }
+
+    /// Build a `SearcherConfig` carrying every cache/limit size from the
+    /// Java-supplied `GlobalCacheConfig`. `SearcherContext::new_without_invoker`
+    /// constructs its fast-field / footer / leaf-search / list-fields caches, its
+    /// concurrency permit provider, and its aggregation-memory guard from these
+    /// values, so this is what makes those knobs actually take effect.
     pub fn build_searcher_config(&self) -> SearcherConfig {
         let mut config = SearcherConfig::default();
+        config.fast_field_cache = CacheConfig::default_with_capacity(self.fast_field_cache_capacity);
+        config.split_footer_cache =
+            CacheConfig::default_with_capacity(self.split_footer_cache_capacity);
+        config.partial_request_cache =
+            CacheConfig::default_with_capacity(self.partial_request_cache_capacity);
         config.predicate_cache = CacheConfig::default_with_capacity(self.predicate_cache_capacity);
+        config.max_num_concurrent_split_searches = self.max_concurrent_splits;
+        config.aggregation_memory_limit = self.aggregation_memory_limit;
+        config.aggregation_bucket_limit = self.aggregation_bucket_limit;
+        config.warmup_memory_budget = self.warmup_memory_budget;
         config
     }
 
-    /// Create a SearcherContext from these global components
-    /// This ensures all SearcherContext instances share the same cache instances
-    /// FIXED: Now properly shares ALL cache instances including split_footer_cache
+    /// Create a SearcherContext sized from `build_searcher_config`. Cache *sharing*
+    /// across searchers comes from there being a single cached context (see
+    /// `get_global_searcher_context`) — each context still owns its cache instances.
     pub fn create_searcher_context(&self, searcher_config: SearcherConfig) -> Arc<SearcherContext> {
-        debug_println!("RUST DEBUG: Creating SearcherContext from SHARED global components");
-        debug_arc_string_cache_identity(&self.split_footer_cache, "split_footer_cache");
-        debug_cache_summary();
+        debug_println!("RUST DEBUG: Creating SearcherContext (sized from GlobalCacheConfig)");
 
         // Use SearcherContext::new_without_invoker which handles all required fields correctly
         Arc::new(SearcherContext::new_without_invoker(
diff --git a/native/src/global_cache/mod.rs b/native/src/global_cache/mod.rs
index e95d7ecc..d9837ae3 100644
--- a/native/src/global_cache/mod.rs
+++ b/native/src/global_cache/mod.rs
@@ -34,8 +34,8 @@ pub use metrics::{
 
 // Re-export from storage_resolver
 pub use storage_resolver::{
-    generate_storage_cache_key, get_configured_storage_resolver, get_configured_storage_resolver_async,
-    tracked_storage_resolve, GLOBAL_STORAGE_RESOLVER,
+    clear_storage_resolvers, generate_storage_cache_key, get_configured_storage_resolver,
+    get_configured_storage_resolver_async, tracked_storage_resolve, GLOBAL_STORAGE_RESOLVER,
 };
 
 // =====================================================================
@@ -99,6 +99,61 @@ fn get_disk_cache_holder() -> &'static std::sync::RwLock<Option<Arc<L2DiskCache>
     GLOBAL_DISK_CACHE.get_or_init(|| std::sync::RwLock::new(None))
 }
 
+/// Process-global registry of L2 disk caches keyed by root path. A disk cache is a
+/// heavy singleton (background writer + timer threads, its own `manifest.json`,
+/// `total_bytes` accounting). Two managers pointing at the same root must share one
+/// instance, otherwise they fight over the manifest (last-syncer-wins) and each only
+/// counts its own writes — so combined usage can blow past the configured limit
+/// before either evicts. This get-or-create keyed by root path guarantees one
+/// instance per path, mirroring how the L1 cache is a process singleton.
+static DISK_CACHES_BY_ROOT: OnceLock<
+    std::sync::Mutex<std::collections::HashMap<std::path::PathBuf, Arc<L2DiskCache>>>,
+> = OnceLock::new();
+
+fn get_disk_caches_by_root() -> &'static std::sync::Mutex<
+    std::collections::HashMap<std::path::PathBuf, Arc<L2DiskCache>>,
+> {
+    DISK_CACHES_BY_ROOT.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
+}
+
+/// Get-or-create the process-global L2 disk cache for a given root path. Returns the
+/// existing instance if one is already registered for `config.root_path`, otherwise
+/// constructs it. Returns `None` if construction fails.
+pub fn get_or_create_disk_cache(
+    config: crate::disk_cache::DiskCacheConfig,
+) -> Option<Arc<L2DiskCache>> {
+    let key = config.root_path.clone();
+    let mut map = get_disk_caches_by_root().lock().unwrap();
+    if let Some(existing) = map.get(&key) {
+        debug_println!(
+            "🟢 DISK_CACHE_REGISTRY: Reusing existing L2 disk cache for root {:?}",
+            key
+        );
+        return Some(existing.clone());
+    }
+    match L2DiskCache::new(config) {
+        Ok(cache) => {
+            map.insert(key, cache.clone());
+            Some(cache)
+        }
+        Err(e) => {
+            debug_println!("🔴 DISK_CACHE_REGISTRY: Failed to create L2 disk cache: {}", e);
+            None
+        }
+    }
+}
+
+/// Drop all entries from the by-root disk cache registry. Called from the
+/// last-manager-close cleanup so the disk caches (and their background threads) are
+/// released for test isolation.
+pub fn clear_disk_cache_registry() {
+    let mut map = get_disk_caches_by_root().lock().unwrap();
+    if !map.is_empty() {
+        debug_println!("🔴 DISK_CACHE_REGISTRY: Clearing {} disk cache(s)", map.len());
+        map.clear();
+    }
+}
+
 /// Set the global L2 disk cache (called by SplitCacheManager when TieredCacheConfig is provided)
 pub fn set_global_disk_cache(cache: Arc<L2DiskCache>) {
     let holder = get_disk_cache_holder();
@@ -288,6 +343,14 @@ pub fn get_credential_searcher_context(credential_key: &str) -> Arc<SearcherCont
     context
 }
 
+/// Configured aggregation limits `(memory_bytes, bucket_limit)` from the global
+/// cache config. Used by aggregation code paths that build their own
+/// `AggregationLimitsGuard` so the Java-configured limits actually apply instead of
+/// a hardcoded default.
+pub fn get_configured_aggregation_limits() -> (u64, u32) {
+    get_global_components().aggregation_limits()
+}
+
 /// Get a SearcherContext with custom configuration but using global caches
 pub fn get_searcher_context_with_config(searcher_config: SearcherConfig) -> Arc<SearcherContext> {
     debug_println!("RUST DEBUG: Getting SearcherContext with custom config");
diff --git a/native/src/global_cache/storage_resolver.rs b/native/src/global_cache/storage_resolver.rs
index b6e526dd..41616113 100644
--- a/native/src/global_cache/storage_resolver.rs
+++ b/native/src/global_cache/storage_resolver.rs
@@ -1,31 +1,37 @@
 // storage_resolver.rs - Storage resolver caching functions
 // Extracted from global_cache.rs during refactoring
 
-use std::collections::HashMap;
-use std::sync::Arc;
+use std::collections::hash_map::DefaultHasher;
+use std::collections::{HashMap, VecDeque};
+use std::hash::{Hash, Hasher};
+use std::sync::{Arc, Mutex};
 
 use once_cell::sync::Lazy;
-use quickwit_config::{AzureStorageConfig, S3StorageConfig, StorageConfigs};
+use quickwit_config::{AzureStorageConfig, S3StorageConfig, StorageConfig, StorageConfigs};
 use quickwit_storage::StorageResolver;
-use tokio::sync::RwLock as TokioRwLock;
 
 use crate::debug_println;
 
-/// Generate a cache key for storage resolver that includes all credential components.
-/// This ensures different credentials result in different cached StorageResolver instances.
+/// Maximum number of distinct configured `StorageResolver`s to keep cached. Each
+/// resolver holds S3/Azure clients with connection pools, so an unbounded cache
+/// leaks clients under credential rotation (e.g. Spark STS session tokens, which
+/// mint a new key on every refresh). A bounded FIFO keeps reuse for the hot set
+/// while ensuring stale rotated credentials are eventually dropped.
+const MAX_CACHED_RESOLVERS: usize = 64;
+
+/// Generate a cache key for a storage resolver that includes all credential
+/// components. Different credentials therefore map to different cached resolvers,
+/// and — critically — a config carrying BOTH S3 and Azure credentials produces a
+/// key distinct from either alone, so mixed-cloud resolvers are not aliased onto an
+/// S3-only (or Azure-only) cached instance.
 ///
-/// The key includes:
-/// - For S3: region, endpoint, full access_key_id, force_path_style, and a hash of secret+session_token
-/// - For Azure: account_name and a hash of access_key+bearer_token
-///
-/// Sensitive credentials (secrets, tokens) are hashed to avoid leaking to logs while still
-/// ensuring different credentials produce different cache keys.
+/// Sensitive credentials (secrets, tokens) are hashed so they never reach logs while
+/// still distinguishing different credential sets.
 pub fn generate_storage_cache_key(
     s3_config: Option<&S3StorageConfig>,
     azure_config: Option<&AzureStorageConfig>,
 ) -> String {
-    use std::collections::hash_map::DefaultHasher;
-    use std::hash::{Hash, Hasher};
+    let mut parts: Vec<String> = Vec::new();
 
     if let Some(s3) = s3_config {
         let mut hasher = DefaultHasher::new();
@@ -33,27 +39,33 @@ pub fn generate_storage_cache_key(
         s3.session_token.hash(&mut hasher);
         let cred_hash = hasher.finish();
 
-        format!(
+        parts.push(format!(
             "s3:{}:{}:{}:{}:{:x}",
             s3.region.as_deref().unwrap_or("default"),
             s3.endpoint.as_deref().unwrap_or("default"),
             s3.access_key_id.as_deref().unwrap_or("none"),
             s3.force_path_style_access,
             cred_hash
-        )
-    } else if let Some(az) = azure_config {
+        ));
+    }
+
+    if let Some(az) = azure_config {
         let mut hasher = DefaultHasher::new();
         az.access_key.hash(&mut hasher);
         az.bearer_token.hash(&mut hasher);
         let cred_hash = hasher.finish();
 
-        format!(
+        parts.push(format!(
             "azure:{}:{:x}",
             az.account_name.as_deref().unwrap_or("default"),
             cred_hash
-        )
-    } else {
+        ));
+    }
+
+    if parts.is_empty() {
         "global".to_string()
+    } else {
+        parts.join("|")
     }
 }
 
@@ -65,11 +77,114 @@ pub static GLOBAL_STORAGE_RESOLVER: Lazy<StorageResolver> = Lazy::new(|| {
     StorageResolver::configured(&storage_configs)
 });
 
-/// Global storage resolver cache for configured S3 instances (async-compatible)
-/// Uses tokio::sync::RwLock to prevent deadlocks in async context
-static CONFIGURED_STORAGE_RESOLVERS: std::sync::OnceLock<
-    TokioRwLock<std::collections::HashMap<String, StorageResolver>>,
-> = std::sync::OnceLock::new();
+/// Bounded FIFO cache of configured resolvers, keyed by credential cache key.
+struct ResolverCache {
+    map: HashMap<String, StorageResolver>,
+    order: VecDeque<String>,
+}
+
+impl ResolverCache {
+    fn new() -> Self {
+        Self {
+            map: HashMap::new(),
+            order: VecDeque::new(),
+        }
+    }
+
+    fn get(&self, key: &str) -> Option<StorageResolver> {
+        self.map.get(key).cloned()
+    }
+
+    fn insert(&mut self, key: String, resolver: StorageResolver) {
+        if self.map.contains_key(&key) {
+            return;
+        }
+        // Evict oldest entries until there is room for the newcomer.
+        while self.order.len() >= MAX_CACHED_RESOLVERS {
+            if let Some(evicted) = self.order.pop_front() {
+                self.map.remove(&evicted);
+                debug_println!("♻️  STORAGE_RESOLVER_EVICT: dropped cached resolver '{}'", evicted);
+            } else {
+                break;
+            }
+        }
+        self.order.push_back(key.clone());
+        self.map.insert(key, resolver);
+    }
+
+    fn clear(&mut self) {
+        self.map.clear();
+        self.order.clear();
+    }
+}
+
+/// Single process-global resolver cache shared by both the sync and async entry
+/// points (previously two independent maps, so the same credential could be
+/// materialized twice). A `std::sync::Mutex` is safe in async callers here because
+/// no `.await` happens while the lock is held — resolver construction occurs after
+/// the lock is released.
+static STORAGE_RESOLVERS: Lazy<Arc<Mutex<ResolverCache>>> =
+    Lazy::new(|| Arc::new(Mutex::new(ResolverCache::new())));
+
+/// Drop all cached configured resolvers. Called from the last-manager-close cleanup
+/// so rotated credentials and their clients don't survive between test runs.
+pub fn clear_storage_resolvers() {
+    let mut cache = STORAGE_RESOLVERS.lock().unwrap();
+    if !cache.map.is_empty() {
+        debug_println!("🧹 CLEAR_STORAGE_RESOLVERS: dropping {} cached resolver(s)", cache.map.len());
+    }
+    cache.clear();
+}
+
+/// Core get-or-create shared by the sync and async wrappers. Builds a resolver from
+/// BOTH the S3 and Azure configs when present (so mixed-cloud operation is
+/// authenticated for both), caches it under a combined credential key, and reuses
+/// the process-global unconfigured resolver when neither is supplied.
+fn get_or_create_resolver(
+    s3_config_opt: Option<S3StorageConfig>,
+    azure_config_opt: Option<AzureStorageConfig>,
+) -> StorageResolver {
+    if s3_config_opt.is_none() && azure_config_opt.is_none() {
+        return GLOBAL_STORAGE_RESOLVER.clone();
+    }
+
+    let cache_key = generate_storage_cache_key(s3_config_opt.as_ref(), azure_config_opt.as_ref());
+
+    // Fast path: return a cached resolver if present.
+    {
+        let cache = STORAGE_RESOLVERS.lock().unwrap();
+        if let Some(resolver) = cache.get(&cache_key) {
+            debug_println!("🎯 STORAGE_RESOLVER_CACHE_HIT: reusing resolver for key: {}", cache_key);
+            return resolver;
+        }
+    }
+
+    // Build the resolver outside the lock (construction may be non-trivial).
+    let mut configs: Vec<StorageConfig> = Vec::new();
+    if let Some(s3) = s3_config_opt {
+        debug_println!(
+            "   📋 S3 Config: region={:?}, endpoint={:?}, path_style={}",
+            s3.region, s3.endpoint, s3.force_path_style_access
+        );
+        configs.push(StorageConfig::S3(s3));
+    }
+    if let Some(az) = azure_config_opt {
+        debug_println!("   📋 Azure Config: account={:?}", az.account_name);
+        configs.push(StorageConfig::Azure(az));
+    }
+    let resolver = StorageResolver::configured(&StorageConfigs::new(configs));
+    debug_println!("✅ STORAGE_RESOLVER_CREATED: new resolver for key: {}", cache_key);
+
+    // Insert, double-checking for a concurrent creator.
+    let mut cache = STORAGE_RESOLVERS.lock().unwrap();
+    if let Some(existing) = cache.get(&cache_key) {
+        debug_println!("🏃 STORAGE_RESOLVER_RACE: using resolver created by another thread");
+        return existing;
+    }
+    cache.insert(cache_key.clone(), resolver.clone());
+    debug_println!("💾 STORAGE_RESOLVER_CACHED: resolver cached for key: {}", cache_key);
+    resolver
+}
 
 /// Helper function to track storage instance creation for debugging
 /// This helps us understand when and where multiple storage instances are created
@@ -78,360 +193,39 @@ pub async fn tracked_storage_resolve(
     uri: &quickwit_common::uri::Uri,
     context: &str,
 ) -> Result<Arc<dyn quickwit_storage::Storage>, quickwit_storage::StorageResolverError> {
-    static STORAGE_COUNTER: std::sync::atomic::AtomicU32 =
-        std::sync::atomic::AtomicU32::new(1);
-    let storage_id =
-        STORAGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    static STORAGE_COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(1);
+    let storage_id = STORAGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
     debug_println!(
         "🏗️  STORAGE_RESOLVE: Starting storage resolve #{} [{}]",
-        storage_id,
-        context
+        storage_id, context
     );
-    debug_println!("   📍 Resolver address: {:p}", resolver);
     debug_println!("   🌐 URI: {}", uri);
 
-    let resolve_start = std::time::Instant::now();
     let result = resolver.resolve(uri).await;
-
     match &result {
-        Ok(storage) => {
-            debug_println!(
-                "✅ STORAGE_RESOLVED: Storage instance #{} created in {}ms [{}]",
-                storage_id,
-                resolve_start.elapsed().as_millis(),
-                context
-            );
-            debug_println!("   🏭 Storage address: {:p}", &**storage);
-            debug_println!(
-                "   📊 Storage type: {}",
-                std::any::type_name::<dyn quickwit_storage::Storage>()
-            );
-        }
-        Err(e) => {
-            debug_println!(
-                "❌ STORAGE_RESOLVE_FAILED: Storage resolve #{} failed in {}ms [{}]: {}",
-                storage_id,
-                resolve_start.elapsed().as_millis(),
-                context,
-                e
-            );
-        }
+        Ok(_) => debug_println!("✅ STORAGE_RESOLVED: #{} [{}]", storage_id, context),
+        Err(e) => debug_println!("❌ STORAGE_RESOLVE_FAILED: #{} [{}]: {}", storage_id, context, e),
     }
-
     result
 }
 
-/// Get or create a cached StorageResolver with specific S3/Azure credentials (async version)
-/// This follows Quickwit's pattern but enables caching for optimal storage instance reuse
-///
-/// 🚨 CRITICAL: This function should be used for ALL storage resolver creation in ASYNC contexts
-/// to ensure consistent cache sharing. Direct calls to StorageResolver::configured()
-/// bypass caching and cause multiple storage instances.
+/// Get or create a cached StorageResolver (async wrapper).
 ///
-/// ✅ FIXED: Async-compatible using tokio::sync::RwLock to prevent deadlocks
+/// 🚨 Use this (or the sync variant) for ALL configured storage resolver creation so
+/// resolvers are shared. Direct `StorageResolver::configured()` calls bypass the
+/// cache and create redundant storage instances.
 pub async fn get_configured_storage_resolver_async(
     s3_config_opt: Option<S3StorageConfig>,
     azure_config_opt: Option<AzureStorageConfig>,
 ) -> StorageResolver {
-    static RESOLVER_COUNTER: std::sync::atomic::AtomicU32 =
-        std::sync::atomic::AtomicU32::new(1);
-
-    if let Some(s3_config) = s3_config_opt {
-        // Create a cache key that includes ALL credential components
-        let cache_key = generate_storage_cache_key(Some(&s3_config), None);
-
-        // ✅ FIXED: Use async tokio RwLock to prevent deadlocks in async context
-        // Initialize cache if needed
-        let cache = CONFIGURED_STORAGE_RESOLVERS
-            .get_or_init(|| TokioRwLock::new(std::collections::HashMap::new()));
-
-        // Try to get from cache first (async read lock) - DEADLOCK FIXED
-        {
-            let read_cache = cache.read().await;
-            if let Some(cached_resolver) = read_cache.get(&cache_key) {
-                debug_println!(
-                    "🎯 STORAGE_RESOLVER_CACHE_HIT: Reusing cached resolver for key: {} at address {:p}",
-                    cache_key,
-                    cached_resolver
-                );
-                return cached_resolver.clone();
-            }
-        } // <- async read lock released here
-
-        // Cache miss - create new resolver (outside of any locks to prevent deadlock)
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "❌ STORAGE_RESOLVER_CACHE_MISS: Creating new resolver #{} for key: {}",
-            resolver_id,
-            cache_key
-        );
-        debug_println!(
-            "   📋 S3 Config: region={:?}, endpoint={:?}, path_style={}",
-            s3_config.region,
-            s3_config.endpoint,
-            s3_config.force_path_style_access
-        );
-
-        let storage_configs =
-            StorageConfigs::new(vec![quickwit_config::StorageConfig::S3(s3_config)]);
-        let resolver = StorageResolver::configured(&storage_configs); // <- DEADLOCK SUSPECT: Quickwit resolver creation
-        debug_println!(
-            "✅ STORAGE_RESOLVER_CREATED: Resolver #{} created at address {:p}",
-            resolver_id,
-            &resolver
-        );
-
-        // Cache the new resolver (async write lock) - DEADLOCK FIXED
-        {
-            let mut write_cache = cache.write().await;
-            // Double-check in case another thread created it while we were creating ours
-            if let Some(existing_resolver) = write_cache.get(&cache_key) {
-                debug_println!(
-                    "🏃 STORAGE_RESOLVER_RACE: Another thread created resolver, using existing at {:p}",
-                    existing_resolver
-                );
-                return existing_resolver.clone();
-            }
-            write_cache.insert(cache_key.clone(), resolver.clone());
-            debug_println!(
-                "💾 STORAGE_RESOLVER_CACHED: Resolver #{} cached for key: {}",
-                resolver_id,
-                cache_key
-            );
-        } // <- async write lock released here
-
-        resolver
-    } else if let Some(azure_config) = azure_config_opt {
-        // Create a cache key that includes ALL credential components
-        let cache_key = generate_storage_cache_key(None, Some(&azure_config));
-
-        let cache = CONFIGURED_STORAGE_RESOLVERS
-            .get_or_init(|| TokioRwLock::new(HashMap::new()));
-
-        // Try read lock first (async)
-        {
-            let read_cache = cache.read().await;
-            if let Some(cached_resolver) = read_cache.get(&cache_key) {
-                debug_println!(
-                    "🎯 AZURE_RESOLVER_CACHE_HIT: Reusing resolver for key: {}",
-                    cache_key
-                );
-                return cached_resolver.clone();
-            }
-        } // <- Lock released immediately
-
-        // Create resolver OUTSIDE lock (deadlock prevention)
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "❌ AZURE_RESOLVER_CACHE_MISS: Creating resolver #{} for key: {}",
-            resolver_id,
-            cache_key
-        );
-        debug_println!(
-            "   📋 Azure Config: account={:?}",
-            azure_config.account_name
-        );
-
-        let storage_configs = StorageConfigs::new(vec![
-            quickwit_config::StorageConfig::Azure(azure_config),
-        ]);
-        let resolver = StorageResolver::configured(&storage_configs);
-        debug_println!(
-            "✅ AZURE_RESOLVER_CREATED: Resolver #{} at {:p}",
-            resolver_id,
-            &resolver
-        );
-
-        // Cache with write lock (async)
-        {
-            let mut write_cache = cache.write().await;
-            // Double-check for race condition
-            if let Some(existing_resolver) = write_cache.get(&cache_key) {
-                debug_println!(
-                    "🏃 AZURE_RESOLVER_RACE: Using existing at {:p}",
-                    existing_resolver
-                );
-                return existing_resolver.clone();
-            }
-            write_cache.insert(cache_key.clone(), resolver.clone());
-            debug_println!(
-                "💾 AZURE_RESOLVER_CACHED: Resolver #{} cached",
-                resolver_id
-            );
-        } // <- Lock released immediately
-
-        resolver
-    } else {
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "🌐 STORAGE_RESOLVER_GLOBAL: Resolver #{} - Using global unconfigured StorageResolver",
-            resolver_id
-        );
-        let resolver = GLOBAL_STORAGE_RESOLVER.clone();
-        debug_println!(
-            "♻️  STORAGE_RESOLVER_REUSED: Resolver #{} reused global at address {:p}",
-            resolver_id,
-            &resolver
-        );
-        resolver
-    }
+    get_or_create_resolver(s3_config_opt, azure_config_opt)
 }
 
-/// Get or create a cached StorageResolver with specific S3/Azure credentials (sync version)
-/// This version uses a simple sync-safe caching approach for sync contexts
-///
-/// ✅ FIXED: Now uses deadlock-safe sync caching approach
+/// Get or create a cached StorageResolver (sync).
 pub fn get_configured_storage_resolver(
     s3_config_opt: Option<S3StorageConfig>,
     azure_config_opt: Option<AzureStorageConfig>,
 ) -> StorageResolver {
-    use once_cell::sync::Lazy;
-    use std::sync::{Arc, Mutex};
-
-    static SYNC_STORAGE_RESOLVERS: Lazy<
-        Arc<Mutex<std::collections::HashMap<String, StorageResolver>>>,
-    > = Lazy::new(|| Arc::new(Mutex::new(std::collections::HashMap::new())));
-    static RESOLVER_COUNTER: std::sync::atomic::AtomicU32 =
-        std::sync::atomic::AtomicU32::new(1);
-
-    if let Some(s3_config) = s3_config_opt {
-        // Create a cache key that includes ALL credential components
-        let cache_key = generate_storage_cache_key(Some(&s3_config), None);
-
-        // Try to get from sync cache (simple mutex approach)
-        {
-            let cache = SYNC_STORAGE_RESOLVERS.lock().unwrap();
-            if let Some(cached_resolver) = cache.get(&cache_key) {
-                debug_println!(
-                    "🎯 STORAGE_RESOLVER_SYNC_CACHE_HIT: Reusing cached sync resolver for key: {} at address {:p}",
-                    cache_key,
-                    cached_resolver
-                );
-                return cached_resolver.clone();
-            }
-        }
-
-        // Cache miss - create new resolver
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "❌ STORAGE_RESOLVER_SYNC_CACHE_MISS: Creating new sync resolver #{} for key: {}",
-            resolver_id,
-            cache_key
-        );
-        debug_println!(
-            "   📋 S3 Config: region={:?}, endpoint={:?}, path_style={}",
-            s3_config.region,
-            s3_config.endpoint,
-            s3_config.force_path_style_access
-        );
-
-        let storage_configs =
-            StorageConfigs::new(vec![quickwit_config::StorageConfig::S3(s3_config)]);
-        let resolver = StorageResolver::configured(&storage_configs);
-        debug_println!(
-            "✅ STORAGE_RESOLVER_CREATED: Sync resolver #{} created at address {:p}",
-            resolver_id,
-            &resolver
-        );
-
-        // Cache the new resolver
-        {
-            let mut cache = SYNC_STORAGE_RESOLVERS.lock().unwrap();
-            // Double-check in case another thread created it while we were creating ours
-            if let Some(existing_resolver) = cache.get(&cache_key) {
-                debug_println!(
-                    "🏃 STORAGE_RESOLVER_SYNC_RACE: Another thread created sync resolver, using existing at {:p}",
-                    existing_resolver
-                );
-                return existing_resolver.clone();
-            }
-            cache.insert(cache_key.clone(), resolver.clone());
-            debug_println!(
-                "💾 STORAGE_RESOLVER_SYNC_CACHED: Resolver #{} cached for key: {}",
-                resolver_id,
-                cache_key
-            );
-        }
-
-        resolver
-    } else if let Some(azure_config) = azure_config_opt {
-        // Create a cache key that includes ALL credential components
-        let cache_key = generate_storage_cache_key(None, Some(&azure_config));
-
-        // Try to get from sync cache
-        {
-            let cache = SYNC_STORAGE_RESOLVERS.lock().unwrap();
-            if let Some(cached_resolver) = cache.get(&cache_key) {
-                debug_println!(
-                    "🎯 AZURE_RESOLVER_SYNC_CACHE_HIT: Reusing cached sync resolver for key: {} at address {:p}",
-                    cache_key,
-                    cached_resolver
-                );
-                return cached_resolver.clone();
-            }
-        }
-
-        // Cache miss - create new resolver
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "❌ AZURE_RESOLVER_SYNC_CACHE_MISS: Creating new sync resolver #{} for key: {}",
-            resolver_id,
-            cache_key
-        );
-        debug_println!(
-            "   📋 Azure Config: account={:?}",
-            azure_config.account_name
-        );
-
-        let storage_configs = StorageConfigs::new(vec![
-            quickwit_config::StorageConfig::Azure(azure_config),
-        ]);
-        let resolver = StorageResolver::configured(&storage_configs);
-        debug_println!(
-            "✅ AZURE_RESOLVER_CREATED: Sync resolver #{} created at address {:p}",
-            resolver_id,
-            &resolver
-        );
-
-        // Cache the new resolver
-        {
-            let mut cache = SYNC_STORAGE_RESOLVERS.lock().unwrap();
-            // Double-check in case another thread created it while we were creating ours
-            if let Some(existing_resolver) = cache.get(&cache_key) {
-                debug_println!(
-                    "🏃 AZURE_RESOLVER_SYNC_RACE: Another thread created sync resolver, using existing at {:p}",
-                    existing_resolver
-                );
-                return existing_resolver.clone();
-            }
-            cache.insert(cache_key.clone(), resolver.clone());
-            debug_println!(
-                "💾 AZURE_RESOLVER_SYNC_CACHED: Resolver #{} cached for key: {}",
-                resolver_id,
-                cache_key
-            );
-        }
-
-        resolver
-    } else {
-        let resolver_id =
-            RESOLVER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        debug_println!(
-            "🌐 STORAGE_RESOLVER_GLOBAL_SYNC: Resolver #{} - Using global unconfigured StorageResolver",
-            resolver_id
-        );
-        let resolver = GLOBAL_STORAGE_RESOLVER.clone();
-        debug_println!(
-            "♻️  STORAGE_RESOLVER_REUSED_SYNC: Resolver #{} reused global at address {:p}",
-            resolver_id,
-            &resolver
-        );
-        resolver
-    }
+    get_or_create_resolver(s3_config_opt, azure_config_opt)
 }
diff --git a/native/src/memory_pool/jvm_pool.rs b/native/src/memory_pool/jvm_pool.rs
index bc751410..00ea8de2 100644
--- a/native/src/memory_pool/jvm_pool.rs
+++ b/native/src/memory_pool/jvm_pool.rs
@@ -313,14 +313,25 @@ impl MemoryPool for JvmMemoryPool {
 
         // Check if we need more from JVM
         if self.needs_jvm_acquire(new_used) {
-            let want = std::cmp::max(size, self.config.acquire_increment);
+            // Target the actual deficit (usage above the current grant), not just
+            // this call's `size`. After a denial-then-partial-grant, usage can sit
+            // above the grant; topping up by only `size` would leave it there and
+            // require many incremental acquires to catch up.
+            let deficit = new_used.saturating_sub(self.jvm_granted.load(Relaxed));
+            let want = std::cmp::max(deficit, self.config.acquire_increment);
 
             match self.jni_acquire(want) {
-                Ok(acquired) if acquired >= size => {
+                // Success requires covering the DEFICIT, not this call's `size`: the
+                // existing grant already backs `new_used - deficit`, so acquiring
+                // `deficit` more makes `granted == new_used`. Checking against `size`
+                // here would spuriously deny whenever prior grant already covered part
+                // of the usage (deficit < size). When deficit == 0 (a high-watermark
+                // headroom top-up while still within grant) any result is fine.
+                Ok(acquired) if acquired >= deficit => {
                     self.jvm_granted.fetch_add(acquired, Relaxed);
                 }
                 Ok(acquired) => {
-                    // JVM gave us less than we need — release what we got, undo, fail
+                    // JVM gave us less than the deficit — release what we got, undo, fail
                     if acquired > 0 {
                         let _ = self.jni_release(acquired);
                     }
@@ -368,7 +379,13 @@ impl MemoryPool for JvmMemoryPool {
         // subsequent operations or on pool shutdown.
         if let Some(excess) = self.should_jvm_release() {
             let to_release = excess.min(size);
-            if to_release > 0 {
+            // Skip churny sub-`min_release_amount` JNI releases. The min-release
+            // threshold is meant to bound how often we call into the JVM, but the
+            // `excess.min(size)` cap can shrink an above-threshold `excess` down to a
+            // tiny `to_release`; guard on `to_release` itself. Still release when the
+            // pool has fully drained (used == 0), so the last grant is returned.
+            let fully_drained = self.rust_used.load(Relaxed) == 0;
+            if to_release > 0 && (to_release >= self.config.min_release_amount || fully_drained) {
                 if let Ok(()) = self.jni_release(to_release) {
                     self.jvm_granted.fetch_sub(to_release, Relaxed);
                 }
diff --git a/native/src/persistent_cache_storage.rs b/native/src/persistent_cache_storage.rs
index efbeef6d..ba354566 100644
--- a/native/src/persistent_cache_storage.rs
+++ b/native/src/persistent_cache_storage.rs
@@ -140,12 +140,22 @@ impl StorageWithPersistentCache {
     /// The segments may not cover the full requested range if there are gaps,
     /// but when called from fully_cached path, they should be contiguous.
     fn combine_segments(segments: &[CachedSegment], requested: &Range<u64>) -> OwnedBytes {
+        let total_len = (requested.end - requested.start) as usize;
+
+        // Fast path: a single segment that exactly covers the request (the common
+        // full-hit case). Guard on both start position and length so we never return
+        // a buffer shorter than the requested range.
         if segments.len() == 1 {
-            // Single segment - just return it
-            return segments[0].data.clone();
+            let seg = &segments[0];
+            if seg.range.start == requested.start && seg.data.len() == total_len {
+                return seg.data.clone();
+            }
         }
 
-        let total_len = (requested.end - requested.start) as usize;
+        // General path: assemble into a correctly-sized buffer. This is only reached
+        // on the fully_cached path, where the coalescer guarantees the segments tile
+        // the whole request with no holes; the length-correct buffer is defensive
+        // insurance against any residual coverage gap.
         let mut result = vec![0u8; total_len];
 
         for seg in segments {
diff --git a/native/src/searcher/aggregation/deserialize.rs b/native/src/searcher/aggregation/deserialize.rs
index 488da820..6a083deb 100644
--- a/native/src/searcher/aggregation/deserialize.rs
+++ b/native/src/searcher/aggregation/deserialize.rs
@@ -88,7 +88,8 @@ pub(crate) fn deserialize_aggregation_results(
         postcard::from_bytes(intermediate_agg_bytes)?;
 
     // Step 3: Create aggregation limits (using reasonable defaults like Quickwit)
-    let aggregation_limits = AggregationLimitsGuard::new(Some(50_000_000), Some(65_000));
+    let (agg_mem_limit, agg_bucket_limit) = crate::global_cache::get_configured_aggregation_limits();
+    let aggregation_limits = AggregationLimitsGuard::new(Some(agg_mem_limit), Some(agg_bucket_limit));
 
     // Step 4: Convert to final results using Quickwit's proven method
     let final_results: AggregationResults =
@@ -329,10 +330,8 @@ pub(crate) fn find_specific_aggregation_result(
         };
 
     // Step 3: Create aggregation limits (using reasonable defaults like Quickwit)
-    let aggregation_limits = AggregationLimitsGuard::new(
-        Some(50_000_000), // 50MB memory limit
-        Some(65_000),     // 65k bucket limit
-    );
+    let (agg_mem_limit, agg_bucket_limit) = crate::global_cache::get_configured_aggregation_limits();
+    let aggregation_limits = AggregationLimitsGuard::new(Some(agg_mem_limit), Some(agg_bucket_limit));
 
     // Step 4: Convert to final results using Quickwit's proven method
     let final_results: AggregationResults =
diff --git a/native/src/split_cache_manager/jni_cache_ops.rs b/native/src/split_cache_manager/jni_cache_ops.rs
index d14f7acf..33b6a162 100644
--- a/native/src/split_cache_manager/jni_cache_ops.rs
+++ b/native/src/split_cache_manager/jni_cache_ops.rs
@@ -199,14 +199,12 @@ pub extern "system" fn Java_io_indextables_tantivy4java_split_SplitCacheManager_
         None => return,
     };
 
-    // THEN access the cache managers registry
+    // Validate the manager exists. Manager-level preloading is a no-op: real
+    // component prewarming is driven per-split via SplitSearcher.preloadComponents.
     let managers = CACHE_MANAGERS.lock().unwrap();
-    let manager = match managers.get(&cache_name) {
-        Some(manager_arc) => manager_arc,
-        None => return,
-    };
-    // Simulate preloading by updating cache stats
-    manager.current_size.fetch_add(1024, Ordering::Relaxed);
+    if managers.get(&cache_name).is_none() {
+        return;
+    }
 }
 
 #[no_mangle]
diff --git a/native/src/split_cache_manager/jni_lifecycle.rs b/native/src/split_cache_manager/jni_lifecycle.rs
index 6a1cf69e..0400c8cc 100644
--- a/native/src/split_cache_manager/jni_lifecycle.rs
+++ b/native/src/split_cache_manager/jni_lifecycle.rs
@@ -7,7 +7,7 @@ use std::sync::Arc;
 use jni::objects::{JClass, JObject, JString};
 use jni::sys::jlong;
 use jni::JNIEnv;
-use quickwit_config::S3StorageConfig;
+use quickwit_config::{AzureStorageConfig, S3StorageConfig};
 
 use crate::debug_println;
 use crate::disk_cache::{CompressionAlgorithm, DiskCacheConfig, WriteQueueMode};
@@ -112,6 +112,52 @@ pub extern "system" fn Java_io_indextables_tantivy4java_split_SplitCacheManager_
         }
     }
 
+    // Extract Azure configuration from the Java CacheConfig so manager-level storage
+    // operations (e.g. searchAcrossAllSplits via get_storage_resolver) are
+    // authenticated for azure:// splits, not just the per-split search path.
+    if let Ok(azure_config_map) = env.call_method(&config, "getAzureConfig", "()Ljava/util/Map;", &[]) {
+        if let Ok(azure_map_obj) = azure_config_map.l() {
+            if !azure_map_obj.is_null() {
+                let extract = |env: &mut JNIEnv, key: &str| -> Option<String> {
+                    let key_jstring = env.new_string(key).ok()?;
+                    let value = env
+                        .call_method(
+                            &azure_map_obj,
+                            "get",
+                            "(Ljava/lang/Object;)Ljava/lang/Object;",
+                            &[(&key_jstring).into()],
+                        )
+                        .ok()?
+                        .l()
+                        .ok()?;
+                    if value.is_null() {
+                        return None;
+                    }
+                    let value_jstring = JString::from(value);
+                    let value_string = env.get_string(&value_jstring).ok()?;
+                    Some(value_string.to_string_lossy().to_string())
+                };
+
+                let account_name = extract(&mut env, "account_name");
+                let access_key = extract(&mut env, "access_key");
+                let bearer_token = extract(&mut env, "bearer_token");
+
+                if account_name.is_some() && (access_key.is_some() || bearer_token.is_some()) {
+                    debug_println!(
+                        "RUST DEBUG: Configuring Azure storage (account={:?}, auth={})",
+                        account_name,
+                        if bearer_token.is_some() { "bearer_token" } else { "account_key" }
+                    );
+                    manager.set_azure_config(AzureStorageConfig {
+                        account_name,
+                        access_key,
+                        bearer_token,
+                    });
+                }
+            }
+        }
+    }
+
     // Extract TieredCacheConfig from the Java CacheConfig
     debug_println!("RUST DEBUG: Attempting to extract TieredCacheConfig");
     if let Ok(tiered_config_result) = env.call_method(
diff --git a/native/src/split_cache_manager/manager.rs b/native/src/split_cache_manager/manager.rs
index 2e3a8415..e7484dbb 100644
--- a/native/src/split_cache_manager/manager.rs
+++ b/native/src/split_cache_manager/manager.rs
@@ -4,14 +4,13 @@
 #![allow(dead_code)]
 
 use std::collections::HashMap;
-use std::path::PathBuf;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use quickwit_config::{AzureStorageConfig, S3StorageConfig};
 
 use crate::debug_println;
-use crate::disk_cache::{CompressionAlgorithm, DiskCacheConfig, L2DiskCache};
+use crate::disk_cache::{DiskCacheConfig, L2DiskCache};
 use crate::global_cache::{get_configured_storage_resolver, get_global_searcher_context};
 use crate::batch_retrieval::simple::{BatchOptimizationMetrics, PrefetchStats};
 
@@ -35,11 +34,10 @@ pub struct GlobalSplitCacheManager {
     // L2 Disk cache for persistent caching
     pub(crate) disk_cache: Option<Arc<L2DiskCache>>,
 
-    // Statistics
-    pub(crate) total_hits: AtomicU64,
-    pub(crate) total_misses: AtomicU64,
+    // Count of flush/eviction operations triggered through this manager.
+    // (Hit/miss/size counters are NOT tracked here — those are process-global and
+    // reported by get_cache_stats() straight from quickwit's STORAGE_METRICS.)
     pub(crate) total_evictions: AtomicU64,
-    pub(crate) current_size: AtomicU64,
 
     // Managed splits
     pub(crate) managed_splits: Mutex<HashMap<String, u64>>, // split_path -> last_access_time
@@ -61,8 +59,19 @@ impl GlobalSplitCacheManager {
         // All async operations should use the shared global runtime via QuickwitRuntimeManager
         debug_println!("🔧 RUNTIME_FIX: Eliminating separate Tokio runtime to prevent deadlocks");
 
-        // Set the L1 cache capacity from Java's CacheConfig.withMaxCacheSize()
-        // This ensures the global shared L1 cache uses the Java-configured size
+        // Set the L1 cache capacity from Java's CacheConfig.withMaxCacheSize().
+        // The L1 cache is a process-global singleton built once (first-wins): a later
+        // manager configured with a different size cannot resize it. Warn so this
+        // process-global-first-wins behavior isn't a silent surprise.
+        let existing_l1_capacity = crate::global_cache::get_global_l1_cache_capacity();
+        if existing_l1_capacity != 0 && existing_l1_capacity != max_cache_size {
+            debug_println!(
+                "⚠️ L1_CACHE_CAPACITY: cache '{}' requested maxCacheSize={} but the process-global \
+                 L1 cache was already sized to {} by an earlier manager (first-wins); the requested \
+                 size is ignored.",
+                cache_name, max_cache_size, existing_l1_capacity
+            );
+        }
         crate::global_cache::set_l1_cache_capacity(max_cache_size);
 
         // Note: We're using the global caches from GLOBAL_SEARCHER_COMPONENTS
@@ -76,10 +85,7 @@ impl GlobalSplitCacheManager {
             s3_config: None,
             azure_config: None,
             disk_cache: None,
-            total_hits: AtomicU64::new(0),
-            total_misses: AtomicU64::new(0),
             total_evictions: AtomicU64::new(0),
-            current_size: AtomicU64::new(0),
             managed_splits: Mutex::new(HashMap::new()),
             file_field_stats: Mutex::new(HashMap::new()),
         }
@@ -93,22 +99,24 @@ impl GlobalSplitCacheManager {
             config.root_path
         );
 
-        match L2DiskCache::new(config) {
-            Ok(cache) => {
+        // Get-or-create keyed by root path so multiple managers configured with the
+        // same cache directory share one L2DiskCache instance (one manifest, one
+        // total_bytes accounting) instead of racing over the same files.
+        match crate::global_cache::get_or_create_disk_cache(config) {
+            Some(cache) => {
                 let stats = cache.stats();
                 debug_println!(
-                    "RUST DEBUG: L2DiskCache created successfully. Max size: {} bytes, {} splits cached",
+                    "RUST DEBUG: L2DiskCache ready. Max size: {} bytes, {} splits cached",
                     stats.max_bytes,
                     stats.split_count
                 );
-                // Also set the global disk cache so StandaloneSearcher can access it
+                // Also set the active global disk cache so StandaloneSearcher can access it
                 crate::global_cache::set_global_disk_cache(cache.clone());
                 self.disk_cache = Some(cache);
             }
-            Err(e) => {
+            None => {
                 debug_println!(
-                    "RUST WARNING: Failed to create L2DiskCache: {}. Continuing without disk cache.",
-                    e
+                    "RUST WARNING: Failed to create L2DiskCache. Continuing without disk cache."
                 );
             }
         }
@@ -165,6 +173,10 @@ impl GlobalSplitCacheManager {
         self.managed_splits.lock().unwrap().len()
     }
 
+    /// Returns cache statistics. NOTE: hits/misses/evictions/current_size are
+    /// **process-global** (aggregated from quickwit's `STORAGE_METRICS`), not
+    /// specific to this manager — only `max_size` is per-manager. In a process with
+    /// multiple managers these fields reflect the whole process.
     pub fn get_cache_stats(&self) -> GlobalCacheStats {
         // 🚀 OPTIMIZATION: Access real Quickwit cache metrics instead of basic counters
         // This provides comprehensive per-cache-type metrics with ByteRangeCache-specific tracking
@@ -284,10 +296,22 @@ impl GlobalSplitCacheManager {
         get_global_searcher_context()
     }
 
+    /// Flush the in-memory query-path caches. Backs the Java `flushAllCaches()` /
+    /// `flushCacheTypes(LEAF_SEARCH | BYTE_RANGE)` APIs.
+    ///
+    /// The leaf-search and split-footer caches live inside the global
+    /// `SearcherContext`; dropping it forces them to be rebuilt empty on the next
+    /// search. The L1 byte-range cache is cleared in place, and per-credential
+    /// contexts are dropped too. These caches don't support partial (target-size)
+    /// eviction, so this is a full clear regardless of `target_size_bytes`. The
+    /// persistent L2 disk cache is intentionally left intact.
     pub fn force_eviction(&self, _target_size_bytes: u64) {
-        // Simulate eviction by incrementing counter
+        crate::global_cache::clear_global_l1_cache();
+        crate::global_cache::clear_global_searcher_context();
+        crate::global_cache::clear_credential_contexts();
+        crate::split_query::clear_split_schema_cache();
+        crate::split_searcher::clear_searcher_cache();
         self.total_evictions.fetch_add(1, Ordering::Relaxed);
-        // In a real implementation, this would evict cache entries
     }
 }
 
@@ -359,8 +383,9 @@ fn clear_all_global_caches() {
     // 4. Clear searcher cache (LRU cache of Arc<Searcher>)
     crate::split_searcher::clear_searcher_cache();
 
-    // 5. Clear global disk cache reference
+    // 5. Clear global disk cache reference and the by-root disk cache registry
     crate::global_cache::clear_global_disk_cache();
+    crate::global_cache::clear_disk_cache_registry();
 
     // 6. Reset L1 cache (clear all entries)
     crate::global_cache::reset_global_l1_cache();
@@ -368,6 +393,9 @@ fn clear_all_global_caches() {
     // 7. Reset storage download metrics
     crate::global_cache::reset_storage_download_metrics();
 
+    // 8. Drop cached storage resolvers (releases S3/Azure clients)
+    crate::global_cache::clear_storage_resolvers();
+
     debug_println!("🧹 CLEAR_ALL_GLOBAL_CACHES: Complete");
 }
 
diff --git a/native/src/split_searcher/async_impl.rs b/native/src/split_searcher/async_impl.rs
index 5c3f1716..63b68a5f 100644
--- a/native/src/split_searcher/async_impl.rs
+++ b/native/src/split_searcher/async_impl.rs
@@ -942,8 +942,18 @@ async fn perform_real_quickwit_search(
     // Use cached storage directly (Quickwit lifecycle pattern)
     let storage = cached_storage;
 
-    // CRITICAL FIX: Use shared global context for cache hits but create individual permit provider
-    // This preserves cache efficiency while eliminating SearchPermitProvider permit exhaustion
+    // Use shared global context for cache hits but create individual permit provider
+    // to preserve cache efficiency while avoiding SearchPermitProvider permit exhaustion.
+    //
+    // KNOWN LIMITATIONS (see CACHING_MEMORY_INFRA_REVIEW.md §1.9, §2.6):
+    // - This query path reads the *global* SearcherContext, so its leaf-search/footer
+    //   caches are shared across credential sets, whereas the index-open path keys the
+    //   context by credentials. Split IDs are unguessable UUIDs, which bounds the
+    //   practical exposure; full isolation would require threading the credential key
+    //   (S3 + Azure) into this function and every sibling search/aggregation entry point.
+    // - The per-search permit provider below intentionally bypasses the process-global
+    //   concurrency/warmup budget to avoid a permit-exhaustion deadlock; as a result
+    //   max_concurrent_splits / warmup_memory_budget are not enforced here.
     debug_println!("🔍 PERMIT_FIX: Using global context for cache hits but individual permit provider");
 
     let searcher_context = crate::global_cache::get_global_searcher_context();
diff --git a/native/src/split_searcher/byterange_cache.rs b/native/src/split_searcher/byterange_cache.rs
deleted file mode 100644
index 82fd5d98..00000000
--- a/native/src/split_searcher/byterange_cache.rs
+++ /dev/null
@@ -1,136 +0,0 @@
-// byterange_cache.rs - ByteRange cache merging support
-// Enables partial cache hits and range coalescing for efficient storage access
-
-use std::sync::Arc;
-use super::cache_config::MAX_ACCEPTABLE_GAPS;
-
-/// Represents a cached byte range with metadata
-#[derive(Debug, Clone)]
-pub struct CachedRange {
-    pub start: usize,
-    pub end: usize,
-    pub data: Arc<Vec<u8>>,
-    pub last_accessed: std::time::SystemTime,
-}
-
-impl CachedRange {
-    /// Check if this range overlaps with the requested range
-    pub fn overlaps_with(&self, start: usize, end: usize) -> bool {
-        !(self.end <= start || self.start >= end)
-    }
-
-    /// Get the intersection of this range with the requested range
-    pub fn intersection(&self, start: usize, end: usize) -> Option<(usize, usize)> {
-        if self.overlaps_with(start, end) {
-            Some((self.start.max(start), self.end.min(end)))
-        } else {
-            None
-        }
-    }
-
-    /// Get data slice for the specified range (relative to this cached range)
-    pub fn get_slice(&self, start: usize, end: usize) -> Option<&[u8]> {
-        if start >= self.start && end <= self.end {
-            let relative_start = start - self.start;
-            let relative_end = end - self.start;
-            Some(&self.data[relative_start..relative_end])
-        } else {
-            None
-        }
-    }
-}
-
-/// Result of attempting to serve a request from cache with range merging
-pub enum CacheResult {
-    /// Complete cache hit - all data available from cache
-    Hit(Vec<u8>),
-    /// Partial cache hit - some data cached, some needs to be fetched
-    PartialHit {
-        cached_segments: Vec<CachedRange>,
-        missing_gaps: Vec<(usize, usize)>,
-    },
-    /// Cache miss - no useful cached data
-    Miss,
-}
-
-/// Calculate missing gaps between cached ranges for a requested range
-pub fn calculate_missing_gaps(
-    requested_start: usize,
-    requested_end: usize,
-    cached_ranges: &[CachedRange]
-) -> Vec<(usize, usize)> {
-    let mut gaps = Vec::new();
-    let mut current_pos = requested_start;
-
-    // Sort cached ranges by start position
-    let mut sorted_ranges: Vec<_> = cached_ranges.iter()
-        .filter(|r| r.overlaps_with(requested_start, requested_end))
-        .collect();
-    sorted_ranges.sort_by_key(|r| r.start);
-
-    for range in sorted_ranges {
-        let range_start = range.start.max(requested_start);
-        let range_end = range.end.min(requested_end);
-
-        // Add gap before this range if it exists
-        if current_pos < range_start {
-            gaps.push((current_pos, range_start));
-        }
-
-        // Move past this range
-        current_pos = current_pos.max(range_end);
-    }
-
-    // Add final gap if needed
-    if current_pos < requested_end {
-        gaps.push((current_pos, requested_end));
-    }
-
-    gaps
-}
-
-/// Try to merge cached ranges to serve a complete request
-pub fn try_merge_cached_ranges(
-    requested_start: usize,
-    requested_end: usize,
-    cached_ranges: &[CachedRange]
-) -> CacheResult {
-    let gaps = calculate_missing_gaps(requested_start, requested_end, cached_ranges);
-
-    if gaps.is_empty() {
-        // Complete cache hit possible - merge the data
-        let mut result_data = vec![0u8; requested_end - requested_start];
-        let mut covered = false;
-
-        for range in cached_ranges {
-            if let Some((int_start, int_end)) = range.intersection(requested_start, requested_end) {
-                if let Some(slice) = range.get_slice(int_start, int_end) {
-                    let result_start = int_start - requested_start;
-                    let result_end = result_start + slice.len();
-                    result_data[result_start..result_end].copy_from_slice(slice);
-                    covered = true;
-                }
-            }
-        }
-
-        if covered {
-            CacheResult::Hit(result_data)
-        } else {
-            CacheResult::Miss
-        }
-    } else if gaps.len() <= MAX_ACCEPTABLE_GAPS {
-        // Partial hit with acceptable number of gaps
-        let relevant_ranges: Vec<_> = cached_ranges.iter()
-            .filter(|r| r.overlaps_with(requested_start, requested_end))
-            .cloned()
-            .collect();
-
-        CacheResult::PartialHit {
-            cached_segments: relevant_ranges,
-            missing_gaps: gaps,
-        }
-    } else {
-        // Too many gaps - treat as miss
-        CacheResult::Miss
-    }
-}
diff --git a/native/src/split_searcher/jni_lifecycle.rs b/native/src/split_searcher/jni_lifecycle.rs
index f7b3b98a..13daa68a 100644
--- a/native/src/split_searcher/jni_lifecycle.rs
+++ b/native/src/split_searcher/jni_lifecycle.rs
@@ -475,6 +475,12 @@ pub extern "system" fn Java_io_indextables_tantivy4java_split_SplitSearcher_crea
                         s3_config_for_key.secret_access_key = aws_config.get("secret_key").cloned();
                         s3_config_for_key.session_token = aws_config.get("session_token").cloned();
                         s3_config_for_key.region = aws_config.get("region").cloned();
+                        // Include endpoint and path-style so two different endpoints (or
+                        // path-style settings) with the same access key don't collapse onto
+                        // one credential-scoped SearcherContext.
+                        s3_config_for_key.endpoint = aws_config.get("endpoint").cloned();
+                        s3_config_for_key.force_path_style_access =
+                            aws_config.get("path_style_access").map(|s| s == "true").unwrap_or(false);
                         crate::global_cache::generate_storage_cache_key(Some(&s3_config_for_key), None)
                     } else if azure_config.contains_key("account_name") {
                         let azure_config_for_key = quickwit_config::AzureStorageConfig {
diff --git a/native/src/split_searcher/mod.rs b/native/src/split_searcher/mod.rs
index 1ccda317..aa89bc80 100644
--- a/native/src/split_searcher/mod.rs
+++ b/native/src/split_searcher/mod.rs
@@ -3,7 +3,6 @@
 
 // Submodules
 pub mod cache_config;
-pub mod byterange_cache;
 pub mod searcher_cache;
 pub mod types;
 pub mod query_utils;
@@ -36,8 +35,6 @@ pub use async_impl::{
 
 // Cache configuration is now in cache_config.rs submodule
 
-// ByteRange cache merging is now in byterange_cache.rs submodule
-
 // Searcher cache and types are now in searcher_cache.rs and types.rs submodules
 
 // JNI lifecycle functions (createNativeWithSharedCache, closeNative, validateSplitNative, getCacheStatsNative)
diff --git a/src/main/java/io/indextables/tantivy4java/split/SplitCacheManager.java b/src/main/java/io/indextables/tantivy4java/split/SplitCacheManager.java
index 3ea3c2db..e11c869a 100644
--- a/src/main/java/io/indextables/tantivy4java/split/SplitCacheManager.java
+++ b/src/main/java/io/indextables/tantivy4java/split/SplitCacheManager.java
@@ -98,6 +98,14 @@ public class SplitCacheManager implements AutoCloseable {
     private final String cacheName;
     private final String cacheKey; // Full cache key used for storage/retrieval
     private final long maxCacheSize;
+    // Number of getInstance() callers currently sharing this singleton. The native
+    // manager is torn down only when this reaches zero, so one holder's close()
+    // cannot pull the shared instance out from under the others.
+    private final java.util.concurrent.atomic.AtomicInteger refCount =
+            new java.util.concurrent.atomic.AtomicInteger(1);
+    // Set once the instance has been fully torn down; guards against double-close
+    // and use-after-close.
+    private volatile boolean closed = false;
     private final Map<String, SplitSearcher> managedSearchers;
     private final AtomicLong totalCacheSize;
     private final long nativePtr;
@@ -389,14 +397,12 @@ public BatchOptimizationConfig getBatchOptimization() {
          *     .withMaxCacheSize(500_000_000)  // 500MB L1 memory cache
          *     .withTieredCache(new TieredCacheConfig()
          *         .withDiskCachePath("/mnt/nvme/cache")
-         *         .withMaxDiskSize(100_000_000_000L)  // 100GB disk cache
-         *         .withCompression(CompressionAlgorithm.LZ4));  // Fast compression
+         *         .withMaxDiskSize(100_000_000_000L));  // 100GB disk cache
          * }</pre>
          *
          * @param tieredCacheConfig the tiered cache configuration
          * @return this CacheConfig for method chaining
          * @see TieredCacheConfig
-         * @see CompressionAlgorithm
          */
         public CacheConfig withTieredCache(TieredCacheConfig tieredCacheConfig) {
             this.tieredCacheConfig = tieredCacheConfig;
@@ -424,85 +430,105 @@ public String getCacheKey() {
             keyBuilder.append(",maxSize=").append(maxCacheSize);
             keyBuilder.append(",maxLoads=").append(maxConcurrentLoads);
             keyBuilder.append(",queryCache=").append(enableQueryCache);
-            
-            // Add AWS config to key (sorted for consistency)
-            if (!awsConfig.isEmpty()) {
-                keyBuilder.append(",aws={");
-                awsConfig.entrySet().stream()
-                    .sorted(Map.Entry.comparingByKey())
-                    .forEach(entry -> keyBuilder.append(entry.getKey()).append("=").append(entry.getValue()).append(","));
-                keyBuilder.setLength(keyBuilder.length() - 1); // Remove last comma
-                keyBuilder.append("}");
-            }
-            
-            // Add Azure config to key (sorted for consistency)
-            if (!azureConfig.isEmpty()) {
-                keyBuilder.append(",azure={");
-                azureConfig.entrySet().stream()
-                    .sorted(Map.Entry.comparingByKey())
-                    .forEach(entry -> keyBuilder.append(entry.getKey()).append("=").append(entry.getValue()).append(","));
-                keyBuilder.setLength(keyBuilder.length() - 1); // Remove last comma
-                keyBuilder.append("}");
-            }
-            
+
+            appendConfigToKey(keyBuilder, "aws", awsConfig);
+            appendConfigToKey(keyBuilder, "azure", azureConfig);
+
             // Note: parquetTableRoot and parquetStorageConfig are intentionally excluded
             // from the cache key. They are per-split parameters passed to createSplitSearcher()
             // and do not affect the identity of the shared cache manager instance.
 
-            // Add GCP config to key (sorted for consistency)
-            if (!gcpConfig.isEmpty()) {
-                keyBuilder.append(",gcp={");
-                gcpConfig.entrySet().stream()
-                    .sorted(Map.Entry.comparingByKey())
-                    .forEach(entry -> keyBuilder.append(entry.getKey()).append("=").append(entry.getValue()).append(","));
-                keyBuilder.setLength(keyBuilder.length() - 1); // Remove last comma
-                keyBuilder.append("}");
-            }
-            
+            appendConfigToKey(keyBuilder, "gcp", gcpConfig);
+
             return keyBuilder.toString();
         }
+
+        /** Credential map keys whose values are secrets and must be hashed, never
+         *  embedded verbatim, in the (long-lived, map-keying) cache key. */
+        private static final Set<String> SENSITIVE_CONFIG_KEYS = new HashSet<>(Arrays.asList(
+            "access_key", "secret_key", "session_token", "bearer_token", "connection_string"));
+
+        /**
+         * Append a credential config map to the cache key, hashing sensitive values so
+         * that plaintext secrets/session tokens/bearer tokens never live inside a
+         * long-lived map key (one debug log or toString() away from leaking). Hashing
+         * still uniquely distinguishes credential sets.
+         */
+        private static void appendConfigToKey(StringBuilder keyBuilder, String label, Map<String, String> config) {
+            if (config == null || config.isEmpty()) {
+                return;
+            }
+            keyBuilder.append(",").append(label).append("={");
+            config.entrySet().stream()
+                .sorted(Map.Entry.comparingByKey())
+                .forEach(entry -> {
+                    String value = entry.getValue();
+                    if (value != null && SENSITIVE_CONFIG_KEYS.contains(entry.getKey())) {
+                        value = "sha256:" + sha256Hex(value);
+                    }
+                    keyBuilder.append(entry.getKey()).append("=").append(value).append(",");
+                });
+            keyBuilder.setLength(keyBuilder.length() - 1); // Remove last comma
+            keyBuilder.append("}");
+        }
+
+        private static String sha256Hex(String input) {
+            try {
+                java.security.MessageDigest md = java.security.MessageDigest.getInstance("SHA-256");
+                byte[] digest = md.digest(input.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+                StringBuilder sb = new StringBuilder(digest.length * 2);
+                for (byte b : digest) {
+                    sb.append(Character.forDigit((b >> 4) & 0xF, 16));
+                    sb.append(Character.forDigit(b & 0xF, 16));
+                }
+                return sb.toString();
+            } catch (java.security.NoSuchAlgorithmException e) {
+                // SHA-256 is guaranteed present on every JVM; fall back to a non-plaintext
+                // representation just in case so a secret is never embedded verbatim.
+                return Integer.toHexString(input.hashCode());
+            }
+        }
     }
     
     /**
-     * Compression algorithm for L2 disk cache.
+     * Compression algorithm for the L2 disk cache.
      *
-     * <p>Determines how cached data is compressed on disk:
-     * <ul>
-     *   <li>{@link #NONE} - No compression, fastest I/O but largest disk usage</li>
-     *   <li>{@link #LZ4} - Fast compression (~400 MB/s), good compression ratio (default)</li>
-     *   <li>{@link #ZSTD} - Better compression, slower (falls back to LZ4 currently)</li>
-     * </ul>
+     * <p><b>Currently a no-op.</b> The disk cache stores every component
+     * <em>uncompressed</em> regardless of the value selected here. Index components
+     * are accessed as small sub-ranges (postings in 128-doc blocks, fast fields by
+     * doc id, sstable/store blocks that Tantivy already compresses), so whole-file
+     * decompression would badly hurt read latency without a meaningful space win.
+     * This enum is retained for source/binary compatibility; the setting is ignored.
      *
-     * <p>The cache uses intelligent compression decisions based on component type:
-     * <ul>
-     *   <li>Small data (&lt;4KB): Never compressed (overhead exceeds benefit)</li>
-     *   <li>Hot data (footer, metadata): Never compressed (CPU cost not worth it)</li>
-     *   <li>Large components (.term, .idx, .pos): Always compressed (50-70% savings)</li>
-     * </ul>
+     * @deprecated Compression is not applied by the disk cache. The value is accepted
+     *     but has no effect; all cached data is stored uncompressed.
      */
+    @Deprecated
     public enum CompressionAlgorithm {
-        /** No compression - use for already-compressed or small data */
+        /** No compression (this is the effective behavior for all values). */
         NONE,
-        /** LZ4 compression - fast, good for index data (default) */
+        /** Ignored — data is stored uncompressed. Retained for compatibility. */
         LZ4,
-        /** Zstd compression - better ratio, slower (currently falls back to LZ4) */
+        /** Ignored — data is stored uncompressed. Retained for compatibility. */
         ZSTD
     }
 
     /**
      * Configuration for L2 tiered disk cache.
      *
-     * <p>Provides persistent disk caching with intelligent compression for Quickwit split
-     * components. The disk cache sits between the L1 memory cache and remote storage (S3/Azure).
+     * <p>Provides persistent disk caching for Quickwit split components. The disk cache
+     * sits between the L1 memory cache and remote storage (S3/Azure).
      *
      * <p><b>Features:</b>
      * <ul>
      *   <li>Persistent across restarts - survives JVM shutdown</li>
-     *   <li>Intelligent compression - LZ4 for large components, skips small/hot data</li>
      *   <li>Split-level LRU eviction - removes entire splits when disk space is needed</li>
      *   <li>Crash-safe manifest - periodic sync with backup for recovery</li>
      * </ul>
      *
+     * <p><b>Note:</b> Components are stored uncompressed; the compression settings on
+     * this config are accepted but ignored (see {@link CompressionAlgorithm}).
+     *
      * <p><b>Directory Structure:</b>
      * <pre>
      * {diskCachePath}/
@@ -518,9 +544,7 @@ public enum CompressionAlgorithm {
      * <pre>{@code
      * TieredCacheConfig tieredConfig = new TieredCacheConfig()
      *     .withDiskCachePath("/mnt/nvme/tantivy_cache")
-     *     .withMaxDiskSize(100_000_000_000L)  // 100GB
-     *     .withCompression(CompressionAlgorithm.LZ4)
-     *     .withMinCompressSize(4096);  // Don't compress data < 4KB
+     *     .withMaxDiskSize(100_000_000_000L);  // 100GB
      *
      * CacheConfig config = new CacheConfig("prod-cache")
      *     .withMaxCacheSize(500_000_000)  // 500MB L1
@@ -580,16 +604,14 @@ public TieredCacheConfig withMaxDiskSize(long bytes) {
         /**
          * Set the compression algorithm for cached data.
          *
-         * <p>The default is {@link CompressionAlgorithm#LZ4} which provides fast
-         * compression (~400 MB/s) with good ratios (50-70% for index data).
-         *
-         * <p>Note: Compression is only applied to components where it provides benefit.
-         * Small data (&lt;4KB), hot data (footer), and already-compact data (numeric fast fields)
-         * are not compressed regardless of this setting.
+         * <p><b>No effect.</b> The disk cache stores all components uncompressed
+         * (see {@link CompressionAlgorithm}); this value is accepted but ignored.
          *
-         * @param compression compression algorithm to use
+         * @param compression compression algorithm (ignored)
          * @return this TieredCacheConfig for method chaining
+         * @deprecated Compression is not applied by the disk cache.
          */
+        @Deprecated
         public TieredCacheConfig withCompression(CompressionAlgorithm compression) {
             this.compression = compression;
             return this;
@@ -598,7 +620,8 @@ public TieredCacheConfig withCompression(CompressionAlgorithm compression) {
         /**
          * Disable compression entirely.
          *
-         * <p>Use this if CPU is limited or if the data is already compressed.
+         * <p>This is already the effective behavior for every configuration — the
+         * disk cache never compresses. Retained for compatibility.
          *
          * @return this TieredCacheConfig for method chaining
          */
@@ -610,12 +633,14 @@ public TieredCacheConfig withoutCompression() {
         /**
          * Set minimum data size for compression.
          *
-         * <p>Data smaller than this threshold will not be compressed, as the
-         * CPU overhead exceeds the I/O savings. Default is 4KB.
+         * <p><b>No effect.</b> The disk cache does not compress data (see
+         * {@link CompressionAlgorithm}); this threshold is accepted but ignored.
          *
-         * @param bytes minimum size in bytes to consider for compression
+         * @param bytes minimum size in bytes (ignored)
          * @return this TieredCacheConfig for method chaining
+         * @deprecated Compression is not applied by the disk cache.
          */
+        @Deprecated
         public TieredCacheConfig withMinCompressSize(int bytes) {
             this.minCompressSizeBytes = bytes;
             return this;
@@ -927,6 +952,7 @@ public static SplitCacheManager getInstance(CacheConfig config) {
         try {
             SplitCacheManager existing = instances.get(cacheKey);
             if (existing != null) {
+                existing.refCount.incrementAndGet();
                 return existing;
             }
         } finally {
@@ -939,9 +965,25 @@ public static SplitCacheManager getInstance(CacheConfig config) {
             // Double-check pattern - another thread might have created it while we were waiting
             SplitCacheManager existing = instances.get(cacheKey);
             if (existing != null) {
+                existing.refCount.incrementAndGet();
                 return existing;
             }
 
+            // Enforce name uniqueness. The native cache-manager registry is keyed by
+            // cache NAME only, so two managers sharing a name but differing in size or
+            // credentials would silently clobber each other natively (the second
+            // create overwrites the first, orphaning its per-manager state). Reject the
+            // conflicting config with a clear message instead of allowing that.
+            for (SplitCacheManager other : instances.values()) {
+                if (other.cacheName.equals(config.getCacheName())) {
+                    throw new IllegalStateException(
+                        "A SplitCacheManager named '" + config.getCacheName() + "' already exists " +
+                        "with a different configuration. Cache names must be unique per configuration " +
+                        "(the native cache registry is keyed by name). Use a distinct name or reuse the " +
+                        "identical config to share the existing instance.");
+                }
+            }
+
             // Create new instance with validation
             validateCacheConfig(config);
             SplitCacheManager newInstance = new SplitCacheManager(config);
@@ -963,10 +1005,25 @@ public static SplitCacheManager getInstance(CacheConfig config) {
     public static SplitCacheManager getGlobalInstance() {
         SplitCacheManager existing = GLOBAL_INSTANCE.get();
         if (existing != null) {
-            return existing;
+            // Bump the refcount before handing the shared instance back — like
+            // getInstance() does. Returning it without incrementing would let a later
+            // close() by any one holder tear down the native manager while other
+            // getGlobalInstance() callers still believe it is alive (use-after-close).
+            // Re-check under the read lock that it is still registered (not closed
+            // concurrently); if it has been removed, fall through and create a default.
+            instancesLock.readLock().lock();
+            try {
+                SplitCacheManager current = instances.get(existing.cacheKey);
+                if (current != null) {
+                    current.refCount.incrementAndGet();
+                    return current;
+                }
+            } finally {
+                instancesLock.readLock().unlock();
+            }
         }
 
-        // Create default instance if none exists
+        // Create default instance if none exists (getInstance handles refcounting)
         CacheConfig defaultConfig = new CacheConfig("global-default")
             .withMaxCacheSize(500_000_000); // 500MB default
         return getInstance(defaultConfig);
@@ -1439,9 +1496,12 @@ public CacheFlushStats(GlobalCacheStats beforeStats, GlobalCacheStats afterStats
             this.beforeStats = beforeStats;
             this.afterStats = afterStats;
             this.splitsAffected = splitsAffected;
-            this.bytesFreedTotal = beforeStats.getCurrentSize() - afterStats.getCurrentSize();
-            this.itemsEvicted = (beforeStats.getTotalHits() + beforeStats.getTotalMisses()) - 
-                               (afterStats.getTotalHits() + afterStats.getTotalMisses());
+            this.bytesFreedTotal = Math.max(0, beforeStats.getCurrentSize() - afterStats.getCurrentSize());
+            // Number of entries evicted during the flush = growth in the (monotonic)
+            // eviction counter. The previous formula used the hits+misses delta, which
+            // is also monotonic and unrelated to eviction, so it could only ever be
+            // <= 0 — a meaningless metric.
+            this.itemsEvicted = Math.max(0, afterStats.getTotalEvictions() - beforeStats.getTotalEvictions());
         }
         
         public long getBytesFreed() { return bytesFreedTotal; }
@@ -1473,7 +1533,39 @@ public enum CacheType {
     
     @Override
     public void close() {
-        // Close all managed searchers
+        // Reference-counted: getInstance() hands the same singleton to multiple
+        // callers, so only the last close() actually tears the manager down. This
+        // prevents one holder from releasing the native handle (and clearing global
+        // caches) out from under the others.
+        //
+        // The refCount decrement, the "am I the last?" decision, and the registry
+        // removal all happen under the write lock, which is mutually exclusive with
+        // getInstance()'s get()+incrementAndGet() under the read lock. That ordering
+        // closes the resurrection race where a concurrent getInstance() could revive
+        // this instance (refCount 0 -> 1) after we decided to tear it down.
+        if (closed) {
+            return;
+        }
+        instancesLock.writeLock().lock();
+        try {
+            if (closed) {
+                return;
+            }
+            if (refCount.decrementAndGet() > 0) {
+                return; // Other holders still active.
+            }
+            closed = true;
+            // Remove from the registry while holding the lock so no new getInstance()
+            // can hand this instance out after this point.
+            instances.remove(cacheKey);
+            GLOBAL_INSTANCE.compareAndSet(this, null);
+        } finally {
+            instancesLock.writeLock().unlock();
+        }
+
+        // Actual teardown runs OUTSIDE the registry lock — closing searchers and the
+        // native manager can be slow and must not block getInstance() for other
+        // (differently-named) caches, nor risk a callback deadlock.
         for (SplitSearcher searcher : managedSearchers.values()) {
             try {
                 searcher.close();
@@ -1482,14 +1574,10 @@ public void close() {
             }
         }
         managedSearchers.clear();
-        
-        // Close native cache manager
+
         if (nativePtr != 0) {
             closeNativeCacheManager(nativePtr);
         }
-        
-        // Remove from instances using the cache key
-        instances.remove(cacheKey);
     }
     
     /**