lean_ctx/core/

content_cache.rs

//! Resident, bounded file-content cache shared across the search-index build and
//! `ctx_search` (issue #148).
//!
//! Before this module the trigram [`search_index`](crate::core::search_index)
//! build read *every* file in the corpus to extract trigrams and then threw the
//! content away, after which `ctx_search` read the narrowed candidate files
//! **again** to run the regex line-by-line — the corpus was read from disk
//! twice. This cache lets the first reader (whichever it is) populate file
//! contents once, keyed by absolute path and validated by `(mtime, size)`, and
//! every subsequent reader reuse them as an in-memory hit.
//!
//! Correctness: an entry is only ever served when the file's *current*
//! `(mtime, size)` exactly matches the stored identity, so any edit (which
//! changes mtime, and usually size) is a guaranteed miss — results can never go
//! stale. A miss simply falls back to a disk read.
//!
//! Bounds & safety:
//! - Total resident bytes are capped (`LEAN_CTX_CONTENT_CACHE_MB`, default
//!   128 MB) with approximate-LRU eviction, so a large corpus cannot grow the
//!   cache without limit.
//! - Inserts are skipped while the process is under memory pressure, and the
//!   eviction orchestrator can [`clear`] the cache on `UnloadIndices` /
//!   `EmergencyDrop`.

use std::fs::Metadata;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::UNIX_EPOCH;

use lru::LruCache;

/// Default resident byte budget when `LEAN_CTX_CONTENT_CACHE_MB` is unset.
const DEFAULT_BUDGET_MB: usize = 128;

/// Identity of one file *version*. A changed mtime or size ⇒ stale ⇒ cache miss.
/// Mirrors the `(mtime, size)` pair the BM25 index already trusts for staleness.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FileState {
    pub mtime_ms: u64,
    pub size_bytes: u64,
}

impl FileState {
    /// Build from an already-`stat`ed [`Metadata`] (no extra syscall) — callers
    /// in the hot path typically have this in hand from their size/regular-file
    /// checks. Returns `None` only when the platform cannot report mtime.
    pub fn from_metadata(meta: &Metadata) -> Option<Self> {
        let mtime_ms = meta
            .modified()
            .ok()
            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)?;
        Some(Self {
            mtime_ms,
            size_bytes: meta.len(),
        })
    }

    /// Convenience: `stat` the path then build the state. Costs one syscall.
    pub fn from_path(path: &Path) -> Option<Self> {
        Self::from_metadata(&path.metadata().ok()?)
    }
}

struct Entry {
    state: FileState,
    content: Arc<str>,
}

struct Cache {
    /// Unbounded by entry count — eviction is driven by `total_bytes` vs
    /// `budget_bytes` below, not by a capacity `lru::LruCache` would enforce on
    /// its own. `LruCache` gives O(1) recency tracking and O(1) LRU-victim
    /// lookup (`pop_lru`), replacing a hand-rolled `last_used` clock plus an
    /// O(n) `min_by_key` scan per eviction — the scan was rare per insert as
    /// originally noted, but `trim_oldest_percent` called it in a loop, making
    /// a big trim O(n²) over the resident set.
    map: LruCache<PathBuf, Entry>,
    total_bytes: usize,
    budget_bytes: usize,
    hits: u64,
    misses: u64,
    inserts: u64,
    evictions: u64,
}

impl Cache {
    fn new(budget_bytes: usize) -> Self {
        Self {
            map: LruCache::unbounded(),
            total_bytes: 0,
            budget_bytes,
            hits: 0,
            misses: 0,
            inserts: 0,
            evictions: 0,
        }
    }

    fn remove_entry(&mut self, path: &Path) {
        if let Some(old) = self.map.pop(path) {
            self.total_bytes = self.total_bytes.saturating_sub(old.content.len());
        }
    }

    /// Evict LRU entries until the budget is satisfied. O(1) per eviction.
    fn evict_to_budget(&mut self) {
        while self.total_bytes > self.budget_bytes {
            let Some((_, victim)) = self.map.pop_lru() else {
                break;
            };
            self.total_bytes = self.total_bytes.saturating_sub(victim.content.len());
            self.evictions += 1;
        }
    }
}

static CACHE: OnceLock<Mutex<Cache>> = OnceLock::new();

fn budget_bytes() -> usize {
    let mb = std::env::var("LEAN_CTX_CONTENT_CACHE_MB")
        .ok()
        .and_then(|v| v.trim().parse::<usize>().ok())
        .unwrap_or(DEFAULT_BUDGET_MB);
    mb.saturating_mul(1024 * 1024)
}

fn disabled() -> bool {
    // A zero byte budget (or the explicit disable flag) turns the cache into a
    // no-op pass-through — every `get` misses and `insert` is dropped.
    std::env::var("LEAN_CTX_DISABLE_CONTENT_CACHE")
        .is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
        || budget_bytes() == 0
}

fn cache() -> &'static Mutex<Cache> {
    CACHE.get_or_init(|| Mutex::new(Cache::new(budget_bytes())))
}

fn lock() -> std::sync::MutexGuard<'static, Cache> {
    cache()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Look up `path`; returns the cached content only when the supplied current
/// `(mtime, size)` matches the stored identity. A mismatch evicts the stale
/// entry and reports a miss. `state` is passed in (not re-`stat`ed) because hot
/// callers already hold the metadata.
pub fn get(path: &Path, current: FileState) -> Option<Arc<str>> {
    if disabled() {
        return None;
    }
    let mut c = lock();
    // `peek` (not `get`) first: a stale hit must not promote to MRU on its way
    // to being evicted.
    let Some(entry) = c.map.peek(path) else {
        c.misses += 1;
        crate::core::telemetry::global_metrics().record_cache(false);
        return None;
    };
    if entry.state != current {
        // Stale version cached — drop it so we don't keep paying for it.
        c.remove_entry(path);
        c.misses += 1;
        crate::core::telemetry::global_metrics().record_cache(false);
        return None;
    }
    c.hits += 1;
    crate::core::telemetry::global_metrics().record_cache(true);
    // `get` promotes to MRU; present under the lock we still hold, but degrade
    // gracefully instead of panicking on the read hot path if that invariant
    // ever changes.
    let entry = c.map.get(path)?;
    Some(Arc::clone(&entry.content))
}

/// Insert (or replace) the content for `path` at version `state`. Skipped while
/// the process is under memory pressure or when the cache is disabled, so the
/// cache never *adds* to a memory problem.
///
/// `content` must be what `admitted_by` admitted: the cache only ever holds
/// admitted text (Context Gateway, G5). Text admitted under a policy that is
/// no longer the observed one is refused, so a build that outlived a policy
/// change cannot repopulate the cache with it.
pub fn insert(
    path: &Path,
    state: FileState,
    content: Arc<str>,
    admitted_by: &crate::core::context_admission::stores::StoreAdmission,
) {
    if disabled() || crate::core::memory_guard::is_under_pressure() {
        return;
    }
    let len = content.len();
    let mut c = lock();
    // Checked under the lock: a policy change advances the epoch before it
    // clears the cache, so this entry is either refused here or cleared.
    if !admitted_by.is_current() {
        return;
    }
    // A single file larger than the whole budget would thrash eviction — skip it.
    if len > c.budget_bytes {
        return;
    }
    c.remove_entry(path);
    c.map.put(path.to_path_buf(), Entry { state, content });
    c.total_bytes += len;
    c.inserts += 1;
    if c.total_bytes > c.budget_bytes {
        c.evict_to_budget();
    }
}

/// Read a file through the cache: returns cached content on a fresh hit, else
/// reads from disk (any text encoding, see `text_decode`), populates the cache,
/// and returns it. `None` on a
/// binary/unreadable/unstatable file. Convenience for callers without their
/// own size/special-file gating (the search-index build and `ctx_search` use
/// the explicit [`get`]/[`insert`] pair so they keep their own skip rules).
pub fn get_or_read(path: &Path) -> Option<Arc<str>> {
    let state = FileState::from_path(path)?;
    // Observed first: a changed policy drops the cache before it can hit.
    let admission = crate::core::context_admission::stores::StoreAdmission::current();
    if let Some(hit) = get(path, state) {
        return Some(hit);
    }
    // The cache only ever holds admitted text (Context Gateway, G5).
    let content = admission.read(path)?;
    let arc: Arc<str> = Arc::from(content);
    insert(path, state, Arc::clone(&arc), &admission);
    Some(arc)
}

/// Drop all entries, freeing the heap. Called by the eviction orchestrator under
/// memory pressure; the cache simply re-warms on subsequent reads.
pub fn clear() {
    if CACHE.get().is_none() {
        return;
    }
    let mut c = lock();
    c.map.clear();
    c.total_bytes = 0;
}

/// Evict the oldest `percent` of entries by LRU clock. Used after index builds
/// to release file contents that were only needed for chunking/tokenization.
/// `percent` is clamped to `[0, 100]`.
pub fn trim_oldest_percent(percent: u8) {
    if CACHE.get().is_none() {
        return;
    }
    let mut c = lock();
    if c.map.is_empty() {
        return;
    }
    let pct = (percent.min(100)) as usize;
    let target_evictions = c.map.len() * pct / 100;
    for _ in 0..target_evictions {
        let Some((_, victim)) = c.map.pop_lru() else {
            break;
        };
        c.total_bytes = c.total_bytes.saturating_sub(victim.content.len());
        c.evictions += 1;
    }
}

/// Approximate resident heap used by cached contents, in bytes.
pub fn memory_usage_bytes() -> usize {
    if CACHE.get().is_none() {
        return 0;
    }
    lock().total_bytes
}

/// Observability snapshot: `(hits, misses, entries, bytes, evictions)`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct CacheStats {
    pub hits: u64,
    pub misses: u64,
    pub entries: usize,
    pub bytes: usize,
    pub inserts: u64,
    pub evictions: u64,
}

pub fn stats() -> CacheStats {
    if CACHE.get().is_none() {
        return CacheStats::default();
    }
    let c = lock();
    CacheStats {
        hits: c.hits,
        misses: c.misses,
        entries: c.map.len(),
        bytes: c.total_bytes,
        inserts: c.inserts,
        evictions: c.evictions,
    }
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use crate::core::context_admission::stores::StoreAdmission;
    use serial_test::serial;

    fn admitted() -> StoreAdmission {
        StoreAdmission::current()
    }

    /// The cache is a process-wide global and tests mutate it (and the budget
    /// env var). Serialize them so they cannot observe each other's state.
    static TEST_LOCK: Mutex<()> = Mutex::new(());

    /// Held for the whole test: configuration and policy are process-wide,
    /// and a change of either clears this cache (see `get_or_read` and
    /// `insert`). Without these guards a parallel test that switches its data
    /// directory or policy empties the cache between two reads of this test —
    /// seen in gate `h15-engine` as a re-read where a hit was asserted.
    type Stable = (
        crate::core::data_dir::IsolatedDataDir,
        crate::core::policy::runtime::TestPolicyOverride,
    );

    #[must_use]
    fn fresh_cache(budget_bytes: usize) -> Stable {
        let stable = (
            crate::core::data_dir::isolated_data_dir(),
            crate::core::policy::runtime::TestPolicyOverride::set(None),
        );
        crate::test_env::remove_var("LEAN_CTX_CONTENT_CACHE_MB");
        crate::test_env::remove_var("LEAN_CTX_DISABLE_CONTENT_CACHE");
        let mut c = lock();
        *c = Cache::new(budget_bytes);
        stable
    }

    fn write(dir: &Path, name: &str, body: &str) -> PathBuf {
        let p = dir.join(name);
        std::fs::write(&p, body).unwrap();
        p
    }

    #[test]
    #[serial(cache_telemetry)]
    fn hit_after_insert_with_matching_state() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "a.rs", "fn main() {}\n");
        let state = FileState::from_path(&p).unwrap();
        assert!(get(&p, state).is_none(), "cold cache must miss");
        insert(&p, state, Arc::from("fn main() {}\n"), &admitted());
        let got = get(&p, state).expect("warm cache must hit");
        assert_eq!(&*got, "fn main() {}\n");
    }

    #[test]
    #[serial(cache_telemetry)]
    fn miss_paths_update_local_and_central_stats() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "cold.rs", "cold\n");
        let state = FileState::from_path(&p).unwrap();
        let stale_state = FileState {
            size_bytes: state.size_bytes + 1,
            ..state
        };
        let local_before = stats();
        let central = crate::core::telemetry::global_metrics();
        let misses_before = central
            .cache_misses
            .load(std::sync::atomic::Ordering::Relaxed);

        assert!(get(&p, state).is_none());
        insert(&p, state, Arc::from("cold\n"), &admitted());
        assert!(get(&p, stale_state).is_none());

        let local_after = stats();
        assert!(local_after.misses >= local_before.misses + 2);
        assert!(
            central
                .cache_misses
                .load(std::sync::atomic::Ordering::Relaxed)
                >= misses_before + 2
        );
    }

    #[test]
    #[serial(cache_telemetry)]
    fn warm_hit_updates_local_and_central_stats() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "warm.rs", "warm\n");
        let state = FileState::from_path(&p).unwrap();
        insert(&p, state, Arc::from("warm\n"), &admitted());
        let local_before = stats();
        let central = crate::core::telemetry::global_metrics();
        let hits_before = central
            .cache_hits
            .load(std::sync::atomic::Ordering::Relaxed);

        assert!(get(&p, state).is_some());

        let local_after = stats();
        assert!(local_after.hits > local_before.hits);
        assert!(
            central
                .cache_hits
                .load(std::sync::atomic::Ordering::Relaxed)
                > hits_before
        );
    }

    #[test]
    #[serial(cache_telemetry)]
    fn mtime_or_size_change_invalidates() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "a.rs", "v1\n");
        let s1 = FileState::from_path(&p).unwrap();
        insert(&p, s1, Arc::from("v1\n"), &admitted());
        assert!(get(&p, s1).is_some());

        // Different size ⇒ different state ⇒ miss, and the stale entry is dropped.
        let s_bigger = FileState {
            size_bytes: s1.size_bytes + 10,
            ..s1
        };
        assert!(get(&p, s_bigger).is_none(), "size change must miss");
        assert!(
            get(&p, s1).is_none(),
            "stale entry must be evicted on mismatch"
        );

        // Different mtime ⇒ miss as well.
        insert(&p, s1, Arc::from("v1\n"), &admitted());
        let s_newer = FileState {
            mtime_ms: s1.mtime_ms + 1,
            ..s1
        };
        assert!(get(&p, s_newer).is_none(), "mtime change must miss");
    }

    #[test]
    #[serial(cache_telemetry)]
    fn get_or_read_populates_then_serves_from_cache() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "a.rs", "hello world\n");

        let state = FileState::from_path(&p).unwrap();
        assert!(get(&p, state).is_none(), "first read starts cold");
        let first = get_or_read(&p).unwrap();
        assert_eq!(&*first, "hello world\n");
        assert!(
            Arc::ptr_eq(&first, &get(&p, state).expect("first read inserts")),
            "the populated entry retains the returned allocation"
        );

        // Other readers can populate unrelated paths between these calls.
        // Global insertion counters cannot prove a hit for this specific file.
        let other = write(dir.path(), "other.rs", "unrelated\n");
        insert(
            &other,
            FileState::from_path(&other).unwrap(),
            Arc::from("unrelated\n"),
            &admitted(),
        );
        let second = get_or_read(&p).unwrap();
        assert_eq!(&*second, "hello world\n");
        // Allocation identity, not counter arithmetic: a cache hit clones the
        // stored `Arc`, a re-read allocates a new one. `stats()` is
        // process-wide, so bracketing a call with `stats()` deltas races every
        // parallel test that touches the content cache — that is what made this
        // test flake in CI (seen on #1736 / #1738).
        assert!(
            Arc::ptr_eq(&first, &second),
            "the second read must serve the cached allocation, not re-read the file"
        );
    }

    #[test]
    #[serial(cache_telemetry)]
    fn eviction_keeps_cache_within_budget() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Budget fits ~2 small files; a third insert must evict the LRU one.
        let _stable = fresh_cache(64);
        let dir = tempfile::tempdir().unwrap();
        let pa = write(dir.path(), "a", "aaaaaaaaaaaaaaaaaaaaaaaaaaaa"); // 28 bytes
        let pb = write(dir.path(), "b", "bbbbbbbbbbbbbbbbbbbbbbbbbbbb");
        let pc = write(dir.path(), "c", "cccccccccccccccccccccccccccc");
        let sa = FileState::from_path(&pa).unwrap();
        let sb = FileState::from_path(&pb).unwrap();
        let sc = FileState::from_path(&pc).unwrap();

        insert(
            &pa,
            sa,
            Arc::from("aaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
            &admitted(),
        );
        // Touch a so b becomes the LRU victim.
        let _ = get(&pa, sa);
        insert(
            &pb,
            sb,
            Arc::from("bbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
            &admitted(),
        );
        let _ = get(&pa, sa);
        insert(
            &pc,
            sc,
            Arc::from("cccccccccccccccccccccccccccc"),
            &admitted(),
        );

        let st = stats();
        assert!(st.bytes <= 64, "cache must respect byte budget: {st:?}");
        assert!(st.evictions >= 1, "an eviction must have occurred: {st:?}");
        assert!(get(&pa, sa).is_some(), "recently-used entry must survive");
    }

    #[test]
    #[serial(cache_telemetry)]
    fn disabled_via_zero_budget_is_passthrough() {
        let _env_lock = crate::core::data_dir::test_env_lock();
        let _g = TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let _stable = fresh_cache(1024 * 1024);
        crate::test_env::set_var("LEAN_CTX_CONTENT_CACHE_MB", "0");
        let dir = tempfile::tempdir().unwrap();
        let p = write(dir.path(), "a.rs", "x\n");
        let state = FileState::from_path(&p).unwrap();
        insert(&p, state, Arc::from("x\n"), &admitted());
        assert!(get(&p, state).is_none(), "zero-budget cache is a no-op");
        crate::test_env::remove_var("LEAN_CTX_CONTENT_CACHE_MB");
    }
}