aptu_coder_core/
cache.rs

1// SPDX-FileCopyrightText: 2026 aptu-coder contributors
2// SPDX-License-Identifier: Apache-2.0
3//! LRU cache for analysis results indexed by path, modification time, and mode.
4//!
5//! Provides thread-safe, capacity-bounded caching of file analysis outputs using LRU eviction.
6//! Recovers gracefully from poisoned mutex conditions.
7
8use crate::analyze::{AnalysisOutput, FileAnalysisOutput, FocusedAnalysisOutput};
9use crate::traversal::WalkEntry;
10use crate::types::{AnalysisMode, SymbolMatchMode};
11use fs2::FileExt;
12use lru::LruCache;
13use rayon::prelude::*;
14use serde::{Serialize, de::DeserializeOwned};
15use std::num::NonZeroUsize;
16#[cfg(unix)]
17use std::os::unix::fs::PermissionsExt;
18use std::path::PathBuf;
19use std::sync::{Arc, Mutex};
20use std::time::SystemTime;
21use tempfile::NamedTempFile;
22use tracing::{debug, error, instrument, warn};
23
24/// Indicates which cache tier served the result.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum CacheTier {
27    L1Memory,
28    L2Disk,
29    Miss,
30}
31
32/// Parse an LRU cache capacity from an environment variable.
33///
34/// Reads `env_key`, parses it as `usize`, and returns the value clamped to a minimum of 1.
35/// Falls back to `default` when the variable is absent or unparseable, then also clamps
36/// the fallback to at least 1.
37///
38/// This helper centralises all three LRU init sites so the `.max(1)` guard lives in one place.
39#[must_use]
40pub fn parse_cache_capacity(env_key: &str, default: usize) -> usize {
41    std::env::var(env_key)
42        .ok()
43        .and_then(|v| v.parse::<usize>().ok())
44        .unwrap_or(default)
45        .max(1)
46}
47
48impl CacheTier {
49    #[must_use]
50    pub fn as_str(&self) -> &'static str {
51        match self {
52            CacheTier::L1Memory => "l1_memory",
53            CacheTier::L2Disk => "l2_disk",
54            CacheTier::Miss => "miss",
55        }
56    }
57}
58
59/// Cache key combining path, modification time, and analysis mode.
60#[derive(Debug, Clone, Eq, PartialEq, Hash)]
61pub struct CacheKey {
62    pub path: PathBuf,
63    pub modified: SystemTime,
64    pub mode: AnalysisMode,
65}
66
67/// Cache key for directory analysis combining file mtimes, mode, and `max_depth`.
68#[derive(Debug, Clone, Eq, PartialEq, Hash)]
69pub struct DirectoryCacheKey {
70    files: Vec<(PathBuf, SystemTime)>,
71    mode: AnalysisMode,
72    max_depth: Option<u32>,
73    git_ref: Option<String>,
74}
75
76impl DirectoryCacheKey {
77    /// Build a cache key from walk entries, capturing mtime for each file.
78    /// Files are sorted by path for deterministic hashing.
79    /// Directories are filtered out; only file entries are processed.
80    /// Metadata collection is parallelized using rayon.
81    /// The `git_ref` is included so that filtered and unfiltered results have distinct keys.
82    #[must_use]
83    pub fn from_entries(
84        entries: &[WalkEntry],
85        max_depth: Option<u32>,
86        mode: AnalysisMode,
87        git_ref: Option<&str>,
88    ) -> Self {
89        let mut files: Vec<(PathBuf, SystemTime)> = entries
90            .par_iter()
91            .filter(|e| !e.is_dir)
92            .map(|e| {
93                let mtime = e.mtime.unwrap_or(SystemTime::UNIX_EPOCH);
94                (e.path.clone(), mtime)
95            })
96            .collect();
97        files.sort_by(|a, b| a.0.cmp(&b.0));
98        Self {
99            files,
100            mode,
101            max_depth,
102            git_ref: git_ref.map(ToOwned::to_owned),
103        }
104    }
105}
106
107/// Recover from a poisoned mutex by clearing the cache.
108/// On poison, creates a new empty cache and returns the recovery value.
109fn lock_or_recover<K, V, T, F>(mutex: &Mutex<LruCache<K, V>>, capacity: usize, recovery: F) -> T
110where
111    K: std::hash::Hash + Eq,
112    F: FnOnce(&mut LruCache<K, V>) -> T,
113{
114    match mutex.lock() {
115        Ok(mut guard) => recovery(&mut guard),
116        Err(poisoned) => {
117            // SAFETY: 100 is a non-zero literal
118            let cache_size = NonZeroUsize::new(capacity)
119                .unwrap_or_else(|| NonZeroUsize::new(100).expect("100 is non-zero"));
120            let new_cache = LruCache::new(cache_size);
121            let mut guard = poisoned.into_inner();
122            *guard = new_cache;
123            recovery(&mut guard)
124        }
125    }
126}
127
128/// Cache key for call graph analysis combining path, parameters, and file mtimes.
129#[derive(Debug, Clone, Eq, PartialEq, Hash)]
130pub struct CallGraphCacheKey {
131    root_path: PathBuf,
132    git_ref: Option<String>,
133    follow_depth: u32,
134    match_mode: SymbolMatchMode,
135    impl_only: bool,
136    ast_recursion_limit: Option<usize>,
137    /// Sorted (path, mtime_as_unix_nanos) pairs for all non-dir entries.
138    file_mtimes: Vec<(PathBuf, u64)>,
139}
140
141impl CallGraphCacheKey {
142    /// Build a `CallGraphCacheKey` from walk entries and analysis parameters.
143    /// Files are sorted by path for deterministic hashing.
144    /// Directories are filtered out; only file entries contribute to the key.
145    #[must_use]
146    pub fn from_entries(
147        root: &std::path::Path,
148        entries: &[WalkEntry],
149        git_ref: Option<&str>,
150        follow_depth: u32,
151        match_mode: &SymbolMatchMode,
152        impl_only: bool,
153        ast_recursion_limit: Option<usize>,
154    ) -> Self {
155        let mut file_mtimes: Vec<(PathBuf, u64)> = entries
156            .par_iter()
157            .filter(|e| !e.is_dir)
158            .map(|e| {
159                let mtime = e
160                    .mtime
161                    .unwrap_or(SystemTime::UNIX_EPOCH)
162                    .duration_since(SystemTime::UNIX_EPOCH)
163                    .map(|d| d.as_nanos() as u64)
164                    .unwrap_or(0);
165                (e.path.clone(), mtime)
166            })
167            .collect();
168        file_mtimes.sort_by(|a, b| a.0.cmp(&b.0));
169        Self {
170            root_path: root.to_path_buf(),
171            git_ref: git_ref.map(ToOwned::to_owned),
172            follow_depth,
173            match_mode: match_mode.clone(),
174            impl_only,
175            ast_recursion_limit,
176            file_mtimes,
177        }
178    }
179}
180
181/// Cached call graph result: the fully-built `FocusedAnalysisOutput`.
182/// `CallGraph` is not serializable, so caching is L1 memory only.
183pub type CallGraphCacheValue = Arc<FocusedAnalysisOutput>;
184
185/// L1 in-memory LRU cache for call graph results.
186/// Capacity is controlled via `APTU_CODER_SYMBOL_CACHE_CAPACITY` env var (default 32).
187pub struct CallGraphCache {
188    capacity: usize,
189    cache: Arc<Mutex<LruCache<CallGraphCacheKey, CallGraphCacheValue>>>,
190}
191
192impl CallGraphCache {
193    /// Create a new `CallGraphCache` with the given capacity.
194    ///
195    /// `capacity` is clamped to a minimum of 1 so a zero value does not panic.
196    #[must_use]
197    pub fn new(capacity: usize) -> Self {
198        let capacity = capacity.max(1);
199        let cache_size = NonZeroUsize::new(capacity).expect("capacity is non-zero after .max(1)");
200        Self {
201            capacity,
202            cache: Arc::new(Mutex::new(LruCache::new(cache_size))),
203        }
204    }
205
206    /// Look up a cached result by key. Returns `None` on miss or mutex poison.
207    #[must_use]
208    pub fn get(&self, key: &CallGraphCacheKey) -> Option<CallGraphCacheValue> {
209        lock_or_recover(&self.cache, self.capacity, |guard| guard.get(key).cloned())
210    }
211
212    /// Store a result in the cache.
213    pub fn put(&self, key: CallGraphCacheKey, value: CallGraphCacheValue) {
214        lock_or_recover(&self.cache, self.capacity, |guard| {
215            guard.put(key, value);
216        });
217    }
218}
219
220impl Clone for CallGraphCache {
221    fn clone(&self) -> Self {
222        Self {
223            capacity: self.capacity,
224            cache: Arc::clone(&self.cache),
225        }
226    }
227}
228
229/// LRU cache for file analysis results with mutex protection.
230pub struct AnalysisCache {
231    file_capacity: usize,
232    dir_capacity: usize,
233    cache: Arc<Mutex<LruCache<CacheKey, Arc<FileAnalysisOutput>>>>,
234    directory_cache: Arc<Mutex<LruCache<DirectoryCacheKey, Arc<AnalysisOutput>>>>,
235}
236
237impl AnalysisCache {
238    /// Create a new cache with the specified file capacity.
239    /// The directory cache capacity is read from the `APTU_CODER_DIR_CACHE_CAPACITY`
240    /// environment variable (default: 20).
241    #[must_use]
242    pub fn new(capacity: usize) -> Self {
243        let file_capacity = capacity.max(1);
244        let dir_capacity = parse_cache_capacity("APTU_CODER_DIR_CACHE_CAPACITY", 20);
245        let cache_size =
246            NonZeroUsize::new(file_capacity).expect("file_capacity is non-zero after .max(1)");
247        let dir_cache_size =
248            NonZeroUsize::new(dir_capacity).expect("dir_capacity is non-zero after .max(1)");
249        Self {
250            file_capacity,
251            dir_capacity,
252            cache: Arc::new(Mutex::new(LruCache::new(cache_size))),
253            directory_cache: Arc::new(Mutex::new(LruCache::new(dir_cache_size))),
254        }
255    }
256
257    /// Get a cached analysis result if it exists.
258    #[instrument(skip(self), fields(path = ?key.path))]
259    pub fn get(&self, key: &CacheKey) -> Option<Arc<FileAnalysisOutput>> {
260        lock_or_recover(&self.cache, self.file_capacity, |guard| {
261            let result = guard.get(key).cloned();
262            let cache_size = guard.len();
263            if let Some(v) = result {
264                debug!(cache_event = "hit", cache_size = cache_size, path = ?key.path);
265                Some(v)
266            } else {
267                debug!(cache_event = "miss", cache_size = cache_size, path = ?key.path);
268                None
269            }
270        })
271    }
272
273    /// Store an analysis result in the cache.
274    #[instrument(skip(self, value), fields(path = ?key.path))]
275    // public API; callers expect owned semantics
276    #[allow(clippy::needless_pass_by_value)]
277    pub fn put(&self, key: CacheKey, value: Arc<FileAnalysisOutput>) {
278        lock_or_recover(&self.cache, self.file_capacity, |guard| {
279            let push_result = guard.push(key.clone(), value);
280            let cache_size = guard.len();
281            match push_result {
282                None => {
283                    debug!(cache_event = "insert", cache_size = cache_size, path = ?key.path);
284                }
285                Some((returned_key, _)) => {
286                    if returned_key == key {
287                        debug!(cache_event = "update", cache_size = cache_size, path = ?key.path);
288                    } else {
289                        debug!(cache_event = "eviction", cache_size = cache_size, path = ?key.path, evicted_path = ?returned_key.path);
290                    }
291                }
292            }
293        });
294    }
295
296    /// Get a cached directory analysis result if it exists.
297    #[instrument(skip(self))]
298    pub fn get_directory(&self, key: &DirectoryCacheKey) -> Option<Arc<AnalysisOutput>> {
299        lock_or_recover(&self.directory_cache, self.dir_capacity, |guard| {
300            let result = guard.get(key).cloned();
301            let cache_size = guard.len();
302            if let Some(v) = result {
303                debug!(cache_event = "hit", cache_size = cache_size);
304                Some(v)
305            } else {
306                debug!(cache_event = "miss", cache_size = cache_size);
307                None
308            }
309        })
310    }
311
312    /// Store a directory analysis result in the cache.
313    #[instrument(skip(self, value))]
314    pub fn put_directory(&self, key: DirectoryCacheKey, value: Arc<AnalysisOutput>) {
315        lock_or_recover(&self.directory_cache, self.dir_capacity, |guard| {
316            let push_result = guard.push(key, value);
317            let cache_size = guard.len();
318            match push_result {
319                None => {
320                    debug!(cache_event = "insert", cache_size = cache_size);
321                }
322                Some((_, _)) => {
323                    debug!(cache_event = "eviction", cache_size = cache_size);
324                }
325            }
326        });
327    }
328
329    /// Returns the configured file-cache capacity.
330    /// Exposed for testing across crate boundaries; not part of the stable API.
331    #[doc(hidden)]
332    #[must_use]
333    pub fn file_capacity(&self) -> usize {
334        self.file_capacity
335    }
336
337    /// Invalidate all cache entries for a given file path.
338    /// Removes all entries regardless of modification time or analysis mode.
339    #[instrument(skip(self), fields(path = ?path))]
340    pub fn invalidate_file(&self, path: &std::path::Path) {
341        lock_or_recover(&self.cache, self.file_capacity, |guard| {
342            let keys: Vec<CacheKey> = guard
343                .iter()
344                .filter(|(k, _)| k.path == path)
345                .map(|(k, _)| k.clone())
346                .collect();
347            for key in keys {
348                guard.pop(&key);
349            }
350            let cache_size = guard.len();
351            debug!(cache_event = "invalidate_file", cache_size = cache_size, path = ?path);
352        });
353    }
354}
355
356impl Clone for AnalysisCache {
357    fn clone(&self) -> Self {
358        Self {
359            file_capacity: self.file_capacity,
360            dir_capacity: self.dir_capacity,
361            cache: Arc::clone(&self.cache),
362            directory_cache: Arc::clone(&self.directory_cache),
363        }
364    }
365}
366
367#[cfg(test)]
368mod tests {
369    use super::*;
370    use crate::types::SemanticAnalysis;
371
372    #[test]
373    fn test_from_entries_skips_dirs() {
374        // Arrange: create a real temp dir and a real temp file for hermetic isolation.
375        let dir = tempfile::tempdir().expect("tempdir");
376        let file = tempfile::NamedTempFile::new_in(dir.path()).expect("tempfile");
377        let file_path = file.path().to_path_buf();
378
379        let entries = vec![
380            WalkEntry {
381                path: dir.path().to_path_buf(),
382                depth: 0,
383                is_dir: true,
384                is_symlink: false,
385                symlink_target: None,
386                mtime: None,
387                canonical_path: PathBuf::new(),
388            },
389            WalkEntry {
390                path: file_path.clone(),
391                depth: 0,
392                is_dir: false,
393                is_symlink: false,
394                symlink_target: None,
395                mtime: None,
396                canonical_path: PathBuf::new(),
397            },
398        ];
399
400        // Act: build cache key from entries
401        let key = DirectoryCacheKey::from_entries(&entries, None, AnalysisMode::Overview, None);
402
403        // Assert: only the file entry should be in the cache key
404        // The directory entry should be filtered out
405        assert_eq!(key.files.len(), 1);
406        assert_eq!(key.files[0].0, file_path);
407    }
408
409    #[test]
410    fn test_invalidate_file_single_mode() {
411        // Arrange: create a cache and insert one entry for a path
412        let cache = AnalysisCache::new(10);
413        let path = PathBuf::from("/test/file.rs");
414        let key = CacheKey {
415            path: path.clone(),
416            modified: SystemTime::UNIX_EPOCH,
417            mode: AnalysisMode::Overview,
418        };
419        let output = Arc::new(FileAnalysisOutput::new(
420            String::new(),
421            SemanticAnalysis::default(),
422            0,
423            None,
424        ));
425        cache.put(key.clone(), output);
426
427        // Act: invalidate the file
428        cache.invalidate_file(&path);
429
430        // Assert: the entry should be removed
431        assert!(cache.get(&key).is_none());
432    }
433
434    #[test]
435    fn test_invalidate_file_multi_mode() {
436        // Arrange: create a cache and insert two entries for the same path with different modes
437        let cache = AnalysisCache::new(10);
438        let path = PathBuf::from("/test/file.rs");
439        let key1 = CacheKey {
440            path: path.clone(),
441            modified: SystemTime::UNIX_EPOCH,
442            mode: AnalysisMode::Overview,
443        };
444        let key2 = CacheKey {
445            path: path.clone(),
446            modified: SystemTime::UNIX_EPOCH,
447            mode: AnalysisMode::FileDetails,
448        };
449        let output = Arc::new(FileAnalysisOutput::new(
450            String::new(),
451            SemanticAnalysis::default(),
452            0,
453            None,
454        ));
455        cache.put(key1.clone(), output.clone());
456        cache.put(key2.clone(), output);
457
458        // Act: invalidate the file
459        cache.invalidate_file(&path);
460
461        // Assert: both entries should be removed
462        assert!(cache.get(&key1).is_none());
463        assert!(cache.get(&key2).is_none());
464    }
465
466    // Mutex serialises the two dir-cache-capacity tests to prevent env var races.
467    static DIR_CACHE_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
468
469    #[test]
470    fn test_dir_cache_capacity_default() {
471        let _guard = DIR_CACHE_ENV_LOCK.lock().unwrap();
472
473        // Arrange: ensure the env var is not set
474        unsafe { std::env::remove_var("APTU_CODER_DIR_CACHE_CAPACITY") };
475
476        // Act
477        let cache = AnalysisCache::new(100);
478
479        // Assert: default dir capacity is 20
480        assert_eq!(cache.dir_capacity, 20);
481    }
482
483    #[test]
484    fn test_dir_cache_capacity_from_env() {
485        let _guard = DIR_CACHE_ENV_LOCK.lock().unwrap();
486
487        // Arrange
488        unsafe { std::env::set_var("APTU_CODER_DIR_CACHE_CAPACITY", "7") };
489
490        // Act
491        let cache = AnalysisCache::new(100);
492
493        // Cleanup before assertions to minimise env pollution window
494        unsafe { std::env::remove_var("APTU_CODER_DIR_CACHE_CAPACITY") };
495
496        // Assert
497        assert_eq!(cache.dir_capacity, 7);
498    }
499
500    // Mutex serialises parse_cache_capacity tests that set env vars.
501    static PARSE_CAP_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
502
503    #[test]
504    fn test_parse_cache_capacity_missing_returns_default() {
505        let _guard = PARSE_CAP_ENV_LOCK.lock().unwrap();
506
507        // Arrange: env var is absent
508        unsafe { std::env::remove_var("_TEST_APTU_PARSE_CAP") };
509
510        // Act
511        let result = parse_cache_capacity("_TEST_APTU_PARSE_CAP", 42);
512
513        // Assert: default is returned as-is
514        assert_eq!(result, 42);
515    }
516
517    #[test]
518    fn test_parse_cache_capacity_valid_returns_value() {
519        let _guard = PARSE_CAP_ENV_LOCK.lock().unwrap();
520
521        // Arrange
522        unsafe { std::env::set_var("_TEST_APTU_PARSE_CAP", "64") };
523
524        // Act
525        let result = parse_cache_capacity("_TEST_APTU_PARSE_CAP", 10);
526
527        // Cleanup
528        unsafe { std::env::remove_var("_TEST_APTU_PARSE_CAP") };
529
530        // Assert: parsed value is returned
531        assert_eq!(result, 64);
532    }
533
534    #[test]
535    fn test_parse_cache_capacity_zero_returns_one() {
536        let _guard = PARSE_CAP_ENV_LOCK.lock().unwrap();
537
538        // Arrange: zero is below the minimum of 1
539        unsafe { std::env::set_var("_TEST_APTU_PARSE_CAP", "0") };
540
541        // Act
542        let result = parse_cache_capacity("_TEST_APTU_PARSE_CAP", 10);
543
544        // Cleanup
545        unsafe { std::env::remove_var("_TEST_APTU_PARSE_CAP") };
546
547        // Assert: clamped to 1
548        assert_eq!(result, 1);
549    }
550
551    #[test]
552    fn test_parse_cache_capacity_garbage_returns_default() {
553        let _guard = PARSE_CAP_ENV_LOCK.lock().unwrap();
554
555        // Arrange: unparseable string
556        unsafe { std::env::set_var("_TEST_APTU_PARSE_CAP", "not_a_number") };
557
558        // Act
559        let result = parse_cache_capacity("_TEST_APTU_PARSE_CAP", 8);
560
561        // Cleanup
562        unsafe { std::env::remove_var("_TEST_APTU_PARSE_CAP") };
563
564        // Assert: falls back to default
565        assert_eq!(result, 8);
566    }
567}
568
569/// Persistent content-addressable disk cache for analyze_* tools.
570/// All methods are infallible from the caller's perspective: errors are silently dropped.
571/// Number of consecutive L2 write failures that triggers an `error!` log escalation.
572/// Below this threshold each failure logs at `warn!`. At or above it the cache is
573/// considered degraded and a single `error!` is emitted so operators are alerted
574/// without flooding logs on a sustained disk-full condition.
575const DISK_CACHE_DEGRADED_THRESHOLD: u64 = 3;
576
577pub struct DiskCache {
578    base: std::path::PathBuf,
579    disabled: bool,
580    /// Counts write failures since last drain. Incremented inside `put` on any I/O error.
581    write_failures: std::sync::atomic::AtomicU64,
582    /// Cumulative write failures across all drains. Never reset; used for threshold checks.
583    total_write_failures: std::sync::atomic::AtomicU64,
584}
585
586impl DiskCache {
587    /// Returns the number of write failures accumulated since the last call and resets the
588    /// per-drain counter. The cumulative `total_write_failures` is never reset.
589    pub fn drain_write_failures(&self) -> u64 {
590        self.write_failures
591            .swap(0, std::sync::atomic::Ordering::Relaxed)
592    }
593
594    /// Returns true when cumulative write failures have reached `DISK_CACHE_DEGRADED_THRESHOLD`.
595    /// Callers can use this to emit a degraded health signal without polling the counter.
596    pub fn is_degraded(&self) -> bool {
597        self.total_write_failures
598            .load(std::sync::atomic::Ordering::Relaxed)
599            >= DISK_CACHE_DEGRADED_THRESHOLD
600    }
601}
602
603impl DiskCache {
604    /// Creates the cache directory (mode 0700) and returns a new instance.
605    /// If `disabled` is true, or if directory creation fails, all operations are no-ops.
606    pub fn new(base: std::path::PathBuf, disabled: bool) -> Self {
607        if disabled {
608            return Self {
609                base,
610                disabled: true,
611                write_failures: std::sync::atomic::AtomicU64::new(0),
612                total_write_failures: std::sync::atomic::AtomicU64::new(0),
613            };
614        }
615        if let Err(e) = std::fs::create_dir_all(&base) {
616            warn!(path = %base.display(), error = %e, "disk cache disabled: failed to create cache directory");
617            return Self {
618                base,
619                disabled: true,
620                write_failures: std::sync::atomic::AtomicU64::new(0),
621                total_write_failures: std::sync::atomic::AtomicU64::new(0),
622            };
623        }
624        #[cfg(unix)]
625        if let Err(e) = std::fs::set_permissions(&base, std::fs::Permissions::from_mode(0o700)) {
626            warn!(path = %base.display(), error = %e, "disk cache: failed to set directory permissions to 0700");
627        }
628        #[cfg(not(unix))]
629        let _ = &base; // permissions not supported on this platform
630        Self {
631            base,
632            disabled: false,
633            write_failures: std::sync::atomic::AtomicU64::new(0),
634            total_write_failures: std::sync::atomic::AtomicU64::new(0),
635        }
636    }
637
638    pub fn entry_path(&self, tool: &str, key: &blake3::Hash) -> std::path::PathBuf {
639        let hex = format!("{}", key);
640        self.base
641            .join(tool)
642            .join(&hex[..2])
643            .join(format!("{}.json.snap", hex))
644    }
645
646    /// Returns None if entry is absent or corrupt. Never propagates errors.
647    pub fn get<T: DeserializeOwned>(&self, tool: &str, key: &blake3::Hash) -> Option<T> {
648        if self.disabled {
649            return None;
650        }
651        let path = self.entry_path(tool, key);
652        // Acquire shared lock on per-shard .lock sentinel before reading
653        let _lock = lock_shard_shared(path.parent()?);
654        let compressed = match std::fs::read(&path) {
655            Ok(b) => b,
656            Err(_) => return None,
657        };
658        let bytes = match snap::raw::Decoder::new().decompress_vec(&compressed) {
659            Ok(b) => b,
660            Err(e) => {
661                debug!(tool, error = %e, "disk cache decompression failed");
662                return None;
663            }
664        };
665        match serde_json::from_slice(&bytes) {
666            Ok(v) => Some(v),
667            Err(e) => {
668                debug!(tool, error = %e, "disk cache deserialization failed");
669                None
670            }
671        }
672    }
673
674    /// Serialize and compress a value. Returns None if serialization or compression fails.
675    fn serialize_entry<T: Serialize>(value: &T) -> Option<Vec<u8>> {
676        let bytes = serde_json::to_vec(value).ok()?;
677        snap::raw::Encoder::new().compress_vec(&bytes).ok()
678    }
679
680    /// Write compressed data to a temporary file and atomically rename it to the target path.
681    /// Acquires an exclusive lock on the per-shard .lock sentinel before the persist (rename)
682    /// to prevent concurrent writes from corrupting the cache entry.
683    /// Returns Err if any step fails; caller silently drops the error.
684    fn write_entry_atomically(
685        dir: &std::path::Path,
686        path: &std::path::Path,
687        compressed: &[u8],
688    ) -> Result<(), std::io::Error> {
689        use std::io::Write;
690        // Acquire exclusive lock on per-shard .lock sentinel before writing
691        let _lock = lock_shard_exclusive(dir)?;
692        let mut tmp = NamedTempFile::new_in(dir)?;
693        tmp.write_all(compressed)?;
694        tmp.persist(path).map(|_| ()).map_err(|e| e.error)
695    }
696
697    /// Atomic write via NamedTempFile::persist (rename(2)). Silently drops all errors.
698    pub fn put<T: Serialize>(&self, tool: &str, key: &blake3::Hash, value: &T) {
699        if self.disabled {
700            return;
701        }
702        let path = self.entry_path(tool, key);
703        let dir = match path.parent() {
704            Some(d) => d.to_path_buf(),
705            None => return,
706        };
707        if let Err(e) = std::fs::create_dir_all(&dir) {
708            warn!(tool, error = %e, "disk cache: failed to create cache directory");
709            self.record_write_failure();
710            return;
711        }
712        let compressed = match Self::serialize_entry(value) {
713            Some(c) => c,
714            None => return,
715        };
716        if Self::write_entry_atomically(&dir, &path, &compressed)
717            .ok()
718            .is_none()
719        {
720            self.record_write_failure();
721        }
722    }
723
724    /// Increments both the per-drain and cumulative failure counters. Escalates to `error!`
725    /// once cumulative failures reach `DISK_CACHE_DEGRADED_THRESHOLD` so a sustained
726    /// disk-full or permission problem surfaces above the noise of individual `warn!` entries.
727    fn record_write_failure(&self) {
728        self.write_failures
729            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
730        let total = self
731            .total_write_failures
732            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
733            + 1;
734        if total == DISK_CACHE_DEGRADED_THRESHOLD {
735            error!(
736                path = %self.base.display(),
737                total,
738                threshold = DISK_CACHE_DEGRADED_THRESHOLD,
739                "disk cache is degraded: consecutive write failures have reached the alert threshold; \
740                 check disk space and permissions at the cache directory"
741            );
742        }
743    }
744
745    /// Removes files not accessed within retention_days. Best-effort; silently drops errors.
746    pub fn evict_stale(&self, retention_days: u64) {
747        if self.disabled {
748            return;
749        }
750        let cutoff = std::time::SystemTime::now()
751            .checked_sub(std::time::Duration::from_secs(retention_days * 86_400))
752            .unwrap_or(std::time::UNIX_EPOCH);
753        let _ = evict_dir_recursive(&self.base, cutoff);
754    }
755}
756
757fn evict_dir_recursive(
758    dir: &std::path::Path,
759    cutoff: std::time::SystemTime,
760) -> std::io::Result<()> {
761    for entry in std::fs::read_dir(dir)? {
762        let entry = entry?;
763        let meta = entry.metadata()?;
764        let path = entry.path();
765        if meta.is_dir() {
766            let _ = evict_dir_recursive(&path, cutoff);
767        } else if meta.is_file()
768            && let Ok(mtime) = meta.modified()
769            && mtime < cutoff
770        {
771            let _ = std::fs::remove_file(&path);
772        }
773    }
774    Ok(())
775}
776
777/// Acquire a shared (read) lock on the per-shard `.lock` sentinel.
778/// Creates the lock file if it does not exist. Lock failures degrade
779/// gracefully (warn and return None) so that read availability is
780/// never blocked by lock infrastructure issues.
781fn lock_shard_shared(shard_dir: &std::path::Path) -> Option<ShardLockGuard> {
782    let lock_path = shard_dir.join(".lock");
783    let file = std::fs::OpenOptions::new()
784        .create(true)
785        .write(true)
786        .truncate(false)
787        .open(&lock_path)
788        .ok()?;
789    match file.lock_shared() {
790        Ok(()) => Some(ShardLockGuard(file)),
791        Err(e) => {
792            warn!(
793                error = %e, lock_path = %lock_path.display(),
794                "disk cache: failed to acquire shared lock on shard; proceeding without lock"
795            );
796            None
797        }
798    }
799}
800
801/// Acquire an exclusive (write) lock on the per-shard `.lock` sentinel.
802/// Creates the lock file if it does not exist. Returns Err if the lock
803/// file cannot be opened or if the lock acquisition fails, propagating
804/// the error to the caller (which typically degrades gracefully).
805fn lock_shard_exclusive(shard_dir: &std::path::Path) -> Result<ShardLockGuard, std::io::Error> {
806    let lock_path = shard_dir.join(".lock");
807    let file = std::fs::OpenOptions::new()
808        .create(true)
809        .write(true)
810        .truncate(false)
811        .open(&lock_path)?;
812    file.lock_exclusive()?;
813    Ok(ShardLockGuard(file))
814}
815
816/// RAII guard that releases a per-shard flock when dropped.
817/// Closing the underlying file descriptor releases the BSD/OFC lock.
818struct ShardLockGuard(
819    /// Held exclusively for its `Drop` implementation: closing the file
820    /// descriptor releases the advisory flock. Never read directly.
821    #[expect(dead_code)]
822    std::fs::File,
823);
824
825#[cfg(test)]
826mod disk_cache_tests {
827    use super::*;
828    use tempfile::TempDir;
829
830    #[test]
831    fn test_disk_cache_roundtrip() {
832        let dir = TempDir::new().unwrap();
833        let cache1 = DiskCache::new(dir.path().to_path_buf(), false);
834        let key = blake3::hash(b"test-key");
835        let value = serde_json::json!({"result": "hello", "count": 42});
836        cache1.put("analyze_file", &key, &value);
837        let cache2 = DiskCache::new(dir.path().to_path_buf(), false);
838        let result: Option<serde_json::Value> = cache2.get("analyze_file", &key);
839        assert_eq!(result, Some(value));
840    }
841
842    #[cfg(unix)]
843    #[test]
844    fn test_disk_cache_permissions() {
845        use std::os::unix::fs::PermissionsExt;
846        let dir = TempDir::new().unwrap();
847        let cache_dir = dir.path().join("analysis-cache");
848        let _cache = DiskCache::new(cache_dir.clone(), false);
849        let meta = std::fs::metadata(&cache_dir).unwrap();
850        let mode = meta.permissions().mode() & 0o777;
851        assert_eq!(mode, 0o700, "cache dir must be mode 0700");
852    }
853
854    #[test]
855    fn test_disk_cache_corrupt_entry_returns_none() {
856        let dir = TempDir::new().unwrap();
857        let cache = DiskCache::new(dir.path().to_path_buf(), false);
858        let key = blake3::hash(b"corrupt-key");
859        let path = cache.entry_path("analyze_file", &key);
860        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
861        std::fs::write(&path, b"not valid snappy data").unwrap();
862        let result: Option<serde_json::Value> = cache.get("analyze_file", &key);
863        assert!(result.is_none(), "corrupt entry must return None");
864    }
865
866    #[test]
867    fn test_disk_cache_disabled_on_dir_creation_failure() {
868        let dir = TempDir::new().unwrap();
869        // Place a regular file where DiskCache::new() would create a directory.
870        // create_dir_all fails with ENOTDIR; new() must flip disabled=true.
871        let blocked = dir.path().join("blocked");
872        std::fs::write(&blocked, b"").unwrap();
873        let cache = DiskCache::new(blocked, false);
874        // disabled=true: put is a no-op, get always returns None
875        let key = blake3::hash(b"should-not-exist");
876        cache.put("analyze_file", &key, &serde_json::json!({"x": 1}));
877        let result: Option<serde_json::Value> = cache.get("analyze_file", &key);
878        assert!(
879            result.is_none(),
880            "cache must be disabled after dir creation failure"
881        );
882        assert!(
883            cache.disabled,
884            "disabled flag must be true after dir creation failure"
885        );
886    }
887
888    #[test]
889    fn test_concurrent_get_put_same_shard() {
890        // Edge case: concurrent get() + put() on the same shard from two threads
891        // must not panic and must return consistent results.
892        let dir = TempDir::new().unwrap();
893        let cache = std::sync::Arc::new(DiskCache::new(dir.path().to_path_buf(), false));
894        let key = blake3::hash(b"concurrent-test-key");
895        let value = serde_json::json!({"result": "from put thread", "n": 42});
896
897        // Pre-populate so get() has a chance to read something
898        cache.put("analyze_file", &key, &value);
899
900        let cache_get = cache.clone();
901        let cache_put = cache.clone();
902        let key_put = key;
903        let key_get = key;
904        let value_put = serde_json::json!({"result": "from put thread", "n": 100});
905
906        std::thread::scope(|scope| {
907            scope.spawn(|| {
908                // Write thread: perform put
909                cache_put.put("analyze_file", &key_put, &value_put);
910            });
911            scope.spawn(|| {
912                // Read thread: perform get concurrently
913                let _: Option<serde_json::Value> = cache_get.get("analyze_file", &key_get);
914            });
915        });
916
917        // After both threads complete, verify the cache is in a consistent state
918        let result: Option<serde_json::Value> = cache.get("analyze_file", &key);
919        assert!(
920            result.is_some(),
921            "entry must still be retrievable after concurrent access"
922        );
923    }
924
925    #[test]
926    fn test_concurrent_puts_same_shard() {
927        // Edge case: write_entry_atomically acquires exclusive lock before persist;
928        // concurrent writes from two threads must not corrupt the cache entry.
929        let dir = TempDir::new().unwrap();
930        let cache = std::sync::Arc::new(DiskCache::new(dir.path().to_path_buf(), false));
931        let key = blake3::hash(b"concurrent-put-key");
932        let value_a = serde_json::json!({"writer": "A", "data": "hello from A"});
933        let value_b = serde_json::json!({"writer": "B", "data": "hello from B"});
934
935        let cache_a = cache.clone();
936        let cache_b = cache.clone();
937        let key_a = key;
938        let key_b = key;
939
940        std::thread::scope(|scope| {
941            scope.spawn(|| {
942                cache_a.put("analyze_file", &key_a, &value_a);
943            });
944            scope.spawn(|| {
945                cache_b.put("analyze_file", &key_b, &value_b);
946            });
947        });
948
949        // After both writes complete, the entry must be deserializable (not corrupt)
950        let result: Option<serde_json::Value> = cache.get("analyze_file", &key);
951        assert!(
952            result.is_some(),
953            "entry must be retrievable after concurrent puts"
954        );
955        // Either value is acceptable; the key invariant is that the data is uncorrupted
956        let v = result.unwrap();
957        let writer = v.get("writer").and_then(|w| w.as_str());
958        assert!(
959            writer == Some("A") || writer == Some("B"),
960            "entry must contain data from one of the concurrent writers, got {writer:?}"
961        );
962    }
963}
aptu_coder_core/cache.rs

aptu_coder_core/
cache.rs
