Skip to main content

znippy_plugin_git/
exploded_cache.rs

1//! **The exploded tier is a CACHE, not a tier — and this is the bound.**
2//!
3//! Rickard, 2026-08-30, on being shown the lazy-explode design: *"does lazy
4//! explode mean it eventually takes the disk anyway?"* — and the honest answer
5//! is **yes, unless the tier is capped**, because a client that browses a whole
6//! history reads every object once and a write-back-on-first-read tier would
7//! then hold exactly what the eager one held. MEASURED on the rig 2026-08-30:
8//! the 8.12 GiB `torvalds/linux` pack drove the eager tier to **141.7 GiB**
9//! into `/data`, ~17× the pack. *"We cannot have one repo take all 300 GB
10//! disk."*
11//!
12//! So the tier has a **byte budget** and evicts. Nothing here is data: every
13//! row is re-derivable from the verbatim pack bytes by
14//! [`crate::resolve::materialize_at`], so an eviction is never a loss and never
15//! an error a client can see — it is a later read paying ~0.5–2 ms to inflate
16//! its own delta chain again.
17//!
18//! # Two pools, and the split is the whole design
19//!
20//! ```text
21//!   objects.exploded         the EAGER pool — what the indexer writes at push
22//!                            time. Under the shipped policy that is commits
23//!                            and tags: what `graph`/`reach`/tag-peel need, so
24//!                            visibility and pull checks never stall.
25//!                            NEVER evicted — it is the "commits last" rule,
26//!                            held by construction rather than by a comparator.
27//!
28//!   objects.exploded.c1      the LAZY pool — first-read write-backs, appended
29//!   objects.exploded.c2      in read order, one file per segment.
30//!   …                        EVICTION IS ONE `unlink` of the oldest segment.
31//! ```
32//!
33//! Both pools are counted against the budget and both are named on the
34//! dashboard; only the lazy one is evictable. If the eager pool alone exceeds
35//! the budget the lazy pool is evicted to nothing and the eager pool is still
36//! kept — a store that cannot answer `reach` is broken, and a store that reads
37//! slowly is not.
38//!
39//! # Why segments, and why eviction is an `unlink`
40//!
41//! [`ExplodedArchive`] is one append-only Arrow IPC stream: `[schema][batch]…`,
42//! nothing ever rewritten. That is what makes it 1.0× its payload instead of
43//! redb's 12×, and it is also what makes *deleting a row in the middle* cost a
44//! whole-file rewrite — up to 17 GB, which the module docs already refuse to
45//! smuggle into `gc`. Cutting the pool into segment files turns "reclaim the
46//! coldest bytes" into `remove_file`: O(1), no rewrite, no read amplification,
47//! and a crash during it leaves a pool that is merely shorter.
48//!
49//! # The recency argument, stated honestly
50//!
51//! Segments are appended in **first-read order**, so dropping the oldest drops
52//! the objects least recently *admitted*. That is not the same as least
53//! recently *used* — an object read at minute 1 and again at minute 90 sits in
54//! an old segment. It is not promoted, and it does not need to be: when its
55//! segment is dropped, the next read of it misses, re-inflates its chain, and
56//! **writes it back into the newest segment**. Re-entry-on-miss is what makes
57//! this LRU-shaped, at the cost of one chain inflate per hot object per eviction
58//! cycle rather than a copy per hit. So it is FIFO over segments with
59//! re-entry-on-miss, and the thing an LRU is wanted for — a working set that
60//! stays resident under a scan of everything else — holds, because the working
61//! set is what keeps re-entering.
62//!
63//! # What a full-history read now costs on disk
64//!
65//! The plateau, not the climb. A reader that walks every object of a
66//! Linux-sized repository writes back every object it inflates; the pool grows
67//! to the budget and then stops, because each rotation past it drops the oldest
68//! segment. That is the number lane Q reports as its headline, and
69//! `the_tier_plateaus_at_its_budget_under_a_full_scan` is the test that asserts
70//! it rather than the sentence.
71
72use std::path::{Path, PathBuf};
73use std::sync::Mutex;
74use std::sync::atomic::{AtomicU64, Ordering};
75
76use anyhow::Result;
77
78use crate::exploded::ExplodedStats;
79use crate::exploded_arrow::{ExplodePolicy, ExplodedArchive};
80use crate::object::GitObjectKind;
81
82/// One lazy segment is closed and a new one started once it passes this many
83/// bytes. Eviction granularity is one segment, so this is the quantum a budget
84/// is spent and reclaimed in: 1/16 of the budget, floored so a small budget
85/// still gets whole batches and ceilinged so a huge one does not accumulate
86/// thousands of open indexes.
87const MIN_SEGMENT: u64 = 16 << 20;
88const MAX_SEGMENT: u64 = 1 << 30;
89const SEGMENTS_PER_BUDGET: u64 = 16;
90
91/// The lazy pool buffers this much before it writes. Far under
92/// [`crate::exploded_arrow::FLUSH_BYTES`] on purpose: that 64 MiB buffer is
93/// sized for an indexer draining a pack, and a **reader's** write-back holding
94/// 64 MiB per store is memory the box did not agree to spend.
95pub const CACHE_FLUSH_BYTES: u64 = 4 << 20;
96
97// ════════════════════════════════════════════════════════════════════════════
98// A VALIDATION READ IS NOT A READER
99// ════════════════════════════════════════════════════════════════════════════
100
101thread_local! {
102    /// Depth, not a flag: guards nest, and the innermost one ending must not
103    /// re-open the cache for an outer scope that is still validating.
104    static SUPPRESS_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
105}
106
107/// **Declare that reads on this thread are validation, not a reader's.**
108///
109/// `receive-pack` reads every commit and tree it was sent — once — to prove the
110/// push closes, and then never reads them again. Those reads look identical to
111/// a client's from inside the store, so the store cannot tell them apart and
112/// the caller must say.
113///
114/// MEASURED on `torvalds/linux`, oden 2026-09-03: the lazy pool is **empty
115/// until the instant indexing completes** and then grows to
116/// **6 778 047 872 B** (13 segments) before the push returns — 31 % of the
117/// whole store and the largest non-pack thing in it, written for a reader that
118/// does not exist. Full table in
119/// `gunnar/.nornir/storage-levers-2026-09-03.md`.
120///
121/// # Scope
122///
123/// **Thread-local, and that is the point.** A `receive-pack` connection does
124/// its ingest, its index and its connectivity walk on one thread and serves
125/// nobody; the background absorber runs elsewhere and is untouched, and so is
126/// every serving connection. Nothing global is switched off, so a store that is
127/// simultaneously answering a clone still warms its cache from that clone.
128///
129/// The eager pool is not affected either: the indexer writes that one directly,
130/// not through [`ExplodedCache::write_back`].
131///
132/// # It is a guard, not a pair of calls
133///
134/// Restored on drop, including on unwind — a panic inside a connectivity walk
135/// must not leave a thread that never caches again.
136#[must_use = "the suppression lasts only as long as the guard is held"]
137pub fn suppress_write_back() -> SuppressWriteBack {
138    SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
139    SuppressWriteBack { _private: () }
140}
141
142/// Whether [`suppress_write_back`] is in force on this thread.
143pub fn write_back_suppressed() -> bool {
144    SUPPRESS_DEPTH.with(|d| d.get()) > 0
145}
146
147/// The guard [`suppress_write_back`] returns. See it for what this is for.
148pub struct SuppressWriteBack {
149    _private: (),
150}
151
152impl Drop for SuppressWriteBack {
153    fn drop(&mut self) {
154        SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
155    }
156}
157
158// ════════════════════════════════════════════════════════════════════════════
159// The admission policy: DO NOT EXPLODE UNIFORMLY
160// ════════════════════════════════════════════════════════════════════════════
161
162/// **The cheapest thing a cache can do is not cache.**
163///
164/// Rickard, 2026-08-30: *"can the index point at the original git object and
165/// decompress it directly instead of materialising everything?"* — and it can,
166/// because a packfile is not one stream. Every object is its own zlib stream at
167/// its own offset, so `oid → (offset, type, base, inflated size)` is enough to
168/// inflate exactly one object ([`crate::resolve::materialize_at`]). A **whole**
169/// object is then one `inflate` of a mapped byte range: microseconds, no chain,
170/// no delta application. Writing that into a cache costs a copy, a flush, an
171/// fsync and a slice of somebody's quota — to save microseconds it did not
172/// need saving.
173///
174/// So the cache admits on **rebuild cost**, not on being read:
175///
176/// # It admits on cost DENSITY, not on cost
177///
178/// The first cut admitted anything over 16 KiB "because the inflate itself is
179/// the cost". That is true and it is the wrong rule, because the budget is
180/// **shared**: what a cache should hold is what saves the most time *per byte
181/// held*. A 3 MiB whole blob costs one linear inflate — bandwidth-bound,
182/// microseconds a kilobyte — and would occupy the same 3 MiB as ~250 deep tree
183/// chains that each cost a chain walk. Admitting it evicts all of them.
184///
185/// | what the read paid | admitted? |
186/// |---|---|
187/// | over [`MAX_CACHED_BYTES`] inflated | **no** — a bad tenant of a shared budget whatever its depth; its direct read is bandwidth-bound anyway |
188/// | 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 |
189/// | deeper than [`CHEAP_DEPTH`] | yes — every extra link is another inflate plus a delta application, and git allows chains to 50 |
190/// | a base outside this archive | yes — it went through a [`BaseSource`](crate::resolve::BaseSource), which is never the cheap path |
191///
192/// The thresholds are named, not tuned: `CHEAP_DEPTH` is 2 because git's own
193/// default `pack.depth` is 50 and a 0–2 chain is at most two delta applications
194/// over a mapped buffer; `MAX_CACHED_BYTES` is 1 MiB because that is where one
195/// object starts costing what hundreds of chain-heavy ones do.
196///
197/// **This is also the zero-copy rule.** A declined object is declined *before*
198/// anything is copied, so the commonest reads — and every large one — cost the
199/// cache exactly zero bytes. MEASURED by lane AA 2026-08-30: one 3 145 728 B
200/// object read through `gunnar-cli` allocated 6 340 693 B, exactly two copies,
201/// and the second was the cache taking a blob it should never have wanted.
202pub const CHEAP_DEPTH: u32 = 2;
203pub const MAX_CACHED_BYTES: u64 = 1 << 20;
204
205/// What it cost to produce an object, as the cache scores it.
206///
207/// `depth + 1` inflates and `depth` delta applications over `inflated` bytes.
208/// Deliberately a coarse integer and not a timing: a duration measured on the
209/// reader's thread carries that thread's scheduling, and two identical objects
210/// would then score differently depending on how busy the box was.
211pub fn rebuild_cost(depth: u32, inflated: u64, external_base: bool) -> u64 {
212    // Bytes inflated across the whole chain, roughly: every link inflates
213    // something of about the object's order, plus one delta application each.
214    let links = depth as u64 + 1;
215    let base = links * inflated.max(1);
216    if external_base { base.saturating_mul(4) } else { base }
217}
218
219/// Is this object worth a slice of the quota?
220pub fn worth_caching(depth: u32, inflated: u64, external_base: bool) -> bool {
221    inflated <= MAX_CACHED_BYTES && (external_base || depth > CHEAP_DEPTH)
222}
223
224/// Environment door for the budget. Named in [`crate::arms::ENV_EXPLODE_BUDGET`]
225/// and read through that crate's one counted door, so
226/// `every_environment_read_in_this_crate_is_named_in_all_env` can see it —
227/// a key spelled locally is a key that roster cannot check.
228pub use crate::arms::ENV_EXPLODE_BUDGET as ENV_BUDGET;
229
230/// **The default budget, and the arithmetic that makes it safe.**
231///
232/// `min(10 % of the filesystem holding the store, 8 GiB)` per repository.
233///
234/// On the shipped appliance — 300 GB `/data`, the shape Rickard named — 10 %
235/// is 30 GB, so the 8 GiB ceiling binds and every repository is capped at
236/// 8 GiB. **Twenty** repositories of any size therefore cost at most 160 GB of
237/// exploded cache against 300 GB, leaving the packs themselves (the truth,
238/// which is never evictable) room; and a repository only reaches its cap if
239/// somebody actually reads all of it. The same appliance with the eager tier
240/// took 141.7 GB for **one** repository.
241///
242/// On a small box the percentage binds instead: a 64 GiB `/data` gives 6.4 GiB,
243/// a 20 GiB one gives 2 GiB. The floor is [`MIN_BUDGET`] so a tiny volume still
244/// gets a cache rather than thrashing at zero.
245pub const BUDGET_FRACTION_PERCENT: u64 = 10;
246pub const BUDGET_CEILING: u64 = 8 << 30;
247pub const MIN_BUDGET: u64 = 64 << 20;
248
249/// The default budget for a store rooted at `dir`, from that filesystem's size.
250pub fn default_budget(dir: &Path) -> u64 {
251    let total = filesystem_bytes(dir).unwrap_or(0);
252    if total == 0 {
253        return BUDGET_CEILING;
254    }
255    (total / 100 * BUDGET_FRACTION_PERCENT)
256        .min(BUDGET_CEILING)
257        .max(MIN_BUDGET)
258}
259
260/// The process-wide line for a store rooted at `dir`: the same tenth of the
261/// filesystem as [`default_budget`], **uncapped** — the cap is per store, the
262/// tenth is per box. Every cache on one filesystem computes the same figure;
263/// caches on different filesystems each hold the sum to their own, which
264/// over-counts across mounts and is the conservative direction. `0` (no
265/// filesystem figure) turns the global check off rather than inventing one.
266pub fn default_global_line(dir: &Path) -> u64 {
267    filesystem_bytes(dir).unwrap_or(0) / 100 * BUDGET_FRACTION_PERCENT
268}
269
270/// Total bytes of the filesystem holding `dir`, via `statvfs`.
271fn filesystem_bytes(dir: &Path) -> Option<u64> {
272    use std::os::unix::ffi::OsStrExt;
273    let mut c = Vec::with_capacity(dir.as_os_str().len() + 1);
274    c.extend_from_slice(dir.as_os_str().as_bytes());
275    c.push(0);
276    let mut s: libc::statvfs = unsafe { std::mem::zeroed() };
277    let rc = unsafe { libc::statvfs(c.as_ptr().cast(), &mut s) };
278    (rc == 0).then(|| s.f_blocks as u64 * s.f_frsize as u64)
279}
280
281/// The budget a store opens with: the environment door if it names one, the
282/// filesystem-derived default otherwise.
283pub fn budget_from_env(dir: &Path) -> Result<u64> {
284    match crate::arms::read_env(crate::arms::ENV_EXPLODE_BUDGET) {
285        Some(v) => parse_budget(&v),
286        None => Ok(budget_default().unwrap_or_else(|| default_budget(dir))),
287    }
288}
289
290/// `u64::MAX` is "no embedder default" and not a budget: `0` already MEANS
291/// unlimited ([`parse_budget`]), so the sentinel cannot be zero without making
292/// "unbounded" unsayable from code.
293const NO_BUDGET_DEFAULT: u64 = u64::MAX;
294
295/// What [`budget_from_env`] answers when [`crate::arms::ENV_EXPLODE_BUDGET`] is unset:
296/// the embedder's figure, or the filesystem-derived [`default_budget`] when it set none.
297static BUDGET_DEFAULT: AtomicU64 = AtomicU64::new(NO_BUDGET_DEFAULT);
298
299/// The embedder's process-wide budget, if one was set.
300pub fn budget_default() -> Option<u64> {
301    match BUDGET_DEFAULT.load(Ordering::Relaxed) {
302        NO_BUDGET_DEFAULT => None,
303        n => Some(n),
304    }
305}
306
307/// **Set the process-wide default explode budget** for every store opened AFTER this call
308/// (an explicit [`crate::arms::ENV_EXPLODE_BUDGET`] still wins). `None` restores the
309/// filesystem-derived [`default_budget`]. The twin of
310/// [`crate::arms::set_graph_cache_default`], for an embedder that learns its role at runtime
311/// and cannot set the environment — edition 2024 makes `set_var` unsafe, and gunnar forbids
312/// `unsafe` — such as gunnar's twin, which caps the pool at 16 MiB so a later policy change
313/// cannot walk back over its measured memory bound.
314pub fn set_explode_budget_default(bytes: Option<u64>) {
315    BUDGET_DEFAULT.store(bytes.unwrap_or(NO_BUDGET_DEFAULT), Ordering::Relaxed);
316}
317
318/// `12G`, `512M`, `1048576`, `0`, `unlimited`. An unparseable value is an
319/// ERROR and never a silent default — a typo in a quota is a bill that is
320/// wrong, and nobody notices until it is.
321pub fn parse_budget(s: &str) -> Result<u64> {
322    let t = s.trim().to_ascii_lowercase();
323    if t == "unlimited" || t == "none" || t == "off" {
324        return Ok(0);
325    }
326    let (digits, mult) = match t.as_bytes().last() {
327        Some(b'k') => (&t[..t.len() - 1], 1u64 << 10),
328        Some(b'm') => (&t[..t.len() - 1], 1u64 << 20),
329        Some(b'g') => (&t[..t.len() - 1], 1u64 << 30),
330        Some(b't') => (&t[..t.len() - 1], 1u64 << 40),
331        _ => (t.as_str(), 1),
332    };
333    let n: u64 = digits
334        .trim()
335        .parse()
336        .map_err(|_| anyhow::anyhow!("unreadable explode budget {s:?}; expected 0, 512M, 8G, …"))?;
337    n.checked_mul(mult)
338        .ok_or_else(|| anyhow::anyhow!("explode budget {s:?} overflows"))
339}
340
341/// What the tier is holding and what it is allowed to hold — the row a
342/// dashboard prints and `Admin.StorageInfo` carries.
343#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
344pub struct CacheUse {
345    /// Bytes in the eager pool (commits and tags under the shipped policy).
346    pub eager_bytes: u64,
347    /// Bytes in the lazy pool, across every live segment.
348    pub lazy_bytes: u64,
349    /// The cap. `0` is unlimited.
350    pub budget: u64,
351    /// Live lazy segments.
352    pub segments: u64,
353    /// Segments unlinked since this store was opened.
354    pub evicted_segments: u64,
355    /// Bytes those segments held. **Applied output**: what actually left the
356    /// filesystem, measured before the unlink, not a projection.
357    pub evicted_bytes: u64,
358    /// Rows those segments held.
359    pub evicted_rows: u64,
360    /// Objects written back by a first read.
361    pub written_back: u64,
362    /// **Reads served by inflating the object directly out of the pack and
363    /// caching NOTHING** — the cheap ones the admission policy declined.
364    /// Applied output: the counter says the policy ran, which a size alone
365    /// could not.
366    pub direct_reads: u64,
367    /// **Write-backs declined because the caller declared the read a
368    /// validation read** ([`suppress_write_back`]) — a `receive-pack`'s
369    /// connectivity walk, not a reader.
370    ///
371    /// Applied output, and the only way to tell the suppression ran from a
372    /// store that simply had nothing to cache: a pool that stays empty proves
373    /// nothing on its own, and this counter moving proves the gate fired.
374    pub suppressed_write_backs: u64,
375}
376
377impl CacheUse {
378    pub fn used(&self) -> u64 {
379        self.eager_bytes + self.lazy_bytes
380    }
381    /// `3.1G/8G` — what the dashboard cell prints.
382    pub fn band(&self) -> String {
383        if self.budget == 0 {
384            format!("{}/unlimited", human(self.used()))
385        } else {
386            format!("{}/{}", human(self.used()), human(self.budget))
387        }
388    }
389    /// Over budget right now — true only while the eager pool alone exceeds it,
390    /// because the lazy pool is evicted back under after every rotation.
391    pub fn over(&self) -> bool {
392        self.budget != 0 && self.used() > self.budget
393    }
394}
395
396/// `3.1G`, `812M`, `40K` — one significant fraction, the same shape the
397/// dashboard's other byte cells use.
398pub fn human(bytes: u64) -> String {
399    const UNITS: [(u64, &str); 4] = [(1 << 40, "T"), (1 << 30, "G"), (1 << 20, "M"), (1 << 10, "K")];
400    for (scale, suffix) in UNITS {
401        if bytes >= scale {
402            let v = bytes as f64 / scale as f64;
403            return if v < 10.0 {
404                format!("{v:.1}{suffix}")
405            } else {
406                format!("{v:.0}{suffix}")
407            };
408        }
409    }
410    format!("{bytes}B")
411}
412
413/// What one live lazy segment is holding, for cost-weighted eviction.
414#[derive(Debug, Clone, Copy, Default)]
415struct SegStat {
416    rows: u64,
417    /// Summed [`rebuild_cost`] of the rows written into it.
418    cost: u64,
419    /// Monotonic tick of the last write into it — the recency half of the rank.
420    touched: u64,
421}
422
423/// **The two pools, one door.** Everything that used to call an
424/// [`ExplodedArchive`] calls this instead; the eager pool *is* an
425/// `ExplodedArchive` and behaves exactly as it did.
426pub struct ExplodedCache {
427    eager: ExplodedArchive,
428    /// Oldest first. `Mutex` and not `RwLock`: it is touched once per rotation
429    /// and once per read miss, never per object.
430    lazy: Mutex<Vec<std::sync::Arc<ExplodedArchive>>>,
431    base: PathBuf,
432    policy: ExplodePolicy,
433    budget: AtomicU64,
434    /// **The process-wide line**, in bytes, that the SUM of every registered
435    /// cache's disk use is held under; `0` = no global line. `budget` above is
436    /// per store — 10 % of the filesystem, capped at 8 GiB — and MEASURED on
437    /// 2026-09-05 that meant N repositories read hard could hold N × 8 GiB:
438    /// ten busy tenants on a 1 TB appliance, 80 GB of cache. This is the same
439    /// 10 % of the filesystem, uncapped, applied once to the box: a cache whose
440    /// own use is above its fair share of it (`line / live caches`) evicts its
441    /// cheapest segment on its tick even while under its own budget, so the
442    /// biggest pools shrink first and the sum converges under the line.
443    global_line: AtomicU64,
444    segment_bytes: AtomicU64,
445    next_seg: AtomicU64,
446    written_back: AtomicU64,
447    direct_reads: AtomicU64,
448    /// See [`CacheUse::suppressed_write_backs`].
449    suppressed_write_backs: AtomicU64,
450    /// Per live segment, parallel to `lazy`: rows, and the summed rebuild cost
451    /// of what is in it. What cost-weighted eviction ranks on.
452    seg_cost: Mutex<Vec<SegStat>>,
453    evicted_segments: AtomicU64,
454    evicted_bytes: AtomicU64,
455    evicted_rows: AtomicU64,
456    /// Bytes buffered in the newest lazy segment since its last flush. The
457    /// flusher reads it; a reader adds to it and never blocks on it.
458    dirty: AtomicU64,
459    /// Where this cache's store lives, for the registry's rows.
460    root: String,
461    /// A maintenance thread is running. **The whole of "nobody waits"**: the
462    /// reader that fills the buffer starts one and walks away, and a second
463    /// reader that arrives while it runs does nothing at all.
464    maintaining: std::sync::atomic::AtomicBool,
465}
466
467impl ExplodedCache {
468    /// Open both pools at `base` (`…/objects.exploded`), adopting whatever
469    /// lazy segments a previous process left beside it.
470    pub fn open(base: &Path, policy: ExplodePolicy, budget: u64) -> Result<Self> {
471        let eager = ExplodedArchive::open_with_policy(base, policy)?;
472        let mut found: Vec<(u64, PathBuf)> = Vec::new();
473        if let Some(dir) = base.parent()
474            && let Ok(rd) = std::fs::read_dir(dir)
475        {
476            let stem = base
477                .file_name()
478                .map(|s| s.to_string_lossy().into_owned())
479                .unwrap_or_default();
480            let prefix = format!("{stem}.c");
481            for e in rd.flatten() {
482                let name = e.file_name().to_string_lossy().into_owned();
483                if let Some(n) = name.strip_prefix(&prefix)
484                    && let Ok(n) = n.parse::<u64>()
485                {
486                    found.push((n, e.path()));
487                }
488            }
489        }
490        found.sort_unstable_by_key(|(n, _)| *n);
491        let next = found.last().map(|(n, _)| n + 1).unwrap_or(1);
492        // A segment adopted from a previous process has no cost history — the
493        // scores are per-process and are not written down. It is scored by its
494        // position instead, oldest first, which is what plain FIFO would do
495        // and is the honest reading of "we do not know".
496        let seg_stats: Vec<SegStat> = (0..found.len())
497            .map(|i| SegStat { rows: 0, cost: 0, touched: i as u64 })
498            .collect();
499        let mut lazy = Vec::with_capacity(found.len());
500        for (_, p) in found {
501            // The lazy pool takes EVERY kind: it is where an object the eager
502            // policy declined comes to rest, so its own policy is `Full`.
503            lazy.push(std::sync::Arc::new(ExplodedArchive::open_with_policy(
504                &p,
505                ExplodePolicy::Full,
506            )?));
507        }
508        Ok(Self {
509            eager,
510            lazy: Mutex::new(lazy),
511            base: base.to_path_buf(),
512            policy,
513            budget: AtomicU64::new(budget),
514            global_line: AtomicU64::new(default_global_line(base)),
515            segment_bytes: AtomicU64::new(segment_for(budget)),
516            next_seg: AtomicU64::new(next),
517            written_back: AtomicU64::new(0),
518            direct_reads: AtomicU64::new(0),
519            suppressed_write_backs: AtomicU64::new(0),
520            seg_cost: Mutex::new(seg_stats),
521            evicted_segments: AtomicU64::new(0),
522            evicted_bytes: AtomicU64::new(0),
523            evicted_rows: AtomicU64::new(0),
524            dirty: AtomicU64::new(0),
525            root: base.parent().unwrap_or(base).display().to_string(),
526            maintaining: std::sync::atomic::AtomicBool::new(false),
527        })
528    }
529
530    /// Register this cache so a dashboard tick can read every open store's
531    /// band without holding a store lock — the same `Weak` registry
532    /// [`crate::progress`] keeps, for the same reason and with the same sweep.
533    pub fn registered(self: std::sync::Arc<Self>) -> std::sync::Arc<Self> {
534        registry()
535            .lock()
536            .unwrap_or_else(|e| e.into_inner())
537            .push(std::sync::Arc::downgrade(&self));
538        self
539    }
540
541    pub fn root(&self) -> &str {
542        &self.root
543    }
544
545    /// The quota, changed live — what the control plane sets when an account's
546    /// plan changes. Takes effect at the next rotation; it never refuses a
547    /// write and never deletes on the caller's thread.
548    pub fn set_budget(&self, bytes: u64) {
549        self.budget.store(bytes, Ordering::Release);
550        self.segment_bytes.store(segment_for(bytes), Ordering::Release);
551    }
552
553    /// The process-wide line this cache holds itself to its fair share of.
554    /// `0` turns the global check off. Tests set it small; production takes
555    /// [`default_global_line`] at open.
556    pub fn set_global_line(&self, bytes: u64) {
557        self.global_line.store(bytes, Ordering::Release);
558    }
559
560    /// True when the registered caches together exceed the global line AND this
561    /// cache holds more than its fair share of it — the condition under which
562    /// this cache evicts for the box rather than for itself. Computed without
563    /// holding this cache's own lock, so the registry sweep can lock the others.
564    fn over_global_share(&self) -> bool {
565        let line = self.global_line.load(Ordering::Acquire);
566        if line == 0 {
567            return false;
568        }
569        let all = use_all();
570        // The share is among caches that HOLD something: an idle repository's
571        // empty pool must not dilute the line for the ones actually using it.
572        let live = all
573            .iter()
574            .filter(|(_, u)| u.eager_bytes + u.lazy_bytes > 0)
575            .count()
576            .max(1) as u64;
577        let sum: u64 = all.iter().map(|(_, u)| u.eager_bytes + u.lazy_bytes).sum();
578        if sum <= line {
579            return false;
580        }
581        let mine = self.use_now();
582        mine.eager_bytes + mine.lazy_bytes > line / live
583    }
584
585    pub fn budget(&self) -> u64 {
586        self.budget.load(Ordering::Acquire)
587    }
588
589    pub fn policy(&self) -> ExplodePolicy {
590        self.policy
591    }
592
593    pub fn path(&self) -> &Path {
594        &self.base
595    }
596
597    /// **The eager write** — the indexer's sink, at push time. Unchanged
598    /// behaviour: the policy decides, and what it declines is counted as
599    /// skipped rather than silently dropped.
600    pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
601        self.eager.explode(oid, kind, payload)
602    }
603
604    pub fn explode_at(
605        &self,
606        oid: &[u8],
607        kind: GitObjectKind,
608        mode: Option<u32>,
609        path: Option<&str>,
610        payload: &[u8],
611    ) -> Result<()> {
612        self.eager.explode_at(oid, kind, mode, path, payload)
613    }
614
615    /// **The lazy write-back** — one object a reader has just inflated from the
616    /// pack, kept for the next reader.
617    ///
618    /// Buffers only: it appends to the newest segment's in-memory rows and
619    /// returns. Nothing is written, nothing is fsynced, no eviction runs. The
620    /// reader that called it has its bytes already and is not made to wait for
621    /// anybody — [`Self::maintain`] does the I/O on the store's own thread.
622    /// Two readers racing on one oid is benign: the same bytes are buffered
623    /// twice and the newest row wins, which is what
624    /// [`ExplodedArchive::explode_at`] already promises.
625    pub fn write_back(
626        &self,
627        oid: &[u8],
628        kind: GitObjectKind,
629        payload: &[u8],
630        cost: crate::resolve::ReadCost,
631    ) -> Result<bool> {
632        // **A validation read is not a reader.** A `receive-pack` reads every
633        // commit and tree it was sent, once, to prove the push closes — and
634        // then never reads them again. Caching those is writing a read-cache
635        // for a reader that does not exist. MEASURED on torvalds/linux
636        // (2026-09-03, oden): the lazy pool is EMPTY until the instant indexing
637        // completes and then grows to **6 778 047 872 B** before the push
638        // returns, which is 31 % of the whole store and the largest non-pack
639        // thing in it. See `gunnar/.nornir/storage-levers-2026-09-03.md`.
640        //
641        // The caller declares the window ([`suppress_write_back`]) because the
642        // caller is the only thing that knows what kind of read this is: the
643        // store sees the same `read` either way.
644        if write_back_suppressed() {
645            self.suppressed_write_backs.fetch_add(1, Ordering::Relaxed);
646            return Ok(false);
647        }
648        // **The admission gate.** A shallow, small object is cheaper to inflate
649        // again than to keep: caching it would spend somebody's quota to save
650        // microseconds, and evict something that costs milliseconds.
651        if !worth_caching(cost.depth, cost.inflated, cost.external_base) {
652            self.direct_reads.fetch_add(1, Ordering::Relaxed);
653            return Ok(false);
654        }
655        let seg = {
656            let mut l = self
657                .lazy
658                .lock()
659                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
660            match l.last() {
661                Some(s) => s.clone(),
662                None => {
663                    let s = self.new_segment()?;
664                    l.push(s.clone());
665                    s
666                }
667            }
668        };
669        seg.explode(oid, kind, payload)?;
670        {
671            let tick = self.written_back.load(Ordering::Relaxed);
672            let mut st = self
673                .seg_cost
674                .lock()
675                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
676            if st.is_empty() {
677                st.push(SegStat::default());
678            }
679            if let Some(last) = st.last_mut() {
680                last.rows += 1;
681                last.cost += rebuild_cost(cost.depth, cost.inflated, cost.external_base);
682                last.touched = tick;
683            }
684        }
685        self.written_back.fetch_add(1, Ordering::Relaxed);
686        let n = self
687            .dirty
688            .fetch_add(payload.len() as u64, Ordering::AcqRel)
689            + payload.len() as u64;
690        Ok(n >= CACHE_FLUSH_BYTES)
691    }
692
693    /// **Write down what is buffered, rotate if the newest segment is full, and
694    /// evict until the pools fit the budget.** The whole of the tier's I/O, on
695    /// one thread that is nobody's reader.
696    ///
697    /// Returns the bytes reclaimed, so a caller can log applied output rather
698    /// than an intention.
699    pub fn maintain(&self) -> Result<u64> {
700        let newest = {
701            let l = self
702                .lazy
703                .lock()
704                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
705            l.last().cloned()
706        };
707        if let Some(seg) = newest {
708            seg.flush()?;
709            self.dirty.store(0, Ordering::Release);
710            if seg.disk_bytes() >= self.segment_bytes.load(Ordering::Acquire) {
711                let fresh = self.new_segment()?;
712                let mut l = self
713                    .lazy
714                    .lock()
715                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
716                l.push(fresh);
717                if let Ok(mut st) = self.seg_cost.lock() {
718                    st.push(SegStat {
719                        rows: 0,
720                        cost: 0,
721                        touched: self.written_back.load(Ordering::Relaxed),
722                    });
723                }
724            }
725        }
726        self.evict_to_budget()
727    }
728
729    /// Drop segments until the two pools fit, **cheapest-to-rebuild first**.
730    ///
731    /// Not plain FIFO. A segment is ranked by
732    /// `mean rebuild cost × recency`, and the lowest score goes: a segment of
733    /// shallow 20 KiB blobs is worth less than one of depth-40 chains even if
734    /// it was written more recently, because re-reading the first costs
735    /// microseconds an object and the second costs milliseconds. Recency is
736    /// still half the rank, so a cold segment of expensive objects does
737    /// eventually go rather than pinning the pool.
738    ///
739    /// **The eager pool is never touched**: commits are what `graph`, `reach`
740    /// and a tag peel read, and a store that answers those slowly is worse than
741    /// one that reads a blob slowly. If the eager pool alone is over budget the
742    /// lazy pool goes to nothing and the store is reported `over`.
743    fn evict_to_budget(&self) -> Result<u64> {
744        let budget = self.budget.load(Ordering::Acquire);
745        if budget == 0 && self.global_line.load(Ordering::Acquire) == 0 {
746            return Ok(0);
747        }
748        let eager = self.eager.disk_bytes();
749        let mut freed = 0u64;
750        loop {
751            // Re-read per iteration: each eviction moves this cache's share.
752            let over_share = self.over_global_share();
753            let (used, victim, at) = {
754                let l = self
755                    .lazy
756                    .lock()
757                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
758                let lazy: u64 = l.iter().map(|s| s.disk_bytes()).sum();
759                // The newest segment is the one being written into; evicting it
760                // would drop rows a reader just paid for and start the same
761                // segment again. There is always at least one live segment.
762                // Within its own budget (`0` = unlimited) AND not over its
763                // share of the box: nothing to do. Otherwise the cheapest
764                // segment goes, for whichever reason applies.
765                let within_own = budget == 0 || eager + lazy <= budget;
766                if (within_own && !over_share) || l.len() < 2 {
767                    (eager + lazy, None, 0)
768                } else {
769                    let at = self.cheapest_segment(l.len() - 1);
770                    (eager + lazy, Some(l[at].clone()), at)
771                }
772            };
773            let Some(victim) = victim else {
774                let _ = used;
775                break;
776            };
777            let bytes = victim.disk_bytes();
778            let rows = victim.rows().unwrap_or(0);
779            let path = victim.path().to_path_buf();
780            {
781                let mut l = self
782                    .lazy
783                    .lock()
784                    .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
785                if at < l.len() {
786                    l.remove(at);
787                }
788                if let Ok(mut st) = self.seg_cost.lock()
789                    && at < st.len()
790                {
791                    st.remove(at);
792                }
793            }
794            drop(victim);
795            // The file goes even if some reader still holds an `Arc` on the
796            // segment: an unlinked file keeps answering the handles that are
797            // open, and the blocks come back when the last one closes. So the
798            // budget is honoured immediately and no read in flight fails.
799            std::fs::remove_file(&path).ok();
800            self.evicted_segments.fetch_add(1, Ordering::Relaxed);
801            self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
802            self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
803            freed += bytes;
804        }
805        Ok(freed)
806    }
807
808    /// The index of the segment with the lowest `mean rebuild cost × recency`,
809    /// searched over `[0, exclusive)` so the segment being written into is
810    /// never a candidate. Ties and unknown costs (a segment adopted from a
811    /// previous process, which carries no history) fall back to the oldest,
812    /// which is what plain FIFO would have chosen.
813    fn cheapest_segment(&self, exclusive: usize) -> usize {
814        let Ok(st) = self.seg_cost.lock() else {
815            return 0;
816        };
817        let newest = st.last().map(|s| s.touched).unwrap_or(0).max(1);
818        let mut best = 0usize;
819        let mut best_score = u128::MAX;
820        for i in 0..exclusive.min(st.len()) {
821            let s = st[i];
822            let mean = if s.rows == 0 { 0 } else { s.cost / s.rows };
823            // Recency in [1, 1024]: a segment written at the very start scores
824            // 1, the newest scores 1024. Multiplying rather than adding keeps
825            // the two halves on one scale without a weight nobody can justify.
826            let recency = 1 + (s.touched as u128 * 1023) / newest as u128;
827            let score = (mean as u128 + 1) * recency;
828            if score < best_score {
829                best_score = score;
830                best = i;
831            }
832        }
833        best
834    }
835
836    fn new_segment(&self) -> Result<std::sync::Arc<ExplodedArchive>> {
837        let n = self.next_seg.fetch_add(1, Ordering::AcqRel);
838        let name = format!(
839            "{}.c{n}",
840            self.base.file_name().unwrap_or_default().to_string_lossy()
841        );
842        let path = self.base.with_file_name(name);
843        Ok(std::sync::Arc::new(ExplodedArchive::open_with_policy(
844            &path,
845            ExplodePolicy::Full,
846        )?))
847    }
848
849    /// The object's kind and bytes from either pool, or `None` — *fall back and
850    /// re-derive*, never *wrong*.
851    ///
852    /// Eager first: it is the smaller table, its index is the one already warm,
853    /// and under the shipped policy it holds exactly the objects a graph walk
854    /// asks for. Then the lazy segments **newest first**, so a re-entered
855    /// object is found before the stale copy an older segment may still hold.
856    pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
857        if let Some(hit) = self.eager.content(oid)? {
858            return Ok(Some(hit));
859        }
860        let segs = self.segments()?;
861        for s in segs.iter().rev() {
862            if let Some(hit) = s.content(oid)? {
863                return Ok(Some(hit));
864            }
865        }
866        Ok(None)
867    }
868
869    pub fn has(&self, oid: &[u8]) -> Result<bool> {
870        if self.eager.has(oid)? {
871            return Ok(true);
872        }
873        for s in self.segments()?.iter().rev() {
874            if s.has(oid)? {
875                return Ok(true);
876            }
877        }
878        Ok(false)
879    }
880
881    pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
882        if let Some(n) = self.eager.name(oid)? {
883            return Ok(Some(n));
884        }
885        for s in self.segments()?.iter().rev() {
886            if let Some(n) = s.name(oid)? {
887                return Ok(Some(n));
888            }
889        }
890        Ok(None)
891    }
892
893    /// Every oid of one kind, across both pools, newest spelling winning.
894    ///
895    /// The graph fold's input. Under the shipped policy every commit is in the
896    /// eager pool, so this is one table's scan and the lazy segments add
897    /// nothing — but it unions rather than assuming that, because a store
898    /// opened under a different policy must still fold a whole graph.
899    pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
900        let mut out = self.eager.of_kind(kind)?;
901        let segs = self.segments()?;
902        if segs.is_empty() {
903            return Ok(out);
904        }
905        let mut seen: std::collections::HashSet<Vec<u8>> =
906            out.iter().map(|(o, _)| o.clone()).collect();
907        for s in segs.iter().rev() {
908            for (oid, payload) in s.of_kind(kind)? {
909                if seen.insert(oid.clone()) {
910                    out.push((oid, payload));
911                }
912            }
913        }
914        Ok(out)
915    }
916
917    /// Retire every oid `live` rejects — in **both** pools.
918    ///
919    /// Both, and that is not defensive: a tombstone written only into the eager
920    /// pool would leave the lazy pool's row live, `content` would fall through
921    /// to it, and a `gc`'d object would come back by the second door. That is
922    /// the same defect
923    /// `a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened` found
924    /// the first time, one pool along.
925    pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
926        let mut n = self.eager.retain(live)?;
927        for s in self.segments()? {
928            n += s.retain(live)?;
929        }
930        Ok(n)
931    }
932
933    /// **`repo gc` reclaims the cache.** Every lazy segment is unlinked; the
934    /// eager pool keeps its rows, because they are the graph. Returns the bytes
935    /// that left the filesystem.
936    pub fn drop_cache(&self) -> Result<u64> {
937        let victims = {
938            let mut l = self
939                .lazy
940                .lock()
941                .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
942            if let Ok(mut st) = self.seg_cost.lock() {
943                st.clear();
944            }
945            std::mem::take(&mut *l)
946        };
947        let mut freed = 0;
948        for s in victims {
949            let bytes = s.disk_bytes();
950            let rows = s.rows().unwrap_or(0);
951            let path = s.path().to_path_buf();
952            drop(s);
953            std::fs::remove_file(&path).ok();
954            self.evicted_segments.fetch_add(1, Ordering::Relaxed);
955            self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
956            self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
957            freed += bytes;
958        }
959        self.dirty.store(0, Ordering::Release);
960        Ok(freed)
961    }
962
963    pub fn flush(&self) -> Result<()> {
964        self.eager.flush()?;
965        for s in self.segments()? {
966            s.flush()?;
967        }
968        self.dirty.store(0, Ordering::Release);
969        Ok(())
970    }
971
972    /// What the memory watchdog reaches for: every derived, rebuildable
973    /// structure in every pool.
974    pub fn drop_caches(&self) {
975        self.eager.drop_caches();
976        if let Ok(l) = self.lazy.lock() {
977            for s in l.iter() {
978                s.drop_caches();
979            }
980        }
981    }
982
983    pub fn rows(&self) -> Result<u64> {
984        let mut n = self.eager.rows()?;
985        for s in self.segments()? {
986            n += s.rows()?;
987        }
988        Ok(n)
989    }
990
991    /// Rows in the EAGER pool only — what `adopt_journal` compares against the
992    /// index, because the lazy pool's size is a function of what has been read
993    /// and says nothing about whether a pack was absorbed.
994    pub fn eager_rows(&self) -> Result<u64> {
995        self.eager.rows()
996    }
997
998    pub fn disk_bytes(&self) -> u64 {
999        self.use_now().used()
1000    }
1001
1002    pub fn use_now(&self) -> CacheUse {
1003        let (lazy_bytes, segments) = match self.lazy.lock() {
1004            Ok(l) => (l.iter().map(|s| s.disk_bytes()).sum(), l.len() as u64),
1005            Err(_) => (0, 0),
1006        };
1007        CacheUse {
1008            eager_bytes: self.eager.disk_bytes(),
1009            lazy_bytes,
1010            budget: self.budget.load(Ordering::Acquire),
1011            segments,
1012            evicted_segments: self.evicted_segments.load(Ordering::Relaxed),
1013            evicted_bytes: self.evicted_bytes.load(Ordering::Relaxed),
1014            evicted_rows: self.evicted_rows.load(Ordering::Relaxed),
1015            written_back: self.written_back.load(Ordering::Relaxed),
1016            direct_reads: self.direct_reads.load(Ordering::Relaxed),
1017            suppressed_write_backs: self.suppressed_write_backs.load(Ordering::Relaxed),
1018        }
1019    }
1020
1021    pub fn note_rederived(&self) {
1022        self.eager.note_rederived();
1023    }
1024
1025    pub fn pread_served(&self) -> u64 {
1026        self.eager.pread_served()
1027            + self
1028                .segments()
1029                .map(|s| s.iter().map(|x| x.pread_served()).sum::<u64>())
1030                .unwrap_or(0)
1031    }
1032
1033    pub fn skipped(&self) -> u64 {
1034        self.eager.skipped()
1035    }
1036
1037    /// The engine-wide counters, summed over both pools, so no caller can tell
1038    /// the two apart by its instruments.
1039    pub fn engine_stats(&self) -> ExplodedStats {
1040        let mut s = self.eager.engine_stats();
1041        if let Ok(segs) = self.segments() {
1042            for x in segs {
1043                let e = x.engine_stats();
1044                s.rows += e.rows;
1045                s.written += e.written;
1046                s.served += e.served;
1047                s.absent += e.absent;
1048            }
1049        }
1050        s
1051    }
1052
1053    /// **Start the write-back thread if one is not already running — and
1054    /// return to the reader immediately either way.**
1055    ///
1056    /// Rickard's law, from the design note: *nobody waits*. The reader already
1057    /// has its bytes; the append, the fsync, the rotation and the eviction all
1058    /// happen here, on a thread that is nobody's request. A thread per 4 MiB of
1059    /// write-back rather than a parked thread per store, because a box holding
1060    /// 1 200 open repositories (lane V) cannot afford 1 200 idle threads and
1061    /// can easily afford one spawn per few thousand objects read.
1062    ///
1063    /// The flag is the whole of the mutual exclusion: a second reader finding
1064    /// it set does nothing, and the running thread loops until the buffer is
1065    /// empty before clearing it, so no work is dropped between the two.
1066    pub fn kick(self: &std::sync::Arc<Self>) {
1067        if self
1068            .maintaining
1069            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
1070            .is_err()
1071        {
1072            return;
1073        }
1074        let me = self.clone();
1075        let spawned = std::thread::Builder::new()
1076            .name("explode-writeback".into())
1077            .stack_size(256 << 10)
1078            .spawn(move || {
1079                loop {
1080                    if let Err(e) = me.maintain() {
1081                        tracing::warn!(target: "znippy.git", error = %e, "explode cache maintenance");
1082                        break;
1083                    }
1084                    if me.dirty.load(Ordering::Acquire) < CACHE_FLUSH_BYTES {
1085                        break;
1086                    }
1087                }
1088                me.maintaining.store(false, Ordering::Release);
1089            });
1090        if spawned.is_err() {
1091            // No thread to be had. Clearing the flag is what makes the next
1092            // reader try again rather than the cache silently stopping.
1093            self.maintaining.store(false, Ordering::Release);
1094        }
1095    }
1096
1097    fn segments(&self) -> Result<Vec<std::sync::Arc<ExplodedArchive>>> {
1098        Ok(self
1099            .lazy
1100            .lock()
1101            .map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?
1102            .clone())
1103    }
1104}
1105
1106impl crate::exploded::PayloadSink for ExplodedCache {
1107    fn wants(&self, kind: GitObjectKind) -> bool {
1108        crate::exploded::PayloadSink::wants(&self.eager, kind)
1109    }
1110    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
1111        ExplodedCache::explode(self, oid, kind, payload)
1112    }
1113    fn flush(&self) -> Result<()> {
1114        self.eager.flush().map(|_| ())
1115    }
1116}
1117
1118fn registry() -> &'static Mutex<Vec<std::sync::Weak<ExplodedCache>>> {
1119    static R: std::sync::OnceLock<Mutex<Vec<std::sync::Weak<ExplodedCache>>>> =
1120        std::sync::OnceLock::new();
1121    R.get_or_init(|| Mutex::new(Vec::new()))
1122}
1123
1124/// **Every open store's explode band**, as `(store root, use)`. What the
1125/// dashboard tick and `Admin.StorageInfo` read; a `metadata` call per pool and
1126/// no store lock. Dead entries are swept on the way past.
1127pub fn use_all() -> Vec<(String, CacheUse)> {
1128    let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
1129    r.retain(|w| w.strong_count() > 0);
1130    r.iter()
1131        .filter_map(|w| w.upgrade())
1132        .map(|c| (c.root.clone(), c.use_now()))
1133        .collect()
1134}
1135
1136/// Set the budget of every open store under `root` (a prefix match, so an
1137/// account's whole tree is one call). Returns how many were changed — applied
1138/// output, so a control plane that set nothing can say so.
1139pub fn set_budget_under(root: &str, bytes: u64) -> usize {
1140    let mut n = 0;
1141    let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
1142    r.retain(|w| w.strong_count() > 0);
1143    for c in r.iter().filter_map(|w| w.upgrade()) {
1144        if c.root.starts_with(root) {
1145            c.set_budget(bytes);
1146            n += 1;
1147        }
1148    }
1149    n
1150}
1151
1152/// One segment is 1/16 of the budget, clamped. An unlimited budget still
1153/// rotates, so a pool that is later given a budget can be evicted without a
1154/// rewrite.
1155fn segment_for(budget: u64) -> u64 {
1156    if budget == 0 {
1157        return MAX_SEGMENT.min(1 << 28);
1158    }
1159    // The floor yields to the budget rather than overriding it: a 16 MiB
1160    // minimum against a 1 MiB quota would mean the pool never rotates and
1161    // therefore never evicts — a cap that silently is not one, which is the
1162    // exact shape of bug LAW 2 is about. So the floor is `min(MIN_SEGMENT,
1163    // budget/4)`: a real budget gets 16 MiB segments, a tiny one gets tiny ones.
1164    let floor = MIN_SEGMENT.min(budget / 4).max(64 << 10);
1165    (budget / SEGMENTS_PER_BUDGET).clamp(floor, MAX_SEGMENT)
1166}
1167
1168#[cfg(test)]
1169mod tests {
1170    use super::*;
1171
1172    fn tmp(name: &str) -> PathBuf {
1173        let d = std::env::temp_dir().join(format!("explode-cache-{}-{name}", std::process::id()));
1174        let _ = std::fs::remove_dir_all(&d);
1175        std::fs::create_dir_all(&d).unwrap();
1176        d.join("objects.exploded")
1177    }
1178
1179    /// A cost the admission gate admits: deep enough and big enough that the
1180    /// cache is the cheaper place for it. Tests that are about eviction must
1181    /// not accidentally be about admission.
1182    /// An INCOMPRESSIBLE payload of `len` bytes (a tiny LCG seeded by `seed`).
1183    /// The eviction tests reason about bytes on disk against a budget, and since
1184    /// lever 3 (2026-09-04) a row is stored zlib-compressed when that is smaller
1185    /// — a `vec![3u8; 64 << 10]` lands as ~80 bytes and the pool never reaches
1186    /// the budget those tests were sized for ("nothing was evicted" ×3). Noise
1187    /// does not compress, so the budget arithmetic means what it did.
1188    fn noise(len: usize, seed: u8) -> Vec<u8> {
1189        let mut x: u32 = 0x9E37_79B9 ^ (seed as u32).wrapping_mul(0x0101_0101);
1190        (0..len)
1191            .map(|_| {
1192                x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
1193                (x >> 24) as u8
1194            })
1195            .collect()
1196    }
1197
1198    fn dear(payload: &[u8]) -> crate::resolve::ReadCost {
1199        crate::resolve::ReadCost {
1200            depth: CHEAP_DEPTH + 8,
1201            inflated: payload.len() as u64,
1202            external_base: false,
1203        }
1204    }
1205
1206    /// Assert at compile time that the eviction fixtures below stay under the
1207    /// size cap — a fixture that grew past it would be silently DECLINED and
1208    /// the test would then be asserting nothing.
1209    const _: () = assert!((64 << 10) < MAX_CACHED_BYTES as usize);
1210
1211    fn oid(n: u32) -> Vec<u8> {
1212        let mut o = vec![0u8; 20];
1213        o[..4].copy_from_slice(&n.to_be_bytes());
1214        o
1215    }
1216
1217    /// ★ **THE ANSWER TO "does it take the disk anyway?"**
1218    ///
1219    /// Every object of a repository is read once, each one written back — the
1220    /// full-history browse that would otherwise reproduce the eager tier. The
1221    /// tier must **plateau at its budget**, not climb.
1222    ///
1223    /// Seen RED by returning `Ok(0)` from `evict_to_budget`: the pool climbs to
1224    /// 9.4 MiB against a 2 MiB budget and the assertion names the number.
1225    #[test]
1226    fn the_tier_plateaus_at_its_budget_under_a_full_scan() {
1227        let base = tmp("plateau");
1228        let budget = 2 << 20;
1229        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, budget).unwrap();
1230        let payload = vec![7u8; 8 << 10];
1231        let mut peak = 0u64;
1232        for i in 0..2000u32 {
1233            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
1234            // The store's own thread, doing what it does between reads.
1235            c.maintain().unwrap();
1236            peak = peak.max(c.use_now().used());
1237        }
1238        let end = c.use_now();
1239        let written = 2000u64 * payload.len() as u64;
1240        assert!(
1241            peak <= budget as u64 + segment_for(budget) + (1 << 20),
1242            "the tier climbed to {peak} B against a {budget} B budget after {written} B of \
1243             write-backs — it is a tier, not a cache"
1244        );
1245        assert!(
1246            end.evicted_bytes > 0 && end.evicted_segments > 0,
1247            "nothing was evicted, so the plateau is an accident of size: {end:?}"
1248        );
1249    }
1250
1251    /// An account at its budget still serves **every** read correctly — only
1252    /// slower. Eviction is never an error a client sees.
1253    #[test]
1254    fn an_evicted_object_is_absent_and_never_wrong() {
1255        let base = tmp("absent");
1256        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
1257        let payload = noise(64 << 10, 3);
1258        for i in 0..200u32 {
1259            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
1260            c.maintain().unwrap();
1261        }
1262        // Whatever survived comes back byte-identical; whatever did not comes
1263        // back `None`, which is the caller's cue to re-inflate the chain.
1264        let mut hits = 0;
1265        for i in 0..200u32 {
1266            match c.content(&oid(i)).unwrap() {
1267                Some((k, b)) => {
1268                    assert_eq!(k, GitObjectKind::Blob);
1269                    assert_eq!(b, payload, "an evicting cache returned the wrong bytes");
1270                    hits += 1;
1271                }
1272                None => {}
1273            }
1274        }
1275        assert!(hits > 0, "the cache evicted everything, including the newest segment");
1276        assert!(hits < 200, "nothing was evicted, so this proves nothing");
1277    }
1278
1279    /// Commits are the last thing standing: the eager pool is never unlinked,
1280    /// whatever the lazy pool does.
1281    /// **The box has one line, and the biggest pool pays for it.** Two caches,
1282    /// each far under its own per-store budget, share a small global line: the
1283    /// one holding most of the box's cache evicts on its tick; the one under
1284    /// its fair share does not, however far over the line the box is.
1285    ///
1286    /// Robust to whatever else is registered in this process: the fairness
1287    /// claim is made with B holding ONE 8 KiB row (over its share only if more
1288    /// than ~128 caches held bytes at once), and the convergence claim is made
1289    /// on A's own bytes, not on a sum that other tests' pools can move.
1290    ///
1291    /// Seen RED by making `over_global_share` return `false` always: A evicted
1292    /// nothing ("the big pool must evict for the box") and stayed at 3 MiB
1293    /// over a 1 MiB line. Restored.
1294    #[test]
1295    fn the_global_line_is_paid_by_the_cache_over_its_share() {
1296        let base_a = tmp("global-a");
1297        let base_b = tmp("global-b");
1298        // 4 MiB per store → 1 MiB segments (the budget/4 floor), so 3 MiB rotates into four and
1299        // the per-store budget itself never evicts anything here.
1300        let a = std::sync::Arc::new(ExplodedCache::open(&base_a, ExplodePolicy::Commits, 4 << 20).unwrap())
1301            .registered();
1302        let b = std::sync::Arc::new(ExplodedCache::open(&base_b, ExplodePolicy::Commits, 4 << 20).unwrap())
1303            .registered();
1304        // Fill with the global check OFF, so the shape under test is set up
1305        // before it is asked about.
1306        a.set_global_line(0);
1307        b.set_global_line(0);
1308        let big = noise(64 << 10, 5);
1309        for i in 0..48u32 {
1310            a.write_back(&oid(i), GitObjectKind::Blob, &big, dear(&big)).unwrap();
1311            a.maintain().unwrap();
1312        }
1313        let small = noise(8 << 10, 6);
1314        b.write_back(&oid(100), GitObjectKind::Blob, &small, dear(&small)).unwrap();
1315        b.maintain().unwrap();
1316        let a0 = a.use_now();
1317        let b0 = b.use_now();
1318        let line = 1u64 << 20;
1319        assert!(a0.lazy_bytes > line, "precondition: A alone is over the line, has {}", a0.lazy_bytes);
1320        assert!(a0.segments >= 3, "precondition: A rotated into segments, has {}", a0.segments);
1321        assert_eq!(a0.evicted_segments + b0.evicted_segments, 0, "precondition: nothing evicted yet");
1322
1323        a.set_global_line(line);
1324        b.set_global_line(line);
1325
1326        // B ticks first: under its share, it must not pay for A.
1327        b.maintain().unwrap();
1328        assert_eq!(b.use_now().evicted_segments, 0, "a cache under its fair share must not evict for the box");
1329
1330        // A ticks: over its share, it evicts for the box until it is under the
1331        // line on its own (it cannot stop above it: then the box is over and
1332        // it is over any share of it).
1333        a.maintain().unwrap();
1334        let a1 = a.use_now();
1335        assert!(a1.evicted_segments > 0, "the big pool must evict for the box");
1336        assert!(
1337            a1.lazy_bytes + a1.eager_bytes <= line,
1338            "A must have shrunk under the line on its own: {} > {line}",
1339            a1.lazy_bytes + a1.eager_bytes
1340        );
1341        assert!(a1.lazy_bytes < 4 << 20, "its own budget was never the reason");
1342        assert_eq!(b.use_now().evicted_segments, 0, "B still untouched");
1343    }
1344    #[test]
1345    fn eviction_never_takes_a_commit() {
1346        let base = tmp("commits");
1347        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
1348        for i in 0..50u32 {
1349            c.explode(&oid(1_000_000 + i), GitObjectKind::Commit, b"tree x\n")
1350                .unwrap();
1351        }
1352        c.flush().unwrap();
1353        let blob = noise(64 << 10, 1);
1354        for i in 0..200u32 {
1355            c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
1356            c.maintain().unwrap();
1357        }
1358        assert!(c.use_now().evicted_segments > 0, "nothing was evicted");
1359        for i in 0..50u32 {
1360            assert!(
1361                c.content(&oid(1_000_000 + i)).unwrap().is_some(),
1362                "a commit was evicted — graph and reach are now blind"
1363            );
1364        }
1365    }
1366
1367    /// `repo gc` reclaims the cache and keeps the graph.
1368    #[test]
1369    fn dropping_the_cache_reclaims_the_lazy_pool_only() {
1370        let base = tmp("gc");
1371        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
1372        c.explode(&oid(9), GitObjectKind::Commit, b"tree y\n").unwrap();
1373        let blob = vec![2u8; 32 << 10];
1374        for i in 0..100u32 {
1375            c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
1376        }
1377        c.maintain().unwrap();
1378        let before = c.use_now();
1379        assert!(before.lazy_bytes > 0);
1380        let freed = c.drop_cache().unwrap();
1381        assert!(freed > 0, "gc reclaimed nothing");
1382        let after = c.use_now();
1383        assert_eq!(after.lazy_bytes, 0, "the lazy pool survived gc");
1384        assert_eq!(after.eager_bytes, before.eager_bytes, "gc took the graph too");
1385        assert!(c.content(&oid(9)).unwrap().is_some());
1386    }
1387
1388    /// The segments a previous process wrote are adopted, so a restart does not
1389    /// lose the cache and does not lose track of its size either.
1390    #[test]
1391    fn a_reopen_adopts_the_segments_it_left() {
1392        let base = tmp("reopen");
1393        let blob = vec![5u8; 32 << 10];
1394        let bytes = {
1395            let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
1396            for i in 0..40u32 {
1397                c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
1398            }
1399            c.maintain().unwrap();
1400            c.use_now().lazy_bytes
1401        };
1402        assert!(bytes > 0);
1403        let again = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
1404        assert_eq!(again.use_now().lazy_bytes, bytes, "the segments were not adopted");
1405        assert_eq!(
1406            again.content(&oid(7)).unwrap().map(|(_, b)| b),
1407            Some(blob),
1408            "an adopted segment does not answer"
1409        );
1410    }
1411
1412    /// ★ **DO NOT EXPLODE UNIFORMLY: a cheap object is never cached.**
1413    ///
1414    /// Rickard, 2026-08-30: the index points at the object's own zlib stream at
1415    /// its own offset, so a whole object is one direct inflate. Caching that
1416    /// spends quota to save microseconds — and evicts something that costs
1417    /// milliseconds. So the gate is rebuild cost, not "was read".
1418    ///
1419    /// Seen RED by making `write_back` admit everything: 400 shallow 1 KiB
1420    /// objects land in the pool, `direct_reads` stays 0, and the assertion
1421    /// names both.
1422    #[test]
1423    fn a_shallow_small_object_is_read_directly_and_never_cached() {
1424        let base = tmp("cheap");
1425        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
1426        let small = vec![1u8; 1 << 10];
1427        let big = vec![2u8; 64 << 10];
1428
1429        // Cheap: shallow and small. Not worth a byte of anybody's quota.
1430        for i in 0..400u32 {
1431            let took = c
1432                .write_back(
1433                    &oid(i),
1434                    GitObjectKind::Blob,
1435                    &small,
1436                    crate::resolve::ReadCost {
1437                        depth: CHEAP_DEPTH,
1438                        inflated: small.len() as u64,
1439                        external_base: false,
1440                    },
1441                )
1442                .unwrap();
1443            assert!(!took, "a cheap object asked for a flush");
1444        }
1445        c.maintain().unwrap();
1446        let after_cheap = c.use_now();
1447        assert_eq!(
1448            after_cheap.lazy_bytes, 0,
1449            "400 shallow 1 KiB objects put {} B in the cache; a direct read is \
1450             microseconds and this is quota spent to save them",
1451            after_cheap.lazy_bytes
1452        );
1453        assert_eq!(
1454            after_cheap.direct_reads, 400,
1455            "the policy declined {} of 400 cheap objects — it is not running",
1456            after_cheap.direct_reads
1457        );
1458        assert_eq!(after_cheap.written_back, 0);
1459
1460        // Expensive by DEPTH, and by an external base. Each on its own admits.
1461        c.write_back(
1462            &oid(1000),
1463            GitObjectKind::Blob,
1464            &small,
1465            crate::resolve::ReadCost { depth: CHEAP_DEPTH + 1, inflated: small.len() as u64, external_base: false },
1466        )
1467        .unwrap();
1468        c.write_back(
1469            &oid(1002),
1470            GitObjectKind::Blob,
1471            &small,
1472            crate::resolve::ReadCost { depth: 0, inflated: small.len() as u64, external_base: true },
1473        )
1474        .unwrap();
1475        // **Big is NOT a reason to cache.** A 64 KiB — or a 3 MiB — whole blob
1476        // is one linear inflate and a bad tenant of a shared budget, and it is
1477        // declined however deep it is. This is lane AA's second copy.
1478        c.write_back(
1479            &oid(1001),
1480            GitObjectKind::Blob,
1481            &big,
1482            crate::resolve::ReadCost { depth: 40, inflated: (MAX_CACHED_BYTES + 1) as u64, external_base: true },
1483        )
1484        .unwrap();
1485        c.maintain().unwrap();
1486        let end = c.use_now();
1487        assert_eq!(end.written_back, 2, "an expensive object was declined");
1488        assert_eq!(
1489            end.direct_reads, 401,
1490            "an object over the size cap was cached instead of read directly"
1491        );
1492        for n in [1000u32, 1002] {
1493            assert!(
1494                c.content(&oid(n)).unwrap().is_some(),
1495                "the expensive object {n} is not in the cache"
1496            );
1497        }
1498        assert!(
1499            c.content(&oid(1001)).unwrap().is_none(),
1500            "an object over the size cap is in the cache"
1501        );
1502        assert!(
1503            c.content(&oid(7)).unwrap().is_none(),
1504            "a declined object is in the cache after all"
1505        );
1506    }
1507
1508    /// ★ **Eviction ranks by rebuild cost, not by age alone.**
1509    ///
1510    /// Two segments, the older one full of expensive objects and the newer one
1511    /// full of barely-admitted ones. Plain FIFO takes the expensive segment
1512    /// because it is older; ranking by `mean rebuild cost × recency` takes the
1513    /// cheap one, which is the whole point — evict what is cheap to recompute.
1514    ///
1515    /// Seen RED by restoring `Some(l[0].clone())`: the depth-40 objects go and
1516    /// the barely-admitted ones stay.
1517    #[test]
1518    fn eviction_takes_the_cheap_segment_before_the_old_expensive_one() {
1519        let base = tmp("costrank");
1520        // Small budget, small segments: three segments fit and the fourth
1521        // forces one out.
1522        let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
1523        let payload = noise(32 << 10, 9);
1524        let dear = crate::resolve::ReadCost { depth: 40, inflated: payload.len() as u64, external_base: false };
1525        // Barely admitted: one link past the cheap depth.
1526        let cheapish = crate::resolve::ReadCost {
1527            depth: CHEAP_DEPTH + 1,
1528            inflated: payload.len() as u64,
1529            external_base: false,
1530        };
1531
1532        // Segment 0: expensive. Rotate it out by filling past the segment size.
1533        for i in 0..16u32 {
1534            c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear).unwrap();
1535        }
1536        c.maintain().unwrap();
1537        // Segment 1+: cheapish.
1538        for i in 100..160u32 {
1539            c.write_back(&oid(i), GitObjectKind::Blob, &payload, cheapish).unwrap();
1540            c.maintain().unwrap();
1541        }
1542        let end = c.use_now();
1543        assert!(end.evicted_segments > 0, "nothing was evicted; this proves nothing");
1544        // The expensive objects are the ones still there.
1545        let expensive_left = (0..16u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
1546        let cheap_left = (100..160u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
1547        assert!(
1548            expensive_left > 0,
1549            "every depth-40 object was evicted while {cheap_left} barely-admitted ones \
1550             survived — eviction is ranking by age alone: {end:?}"
1551        );
1552    }
1553
1554    /// A budget is parsed, and a typo is refused rather than silently taking a
1555    /// default nobody is billed for.
1556    #[test]
1557    fn a_budget_is_read_and_a_typo_is_refused() {
1558        assert_eq!(parse_budget("8G").unwrap(), 8 << 30);
1559        assert_eq!(parse_budget(" 512m ").unwrap(), 512 << 20);
1560        assert_eq!(parse_budget("1048576").unwrap(), 1 << 20);
1561        assert_eq!(parse_budget("unlimited").unwrap(), 0);
1562        assert!(parse_budget("8 gigs").is_err());
1563        assert!(parse_budget("").is_err());
1564    }
1565
1566    /// The default is derived from the filesystem, and the appliance arithmetic
1567    /// in the module docs is the one the code does.
1568    #[test]
1569    fn the_default_budget_is_the_smaller_of_a_tenth_and_the_ceiling() {
1570        let d = default_budget(&std::env::temp_dir());
1571        assert!(d >= MIN_BUDGET && d <= BUDGET_CEILING, "{d}");
1572    }
1573
1574    /// **A validation read caches nothing, and the same reads without the guard
1575    /// fill the pool** — both halves, because either alone proves nothing.
1576    ///
1577    /// The RED half is the first loop: these are objects the admission policy
1578    /// WANTS (deep chains), so a pool that stays empty under the guard cannot be
1579    /// explained by `worth_caching` declining them. The counter is asserted too,
1580    /// because "the pool is empty" is also what a cache that was handed nothing
1581    /// looks like.
1582    #[test]
1583    fn a_suppressed_thread_caches_nothing_and_an_unsuppressed_one_fills_the_pool() {
1584        let payload = vec![7u8; 64 << 10];
1585        let costly = crate::resolve::ReadCost {
1586            depth: CHEAP_DEPTH + 8,
1587            inflated: payload.len() as u64,
1588            external_base: false,
1589        };
1590
1591        // ── RED: no guard. The same objects, admitted. ──────────────────────
1592        let hot = ExplodedCache::open(&tmp("nosuppress"), ExplodePolicy::Commits, 0).unwrap();
1593        for i in 0..200u32 {
1594            hot.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
1595        }
1596        hot.maintain().unwrap();
1597        let hot_use = hot.use_now();
1598        assert!(
1599            hot_use.lazy_bytes > 0,
1600            "without the guard these objects must be cached, or the GREEN half \
1601             below is proving that `worth_caching` declined them and nothing else"
1602        );
1603        assert_eq!(hot_use.written_back, 200);
1604        assert_eq!(hot_use.suppressed_write_backs, 0);
1605
1606        // ── GREEN: the identical loop inside the guard. ─────────────────────
1607        let cold = ExplodedCache::open(&tmp("suppress"), ExplodePolicy::Commits, 0).unwrap();
1608        {
1609            let _guard = suppress_write_back();
1610            assert!(write_back_suppressed());
1611            for i in 0..200u32 {
1612                let took = cold.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
1613                assert!(!took, "a validation read asked for a flush");
1614            }
1615        }
1616        cold.maintain().unwrap();
1617        let cold_use = cold.use_now();
1618        assert_eq!(
1619            cold_use.lazy_bytes, 0,
1620            "a receive-pack's connectivity walk put {} B in the cache",
1621            cold_use.lazy_bytes
1622        );
1623        assert_eq!(cold_use.written_back, 0);
1624        assert_eq!(
1625            cold_use.suppressed_write_backs, 200,
1626            "the gate must be what declined them — {} suppressed, not 200",
1627            cold_use.suppressed_write_backs
1628        );
1629        // The policy was never consulted: a suppressed read is declined BEFORE
1630        // the admission gate, so it is not a `direct_read` either.
1631        assert_eq!(cold_use.direct_reads, 0);
1632    }
1633
1634    /// The guard is released on drop, including out of a panic — a thread that
1635    /// panicked mid-walk must not stop caching for the rest of its life.
1636    #[test]
1637    fn the_guard_is_released_on_drop_and_on_unwind_and_it_nests() {
1638        assert!(!write_back_suppressed(), "a fresh thread starts unsuppressed");
1639        {
1640            let _a = suppress_write_back();
1641            {
1642                let _b = suppress_write_back();
1643                assert!(write_back_suppressed());
1644            }
1645            assert!(
1646                write_back_suppressed(),
1647                "the inner guard ending reopened the cache while the outer scope \
1648                 is still validating"
1649            );
1650        }
1651        assert!(!write_back_suppressed());
1652
1653        let unwound = std::panic::catch_unwind(|| {
1654            let _g = suppress_write_back();
1655            assert!(write_back_suppressed());
1656            panic!("a connectivity walk died");
1657        });
1658        assert!(unwound.is_err());
1659        assert!(
1660            !write_back_suppressed(),
1661            "a panic inside the guard left this thread never caching again"
1662        );
1663    }
1664}