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

aptu_coder_core/
cache.rs
