Skip to main content

hermes_core/directories/
slice_cache.rs

1//! Slice-level caching directory with overlap management
2//!
3//! Caches byte ranges from files, merging overlapping ranges and
4//! evicting least-recently-used slices when the cache limit is reached.
5
6use async_trait::async_trait;
7use parking_lot::RwLock;
8use std::collections::BTreeMap;
9use std::io::{self, Read, Write};
10use std::ops::Range;
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13
14use super::{Directory, FileHandle, OwnedBytes, RangeReadFn};
15
16/// File extension for slice cache files
17pub const SLICE_CACHE_EXTENSION: &str = "slicecache";
18
19/// Magic bytes for slice cache file format
20const SLICE_CACHE_MAGIC: &[u8; 8] = b"HRMSCACH";
21
22/// Current version of the slice cache format
23/// v2: Added file size caching
24const SLICE_CACHE_VERSION: u32 = 2;
25
26/// A cached slice of a file
27#[derive(Debug, Clone)]
28struct CachedSlice {
29    /// Byte range in the file
30    range: Range<u64>,
31    /// Arc-backed cached data. Cache hits return cheap sub-slices instead of
32    /// allocating and copying the requested range.
33    data: OwnedBytes,
34    /// Access counter for LRU eviction
35    access_count: u64,
36}
37
38/// Per-file slice cache using interval tree for overlap detection
39struct FileSliceCache {
40    /// Slices sorted by start offset for efficient overlap detection
41    slices: BTreeMap<u64, CachedSlice>,
42    /// Total bytes cached for this file
43    total_bytes: usize,
44}
45
46impl FileSliceCache {
47    fn new() -> Self {
48        Self {
49            slices: BTreeMap::new(),
50            total_bytes: 0,
51        }
52    }
53
54    /// Serialize this file cache to bytes
55    fn serialize(&self) -> Vec<u8> {
56        let mut buf = Vec::new();
57        // Number of slices
58        buf.extend_from_slice(&(self.slices.len() as u32).to_le_bytes());
59        for slice in self.slices.values() {
60            // Range start and end
61            buf.extend_from_slice(&slice.range.start.to_le_bytes());
62            buf.extend_from_slice(&slice.range.end.to_le_bytes());
63            // Data length and data
64            buf.extend_from_slice(&(slice.data.len() as u32).to_le_bytes());
65            buf.extend_from_slice(slice.data.as_slice());
66        }
67        buf
68    }
69
70    /// Deserialize from bytes, returns (cache, bytes_consumed)
71    fn deserialize(
72        data: &[u8],
73        access_counter: u64,
74        max_bytes: usize,
75    ) -> io::Result<(Self, usize)> {
76        let mut pos = 0;
77        if data.len() < 4 {
78            return Err(io::Error::new(
79                io::ErrorKind::InvalidData,
80                "truncated slice cache",
81            ));
82        }
83        let num_slices = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
84        pos += 4;
85
86        let mut cache = FileSliceCache::new();
87        for _ in 0..num_slices {
88            if pos + 20 > data.len() {
89                return Err(io::Error::new(
90                    io::ErrorKind::InvalidData,
91                    "truncated slice entry",
92                ));
93            }
94            let range_start = u64::from_le_bytes(data[pos..pos + 8].try_into().unwrap());
95            pos += 8;
96            let range_end = u64::from_le_bytes(data[pos..pos + 8].try_into().unwrap());
97            pos += 8;
98            let data_len = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
99            pos += 4;
100
101            let data_end = pos.checked_add(data_len).ok_or_else(|| {
102                io::Error::new(io::ErrorKind::InvalidData, "slice data length overflow")
103            })?;
104            if data_end > data.len() {
105                return Err(io::Error::new(
106                    io::ErrorKind::InvalidData,
107                    "truncated slice data",
108                ));
109            }
110            if range_end < range_start || range_end - range_start != data_len as u64 {
111                return Err(io::Error::new(
112                    io::ErrorKind::InvalidData,
113                    "slice range and data length are inconsistent",
114                ));
115            }
116            let slice_range = range_start..range_end;
117            pos = data_end;
118
119            // Do not duplicate an oversized serialized entry just to evict it
120            // after the complete cache has been reconstructed. Retain at most
121            // one cache budget while parsing each file.
122            if data_len <= max_bytes {
123                let bytes_to_free = cache
124                    .total_bytes
125                    .saturating_add(data_len)
126                    .saturating_sub(max_bytes);
127                cache.evict_lru(bytes_to_free);
128                cache.insert(
129                    slice_range,
130                    OwnedBytes::new(data[data_end - data_len..data_end].to_vec()),
131                    access_counter,
132                );
133                debug_assert!(cache.total_bytes <= max_bytes);
134            }
135        }
136        Ok((cache, pos))
137    }
138
139    /// Get iterator over all slices for serialization
140    #[allow(dead_code)]
141    fn iter_slices(&self) -> impl Iterator<Item = (&u64, &CachedSlice)> {
142        self.slices.iter()
143    }
144
145    /// Try to read from cache, returns None if not fully cached
146    fn try_read(&mut self, range: Range<u64>, access_counter: &mut u64) -> Option<OwnedBytes> {
147        // Find slices that might contain our range
148        let start = range.start;
149        let end = range.end;
150
151        // Look for a slice that contains the entire range
152        let mut found_key = None;
153        for (&slice_start, slice) in self.slices.range(..=start).rev() {
154            if slice_start <= start && slice.range.end >= end {
155                found_key = Some((
156                    slice_start,
157                    (start - slice_start) as usize,
158                    (end - start) as usize,
159                ));
160                break;
161            }
162        }
163
164        if let Some((key, offset, len)) = found_key {
165            // Update access count for LRU
166            *access_counter += 1;
167            if let Some(s) = self.slices.get_mut(&key) {
168                s.access_count = *access_counter;
169                return Some(s.data.slice(offset..offset + len));
170            }
171        }
172
173        None
174    }
175
176    /// Insert a slice, merging with overlapping slices
177    /// Returns the net change in bytes (can be negative if merge reduces size, but typically positive)
178    fn insert(&mut self, range: Range<u64>, data: OwnedBytes, access_counter: u64) -> isize {
179        let start = range.start;
180        let end = range.end;
181        let data_len = data.len();
182
183        // Find and remove overlapping slices
184        let mut to_remove = Vec::new();
185        let mut merged_start = start;
186        let mut merged_end = end;
187        let mut merged_data: Option<OwnedBytes> = None;
188        let mut bytes_removed: usize = 0;
189
190        for (&slice_start, slice) in &self.slices {
191            // Check for overlap
192            if slice_start < end && slice.range.end > start {
193                to_remove.push(slice_start);
194
195                // Extend merged range
196                merged_start = merged_start.min(slice_start);
197                merged_end = merged_end.max(slice.range.end);
198            }
199        }
200
201        // If we have overlaps, merge the data
202        if !to_remove.is_empty() {
203            let merged_len = (merged_end - merged_start) as usize;
204            let mut new_data = vec![0u8; merged_len];
205
206            // Copy existing slices
207            for &slice_start in &to_remove {
208                if let Some(slice) = self.slices.get(&slice_start) {
209                    let offset = (slice_start - merged_start) as usize;
210                    new_data[offset..offset + slice.data.len()]
211                        .copy_from_slice(slice.data.as_slice());
212                    bytes_removed += slice.data.len();
213                    self.total_bytes -= slice.data.len();
214                }
215            }
216
217            // Copy new data (overwrites any overlapping parts)
218            let offset = (start - merged_start) as usize;
219            new_data[offset..offset + data_len].copy_from_slice(data.as_slice());
220
221            // Remove old slices
222            for slice_start in to_remove {
223                self.slices.remove(&slice_start);
224            }
225
226            merged_data = Some(OwnedBytes::new(new_data));
227        }
228
229        // Insert the (possibly merged) slice
230        let (final_start, final_data) = if let Some(md) = merged_data {
231            (merged_start, md)
232        } else {
233            (start, data)
234        };
235
236        let bytes_added = final_data.len();
237        self.total_bytes += bytes_added;
238
239        self.slices.insert(
240            final_start,
241            CachedSlice {
242                range: final_start..final_start + bytes_added as u64,
243                data: final_data,
244                access_count: access_counter,
245            },
246        );
247
248        // Return net change: bytes added minus bytes removed during merge
249        bytes_added as isize - bytes_removed as isize
250    }
251
252    /// Evict least recently used slices to free up space
253    fn evict_lru(&mut self, bytes_to_free: usize) -> usize {
254        let mut freed = 0;
255
256        while freed < bytes_to_free && !self.slices.is_empty() {
257            // Find the slice with lowest access count
258            let lru_key = self
259                .slices
260                .iter()
261                .min_by_key(|(_, s)| s.access_count)
262                .map(|(&k, _)| k);
263
264            if let Some(key) = lru_key {
265                if let Some(slice) = self.slices.remove(&key) {
266                    freed += slice.data.len();
267                    self.total_bytes -= slice.data.len();
268                }
269            } else {
270                break;
271            }
272        }
273
274        freed
275    }
276}
277
278fn evict_cached_slices(
279    caches: &mut std::collections::HashMap<PathBuf, FileSliceCache>,
280    current_bytes: &mut usize,
281    max_bytes: usize,
282    needed: usize,
283) {
284    let target = current_bytes
285        .saturating_add(needed)
286        .saturating_sub(max_bytes);
287    let mut freed = 0;
288
289    while freed < target {
290        let oldest_file = caches
291            .iter()
292            .filter(|(_, cache)| !cache.slices.is_empty())
293            .min_by_key(|(_, cache)| {
294                cache
295                    .slices
296                    .values()
297                    .map(|slice| slice.access_count)
298                    .min()
299                    .unwrap_or(u64::MAX)
300            })
301            .map(|(path, _)| path.clone());
302
303        let Some(path) = oldest_file else {
304            break;
305        };
306        let Some(file_cache) = caches.get_mut(&path) else {
307            break;
308        };
309        freed += file_cache.evict_lru(target - freed);
310    }
311
312    *current_bytes = current_bytes.saturating_sub(freed);
313}
314
315/// Slice-caching directory wrapper
316///
317/// Caches byte ranges from the inner directory, with:
318/// - Overlap detection and merging
319/// - LRU eviction when cache limit is reached
320/// - Bounded total memory usage
321/// - File size caching to avoid HEAD requests
322pub struct SliceCachingDirectory<D: Directory> {
323    inner: Arc<D>,
324    /// Per-file slice caches
325    caches: Arc<RwLock<std::collections::HashMap<PathBuf, FileSliceCache>>>,
326    /// Cached file sizes (avoids HEAD requests on lazy open)
327    file_sizes: Arc<RwLock<std::collections::HashMap<PathBuf, u64>>>,
328    /// Maximum total bytes to cache
329    max_bytes: usize,
330    /// Current total bytes cached
331    current_bytes: Arc<RwLock<usize>>,
332    /// Global access counter for LRU
333    access_counter: Arc<RwLock<u64>>,
334    /// Index name for Directory-layer metric labels (also forwarded to inner)
335    label: super::IndexLabel,
336}
337
338impl<D: Directory> SliceCachingDirectory<D> {
339    /// Create a new slice-caching directory with the given memory limit
340    pub fn new(inner: D, max_bytes: usize) -> Self {
341        Self {
342            inner: Arc::new(inner),
343            caches: Arc::new(RwLock::new(std::collections::HashMap::new())),
344            file_sizes: Arc::new(RwLock::new(std::collections::HashMap::new())),
345            max_bytes,
346            current_bytes: Arc::new(RwLock::new(0)),
347            access_counter: Arc::new(RwLock::new(0)),
348            label: super::IndexLabel::default(),
349        }
350    }
351
352    /// Get a reference to the inner directory
353    pub fn inner(&self) -> &D {
354        &self.inner
355    }
356
357    /// Try to read from cache
358    fn try_cache_read(&self, path: &Path, range: Range<u64>) -> Option<OwnedBytes> {
359        let mut caches = self.caches.write();
360        let mut counter = self.access_counter.write();
361
362        if let Some(file_cache) = caches.get_mut(path) {
363            file_cache.try_read(range, &mut counter)
364        } else {
365            None
366        }
367    }
368
369    /// Insert into cache, evicting if necessary
370    fn cache_insert(&self, path: &Path, range: Range<u64>, data: OwnedBytes) {
371        let data_len = data.len();
372        // An individual entry larger than the entire cache can never fit.
373        // Bypass it instead of evicting useful data and exceeding the cap.
374        if data_len > self.max_bytes {
375            return;
376        }
377
378        let mut caches = self.caches.write();
379        let mut current = self.current_bytes.write();
380        let counter = *self.access_counter.read();
381
382        // Free enough space before merging. Besides keeping the retained size
383        // bounded, this avoids constructing a large merged allocation only to
384        // evict it immediately afterward.
385        evict_cached_slices(&mut caches, &mut current, self.max_bytes, data_len);
386        let file_cache = caches
387            .entry(path.to_path_buf())
388            .or_insert_with(FileSliceCache::new);
389
390        let net_change = file_cache.insert(range, data, counter);
391        if net_change >= 0 {
392            *current += net_change as usize;
393        } else {
394            *current = current.saturating_sub((-net_change) as usize);
395        }
396        evict_cached_slices(&mut caches, &mut current, self.max_bytes, 0);
397        debug_assert!(*current <= self.max_bytes);
398    }
399
400    /// Get cache statistics
401    pub fn stats(&self) -> SliceCacheStats {
402        let caches = self.caches.read();
403        let mut total_slices = 0;
404        let mut files_cached = 0;
405
406        for fc in caches.values() {
407            if !fc.slices.is_empty() {
408                files_cached += 1;
409                total_slices += fc.slices.len();
410            }
411        }
412
413        SliceCacheStats {
414            total_bytes: *self.current_bytes.read(),
415            max_bytes: self.max_bytes,
416            total_slices,
417            files_cached,
418        }
419    }
420
421    /// Serialize the entire cache to a single binary blob
422    ///
423    /// Format (v2):
424    /// - Magic: 8 bytes "HRMSCACH"
425    /// - Version: 4 bytes (u32 LE)
426    /// - Num files: 4 bytes (u32 LE)
427    /// - For each file:
428    ///   - Path length: 4 bytes (u32 LE)
429    ///   - Path: UTF-8 bytes
430    ///   - File cache data (see FileSliceCache::serialize)
431    /// - Num file sizes: 4 bytes (u32 LE) [v2+]
432    /// - For each file size: [v2+]
433    ///   - Path length: 4 bytes (u32 LE)
434    ///   - Path: UTF-8 bytes
435    ///   - File size: 8 bytes (u64 LE)
436    pub fn serialize(&self) -> Vec<u8> {
437        let caches = self.caches.read();
438        let file_sizes = self.file_sizes.read();
439        let mut buf = Vec::new();
440
441        // Magic and version
442        buf.extend_from_slice(SLICE_CACHE_MAGIC);
443        buf.extend_from_slice(&SLICE_CACHE_VERSION.to_le_bytes());
444
445        // Count non-empty caches
446        let non_empty: Vec<_> = caches
447            .iter()
448            .filter(|(_, fc)| !fc.slices.is_empty())
449            .collect();
450        buf.extend_from_slice(&(non_empty.len() as u32).to_le_bytes());
451
452        for (path, file_cache) in non_empty {
453            // Path
454            let path_str = path.to_string_lossy();
455            let path_bytes = path_str.as_bytes();
456            buf.extend_from_slice(&(path_bytes.len() as u32).to_le_bytes());
457            buf.extend_from_slice(path_bytes);
458
459            // File cache data
460            let cache_data = file_cache.serialize();
461            buf.extend_from_slice(&cache_data);
462        }
463
464        // v2: File sizes section
465        buf.extend_from_slice(&(file_sizes.len() as u32).to_le_bytes());
466        for (path, &size) in file_sizes.iter() {
467            let path_str = path.to_string_lossy();
468            let path_bytes = path_str.as_bytes();
469            buf.extend_from_slice(&(path_bytes.len() as u32).to_le_bytes());
470            buf.extend_from_slice(path_bytes);
471            buf.extend_from_slice(&size.to_le_bytes());
472        }
473
474        buf
475    }
476
477    /// Deserialize and prefill the cache from a binary blob
478    ///
479    /// This loads cached slices from a previously serialized cache file.
480    /// Existing cache entries are preserved; new entries are merged in.
481    pub fn deserialize(&self, data: &[u8]) -> io::Result<()> {
482        let mut pos = 0;
483
484        // Check magic
485        if data.len() < 16 {
486            return Err(io::Error::new(
487                io::ErrorKind::InvalidData,
488                "slice cache too short",
489            ));
490        }
491        if &data[pos..pos + 8] != SLICE_CACHE_MAGIC {
492            return Err(io::Error::new(
493                io::ErrorKind::InvalidData,
494                "invalid slice cache magic",
495            ));
496        }
497        pos += 8;
498
499        // Check version (v2 only)
500        let version = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap());
501        pos += 4;
502        if version != 2 {
503            return Err(io::Error::new(
504                io::ErrorKind::InvalidData,
505                format!("unsupported slice cache version: {} (expected 2)", version),
506            ));
507        }
508
509        // Number of files
510        let num_files = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
511        pos += 4;
512
513        let mut caches = self.caches.write();
514        let mut current_bytes = self.current_bytes.write();
515        let counter = *self.access_counter.read();
516
517        for _ in 0..num_files {
518            // Path length
519            if pos + 4 > data.len() {
520                return Err(io::Error::new(
521                    io::ErrorKind::InvalidData,
522                    "truncated path length",
523                ));
524            }
525            let path_len = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
526            pos += 4;
527
528            // Path
529            if pos + path_len > data.len() {
530                return Err(io::Error::new(io::ErrorKind::InvalidData, "truncated path"));
531            }
532            let path_str = std::str::from_utf8(&data[pos..pos + path_len])
533                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
534            let path = PathBuf::from(path_str);
535            pos += path_len;
536
537            // File cache
538            let (file_cache, consumed) =
539                FileSliceCache::deserialize(&data[pos..], counter, self.max_bytes)?;
540            pos += consumed;
541
542            let new_bytes = file_cache.total_bytes;
543            if let Some(previous) = caches.insert(path, file_cache) {
544                *current_bytes = current_bytes.saturating_sub(previous.total_bytes);
545            }
546            *current_bytes = current_bytes.saturating_add(new_bytes);
547            evict_cached_slices(&mut caches, &mut current_bytes, self.max_bytes, 0);
548        }
549
550        // Recompute once after loading as a consistency check for serialized
551        // caches containing duplicate paths or overlapping ranges.
552        *current_bytes = caches.values().map(|cache| cache.total_bytes).sum();
553        evict_cached_slices(&mut caches, &mut current_bytes, self.max_bytes, 0);
554
555        // Load file sizes
556        if pos + 4 <= data.len() {
557            let num_sizes = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
558            pos += 4;
559
560            let mut file_sizes = self.file_sizes.write();
561            for _ in 0..num_sizes {
562                if pos + 4 > data.len() {
563                    break;
564                }
565                let path_len = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
566                pos += 4;
567
568                if pos + path_len > data.len() {
569                    break;
570                }
571                let path_str = match std::str::from_utf8(&data[pos..pos + path_len]) {
572                    Ok(s) => s,
573                    Err(_) => break,
574                };
575                let path = PathBuf::from(path_str);
576                pos += path_len;
577
578                if pos + 8 > data.len() {
579                    break;
580                }
581                let size = u64::from_le_bytes(data[pos..pos + 8].try_into().unwrap());
582                pos += 8;
583
584                file_sizes.insert(path, size);
585            }
586        }
587
588        Ok(())
589    }
590
591    /// Serialize the cache to a writer
592    pub fn serialize_to_writer<W: Write>(&self, mut writer: W) -> io::Result<()> {
593        let data = self.serialize();
594        writer.write_all(&data)
595    }
596
597    /// Deserialize the cache from a reader
598    pub fn deserialize_from_reader<R: Read>(&self, mut reader: R) -> io::Result<()> {
599        let mut data = Vec::new();
600        reader.read_to_end(&mut data)?;
601        self.deserialize(&data)
602    }
603
604    /// Check if the cache is empty
605    pub fn is_empty(&self) -> bool {
606        *self.current_bytes.read() == 0
607    }
608
609    /// Clear all cached data
610    pub fn clear(&self) {
611        let mut caches = self.caches.write();
612        let mut current_bytes = self.current_bytes.write();
613        caches.clear();
614        *current_bytes = 0;
615    }
616}
617
618/// Cache statistics
619#[derive(Debug, Clone)]
620pub struct SliceCacheStats {
621    pub total_bytes: usize,
622    pub max_bytes: usize,
623    pub total_slices: usize,
624    pub files_cached: usize,
625}
626
627#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
628#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
629impl<D: Directory> Directory for SliceCachingDirectory<D> {
630    async fn exists(&self, path: &Path) -> io::Result<bool> {
631        self.inner.exists(path).await
632    }
633
634    async fn file_size(&self, path: &Path) -> io::Result<u64> {
635        // Check cache first
636        {
637            let file_sizes = self.file_sizes.read();
638            if let Some(&size) = file_sizes.get(path) {
639                return Ok(size);
640            }
641        }
642
643        // Fetch from inner and cache
644        let size = self.inner.file_size(path).await?;
645        {
646            let mut file_sizes = self.file_sizes.write();
647            file_sizes.insert(path.to_path_buf(), size);
648        }
649        Ok(size)
650    }
651
652    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
653        // Check if we have the full file cached (use our caching file_size)
654        let file_size = self.file_size(path).await?;
655        let full_range = 0..file_size;
656
657        // Try cache first for full file
658        if let Some(data) = self.try_cache_read(path, full_range.clone()) {
659            return Ok(FileHandle::from_bytes(data));
660        }
661
662        // Read from inner
663        let handle = self.inner.open_read(path).await?;
664        let bytes = handle.read_bytes().await?;
665
666        // Cache the full file
667        self.cache_insert(path, full_range, bytes.clone());
668
669        Ok(FileHandle::from_bytes(bytes))
670    }
671
672    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
673        // Try cache first
674        if let Some(data) = self.try_cache_read(path, range.clone()) {
675            return Ok(data);
676        }
677
678        // Read from inner
679        let data = self.inner.read_range(path, range.clone()).await?;
680
681        // Cache the result
682        self.cache_insert(path, range, data.clone());
683
684        Ok(data)
685    }
686
687    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
688        self.inner.list_files(prefix).await
689    }
690
691    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
692        // Get file size (uses cache to avoid HEAD requests)
693        let file_size = self.file_size(path).await?;
694
695        // Create a caching wrapper around the inner directory's read_range
696        let path_buf = path.to_path_buf();
697        let caches = Arc::clone(&self.caches);
698        let current_bytes = Arc::clone(&self.current_bytes);
699        let access_counter = Arc::clone(&self.access_counter);
700        let max_bytes = self.max_bytes;
701        let inner = Arc::clone(&self.inner);
702
703        let read_fn: RangeReadFn = Arc::new(move |range: Range<u64>| {
704            let path = path_buf.clone();
705            let caches = Arc::clone(&caches);
706            let current_bytes = Arc::clone(&current_bytes);
707            let access_counter = Arc::clone(&access_counter);
708            let inner = Arc::clone(&inner);
709
710            Box::pin(async move {
711                // Try cache first
712                {
713                    let mut caches_guard = caches.write();
714                    let mut counter = access_counter.write();
715                    if let Some(file_cache) = caches_guard.get_mut(&path)
716                        && let Some(data) = file_cache.try_read(range.clone(), &mut counter)
717                    {
718                        return Ok(data);
719                    }
720                }
721
722                log::trace!("Cache MISS: {:?} [{}-{}]", path, range.start, range.end);
723
724                // Read from inner
725                let data = inner.read_range(&path, range.clone()).await?;
726
727                // Cache the result
728                let data_len = data.len();
729                if data_len <= max_bytes {
730                    let mut caches_guard = caches.write();
731                    let mut current = current_bytes.write();
732                    let counter = *access_counter.read();
733                    evict_cached_slices(&mut caches_guard, &mut current, max_bytes, data_len);
734                    let file_cache = caches_guard
735                        .entry(path.clone())
736                        .or_insert_with(FileSliceCache::new);
737                    let net_change = file_cache.insert(range, data.clone(), counter);
738                    if net_change >= 0 {
739                        *current += net_change as usize;
740                    } else {
741                        *current = current.saturating_sub((-net_change) as usize);
742                    }
743                    evict_cached_slices(&mut caches_guard, &mut current, max_bytes, 0);
744                    debug_assert!(*current <= max_bytes);
745                }
746
747                Ok(data)
748            })
749        });
750
751        Ok(FileHandle::lazy_labeled(
752            file_size,
753            read_fn,
754            self.label.get(),
755        ))
756    }
757
758    fn local_path(&self, path: &Path) -> Option<PathBuf> {
759        self.inner.local_path(path)
760    }
761
762    fn set_index_label(&self, label: &str) {
763        self.label.set(label);
764        self.inner.set_index_label(label);
765    }
766}
767
768/// DirectoryWriter implementation for SliceCachingDirectory
769/// Delegates to inner directory and invalidates cache entries as needed
770#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
771#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
772impl<D: super::DirectoryWriter> super::DirectoryWriter for SliceCachingDirectory<D> {
773    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
774        // Invalidate cache for this file
775        {
776            let mut caches = self.caches.write();
777            if let Some(file_cache) = caches.remove(path) {
778                let mut current = self.current_bytes.write();
779                *current = current.saturating_sub(file_cache.total_bytes);
780            }
781        }
782        // Invalidate file size cache
783        {
784            let mut file_sizes = self.file_sizes.write();
785            file_sizes.remove(path);
786        }
787        // Delegate to inner
788        self.inner.write(path, data).await
789    }
790
791    async fn delete(&self, path: &Path) -> io::Result<()> {
792        // Invalidate cache for this file
793        {
794            let mut caches = self.caches.write();
795            if let Some(file_cache) = caches.remove(path) {
796                let mut current = self.current_bytes.write();
797                *current = current.saturating_sub(file_cache.total_bytes);
798            }
799        }
800        // Invalidate file size cache
801        {
802            let mut file_sizes = self.file_sizes.write();
803            file_sizes.remove(path);
804        }
805        // Delegate to inner
806        self.inner.delete(path).await
807    }
808
809    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
810        // Move cache entries from old path to new path
811        {
812            let mut caches = self.caches.write();
813            if let Some(file_cache) = caches.remove(from) {
814                caches.insert(to.to_path_buf(), file_cache);
815            }
816        }
817        // Move file size cache
818        {
819            let mut file_sizes = self.file_sizes.write();
820            if let Some(size) = file_sizes.remove(from) {
821                file_sizes.insert(to.to_path_buf(), size);
822            }
823        }
824        // Delegate to inner
825        self.inner.rename(from, to).await
826    }
827
828    async fn link(&self, from: &Path, to: &Path) -> io::Result<()> {
829        // A link creates an immutable alias. Do not copy cache entries: the
830        // destination starts cold and is populated under its own path.
831        self.inner.link(from, to).await
832    }
833
834    async fn sync(&self) -> io::Result<()> {
835        self.inner.sync().await
836    }
837
838    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn super::StreamingWriter>> {
839        // Invalidate cache for this file before writing
840        {
841            let mut caches = self.caches.write();
842            if let Some(file_cache) = caches.remove(path) {
843                let mut current = self.current_bytes.write();
844                *current = current.saturating_sub(file_cache.total_bytes);
845            }
846        }
847        {
848            let mut file_sizes = self.file_sizes.write();
849            file_sizes.remove(path);
850        }
851        self.inner.streaming_writer(path).await
852    }
853}
854
855#[cfg(test)]
856mod tests {
857    use super::*;
858    use crate::directories::{DirectoryWriter, RamDirectory};
859
860    #[tokio::test]
861    async fn test_slice_cache_basic() {
862        let ram = RamDirectory::new();
863        ram.write(Path::new("test.bin"), &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
864            .await
865            .unwrap();
866
867        let cached = SliceCachingDirectory::new(ram, 1024);
868
869        // First read - cache miss
870        let data = cached
871            .read_range(Path::new("test.bin"), 2..5)
872            .await
873            .unwrap();
874        assert_eq!(data.as_slice(), &[2, 3, 4]);
875
876        // Second read - should be cache hit
877        let data = cached
878            .read_range(Path::new("test.bin"), 2..5)
879            .await
880            .unwrap();
881        assert_eq!(data.as_slice(), &[2, 3, 4]);
882
883        let stats = cached.stats();
884        assert_eq!(stats.total_slices, 1);
885        assert_eq!(stats.total_bytes, 3);
886    }
887
888    #[tokio::test]
889    async fn slice_cache_hits_reuse_the_cached_backing_allocation() {
890        let ram = RamDirectory::new();
891        ram.write(Path::new("test.bin"), &[7; 64]).await.unwrap();
892        let cached = SliceCachingDirectory::new(ram, 64);
893
894        let miss = cached
895            .read_range(Path::new("test.bin"), 8..56)
896            .await
897            .unwrap();
898        let hit = cached
899            .read_range(Path::new("test.bin"), 8..56)
900            .await
901            .unwrap();
902
903        assert_eq!(miss.as_slice(), hit.as_slice());
904        assert_eq!(miss.as_slice().as_ptr(), hit.as_slice().as_ptr());
905    }
906
907    #[tokio::test]
908    async fn oversized_slice_bypasses_cache_instead_of_exceeding_limit() {
909        let ram = RamDirectory::new();
910        ram.write(Path::new("test.bin"), &[3; 32]).await.unwrap();
911        let cached = SliceCachingDirectory::new(ram, 8);
912
913        let data = cached
914            .read_range(Path::new("test.bin"), 0..32)
915            .await
916            .unwrap();
917        assert_eq!(data.len(), 32);
918        assert_eq!(cached.stats().total_bytes, 0);
919    }
920
921    #[tokio::test]
922    async fn test_slice_cache_overlap_merge() {
923        let ram = RamDirectory::new();
924        ram.write(Path::new("test.bin"), &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
925            .await
926            .unwrap();
927
928        let cached = SliceCachingDirectory::new(ram, 1024);
929
930        // Read [2..5]
931        cached
932            .read_range(Path::new("test.bin"), 2..5)
933            .await
934            .unwrap();
935
936        // Read [4..7] - overlaps with previous
937        cached
938            .read_range(Path::new("test.bin"), 4..7)
939            .await
940            .unwrap();
941
942        let stats = cached.stats();
943        // Should be merged into one slice [2..7]
944        assert_eq!(stats.total_slices, 1);
945        assert_eq!(stats.total_bytes, 5); // bytes 2,3,4,5,6
946
947        // Reading from merged range should work
948        let data = cached
949            .read_range(Path::new("test.bin"), 3..6)
950            .await
951            .unwrap();
952        assert_eq!(data.as_slice(), &[3, 4, 5]);
953    }
954
955    #[tokio::test]
956    async fn test_slice_cache_eviction() {
957        let ram = RamDirectory::new();
958        ram.write(Path::new("test.bin"), &[0; 100]).await.unwrap();
959
960        // Small cache limit
961        let cached = SliceCachingDirectory::new(ram, 50);
962
963        // Fill cache
964        cached
965            .read_range(Path::new("test.bin"), 0..30)
966            .await
967            .unwrap();
968
969        // This should trigger eviction
970        cached
971            .read_range(Path::new("test.bin"), 50..80)
972            .await
973            .unwrap();
974
975        let stats = cached.stats();
976        assert!(stats.total_bytes <= 50);
977    }
978
979    #[tokio::test]
980    async fn test_slice_cache_serialize_deserialize() {
981        let ram = RamDirectory::new();
982        ram.write(Path::new("file1.bin"), &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
983            .await
984            .unwrap();
985        ram.write(Path::new("file2.bin"), &[10, 11, 12, 13, 14, 15])
986            .await
987            .unwrap();
988
989        let cached = SliceCachingDirectory::new(ram.clone(), 1024);
990
991        // Read some ranges to populate cache
992        cached
993            .read_range(Path::new("file1.bin"), 2..6)
994            .await
995            .unwrap();
996        cached
997            .read_range(Path::new("file2.bin"), 1..4)
998            .await
999            .unwrap();
1000
1001        let stats = cached.stats();
1002        assert_eq!(stats.files_cached, 2);
1003        assert_eq!(stats.total_bytes, 7); // 4 + 3
1004
1005        // Serialize
1006        let serialized = cached.serialize();
1007        assert!(!serialized.is_empty());
1008
1009        // Create new cache and deserialize
1010        let cached2 = SliceCachingDirectory::new(ram.clone(), 1024);
1011        assert!(cached2.is_empty());
1012
1013        cached2.deserialize(&serialized).unwrap();
1014
1015        let stats2 = cached2.stats();
1016        assert_eq!(stats2.files_cached, 2);
1017        assert_eq!(stats2.total_bytes, 7);
1018
1019        // Verify cached data is correct by reading (should be cache hits)
1020        let data = cached2
1021            .read_range(Path::new("file1.bin"), 2..6)
1022            .await
1023            .unwrap();
1024        assert_eq!(data.as_slice(), &[2, 3, 4, 5]);
1025
1026        let data = cached2
1027            .read_range(Path::new("file2.bin"), 1..4)
1028            .await
1029            .unwrap();
1030        assert_eq!(data.as_slice(), &[11, 12, 13]);
1031    }
1032
1033    #[tokio::test]
1034    async fn test_slice_cache_serialize_empty() {
1035        let ram = RamDirectory::new();
1036        let cached = SliceCachingDirectory::new(ram, 1024);
1037
1038        // Serialize empty cache
1039        let serialized = cached.serialize();
1040        assert!(!serialized.is_empty()); // Should have header
1041
1042        // Deserialize into new cache
1043        let cached2 = SliceCachingDirectory::new(RamDirectory::new(), 1024);
1044        cached2.deserialize(&serialized).unwrap();
1045        assert!(cached2.is_empty());
1046    }
1047
1048    #[tokio::test]
1049    async fn deserialization_enforces_the_destination_cache_limit() {
1050        let ram = RamDirectory::new();
1051        ram.write(Path::new("test.bin"), &[1; 64]).await.unwrap();
1052        let source = SliceCachingDirectory::new(ram.clone(), 64);
1053        source
1054            .read_range(Path::new("test.bin"), 0..64)
1055            .await
1056            .unwrap();
1057
1058        let destination = SliceCachingDirectory::new(ram, 8);
1059        destination.deserialize(&source.serialize()).unwrap();
1060        assert!(destination.stats().total_bytes <= 8);
1061    }
1062}