znippy_plugin_git/

exploded_cache.rs

//! **The exploded tier is a CACHE, not a tier — and this is the bound.**
//!
//! The owner, 2026-08-30, on being shown the lazy-explode design: *"does lazy
//! explode mean it eventually takes the disk anyway?"* — and the honest answer
//! is **yes, unless the tier is capped**, because a client that browses a whole
//! history reads every object once and a write-back-on-first-read tier would
//! then hold exactly what the eager one held. MEASURED on the rig 2026-08-30:
//! the 8.12 GiB `torvalds/linux` pack drove the eager tier to **141.7 GiB**
//! into `/data`, ~17× the pack. *"We cannot have one repo take all 300 GB
//! disk."*
//!
//! So the tier has a **byte budget** and evicts. Nothing here is data: every
//! row is re-derivable from the verbatim pack bytes by
//! [`crate::resolve::materialize_at`], so an eviction is never a loss and never
//! an error a client can see — it is a later read paying ~0.5–2 ms to inflate
//! its own delta chain again.
//!
//! # Two pools, and the split is the whole design
//!
//! ```text
//!   objects.exploded         the EAGER pool — what the indexer writes at push
//!                            time. Under the shipped policy that is commits
//!                            and tags: what `graph`/`reach`/tag-peel need, so
//!                            visibility and pull checks never stall.
//!                            NEVER evicted — it is the "commits last" rule,
//!                            held by construction rather than by a comparator.
//!
//!   objects.exploded.c1      the LAZY pool — first-read write-backs, appended
//!   objects.exploded.c2      in read order, one file per segment.
//!   …                        EVICTION IS ONE `unlink` of the oldest segment.
//! ```
//!
//! Both pools are counted against the budget and both are named on the
//! dashboard; only the lazy one is evictable. If the eager pool alone exceeds
//! the budget the lazy pool is evicted to nothing and the eager pool is still
//! kept — a store that cannot answer `reach` is broken, and a store that reads
//! slowly is not.
//!
//! # Why segments, and why eviction is an `unlink`
//!
//! [`ExplodedArchive`] is one append-only Arrow IPC stream: `[schema][batch]…`,
//! nothing ever rewritten. That is what makes it 1.0× its payload instead of
//! redb's 12×, and it is also what makes *deleting a row in the middle* cost a
//! whole-file rewrite — up to 17 GB, which the module docs already refuse to
//! smuggle into `gc`. Cutting the pool into segment files turns "reclaim the
//! coldest bytes" into `remove_file`: O(1), no rewrite, no read amplification,
//! and a crash during it leaves a pool that is merely shorter.
//!
//! # The recency argument, stated honestly
//!
//! Segments are appended in **first-read order**, so dropping the oldest drops
//! the objects least recently *admitted*. That is not the same as least
//! recently *used* — an object read at minute 1 and again at minute 90 sits in
//! an old segment. It is not promoted, and it does not need to be: when its
//! segment is dropped, the next read of it misses, re-inflates its chain, and
//! **writes it back into the newest segment**. Re-entry-on-miss is what makes
//! this LRU-shaped, at the cost of one chain inflate per hot object per eviction
//! cycle rather than a copy per hit. So it is FIFO over segments with
//! re-entry-on-miss, and the thing an LRU is wanted for — a working set that
//! stays resident under a scan of everything else — holds, because the working
//! set is what keeps re-entering.
//!
//! # What a full-history read now costs on disk
//!
//! The plateau, not the climb. A reader that walks every object of a
//! Linux-sized repository writes back every object it inflates; the pool grows
//! to the budget and then stops, because each rotation past it drops the oldest
//! segment. That is the number lane Q reports as its headline, and
//! `the_tier_plateaus_at_its_budget_under_a_full_scan` is the test that asserts
//! it rather than the sentence.

use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};

use anyhow::Result;

use crate::exploded::ExplodedStats;
use crate::exploded_arrow::{ExplodePolicy, ExplodedArchive};
use crate::object::GitObjectKind;

/// One lazy segment is closed and a new one started once it passes this many
/// bytes. Eviction granularity is one segment, so this is the quantum a budget
/// is spent and reclaimed in: 1/16 of the budget, floored so a small budget
/// still gets whole batches and ceilinged so a huge one does not accumulate
/// thousands of open indexes.
const MIN_SEGMENT: u64 = 16 << 20;
const MAX_SEGMENT: u64 = 1 << 30;
const SEGMENTS_PER_BUDGET: u64 = 16;

/// The lazy pool buffers this much before it writes. Far under
/// [`crate::exploded_arrow::FLUSH_BYTES`] on purpose: that 64 MiB buffer is
/// sized for an indexer draining a pack, and a **reader's** write-back holding
/// 64 MiB per store is memory the box did not agree to spend.
pub const CACHE_FLUSH_BYTES: u64 = 4 << 20;

// ════════════════════════════════════════════════════════════════════════════
// A VALIDATION READ IS NOT A READER
// ════════════════════════════════════════════════════════════════════════════

thread_local! {
    /// Depth, not a flag: guards nest, and the innermost one ending must not
    /// re-open the cache for an outer scope that is still validating.
    static SUPPRESS_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}

/// **Declare that reads on this thread are validation, not a reader's.**
///
/// `receive-pack` reads every commit and tree it was sent — once — to prove the
/// push closes, and then never reads them again. Those reads look identical to
/// a client's from inside the store, so the store cannot tell them apart and
/// the caller must say.
///
/// MEASURED on `torvalds/linux`, oden 2026-09-03: the lazy pool is **empty
/// until the instant indexing completes** and then grows to
/// **6 778 047 872 B** (13 segments) before the push returns — 31 % of the
/// whole store and the largest non-pack thing in it, written for a reader that
/// does not exist. Full table in
/// `gunnar/.nornir/storage-levers-2026-09-03.md`.
///
/// # Scope
///
/// **Thread-local, and that is the point.** A `receive-pack` connection does
/// its ingest, its index and its connectivity walk on one thread and serves
/// nobody; the background absorber runs elsewhere and is untouched, and so is
/// every serving connection. Nothing global is switched off, so a store that is
/// simultaneously answering a clone still warms its cache from that clone.
///
/// The eager pool is not affected either: the indexer writes that one directly,
/// not through [`ExplodedCache::write_back`].
///
/// # It is a guard, not a pair of calls
///
/// Restored on drop, including on unwind — a panic inside a connectivity walk
/// must not leave a thread that never caches again.
#[must_use = "the suppression lasts only as long as the guard is held"]
pub fn suppress_write_back() -> SuppressWriteBack {
    SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
    SuppressWriteBack { _private: () }
}

/// Whether [`suppress_write_back`] is in force on this thread.
pub fn write_back_suppressed() -> bool {
    SUPPRESS_DEPTH.with(|d| d.get()) > 0
}

/// The guard [`suppress_write_back`] returns. See it for what this is for.
pub struct SuppressWriteBack {
    _private: (),
}

impl Drop for SuppressWriteBack {
    fn drop(&mut self) {
        SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
    }
}

// ════════════════════════════════════════════════════════════════════════════
// The admission policy: DO NOT EXPLODE UNIFORMLY
// ════════════════════════════════════════════════════════════════════════════

/// **The cheapest thing a cache can do is not cache.**
///
/// The owner, 2026-08-30: *"can the index point at the original git object and
/// decompress it directly instead of materialising everything?"* — and it can,
/// because a packfile is not one stream. Every object is its own zlib stream at
/// its own offset, so `oid → (offset, type, base, inflated size)` is enough to
/// inflate exactly one object ([`crate::resolve::materialize_at`]). A **whole**
/// object is then one `inflate` of a mapped byte range: microseconds, no chain,
/// no delta application. Writing that into a cache costs a copy, a flush, an
/// fsync and a slice of somebody's quota — to save microseconds it did not
/// need saving.
///
/// So the cache admits on **rebuild cost**, not on being read:
///
/// # It admits on cost DENSITY, not on cost
///
/// The first cut admitted anything over 16 KiB "because the inflate itself is
/// the cost". That is true and it is the wrong rule, because the budget is
/// **shared**: what a cache should hold is what saves the most time *per byte
/// held*. A 3 MiB whole blob costs one linear inflate — bandwidth-bound,
/// microseconds a kilobyte — and would occupy the same 3 MiB as ~250 deep tree
/// chains that each cost a chain walk. Admitting it evicts all of them.
///
/// | what the read paid | admitted? |
/// |---|---|
/// | over [`MAX_CACHED_BYTES`] inflated | **no** — a bad tenant of a shared budget whatever its depth; its direct read is bandwidth-bound anyway |
/// | depth 0–[`CHEAP_DEPTH`] | **no** — one inflate of one mapped zlib stream, plus at most two delta applications. Caching it spends quota to save microseconds |
/// | deeper than [`CHEAP_DEPTH`] | yes — every extra link is another inflate plus a delta application, and git allows chains to 50 |
/// | a base outside this archive | yes — it went through a [`BaseSource`](crate::resolve::BaseSource), which is never the cheap path |
///
/// The thresholds are named, not tuned: `CHEAP_DEPTH` is 2 because git's own
/// default `pack.depth` is 50 and a 0–2 chain is at most two delta applications
/// over a mapped buffer; `MAX_CACHED_BYTES` is 1 MiB because that is where one
/// object starts costing what hundreds of chain-heavy ones do.
///
/// **This is also the zero-copy rule.** A declined object is declined *before*
/// anything is copied, so the commonest reads — and every large one — cost the
/// cache exactly zero bytes. MEASURED by lane AA 2026-08-30: one 3 145 728 B
/// object read through `gunnar-cli` allocated 6 340 693 B, exactly two copies,
/// and the second was the cache taking a blob it should never have wanted.
pub const CHEAP_DEPTH: u32 = 2;
pub const MAX_CACHED_BYTES: u64 = 1 << 20;

/// What it cost to produce an object, as the cache scores it.
///
/// `depth + 1` inflates and `depth` delta applications over `inflated` bytes.
/// Deliberately a coarse integer and not a timing: a duration measured on the
/// reader's thread carries that thread's scheduling, and two identical objects
/// would then score differently depending on how busy the box was.
pub fn rebuild_cost(depth: u32, inflated: u64, external_base: bool) -> u64 {
    // Bytes inflated across the whole chain, roughly: every link inflates
    // something of about the object's order, plus one delta application each.
    let links = depth as u64 + 1;
    let base = links * inflated.max(1);
    if external_base { base.saturating_mul(4) } else { base }
}

/// Is this object worth a slice of the quota?
pub fn worth_caching(depth: u32, inflated: u64, external_base: bool) -> bool {
    inflated <= MAX_CACHED_BYTES && (external_base || depth > CHEAP_DEPTH)
}

/// Environment door for the budget. Named in [`crate::arms::ENV_EXPLODE_BUDGET`]
/// and read through that crate's one counted door, so
/// `every_environment_read_in_this_crate_is_named_in_all_env` can see it —
/// a key spelled locally is a key that roster cannot check.
pub use crate::arms::ENV_EXPLODE_BUDGET as ENV_BUDGET;

/// **The default budget, and the arithmetic that makes it safe.**
///
/// `min(10 % of the filesystem holding the store, 8 GiB)` per repository.
///
/// On the shipped appliance — 300 GB `/data`, the shape the owner named — 10 %
/// is 30 GB, so the 8 GiB ceiling binds and every repository is capped at
/// 8 GiB. **Twenty** repositories of any size therefore cost at most 160 GB of
/// exploded cache against 300 GB, leaving the packs themselves (the truth,
/// which is never evictable) room; and a repository only reaches its cap if
/// somebody actually reads all of it. The same appliance with the eager tier
/// took 141.7 GB for **one** repository.
///
/// On a small box the percentage binds instead: a 64 GiB `/data` gives 6.4 GiB,
/// a 20 GiB one gives 2 GiB. The floor is [`MIN_BUDGET`] so a tiny volume still
/// gets a cache rather than thrashing at zero.
pub const BUDGET_FRACTION_PERCENT: u64 = 10;
pub const BUDGET_CEILING: u64 = 8 << 30;
pub const MIN_BUDGET: u64 = 64 << 20;

/// The default budget for a store rooted at `dir`, from that filesystem's size.
pub fn default_budget(dir: &Path) -> u64 {
    let total = filesystem_bytes(dir).unwrap_or(0);
    if total == 0 {
        return BUDGET_CEILING;
    }
    (total / 100 * BUDGET_FRACTION_PERCENT)
        .min(BUDGET_CEILING)
        .max(MIN_BUDGET)
}

/// The process-wide line for a store rooted at `dir`: the same tenth of the
/// filesystem as [`default_budget`], **uncapped** — the cap is per store, the
/// tenth is per box. Every cache on one filesystem computes the same figure;
/// caches on different filesystems each hold the sum to their own, which
/// over-counts across mounts and is the conservative direction. `0` (no
/// filesystem figure) turns the global check off rather than inventing one.
pub fn default_global_line(dir: &Path) -> u64 {
    filesystem_bytes(dir).unwrap_or(0) / 100 * BUDGET_FRACTION_PERCENT
}

/// **The disk line**: once the filesystem holding a store is this full, the
/// lazy pool gives its OLDEST segments back, whatever its own budget says.
///
/// Rickard, 2026-09-28: *"add a deletion of the old objects in case 80 % disk
/// is taken"*. The budget and the global line are both fractions of the
/// filesystem's SIZE, so a disk filled by packs — the truth, never evictable —
/// never moved either of them: the cache kept its full 10 % while pushes were
/// refused at the floor. 80 % is also where the appliance's grow batcher
/// queues a growth (monetize `grow_high_water_percent`), so the cache is
/// handing space back at the moment the disk is asked to grow, not after.
///
/// Oldest first, not cheapest first: under pressure the question is "what has
/// nobody needed for longest", and the rebuild cost of an object nobody reads
/// is paid by nobody. The eager pool is still never touched. `0` = off.
pub const DISK_LINE_PERCENT: u64 = 80;
/// **Hysteresis under the disk line** (owner 2026-09-28): once the line trips,
/// the pressure holds until the disk is this many points BELOW it (80 → 75).
/// Without it a disk that stays over 80 % because of packs — not cache — made
/// every tick drop every lazy segment but the live one, in every store, while
/// write-back kept refilling them: the cache rewrote and threw away up to a
/// segment per store per tick.
pub const DISK_LINE_HYSTERESIS_PERCENT: u64 = 5;

/// How full the filesystem holding `dir` is, in percent, the way `df` counts
/// it: used / (used + available to an unprivileged writer). `None` when
/// `statvfs` has no answer — and then the disk line does nothing, because a
/// reading that did not happen is not a full disk.
/// Pure: the latched pressure after a reading of `used` % against `line` %.
/// It trips at `used >= line` and releases only at
/// `used < line - DISK_LINE_HYSTERESIS_PERCENT`; in between it keeps `was`.
pub fn pressure_after(was: bool, used: u64, line: u64) -> bool {
    if used >= line {
        true
    } else if used < line.saturating_sub(DISK_LINE_HYSTERESIS_PERCENT) {
        false
    } else {
        was
    }
}

pub fn disk_used_percent(dir: &Path) -> Option<u64> {
    use std::os::unix::ffi::OsStrExt;
    let mut c = Vec::with_capacity(dir.as_os_str().len() + 1);
    c.extend_from_slice(dir.as_os_str().as_bytes());
    c.push(0);
    let mut s: libc::statvfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statvfs(c.as_ptr().cast(), &mut s) };
    if rc != 0 {
        return None;
    }
    let used = (s.f_blocks as u64).saturating_sub(s.f_bfree as u64);
    let seen = used + s.f_bavail as u64;
    (seen > 0).then(|| used * 100 / seen)
}

/// Total bytes of the filesystem holding `dir`, via `statvfs`.
fn filesystem_bytes(dir: &Path) -> Option<u64> {
    use std::os::unix::ffi::OsStrExt;
    let mut c = Vec::with_capacity(dir.as_os_str().len() + 1);
    c.extend_from_slice(dir.as_os_str().as_bytes());
    c.push(0);
    let mut s: libc::statvfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statvfs(c.as_ptr().cast(), &mut s) };
    (rc == 0).then(|| s.f_blocks as u64 * s.f_frsize as u64)
}

/// The budget a store opens with: the environment door if it names one, the
/// filesystem-derived default otherwise.
pub fn budget_from_env(dir: &Path) -> Result<u64> {
    match crate::arms::read_env(crate::arms::ENV_EXPLODE_BUDGET) {
        Some(v) => parse_budget(&v),
        None => Ok(budget_default().unwrap_or_else(|| default_budget(dir))),
    }
}

/// `u64::MAX` is "no embedder default" and not a budget: `0` already MEANS
/// unlimited ([`parse_budget`]), so the sentinel cannot be zero without making
/// "unbounded" unsayable from code.
const NO_BUDGET_DEFAULT: u64 = u64::MAX;

/// What [`budget_from_env`] answers when [`crate::arms::ENV_EXPLODE_BUDGET`] is unset:
/// the embedder's figure, or the filesystem-derived [`default_budget`] when it set none.
static BUDGET_DEFAULT: AtomicU64 = AtomicU64::new(NO_BUDGET_DEFAULT);

/// The embedder's process-wide budget, if one was set.
pub fn budget_default() -> Option<u64> {
    match BUDGET_DEFAULT.load(Ordering::Relaxed) {
        NO_BUDGET_DEFAULT => None,
        n => Some(n),
    }
}

/// **Set the process-wide default explode budget** for every store opened AFTER this call
/// (an explicit [`crate::arms::ENV_EXPLODE_BUDGET`] still wins). `None` restores the
/// filesystem-derived [`default_budget`]. The twin of
/// [`crate::arms::set_graph_cache_default`], for an embedder that learns its role at runtime
/// and cannot set the environment — edition 2024 makes `set_var` unsafe, and gunnar forbids
/// `unsafe` — such as gunnar's twin, which caps the pool at 16 MiB so a later policy change
/// cannot walk back over its measured memory bound.
pub fn set_explode_budget_default(bytes: Option<u64>) {
    BUDGET_DEFAULT.store(bytes.unwrap_or(NO_BUDGET_DEFAULT), Ordering::Relaxed);
}

/// `12G`, `512M`, `1048576`, `0`, `unlimited`. An unparseable value is an
/// ERROR and never a silent default — a typo in a quota is a bill that is
/// wrong, and nobody notices until it is.
pub fn parse_budget(s: &str) -> Result<u64> {
    let t = s.trim().to_ascii_lowercase();
    if t == "unlimited" || t == "none" || t == "off" {
        return Ok(0);
    }
    let (digits, mult) = match t.as_bytes().last() {
        Some(b'k') => (&t[..t.len() - 1], 1u64 << 10),
        Some(b'm') => (&t[..t.len() - 1], 1u64 << 20),
        Some(b'g') => (&t[..t.len() - 1], 1u64 << 30),
        Some(b't') => (&t[..t.len() - 1], 1u64 << 40),
        _ => (t.as_str(), 1),
    };
    let n: u64 = digits
        .trim()
        .parse()
        .map_err(|_| anyhow::anyhow!("unreadable explode budget {s:?}; expected 0, 512M, 8G, …"))?;
    n.checked_mul(mult)
        .ok_or_else(|| anyhow::anyhow!("explode budget {s:?} overflows"))
}

/// What the tier is holding and what it is allowed to hold — the row a
/// dashboard prints and `Admin.StorageInfo` carries.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct CacheUse {
    /// Bytes in the eager pool (commits and tags under the shipped policy).
    pub eager_bytes: u64,
    /// Bytes in the lazy pool, across every live segment.
    pub lazy_bytes: u64,
    /// The cap. `0` is unlimited.
    pub budget: u64,
    /// Live lazy segments.
    pub segments: u64,
    /// Segments unlinked since this store was opened.
    pub evicted_segments: u64,
    /// Bytes those segments held. **Applied output**: what actually left the
    /// filesystem, measured before the unlink, not a projection.
    pub evicted_bytes: u64,
    /// Rows those segments held.
    pub evicted_rows: u64,
    /// Objects written back by a first read.
    pub written_back: u64,
    /// **Reads served by inflating the object directly out of the pack and
    /// caching NOTHING** — the cheap ones the admission policy declined.
    /// Applied output: the counter says the policy ran, which a size alone
    /// could not.
    pub direct_reads: u64,
    /// **Write-backs declined because the caller declared the read a
    /// validation read** ([`suppress_write_back`]) — a `receive-pack`'s
    /// connectivity walk, not a reader.
    ///
    /// Applied output, and the only way to tell the suppression ran from a
    /// store that simply had nothing to cache: a pool that stays empty proves
    /// nothing on its own, and this counter moving proves the gate fired.
    pub suppressed_write_backs: u64,
}

impl CacheUse {
    pub fn used(&self) -> u64 {
        self.eager_bytes + self.lazy_bytes
    }
    /// `3.1G/8G` — what the dashboard cell prints.
    pub fn band(&self) -> String {
        if self.budget == 0 {
            format!("{}/unlimited", human(self.used()))
        } else {
            format!("{}/{}", human(self.used()), human(self.budget))
        }
    }
    /// Over budget right now — true only while the eager pool alone exceeds it,
    /// because the lazy pool is evicted back under after every rotation.
    pub fn over(&self) -> bool {
        self.budget != 0 && self.used() > self.budget
    }
}

/// `3.1G`, `812M`, `40K` — one significant fraction, the same shape the
/// dashboard's other byte cells use.
pub fn human(bytes: u64) -> String {
    const UNITS: [(u64, &str); 4] = [(1 << 40, "T"), (1 << 30, "G"), (1 << 20, "M"), (1 << 10, "K")];
    for (scale, suffix) in UNITS {
        if bytes >= scale {
            let v = bytes as f64 / scale as f64;
            return if v < 10.0 {
                format!("{v:.1}{suffix}")
            } else {
                format!("{v:.0}{suffix}")
            };
        }
    }
    format!("{bytes}B")
}

/// What one live lazy segment is holding, for cost-weighted eviction.
#[derive(Debug, Clone, Copy, Default)]
struct SegStat {
    rows: u64,
    /// Summed [`rebuild_cost`] of the rows written into it.
    cost: u64,
    /// Monotonic tick of the last write into it — the recency half of the rank.
    touched: u64,
}

/// **The two pools, one door.** Everything that used to call an
/// [`ExplodedArchive`] calls this instead; the eager pool *is* an
/// `ExplodedArchive` and behaves exactly as it did.
pub struct ExplodedCache {
    eager: ExplodedArchive,
    /// Oldest first. `Mutex` and not `RwLock`: it is touched once per rotation
    /// and once per read miss, never per object.
    lazy: Mutex<Vec<std::sync::Arc<ExplodedArchive>>>,
    base: PathBuf,
    policy: ExplodePolicy,
    budget: AtomicU64,
    /// **The process-wide line**, in bytes, that the SUM of every registered
    /// cache's disk use is held under; `0` = no global line. `budget` above is
    /// per store — 10 % of the filesystem, capped at 8 GiB — and MEASURED on
    /// 2026-09-05 that meant N repositories read hard could hold N × 8 GiB:
    /// ten busy tenants on a 1 TB appliance, 80 GB of cache. This is the same
    /// 10 % of the filesystem, uncapped, applied once to the box: a cache whose
    /// own use is above its fair share of it (`line / live caches`) evicts its
    /// cheapest segment on its tick even while under its own budget, so the
    /// biggest pools shrink first and the sum converges under the line.
    global_line: AtomicU64,
    /// [`DISK_LINE_PERCENT`], changeable live; `0` = off.
    disk_line_percent: AtomicU64,
    /// Latched disk pressure: set at the line, cleared only below
    /// `line − DISK_LINE_HYSTERESIS_PERCENT`. While set, eviction runs
    /// oldest-first AND write-back admits nothing (see [`Self::write_back`]).
    disk_pressure: std::sync::atomic::AtomicBool,
    /// Write-backs refused because the disk line was latched.
    pressure_refused_write_backs: AtomicU64,
    /// A test's disk reading in place of `statvfs`; [`Self::NO_DISK_READING`] = read
    /// the real filesystem. Starts at `0` under test (see `open`).
    #[cfg(test)]
    disk_used_override: AtomicU64,
    segment_bytes: AtomicU64,
    next_seg: AtomicU64,
    written_back: AtomicU64,
    direct_reads: AtomicU64,
    /// See [`CacheUse::suppressed_write_backs`].
    suppressed_write_backs: AtomicU64,
    /// Per live segment, parallel to `lazy`: rows, and the summed rebuild cost
    /// of what is in it. What cost-weighted eviction ranks on.
    seg_cost: Mutex<Vec<SegStat>>,
    evicted_segments: AtomicU64,
    evicted_bytes: AtomicU64,
    evicted_rows: AtomicU64,
    /// Bytes buffered in the newest lazy segment since its last flush. The
    /// flusher reads it; a reader adds to it and never blocks on it.
    dirty: AtomicU64,
    /// Where this cache's store lives, for the registry's rows.
    root: String,
    /// A maintenance thread is running. **The whole of "nobody waits"**: the
    /// reader that fills the buffer starts one and walks away, and a second
    /// reader that arrives while it runs does nothing at all.
    maintaining: std::sync::atomic::AtomicBool,
}

impl ExplodedCache {
    /// "No test reading": the disk line reads the real filesystem.
    #[cfg(test)]
    const NO_DISK_READING: u64 = u64::MAX;

    /// Open both pools at `base` (`…/objects.exploded`), adopting whatever
    /// lazy segments a previous process left beside it.
    pub fn open(base: &Path, policy: ExplodePolicy, budget: u64) -> Result<Self> {
        let eager = ExplodedArchive::open_with_policy(base, policy)?;
        let mut found: Vec<(u64, PathBuf)> = Vec::new();
        if let Some(dir) = base.parent()
            && let Ok(rd) = std::fs::read_dir(dir)
        {
            let stem = base
                .file_name()
                .map(|s| s.to_string_lossy().into_owned())
                .unwrap_or_default();
            let prefix = format!("{stem}.c");
            for e in rd.flatten() {
                let name = e.file_name().to_string_lossy().into_owned();
                if let Some(n) = name.strip_prefix(&prefix)
                    && let Ok(n) = n.parse::<u64>()
                {
                    found.push((n, e.path()));
                }
            }
        }
        found.sort_unstable_by_key(|(n, _)| *n);
        let next = found.last().map(|(n, _)| n + 1).unwrap_or(1);
        // A segment adopted from a previous process has no cost history — the
        // scores are per-process and are not written down. It is scored by its
        // position instead, oldest first, which is what plain FIFO would do
        // and is the honest reading of "we do not know".
        let seg_stats: Vec<SegStat> = (0..found.len())
            .map(|i| SegStat { rows: 0, cost: 0, touched: i as u64 })
            .collect();
        let mut lazy = Vec::with_capacity(found.len());
        for (_, p) in found {
            // The lazy pool takes EVERY kind: it is where an object the eager
            // policy declined comes to rest, so its own policy is `Full`.
            lazy.push(std::sync::Arc::new(ExplodedArchive::open_with_policy(
                &p,
                ExplodePolicy::Full,
            )?));
        }
        Ok(Self {
            eager,
            lazy: Mutex::new(lazy),
            base: base.to_path_buf(),
            policy,
            budget: AtomicU64::new(budget),
            global_line: AtomicU64::new(default_global_line(base)),
            disk_line_percent: AtomicU64::new(DISK_LINE_PERCENT),
            disk_pressure: std::sync::atomic::AtomicBool::new(false),
            pressure_refused_write_backs: AtomicU64::new(0),
            // Hermetic by default: a test box at 85 % would otherwise run every
            // eviction test in this crate — and every `GitStore` test — under the
            // disk line, turning "cheapest first" into "oldest first" and a
            // store's lazy pool into one segment, depending on whose disk it is.
            // A test that wants the real `statvfs` sets `NO_DISK_READING`.
            #[cfg(test)]
            disk_used_override: AtomicU64::new(0),
            segment_bytes: AtomicU64::new(segment_for(budget)),
            next_seg: AtomicU64::new(next),
            written_back: AtomicU64::new(0),
            direct_reads: AtomicU64::new(0),
            suppressed_write_backs: AtomicU64::new(0),
            seg_cost: Mutex::new(seg_stats),
            evicted_segments: AtomicU64::new(0),
            evicted_bytes: AtomicU64::new(0),
            evicted_rows: AtomicU64::new(0),
            dirty: AtomicU64::new(0),
            root: base.parent().unwrap_or(base).display().to_string(),
            maintaining: std::sync::atomic::AtomicBool::new(false),
        })
    }

    /// Register this cache so a dashboard tick can read every open store's
    /// band without holding a store lock — the same `Weak` registry
    /// [`crate::progress`] keeps, for the same reason and with the same sweep.
    pub fn registered(self: std::sync::Arc<Self>) -> std::sync::Arc<Self> {
        registry()
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(std::sync::Arc::downgrade(&self));
        self
    }

    pub fn root(&self) -> &str {
        &self.root
    }

    /// The quota, changed live — what the control plane sets when an account's
    /// plan changes. Takes effect at the next rotation; it never refuses a
    /// write and never deletes on the caller's thread.
    pub fn set_budget(&self, bytes: u64) {
        self.budget.store(bytes, Ordering::Release);
        self.segment_bytes.store(segment_for(bytes), Ordering::Release);
    }

    /// The process-wide line this cache holds itself to its fair share of.
    /// `0` turns the global check off. Tests set it small; production takes
    /// [`default_global_line`] at open.
    pub fn set_global_line(&self, bytes: u64) {
        self.global_line.store(bytes, Ordering::Release);
    }

    /// The disk line, changed live. `0` turns it off. Takes effect at the next
    /// rotation, like the budget.
    pub fn set_disk_line_percent(&self, percent: u64) {
        self.disk_line_percent.store(percent, Ordering::Release);
    }

    /// A test's disk reading, in percent, in place of `statvfs`.
    #[cfg(test)]
    pub(crate) fn set_disk_reading_for_test(&self, percent: u64) {
        self.disk_used_override.store(percent, Ordering::Release);
    }

    /// A test's segment size, so a few write-backs rotate into several segments.
    #[cfg(test)]
    pub(crate) fn set_segment_bytes_for_test(&self, bytes: u64) {
        self.segment_bytes.store(bytes, Ordering::Release);
    }

    /// True while the filesystem holding this store is at or past the disk
    /// line. Read afresh per eviction, because each unlink moves it.
    fn over_disk_line(&self) -> bool {
        let line = self.disk_line_percent.load(Ordering::Acquire);
        if line == 0 {
            self.disk_pressure.store(false, Ordering::Release);
            return false;
        }
        let used = self.disk_used_reading();
        let Some(used) = used else {
            // No reading: keep whatever was latched — a statvfs that fails
            // once must neither start nor end a pressure episode.
            return self.disk_pressure.load(Ordering::Acquire);
        };
        let latched = pressure_after(self.disk_pressure.load(Ordering::Acquire), used, line);
        self.disk_pressure.store(latched, Ordering::Release);
        latched
    }

    /// The disk's used percent: the test override, else `statvfs` on the pool.
    fn disk_used_reading(&self) -> Option<u64> {
        #[cfg(test)]
        {
            let o = self.disk_used_override.load(Ordering::Acquire);
            if o != Self::NO_DISK_READING {
                return Some(o);
            }
        }
        disk_used_percent(self.pool_dir())
    }

    /// Write-backs refused while the disk line was latched.
    pub fn pressure_refused_write_backs(&self) -> u64 {
        self.pressure_refused_write_backs.load(Ordering::Relaxed)
    }

    /// The directory both pools live in: the eager file is `base` and every lazy
    /// segment is `base.cN` beside it, so this is the filesystem an unlink here
    /// gives back to. Not `base` itself: the eager file does not exist until the
    /// first commit is flushed, `statvfs` on a missing path is `ENOENT`, and a
    /// disk line read through it was silently off for a store that had not yet
    /// indexed a commit.
    fn pool_dir(&self) -> &Path {
        match self.base.parent() {
            Some(p) if !p.as_os_str().is_empty() => p,
            _ => Path::new("."),
        }
    }

    /// A lazy segment's read, where an error from a segment an eviction has
    /// already retired is a MISS. The segments are cloned out of the list and
    /// read with the lock released, and a segment opens its file by path per
    /// read (`File::open` in `ExplodedArchive::read_batch`/`content`), so an
    /// eviction between the clone and the open made the read `ENOENT` — which
    /// `?` turned into an error for a reader, or into a failed absorb through
    /// `refold`'s `of_kind`. A retired segment is by definition not in the
    /// cache any more, so "absent" is the true answer: fall back and re-derive.
    fn lazy_read<T>(seg: &ExplodedArchive, r: Result<T>) -> Result<Option<T>> {
        match r {
            Ok(v) => Ok(Some(v)),
            Err(_) if seg.is_retired() => Ok(None),
            Err(e) => Err(e),
        }
    }

    /// True when the registered caches together exceed the global line AND this
    /// cache holds more than its fair share of it — the condition under which
    /// this cache evicts for the box rather than for itself. Computed without
    /// holding this cache's own lock, so the registry sweep can lock the others.
    fn over_global_share(&self) -> bool {
        let line = self.global_line.load(Ordering::Acquire);
        if line == 0 {
            return false;
        }
        let all = use_all();
        // The share is among caches that HOLD something: an idle repository's
        // empty pool must not dilute the line for the ones actually using it.
        let live = all
            .iter()
            .filter(|(_, u)| u.eager_bytes + u.lazy_bytes > 0)
            .count()
            .max(1) as u64;
        let sum: u64 = all.iter().map(|(_, u)| u.eager_bytes + u.lazy_bytes).sum();
        if sum <= line {
            return false;
        }
        let mine = self.use_now();
        mine.eager_bytes + mine.lazy_bytes > line / live
    }

    pub fn budget(&self) -> u64 {
        self.budget.load(Ordering::Acquire)
    }

    pub fn policy(&self) -> ExplodePolicy {
        self.policy
    }

    pub fn path(&self) -> &Path {
        &self.base
    }

    /// **The eager write** — the indexer's sink, at push time. Unchanged
    /// behaviour: the policy decides, and what it declines is counted as
    /// skipped rather than silently dropped.
    pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
        self.eager.explode(oid, kind, payload)
    }

    pub fn explode_at(
        &self,
        oid: &[u8],
        kind: GitObjectKind,
        mode: Option<u32>,
        path: Option<&str>,
        payload: &[u8],
    ) -> Result<()> {
        self.eager.explode_at(oid, kind, mode, path, payload)
    }

    /// **The lazy write-back** — one object a reader has just inflated from the
    /// pack, kept for the next reader.
    ///
    /// Buffers only: it appends to the newest segment's in-memory rows and
    /// returns. Nothing is written, nothing is fsynced, no eviction runs. The
    /// reader that called it has its bytes already and is not made to wait for
    /// anybody — [`Self::maintain`] does the I/O on the store's own thread.
    /// Two readers racing on one oid is benign: the same bytes are buffered
    /// twice and the newest row wins, which is what
    /// [`ExplodedArchive::explode_at`] already promises.
    pub fn write_back(
        &self,
        oid: &[u8],
        kind: GitObjectKind,
        payload: &[u8],
        cost: crate::resolve::ReadCost,
    ) -> Result<bool> {
        // **A validation read is not a reader.** A `receive-pack` reads every
        // commit and tree it was sent, once, to prove the push closes — and
        // then never reads them again. Caching those is writing a read-cache
        // for a reader that does not exist. MEASURED on torvalds/linux
        // (2026-09-03, oden): the lazy pool is EMPTY until the instant indexing
        // completes and then grows to **6 778 047 872 B** before the push
        // returns, which is 31 % of the whole store and the largest non-pack
        // thing in it. See `gunnar/.nornir/storage-levers-2026-09-03.md`.
        //
        // The caller declares the window ([`suppress_write_back`]) because the
        // caller is the only thing that knows what kind of read this is: the
        // store sees the same `read` either way.
        if write_back_suppressed() {
            self.suppressed_write_backs.fetch_add(1, Ordering::Relaxed);
            return Ok(false);
        }
        // **No refill under disk pressure.** Eviction is dropping the oldest
        // segments to give the disk back; caching now would write what the next
        // tick throws away. The read itself is served; only the copy is skipped.
        if self.over_disk_line() {
            self.pressure_refused_write_backs.fetch_add(1, Ordering::Relaxed);
            return Ok(false);
        }
        // **The admission gate.** A shallow, small object is cheaper to inflate
        // again than to keep: caching it would spend somebody's quota to save
        // microseconds, and evict something that costs milliseconds.
        if !worth_caching(cost.depth, cost.inflated, cost.external_base) {
            self.direct_reads.fetch_add(1, Ordering::Relaxed);
            return Ok(false);
        }
        let seg = {
            let mut l = self
                .lazy
                .lock()
                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
            match l.last() {
                Some(s) => s.clone(),
                None => {
                    let s = self.new_segment()?;
                    l.push(s.clone());
                    s
                }
            }
        };
        seg.explode(oid, kind, payload)?;
        {
            let tick = self.written_back.load(Ordering::Relaxed);
            let mut st = self
                .seg_cost
                .lock()
                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
            if st.is_empty() {
                st.push(SegStat::default());
            }
            if let Some(last) = st.last_mut() {
                last.rows += 1;
                last.cost += rebuild_cost(cost.depth, cost.inflated, cost.external_base);
                last.touched = tick;
            }
        }
        self.written_back.fetch_add(1, Ordering::Relaxed);
        let n = self
            .dirty
            .fetch_add(payload.len() as u64, Ordering::AcqRel)
            + payload.len() as u64;
        Ok(n >= CACHE_FLUSH_BYTES)
    }

    /// **Write down what is buffered, rotate if the newest segment is full, and
    /// evict until the pools fit the budget.** The whole of the tier's I/O, on
    /// one thread that is nobody's reader.
    ///
    /// Returns the bytes reclaimed, so a caller can log applied output rather
    /// than an intention.
    pub fn maintain(&self) -> Result<u64> {
        let newest = {
            let l = self
                .lazy
                .lock()
                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
            l.last().cloned()
        };
        if let Some(seg) = newest {
            seg.flush()?;
            self.dirty.store(0, Ordering::Release);
            if seg.disk_bytes() >= self.segment_bytes.load(Ordering::Acquire) {
                let fresh = self.new_segment()?;
                let mut l = self
                    .lazy
                    .lock()
                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
                l.push(fresh);
                if let Ok(mut st) = self.seg_cost.lock() {
                    st.push(SegStat {
                        rows: 0,
                        cost: 0,
                        touched: self.written_back.load(Ordering::Relaxed),
                    });
                }
            }
        }
        self.evict_to_budget()
    }

    /// Drop segments until the two pools fit, **cheapest-to-rebuild first**.
    ///
    /// Not plain FIFO. A segment is ranked by
    /// `mean rebuild cost × recency`, and the lowest score goes: a segment of
    /// shallow 20 KiB blobs is worth less than one of depth-40 chains even if
    /// it was written more recently, because re-reading the first costs
    /// microseconds an object and the second costs milliseconds. Recency is
    /// still half the rank, so a cold segment of expensive objects does
    /// eventually go rather than pinning the pool.
    ///
    /// **The eager pool is never touched**: commits are what `graph`, `reach`
    /// and a tag peel read, and a store that answers those slowly is worse than
    /// one that reads a blob slowly. If the eager pool alone is over budget the
    /// lazy pool goes to nothing and the store is reported `over`.
    fn evict_to_budget(&self) -> Result<u64> {
        let budget = self.budget.load(Ordering::Acquire);
        if budget == 0
            && self.global_line.load(Ordering::Acquire) == 0
            && self.disk_line_percent.load(Ordering::Acquire) == 0
        {
            return Ok(0);
        }
        let eager = self.eager.disk_bytes();
        let mut freed = 0u64;
        loop {
            // Re-read per iteration: each eviction moves this cache's share.
            let over_share = self.over_global_share();
            let over_disk = self.over_disk_line();
            let (used, victim) = {
                let l = self
                    .lazy
                    .lock()
                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
                let lazy: u64 = l.iter().map(|s| s.disk_bytes()).sum();
                // The newest segment is the one being written into; evicting it
                // would drop rows a reader just paid for and start the same
                // segment again. There is always at least one live segment.
                // Within its own budget (`0` = unlimited) AND not over its
                // share of the box AND the disk under its line: nothing to do.
                // Over the disk line the OLDEST segment goes; otherwise the
                // cheapest, for whichever reason applies.
                let within_own = budget == 0 || eager + lazy <= budget;
                if (within_own && !over_share && !over_disk) || l.len() < 2 {
                    (eager + lazy, None)
                } else {
                    let at = if over_disk { 0 } else { self.cheapest_segment(l.len() - 1) };
                    (eager + lazy, Some(l[at].clone()))
                }
            };
            let Some(victim) = victim else {
                let _ = used;
                break;
            };
            let bytes = victim.disk_bytes();
            let rows = victim.rows().unwrap_or(0);
            let path = victim.path().to_path_buf();
            {
                let mut l = self
                    .lazy
                    .lock()
                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
                // By identity, not by the index read under the first lock: a
                // second evictor (a `maintain` called directly by `gc` or a test
                // while a kicked thread runs) may have removed a segment in
                // between, and `remove(at)` would then drop the WRONG one from
                // the list — never unlinked, never counted, a leak until restart.
                let Some(now) = l.iter().position(|s| std::sync::Arc::ptr_eq(s, &victim)) else {
                    // The other evictor took this one; look again.
                    continue;
                };
                l.remove(now);
                if let Ok(mut st) = self.seg_cost.lock()
                    && now < st.len()
                {
                    st.remove(now);
                }
            }
            // Out of the list, and now closed to writers BEFORE the unlink, so
            // no holder of an older `Arc` can re-create the path.
            victim.retire();
            drop(victim);
            // The file goes even if some reader still holds an `Arc` on the
            // segment: an unlinked file keeps answering the handles that are
            // open, and the blocks come back when the last one closes. So the
            // budget is honoured immediately and no read in flight fails.
            std::fs::remove_file(&path).ok();
            if over_disk {
                tracing::info!(target: "znippy.git", root = %self.root, bytes, "explode cache: oldest segment dropped, disk at or past its line");
            }
            self.evicted_segments.fetch_add(1, Ordering::Relaxed);
            self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
            self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
            freed += bytes;
        }
        Ok(freed)
    }

    /// The index of the segment with the lowest `mean rebuild cost × recency`,
    /// searched over `[0, exclusive)` so the segment being written into is
    /// never a candidate. Ties and unknown costs (a segment adopted from a
    /// previous process, which carries no history) fall back to the oldest,
    /// which is what plain FIFO would have chosen.
    fn cheapest_segment(&self, exclusive: usize) -> usize {
        let Ok(st) = self.seg_cost.lock() else {
            return 0;
        };
        let newest = st.last().map(|s| s.touched).unwrap_or(0).max(1);
        let mut best = 0usize;
        let mut best_score = u128::MAX;
        for i in 0..exclusive.min(st.len()) {
            let s = st[i];
            let mean = if s.rows == 0 { 0 } else { s.cost / s.rows };
            // Recency in [1, 1024]: a segment written at the very start scores
            // 1, the newest scores 1024. Multiplying rather than adding keeps
            // the two halves on one scale without a weight nobody can justify.
            let recency = 1 + (s.touched as u128 * 1023) / newest as u128;
            let score = (mean as u128 + 1) * recency;
            if score < best_score {
                best_score = score;
                best = i;
            }
        }
        best
    }

    fn new_segment(&self) -> Result<std::sync::Arc<ExplodedArchive>> {
        let n = self.next_seg.fetch_add(1, Ordering::AcqRel);
        let name = format!(
            "{}.c{n}",
            self.base.file_name().unwrap_or_default().to_string_lossy()
        );
        let path = self.base.with_file_name(name);
        Ok(std::sync::Arc::new(ExplodedArchive::open_with_policy(
            &path,
            ExplodePolicy::Full,
        )?))
    }

    /// The object's kind and bytes from either pool, or `None` — *fall back and
    /// re-derive*, never *wrong*.
    ///
    /// Eager first: it is the smaller table, its index is the one already warm,
    /// and under the shipped policy it holds exactly the objects a graph walk
    /// asks for. Then the lazy segments **newest first**, so a re-entered
    /// object is found before the stale copy an older segment may still hold.
    pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
        if let Some(hit) = self.eager.content(oid)? {
            return Ok(Some(hit));
        }
        let segs = self.segments()?;
        for s in segs.iter().rev() {
            if let Some(Some(hit)) = Self::lazy_read(s, s.content(oid))? {
                return Ok(Some(hit));
            }
        }
        Ok(None)
    }

    pub fn has(&self, oid: &[u8]) -> Result<bool> {
        if self.eager.has(oid)? {
            return Ok(true);
        }
        for s in self.segments()?.iter().rev() {
            if Self::lazy_read(s, s.has(oid))? == Some(true) {
                return Ok(true);
            }
        }
        Ok(false)
    }

    pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
        if let Some(n) = self.eager.name(oid)? {
            return Ok(Some(n));
        }
        for s in self.segments()?.iter().rev() {
            if let Some(Some(n)) = Self::lazy_read(s, s.name(oid))? {
                return Ok(Some(n));
            }
        }
        Ok(None)
    }

    /// Every oid of one kind, across both pools, newest spelling winning.
    ///
    /// The graph fold's input. Under the shipped policy every commit is in the
    /// eager pool, so this is one table's scan and the lazy segments add
    /// nothing — but it unions rather than assuming that, because a store
    /// opened under a different policy must still fold a whole graph.
    pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        let mut out = self.eager.of_kind(kind)?;
        let segs = self.segments()?;
        if segs.is_empty() {
            return Ok(out);
        }
        let mut seen: std::collections::HashSet<Vec<u8>> =
            out.iter().map(|(o, _)| o.clone()).collect();
        for s in segs.iter().rev() {
            for (oid, payload) in Self::lazy_read(s, s.of_kind(kind))?.unwrap_or_default() {
                if seen.insert(oid.clone()) {
                    out.push((oid, payload));
                }
            }
        }
        Ok(out)
    }

    /// Retire every oid `live` rejects — in **both** pools.
    ///
    /// Both, and that is not defensive: a tombstone written only into the eager
    /// pool would leave the lazy pool's row live, `content` would fall through
    /// to it, and a `gc`'d object would come back by the second door. That is
    /// the same defect
    /// `a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened` found
    /// the first time, one pool along.
    pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
        let mut n = self.eager.retain(live)?;
        for s in self.segments()? {
            n += Self::lazy_read(&s, s.retain(live))?.unwrap_or(0);
        }
        Ok(n)
    }

    /// **`repo gc` reclaims the cache.** Every lazy segment is unlinked; the
    /// eager pool keeps its rows, because they are the graph. Returns the bytes
    /// that left the filesystem.
    pub fn drop_cache(&self) -> Result<u64> {
        let victims = {
            let mut l = self
                .lazy
                .lock()
                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
            if let Ok(mut st) = self.seg_cost.lock() {
                st.clear();
            }
            std::mem::take(&mut *l)
        };
        let mut freed = 0;
        for s in victims {
            let bytes = s.disk_bytes();
            let rows = s.rows().unwrap_or(0);
            let path = s.path().to_path_buf();
            s.retire();
            drop(s);
            std::fs::remove_file(&path).ok();
            self.evicted_segments.fetch_add(1, Ordering::Relaxed);
            self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
            self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
            freed += bytes;
        }
        self.dirty.store(0, Ordering::Release);
        Ok(freed)
    }

    pub fn flush(&self) -> Result<()> {
        self.eager.flush()?;
        for s in self.segments()? {
            Self::lazy_read(&s, s.flush())?;
        }
        self.dirty.store(0, Ordering::Release);
        Ok(())
    }

    /// What the memory watchdog reaches for: every derived, rebuildable
    /// structure in every pool.
    pub fn drop_caches(&self) {
        self.eager.drop_caches();
        if let Ok(l) = self.lazy.lock() {
            for s in l.iter() {
                s.drop_caches();
            }
        }
    }

    pub fn rows(&self) -> Result<u64> {
        let mut n = self.eager.rows()?;
        for s in self.segments()? {
            n += s.rows()?;
        }
        Ok(n)
    }

    /// Rows in the EAGER pool only — what `adopt_journal` compares against the
    /// index, because the lazy pool's size is a function of what has been read
    /// and says nothing about whether a pack was absorbed.
    pub fn eager_rows(&self) -> Result<u64> {
        self.eager.rows()
    }

    pub fn disk_bytes(&self) -> u64 {
        self.use_now().used()
    }

    pub fn use_now(&self) -> CacheUse {
        let (lazy_bytes, segments) = match self.lazy.lock() {
            Ok(l) => (l.iter().map(|s| s.disk_bytes()).sum(), l.len() as u64),
            Err(_) => (0, 0),
        };
        CacheUse {
            eager_bytes: self.eager.disk_bytes(),
            lazy_bytes,
            budget: self.budget.load(Ordering::Acquire),
            segments,
            evicted_segments: self.evicted_segments.load(Ordering::Relaxed),
            evicted_bytes: self.evicted_bytes.load(Ordering::Relaxed),
            evicted_rows: self.evicted_rows.load(Ordering::Relaxed),
            written_back: self.written_back.load(Ordering::Relaxed),
            direct_reads: self.direct_reads.load(Ordering::Relaxed),
            suppressed_write_backs: self.suppressed_write_backs.load(Ordering::Relaxed),
        }
    }

    pub fn note_rederived(&self) {
        self.eager.note_rederived();
    }

    pub fn pread_served(&self) -> u64 {
        self.eager.pread_served()
            + self
                .segments()
                .map(|s| s.iter().map(|x| x.pread_served()).sum::<u64>())
                .unwrap_or(0)
    }

    pub fn skipped(&self) -> u64 {
        self.eager.skipped()
    }

    /// The engine-wide counters, summed over both pools, so no caller can tell
    /// the two apart by its instruments.
    pub fn engine_stats(&self) -> ExplodedStats {
        let mut s = self.eager.engine_stats();
        if let Ok(segs) = self.segments() {
            for x in segs {
                let e = x.engine_stats();
                s.rows += e.rows;
                s.written += e.written;
                s.served += e.served;
                s.absent += e.absent;
            }
        }
        s
    }

    /// **Start the write-back thread if one is not already running — and
    /// return to the reader immediately either way.**
    ///
    /// The owner's law, from the design note: *nobody waits*. The reader already
    /// has its bytes; the append, the fsync, the rotation and the eviction all
    /// happen here, on a thread that is nobody's request. A thread per 4 MiB of
    /// write-back rather than a parked thread per store, because a box holding
    /// 1 200 open repositories (lane V) cannot afford 1 200 idle threads and
    /// can easily afford one spawn per few thousand objects read.
    ///
    /// The flag is the whole of the mutual exclusion: a second reader finding
    /// it set does nothing, and the running thread loops until the buffer is
    /// empty before clearing it, so no work is dropped between the two.
    pub fn kick(self: &std::sync::Arc<Self>) {
        if self
            .maintaining
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return;
        }
        let me = self.clone();
        let spawned = std::thread::Builder::new()
            .name("explode-writeback".into())
            .stack_size(256 << 10)
            .spawn(move || {
                loop {
                    if let Err(e) = me.maintain() {
                        tracing::warn!(target: "znippy.git", error = %e, "explode cache maintenance");
                        break;
                    }
                    if me.dirty.load(Ordering::Acquire) < CACHE_FLUSH_BYTES {
                        break;
                    }
                }
                me.maintaining.store(false, Ordering::Release);
            });
        if spawned.is_err() {
            // No thread to be had. Clearing the flag is what makes the next
            // reader try again rather than the cache silently stopping.
            self.maintaining.store(false, Ordering::Release);
        }
    }

    fn segments(&self) -> Result<Vec<std::sync::Arc<ExplodedArchive>>> {
        Ok(self
            .lazy
            .lock()
            .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?
            .clone())
    }
}

impl crate::exploded::PayloadSink for ExplodedCache {
    fn wants(&self, kind: GitObjectKind) -> bool {
        crate::exploded::PayloadSink::wants(&self.eager, kind)
    }
    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
        ExplodedCache::explode(self, oid, kind, payload)
    }
    fn flush(&self) -> Result<()> {
        self.eager.flush().map(|_| ())
    }
}

fn registry() -> &'static Mutex<Vec<std::sync::Weak<ExplodedCache>>> {
    static R: std::sync::OnceLock<Mutex<Vec<std::sync::Weak<ExplodedCache>>>> =
        std::sync::OnceLock::new();
    R.get_or_init(|| Mutex::new(Vec::new()))
}

/// **Every open store's explode band**, as `(store root, use)`. What the
/// dashboard tick and `Admin.StorageInfo` read; a `metadata` call per pool and
/// no store lock. Dead entries are swept on the way past.
pub fn use_all() -> Vec<(String, CacheUse)> {
    let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
    r.retain(|w| w.strong_count() > 0);
    r.iter()
        .filter_map(|w| w.upgrade())
        .map(|c| (c.root.clone(), c.use_now()))
        .collect()
}

/// Set the budget of every open store under `root` (a prefix match, so an
/// account's whole tree is one call). Returns how many were changed — applied
/// output, so a control plane that set nothing can say so.
pub fn set_budget_under(root: &str, bytes: u64) -> usize {
    let mut n = 0;
    let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
    r.retain(|w| w.strong_count() > 0);
    for c in r.iter().filter_map(|w| w.upgrade()) {
        if c.root.starts_with(root) {
            c.set_budget(bytes);
            n += 1;
        }
    }
    n
}

/// One segment is 1/16 of the budget, clamped. An unlimited budget still
/// rotates, so a pool that is later given a budget can be evicted without a
/// rewrite.
fn segment_for(budget: u64) -> u64 {
    if budget == 0 {
        return MAX_SEGMENT.min(1 << 28);
    }
    // The floor yields to the budget rather than overriding it: a 16 MiB
    // minimum against a 1 MiB quota would mean the pool never rotates and
    // therefore never evicts — a cap that silently is not one, which is the
    // exact shape of bug LAW 2 is about. So the floor is `min(MIN_SEGMENT,
    // budget/4)`: a real budget gets 16 MiB segments, a tiny one gets tiny ones.
    let floor = MIN_SEGMENT.min(budget / 4).max(64 << 10);
    (budget / SEGMENTS_PER_BUDGET).clamp(floor, MAX_SEGMENT)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tmp(name: &str) -> PathBuf {
        let d = std::env::temp_dir().join(format!("explode-cache-{}-{name}", std::process::id()));
        let _ = std::fs::remove_dir_all(&d);
        std::fs::create_dir_all(&d).unwrap();
        d.join("objects.exploded")
    }

    /// A cost the admission gate admits: deep enough and big enough that the
    /// cache is the cheaper place for it. Tests that are about eviction must
    /// not accidentally be about admission.
    /// An INCOMPRESSIBLE payload of `len` bytes (a tiny LCG seeded by `seed`).
    /// The eviction tests reason about bytes on disk against a budget, and since
    /// lever 3 (2026-09-04) a row is stored zlib-compressed when that is smaller
    /// — a `vec![3u8; 64 << 10]` lands as ~80 bytes and the pool never reaches
    /// the budget those tests were sized for ("nothing was evicted" ×3). Noise
    /// does not compress, so the budget arithmetic means what it did.
    fn noise(len: usize, seed: u8) -> Vec<u8> {
        let mut x: u32 = 0x9E37_79B9 ^ (seed as u32).wrapping_mul(0x0101_0101);
        (0..len)
            .map(|_| {
                x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
                (x >> 24) as u8
            })
            .collect()
    }

    fn dear(payload: &[u8]) -> crate::resolve::ReadCost {
        crate::resolve::ReadCost {
            depth: CHEAP_DEPTH + 8,
            inflated: payload.len() as u64,
            external_base: false,
        }
    }

    /// Assert at compile time that the eviction fixtures below stay under the
    /// size cap — a fixture that grew past it would be silently DECLINED and
    /// the test would then be asserting nothing.
    const _: () = assert!((64 << 10) < MAX_CACHED_BYTES as usize);

    fn oid(n: u32) -> Vec<u8> {
        let mut o = vec![0u8; 20];
        o[..4].copy_from_slice(&n.to_be_bytes());
        o
    }

    /// ★ **THE ANSWER TO "does it take the disk anyway?"**
    ///
    /// Every object of a repository is read once, each one written back — the
    /// full-history browse that would otherwise reproduce the eager tier. The
    /// tier must **plateau at its budget**, not climb.
    ///
    /// Seen RED by returning `Ok(0)` from `evict_to_budget`: the pool climbs to
    /// 9.4 MiB against a 2 MiB budget and the assertion names the number.
    #[test]
    fn the_tier_plateaus_at_its_budget_under_a_full_scan() {
        let base = tmp("plateau");
        let budget = 2 << 20;
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, budget).unwrap();
        let payload = vec![7u8; 8 << 10];
        let mut peak = 0u64;
        for i in 0..2000u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
            // The store's own thread, doing what it does between reads.
            c.maintain().unwrap();
            peak = peak.max(c.use_now().used());
        }
        let end = c.use_now();
        let written = 2000u64 * payload.len() as u64;
        assert!(
            peak <= budget as u64 + segment_for(budget) + (1 << 20),
            "the tier climbed to {peak} B against a {budget} B budget after {written} B of \
             write-backs — it is a tier, not a cache"
        );
        assert!(
            end.evicted_bytes > 0 && end.evicted_segments > 0,
            "nothing was evicted, so the plateau is an accident of size: {end:?}"
        );
    }

    /// An account at its budget still serves **every** read correctly — only
    /// slower. Eviction is never an error a client sees.
    #[test]
    fn an_evicted_object_is_absent_and_never_wrong() {
        let base = tmp("absent");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
        let payload = noise(64 << 10, 3);
        for i in 0..200u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
            c.maintain().unwrap();
        }
        // Whatever survived comes back byte-identical; whatever did not comes
        // back `None`, which is the caller's cue to re-inflate the chain.
        let mut hits = 0;
        for i in 0..200u32 {
            match c.content(&oid(i)).unwrap() {
                Some((k, b)) => {
                    assert_eq!(k, GitObjectKind::Blob);
                    assert_eq!(b, payload, "an evicting cache returned the wrong bytes");
                    hits += 1;
                }
                None => {}
            }
        }
        assert!(hits > 0, "the cache evicted everything, including the newest segment");
        assert!(hits < 200, "nothing was evicted, so this proves nothing");
    }

    /// Commits are the last thing standing: the eager pool is never unlinked,
    /// whatever the lazy pool does.
    /// **The box has one line, and the biggest pool pays for it.** Two caches,
    /// each far under its own per-store budget, share a small global line: the
    /// one holding most of the box's cache evicts on its tick; the one under
    /// its fair share does not, however far over the line the box is.
    ///
    /// Robust to whatever else is registered in this process: the fairness
    /// claim is made with B holding ONE 8 KiB row (over its share only if more
    /// than ~128 caches held bytes at once), and the convergence claim is made
    /// on A's own bytes, not on a sum that other tests' pools can move.
    ///
    /// Seen RED by making `over_global_share` return `false` always: A evicted
    /// nothing ("the big pool must evict for the box") and stayed at 3 MiB
    /// over a 1 MiB line. Restored.
    #[test]
    fn the_global_line_is_paid_by_the_cache_over_its_share() {
        let base_a = tmp("global-a");
        let base_b = tmp("global-b");
        // 4 MiB per store → 1 MiB segments (the budget/4 floor), so 3 MiB rotates into four and
        // the per-store budget itself never evicts anything here.
        let a = std::sync::Arc::new(ExplodedCache::open(&base_a, ExplodePolicy::Commits, 4 << 20).unwrap())
            .registered();
        let b = std::sync::Arc::new(ExplodedCache::open(&base_b, ExplodePolicy::Commits, 4 << 20).unwrap())
            .registered();
        // Fill with the global check OFF, so the shape under test is set up
        // before it is asked about.
        a.set_global_line(0);
        b.set_global_line(0);
        let big = noise(64 << 10, 5);
        for i in 0..48u32 {
            a.write_back(&oid(i), GitObjectKind::Blob, &big, dear(&big)).unwrap();
            a.maintain().unwrap();
        }
        let small = noise(8 << 10, 6);
        b.write_back(&oid(100), GitObjectKind::Blob, &small, dear(&small)).unwrap();
        b.maintain().unwrap();
        let a0 = a.use_now();
        let b0 = b.use_now();
        let line = 1u64 << 20;
        assert!(a0.lazy_bytes > line, "precondition: A alone is over the line, has {}", a0.lazy_bytes);
        assert!(a0.segments >= 3, "precondition: A rotated into segments, has {}", a0.segments);
        assert_eq!(a0.evicted_segments + b0.evicted_segments, 0, "precondition: nothing evicted yet");

        a.set_global_line(line);
        b.set_global_line(line);

        // B ticks first: under its share, it must not pay for A.
        b.maintain().unwrap();
        assert_eq!(b.use_now().evicted_segments, 0, "a cache under its fair share must not evict for the box");

        // A ticks: over its share, it evicts for the box until it is under the
        // line on its own (it cannot stop above it: then the box is over and
        // it is over any share of it).
        a.maintain().unwrap();
        let a1 = a.use_now();
        assert!(a1.evicted_segments > 0, "the big pool must evict for the box");
        assert!(
            a1.lazy_bytes + a1.eager_bytes <= line,
            "A must have shrunk under the line on its own: {} > {line}",
            a1.lazy_bytes + a1.eager_bytes
        );
        assert!(a1.lazy_bytes < 4 << 20, "its own budget was never the reason");
        assert_eq!(b.use_now().evicted_segments, 0, "B still untouched");
    }
    /// Fill one store into several lazy segments with its budget and the
    /// global line both far away, so only the disk line can evict.
    fn filled_for_disk(name: &str) -> ExplodedCache {
        let c = ExplodedCache::open(&tmp(name), ExplodePolicy::Commits, 64 << 20).unwrap();
        c.set_global_line(0);
        c.segment_bytes.store(256 << 10, Ordering::Release);
        c.disk_used_override.store(10, Ordering::Release);
        let big = noise(64 << 10, 9);
        for i in 0..24u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &big, dear(&big)).unwrap();
            c.maintain().unwrap();
        }
        let u = c.use_now();
        assert!(u.segments >= 4, "precondition: rotated into segments, has {}", u.segments);
        assert_eq!(u.evicted_segments, 0, "precondition: nothing evicted under a 10 % disk");
        c
    }

    /// ★ **At 80 % disk the oldest cached objects go.** Owner 2026-09-28.
    ///
    /// Seen RED by leaving `over_disk` out of the eviction condition: at 85 %
    /// nothing was evicted.
    #[test]
    fn a_disk_at_its_line_drops_the_oldest_segments() {
        let c = filled_for_disk("disk-line");
        let newest_before = c.segments().unwrap().last().unwrap().path().to_path_buf();
        assert!(c.content(&oid(0)).unwrap().is_some(), "precondition: the oldest write-back is cached");
        c.disk_used_override.store(85, Ordering::Release);
        c.maintain().unwrap();
        let u = c.use_now();
        assert!(u.evicted_segments > 0, "a disk at 85 % must drop cache segments");
        assert_eq!(u.segments, 1, "under pressure the pool gives back all but the live segment");
        assert_eq!(
            c.segments().unwrap()[0].path(),
            newest_before.as_path(),
            "the segment kept is the newest one, so what went was the oldest"
        );
        assert!(c.content(&oid(0)).unwrap().is_none(), "the oldest write-back is gone");
    }

    /// Under the line, or with the line off, the disk reading changes nothing.
    #[test]
    fn a_disk_under_its_line_or_with_it_off_drops_nothing() {
        let c = filled_for_disk("disk-under");
        c.disk_used_override.store(79, Ordering::Release);
        c.maintain().unwrap();
        assert_eq!(c.use_now().evicted_segments, 0, "79 % is under an 80 % line");
        c.set_disk_line_percent(0);
        c.disk_used_override.store(99, Ordering::Release);
        c.maintain().unwrap();
        assert_eq!(c.use_now().evicted_segments, 0, "a line of 0 is off");
    }

    /// The real reading is a percentage, the way `df` prints it.
    #[test]
    fn the_real_disk_reading_is_a_percentage() {
        let p = disk_used_percent(&std::env::temp_dir()).expect("statvfs answers for the temp dir");
        assert!(p <= 100, "{p}");
    }

    /// **The disk line reads the filesystem the pools live on, from the first
    /// open.** Review 2026-09-28: it was read through `base`, the eager FILE,
    /// which does not exist until a commit is flushed — `statvfs` said ENOENT
    /// and the line was off. Seen RED with `disk_used_percent(&self.base)`
    /// restored in `over_disk_line`: a real reading at its own value did not
    /// count as reached.
    #[test]
    fn the_disk_line_reads_the_pool_directory_before_the_eager_file_exists() {
        let base = tmp("disk-fresh");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 64 << 20).unwrap();
        assert!(!base.exists(), "precondition: nothing flushed, no eager file");
        assert!(disk_used_percent(&base).is_none(), "precondition: statvfs of a missing file");
        let p = disk_used_percent(base.parent().unwrap()).expect("statvfs of the pool dir");
        c.disk_used_override.store(ExplodedCache::NO_DISK_READING, Ordering::Release);
        c.set_disk_line_percent(p.max(1));
        assert_eq!(c.over_disk_line(), p >= p.max(1), "the real reading at {p} % against a {p} % line");
    }

    /// **An eviction between a reader's snapshot and its read is a miss, never
    /// an error, and never an orphan file.** Review 2026-09-28: `segments()`
    /// hands out `Arc`s and the lock is released, a segment opens its file BY
    /// PATH per read, so a segment unlinked in between answered `ENOENT` — and
    /// `content`/`of_kind` propagated it with `?` (on the push path that is
    /// `refold` failing the absorb). And a flush through such an `Arc` opened
    /// the path with `O_CREAT`, putting the file back where no pool lists it.
    ///
    /// Seen RED twice, one guard at a time: without `lazy_read`'s retired arm,
    /// "opening …/objects.exploded.c1" came back as an `Err`; without the
    /// retired check in `ExplodedArchive::flush`, "a flush through an evicted
    /// segment re-created …/objects.exploded.c1".
    #[test]
    fn a_read_or_flush_through_an_evicted_segment_is_a_miss_and_leaves_no_file() {
        let base = tmp("evict-race");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 64 << 20).unwrap();
        c.set_global_line(0);
        c.segment_bytes.store(256 << 10, Ordering::Release);
        let big = noise(64 << 10, 9);
        for i in 0..24u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &big, dear(&big)).unwrap();
            c.maintain().unwrap();
        }
        drop(c);
        // A reopen, so no segment has built its index: the first read has to
        // open the file, which is exactly the window.
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 64 << 20).unwrap();
        c.set_global_line(0);
        let snapshot = c.segments().unwrap();
        assert!(snapshot.len() >= 4, "precondition: several segments, has {}", snapshot.len());
        let oldest = snapshot[0].clone();
        let path = oldest.path().to_path_buf();

        c.disk_used_override.store(85, Ordering::Release);
        c.maintain().unwrap();
        assert!(!path.exists(), "precondition: the oldest segment was unlinked");
        assert!(oldest.is_retired(), "an evicted segment is retired");

        // The raw segment read is the hazard …
        assert!(oldest.content(&oid(0)).is_err(), "precondition: the file is gone under the Arc");
        // … and through the cache's door it is a miss.
        assert_eq!(ExplodedCache::lazy_read(&oldest, oldest.content(&oid(0))).unwrap(), None);
        assert_eq!(
            ExplodedCache::lazy_read(&oldest, oldest.of_kind(GitObjectKind::Blob)).unwrap(),
            None
        );

        // A late writer holding the old Arc must not bring the file back.
        oldest.explode(&oid(999), GitObjectKind::Blob, b"late").unwrap();
        oldest.flush().unwrap();
        assert!(!path.exists(), "a flush through an evicted segment re-created {}", path.display());
    }

    #[test]
    fn eviction_never_takes_a_commit() {
        let base = tmp("commits");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
        for i in 0..50u32 {
            c.explode(&oid(1_000_000 + i), GitObjectKind::Commit, b"tree x\n")
                .unwrap();
        }
        c.flush().unwrap();
        let blob = noise(64 << 10, 1);
        for i in 0..200u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
            c.maintain().unwrap();
        }
        assert!(c.use_now().evicted_segments > 0, "nothing was evicted");
        for i in 0..50u32 {
            assert!(
                c.content(&oid(1_000_000 + i)).unwrap().is_some(),
                "a commit was evicted — graph and reach are now blind"
            );
        }
    }

    /// `repo gc` reclaims the cache and keeps the graph.
    #[test]
    fn dropping_the_cache_reclaims_the_lazy_pool_only() {
        let base = tmp("gc");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
        c.explode(&oid(9), GitObjectKind::Commit, b"tree y\n").unwrap();
        let blob = vec![2u8; 32 << 10];
        for i in 0..100u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
        }
        c.maintain().unwrap();
        let before = c.use_now();
        assert!(before.lazy_bytes > 0);
        let freed = c.drop_cache().unwrap();
        assert!(freed > 0, "gc reclaimed nothing");
        let after = c.use_now();
        assert_eq!(after.lazy_bytes, 0, "the lazy pool survived gc");
        assert_eq!(after.eager_bytes, before.eager_bytes, "gc took the graph too");
        assert!(c.content(&oid(9)).unwrap().is_some());
    }

    /// The segments a previous process wrote are adopted, so a restart does not
    /// lose the cache and does not lose track of its size either.
    #[test]
    fn a_reopen_adopts_the_segments_it_left() {
        let base = tmp("reopen");
        let blob = vec![5u8; 32 << 10];
        let bytes = {
            let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
            for i in 0..40u32 {
                c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
            }
            c.maintain().unwrap();
            c.use_now().lazy_bytes
        };
        assert!(bytes > 0);
        let again = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
        assert_eq!(again.use_now().lazy_bytes, bytes, "the segments were not adopted");
        assert_eq!(
            again.content(&oid(7)).unwrap().map(|(_, b)| b),
            Some(blob),
            "an adopted segment does not answer"
        );
    }

    /// ★ **DO NOT EXPLODE UNIFORMLY: a cheap object is never cached.**
    ///
    /// The owner, 2026-08-30: the index points at the object's own zlib stream at
    /// its own offset, so a whole object is one direct inflate. Caching that
    /// spends quota to save microseconds — and evicts something that costs
    /// milliseconds. So the gate is rebuild cost, not "was read".
    ///
    /// Seen RED by making `write_back` admit everything: 400 shallow 1 KiB
    /// objects land in the pool, `direct_reads` stays 0, and the assertion
    /// names both.
    #[test]
    fn a_shallow_small_object_is_read_directly_and_never_cached() {
        let base = tmp("cheap");
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
        let small = vec![1u8; 1 << 10];
        let big = vec![2u8; 64 << 10];

        // Cheap: shallow and small. Not worth a byte of anybody's quota.
        for i in 0..400u32 {
            let took = c
                .write_back(
                    &oid(i),
                    GitObjectKind::Blob,
                    &small,
                    crate::resolve::ReadCost {
                        depth: CHEAP_DEPTH,
                        inflated: small.len() as u64,
                        external_base: false,
                    },
                )
                .unwrap();
            assert!(!took, "a cheap object asked for a flush");
        }
        c.maintain().unwrap();
        let after_cheap = c.use_now();
        assert_eq!(
            after_cheap.lazy_bytes, 0,
            "400 shallow 1 KiB objects put {} B in the cache; a direct read is \
             microseconds and this is quota spent to save them",
            after_cheap.lazy_bytes
        );
        assert_eq!(
            after_cheap.direct_reads, 400,
            "the policy declined {} of 400 cheap objects — it is not running",
            after_cheap.direct_reads
        );
        assert_eq!(after_cheap.written_back, 0);

        // Expensive by DEPTH, and by an external base. Each on its own admits.
        c.write_back(
            &oid(1000),
            GitObjectKind::Blob,
            &small,
            crate::resolve::ReadCost { depth: CHEAP_DEPTH + 1, inflated: small.len() as u64, external_base: false },
        )
        .unwrap();
        c.write_back(
            &oid(1002),
            GitObjectKind::Blob,
            &small,
            crate::resolve::ReadCost { depth: 0, inflated: small.len() as u64, external_base: true },
        )
        .unwrap();
        // **Big is NOT a reason to cache.** A 64 KiB — or a 3 MiB — whole blob
        // is one linear inflate and a bad tenant of a shared budget, and it is
        // declined however deep it is. This is lane AA's second copy.
        c.write_back(
            &oid(1001),
            GitObjectKind::Blob,
            &big,
            crate::resolve::ReadCost { depth: 40, inflated: (MAX_CACHED_BYTES + 1) as u64, external_base: true },
        )
        .unwrap();
        c.maintain().unwrap();
        let end = c.use_now();
        assert_eq!(end.written_back, 2, "an expensive object was declined");
        assert_eq!(
            end.direct_reads, 401,
            "an object over the size cap was cached instead of read directly"
        );
        for n in [1000u32, 1002] {
            assert!(
                c.content(&oid(n)).unwrap().is_some(),
                "the expensive object {n} is not in the cache"
            );
        }
        assert!(
            c.content(&oid(1001)).unwrap().is_none(),
            "an object over the size cap is in the cache"
        );
        assert!(
            c.content(&oid(7)).unwrap().is_none(),
            "a declined object is in the cache after all"
        );
    }

    /// ★ **Eviction ranks by rebuild cost, not by age alone.**
    ///
    /// Two segments, the older one full of expensive objects and the newer one
    /// full of barely-admitted ones. Plain FIFO takes the expensive segment
    /// because it is older; ranking by `mean rebuild cost × recency` takes the
    /// cheap one, which is the whole point — evict what is cheap to recompute.
    ///
    /// Seen RED by restoring `Some(l[0].clone())`: the depth-40 objects go and
    /// the barely-admitted ones stay.
    #[test]
    fn eviction_takes_the_cheap_segment_before_the_old_expensive_one() {
        let base = tmp("costrank");
        // Small budget, small segments: three segments fit and the fourth
        // forces one out.
        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
        let payload = noise(32 << 10, 9);
        let dear = crate::resolve::ReadCost { depth: 40, inflated: payload.len() as u64, external_base: false };
        // Barely admitted: one link past the cheap depth.
        let cheapish = crate::resolve::ReadCost {
            depth: CHEAP_DEPTH + 1,
            inflated: payload.len() as u64,
            external_base: false,
        };

        // Segment 0: expensive. Rotate it out by filling past the segment size.
        for i in 0..16u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear).unwrap();
        }
        c.maintain().unwrap();
        // Segment 1+: cheapish.
        for i in 100..160u32 {
            c.write_back(&oid(i), GitObjectKind::Blob, &payload, cheapish).unwrap();
            c.maintain().unwrap();
        }
        let end = c.use_now();
        assert!(end.evicted_segments > 0, "nothing was evicted; this proves nothing");
        // The expensive objects are the ones still there.
        let expensive_left = (0..16u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
        let cheap_left = (100..160u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
        assert!(
            expensive_left > 0,
            "every depth-40 object was evicted while {cheap_left} barely-admitted ones \
             survived — eviction is ranking by age alone: {end:?}"
        );
    }

    /// A budget is parsed, and a typo is refused rather than silently taking a
    /// default nobody is billed for.
    #[test]
    fn a_budget_is_read_and_a_typo_is_refused() {
        assert_eq!(parse_budget("8G").unwrap(), 8 << 30);
        assert_eq!(parse_budget(" 512m ").unwrap(), 512 << 20);
        assert_eq!(parse_budget("1048576").unwrap(), 1 << 20);
        assert_eq!(parse_budget("unlimited").unwrap(), 0);
        assert!(parse_budget("8 gigs").is_err());
        assert!(parse_budget("").is_err());
    }

    /// The default is derived from the filesystem, and the appliance arithmetic
    /// in the module docs is the one the code does.
    #[test]
    fn the_default_budget_is_the_smaller_of_a_tenth_and_the_ceiling() {
        let d = default_budget(&std::env::temp_dir());
        assert!(d >= MIN_BUDGET && d <= BUDGET_CEILING, "{d}");
    }

    /// **A validation read caches nothing, and the same reads without the guard
    /// fill the pool** — both halves, because either alone proves nothing.
    ///
    /// The RED half is the first loop: these are objects the admission policy
    /// WANTS (deep chains), so a pool that stays empty under the guard cannot be
    /// explained by `worth_caching` declining them. The counter is asserted too,
    /// because "the pool is empty" is also what a cache that was handed nothing
    /// looks like.
    #[test]
    fn a_suppressed_thread_caches_nothing_and_an_unsuppressed_one_fills_the_pool() {
        let payload = vec![7u8; 64 << 10];
        let costly = crate::resolve::ReadCost {
            depth: CHEAP_DEPTH + 8,
            inflated: payload.len() as u64,
            external_base: false,
        };

        // ── RED: no guard. The same objects, admitted. ──────────────────────
        let hot = ExplodedCache::open(&tmp("nosuppress"), ExplodePolicy::Commits, 0).unwrap();
        for i in 0..200u32 {
            hot.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
        }
        hot.maintain().unwrap();
        let hot_use = hot.use_now();
        assert!(
            hot_use.lazy_bytes > 0,
            "without the guard these objects must be cached, or the GREEN half \
             below is proving that `worth_caching` declined them and nothing else"
        );
        assert_eq!(hot_use.written_back, 200);
        assert_eq!(hot_use.suppressed_write_backs, 0);

        // ── GREEN: the identical loop inside the guard. ─────────────────────
        let cold = ExplodedCache::open(&tmp("suppress"), ExplodePolicy::Commits, 0).unwrap();
        {
            let _guard = suppress_write_back();
            assert!(write_back_suppressed());
            for i in 0..200u32 {
                let took = cold.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
                assert!(!took, "a validation read asked for a flush");
            }
        }
        cold.maintain().unwrap();
        let cold_use = cold.use_now();
        assert_eq!(
            cold_use.lazy_bytes, 0,
            "a receive-pack's connectivity walk put {} B in the cache",
            cold_use.lazy_bytes
        );
        assert_eq!(cold_use.written_back, 0);
        assert_eq!(
            cold_use.suppressed_write_backs, 200,
            "the gate must be what declined them — {} suppressed, not 200",
            cold_use.suppressed_write_backs
        );
        // The policy was never consulted: a suppressed read is declined BEFORE
        // the admission gate, so it is not a `direct_read` either.
        assert_eq!(cold_use.direct_reads, 0);
    }

    /// The guard is released on drop, including out of a panic — a thread that
    /// panicked mid-walk must not stop caching for the rest of its life.
    #[test]
    fn the_guard_is_released_on_drop_and_on_unwind_and_it_nests() {
        assert!(!write_back_suppressed(), "a fresh thread starts unsuppressed");
        {
            let _a = suppress_write_back();
            {
                let _b = suppress_write_back();
                assert!(write_back_suppressed());
            }
            assert!(
                write_back_suppressed(),
                "the inner guard ending reopened the cache while the outer scope \
                 is still validating"
            );
        }
        assert!(!write_back_suppressed());

        let unwound = std::panic::catch_unwind(|| {
            let _g = suppress_write_back();
            assert!(write_back_suppressed());
            panic!("a connectivity walk died");
        });
        assert!(unwound.is_err());
        assert!(
            !write_back_suppressed(),
            "a panic inside the guard left this thread never caching again"
        );
    }
}

#[cfg(test)]
mod disk_hysteresis {
    use super::{pressure_after, DISK_LINE_HYSTERESIS_PERCENT};

    #[test]
    fn the_line_trips_at_80_and_holds_until_under_75() {
        assert_eq!(DISK_LINE_HYSTERESIS_PERCENT, 5);
        assert!(!pressure_after(false, 79, 80), "under the line, not latched: no pressure");
        assert!(pressure_after(false, 80, 80), "at the line: trips");
        assert!(pressure_after(true, 78, 80), "between 75 and 80 after tripping: holds");
        assert!(pressure_after(true, 75, 80), "75 is not under 75: holds");
        assert!(!pressure_after(true, 74, 80), "under 75: releases");
        assert!(!pressure_after(false, 77, 80), "between the marks without having tripped: stays off");
    }
}