Skip to main content

mbx_cache_store/
lib.rs

1//! Store inspection and garbage collection.
2//!
3//! The store holds six trees: `cas/v1` for content-addressed objects,
4//! `action-results/v1` for the results that reference them, `task-manifests/v1`
5//! for the prediction index, `checkouts/v1` for the checkouts that have built
6//! each identity, `build-receipts/v1` for exact completed build closures, and
7//! `sessions/v1` for per-build event streams. Only the first two are collected
8//! for size; manifests and receipts are small, checkout records expire with
9//! their checkout claims, and session streams are bounded by age and count
10//! because they are history rather than cache content.
11
12mod events;
13
14use eyre::{Context, Result};
15use mbx_cache_core::{
16    ActionPrediction, CacheDigest, CacheDirectory, LocalCas, RemoteActionResult, RustcMetadata,
17    TaskActionManifest, is_task_identity, task_manifest_actions,
18};
19use serde::{Deserialize, Serialize};
20use std::collections::{BTreeMap, BTreeSet, HashSet};
21use std::path::{Component, Path, PathBuf};
22use std::time::{Duration, SystemTime, UNIX_EPOCH};
23
24const CAS_DIR: &str = "cas/v1";
25const ACTION_RESULTS_DIR: &str = "action-results/v1";
26const CHECKOUTS_DIR: &str = "checkouts/v1";
27const SWEEP_STAMP: &str = "gc/v1/last-sweep";
28const SWEEP_LOCK: &str = "gc/v1/sweep.lock";
29const CHECKOUT_RECORD_VERSION: u8 = 1;
30const BUILD_RECEIPTS_DIR: &str = "build-receipts/v1";
31const BUILD_RECEIPT_VERSION: u8 = 1;
32const EXPORT_MANIFEST: &str = "mbx-cache-export-v1.json";
33const EXPORT_VERSION: u8 = 1;
34
35const SESSION_RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60);
36const MAX_SESSIONS: usize = 256;
37
38/// How long a checkout's claim outlives the last build that renewed it.
39///
40/// Existence alone is not enough to keep a claim alive. An identity covers one
41/// exact command line against one exact lockfile, so a checkout that lives for
42/// years accumulates an identity per lockfile it ever had -- and every one of
43/// them would go on rooting the actions of a build nobody will run again.
44/// Without this bound the rooted set only grows, until everything is rooted and
45/// the ordering means nothing. A build renews the claims it uses, so anything
46/// this stale belongs to a command that has moved on.
47const CHECKOUT_RETENTION: Duration = Duration::from_secs(30 * 24 * 60 * 60);
48
49#[derive(Debug, Default, PartialEq, Eq)]
50pub struct StoreStats {
51    pub objects: u64,
52    pub object_bytes: u64,
53    pub action_results: u64,
54    pub action_result_bytes: u64,
55    pub live_checkouts: u64,
56    /// Claims that root nothing any more: a checkout that is gone, or one that
57    /// is still there but has not renewed this claim inside the retention
58    /// window. Reported as stale rather than gone because those are different
59    /// things and only one of them means the directory is missing.
60    pub stale_checkouts: u64,
61}
62
63impl StoreStats {
64    pub fn total_bytes(&self) -> u64 {
65        self.object_bytes.saturating_add(self.action_result_bytes)
66    }
67}
68
69#[derive(Debug, Default, PartialEq, Eq)]
70pub struct GcOutcome {
71    pub removed_objects: u64,
72    pub removed_action_results: u64,
73    pub removed_checkout_records: u64,
74    /// Event streams dropped on age or count. These bytes are not included in
75    /// `remaining_bytes` because session history is not cache content.
76    pub removed_session_streams: u64,
77    pub removed_bytes: u64,
78    pub remaining_bytes: u64,
79}
80
81#[derive(Debug, PartialEq, Eq)]
82pub struct ProjectUsage {
83    pub workspace_root: PathBuf,
84    pub identities: u64,
85    pub action_bytes: u64,
86    pub target_bytes: u64,
87    pub live: bool,
88}
89
90#[derive(Debug, PartialEq, Eq)]
91pub struct LargestEntry {
92    pub kind: &'static str,
93    pub path: PathBuf,
94    pub bytes: u64,
95}
96
97#[derive(Debug, Default, PartialEq, Eq)]
98pub struct VerifyOutcome {
99    pub checked_objects: u64,
100    pub checked_action_results: u64,
101    pub problems: Vec<PathBuf>,
102}
103
104#[derive(Debug, Default, PartialEq, Eq)]
105pub struct RemoveProjectOutcome {
106    pub removed_checkout_records: u64,
107}
108
109#[derive(Debug, Default, PartialEq, Eq)]
110pub struct TransferOutcome {
111    pub actions: u64,
112    pub objects: u64,
113    pub bytes: u64,
114}
115
116#[derive(Debug, Serialize, Deserialize)]
117#[serde(deny_unknown_fields)]
118struct ExportManifest {
119    version: u8,
120    tasks: Vec<TaskActionManifest>,
121    actions: Vec<CacheDigest>,
122}
123
124/// The exact cache predictions completed by one top-level build command.
125#[derive(Debug, Clone, Serialize, Deserialize)]
126#[serde(deny_unknown_fields)]
127struct BuildReceipt {
128    version: u8,
129    workspace_root: PathBuf,
130    identity: String,
131    completed_nanos: u64,
132    #[serde(default, skip_serializing_if = "Option::is_none")]
133    group: Option<String>,
134    predictions: Vec<ActionPrediction>,
135}
136
137/// Record an exact completed build for checkout and grouped exports.
138pub fn record_build_receipt(
139    store: &Path,
140    run: &str,
141    identity: &str,
142    workspace_root: &Path,
143    group: Option<&str>,
144    predictions: Vec<ActionPrediction>,
145) -> Result<()> {
146    if !is_task_identity(run) || !is_task_identity(identity) {
147        eyre::bail!("invalid build receipt identity");
148    }
149    if !(TaskActionManifest {
150        version: 1,
151        task: identity.to_owned(),
152        predictions: predictions.clone(),
153    })
154    .validate()
155    {
156        eyre::bail!("invalid build receipt prediction");
157    }
158    if let Some(group) = group {
159        validate_export_group(group)?;
160    }
161    let key = workspace_key(workspace_root);
162    let lock_path = store
163        .join(BUILD_RECEIPTS_DIR)
164        .join("locks")
165        .join(format!("{key}.lock"));
166    std::fs::create_dir_all(lock_path.parent().expect("receipt lock has a parent"))?;
167    let mut lock = fslock::LockFile::open(&lock_path)?;
168    lock.lock()?;
169    let completed_nanos = SystemTime::now()
170        .duration_since(UNIX_EPOCH)
171        .map(|elapsed| elapsed.as_nanos().try_into().unwrap_or(u64::MAX))
172        .unwrap_or_default();
173    let receipt = BuildReceipt {
174        version: BUILD_RECEIPT_VERSION,
175        workspace_root: workspace_root.to_path_buf(),
176        identity: identity.to_owned(),
177        completed_nanos,
178        group: group.map(str::to_owned),
179        predictions,
180    };
181    let bytes = serde_json::to_vec(&receipt)?;
182    write_atomic(&latest_receipt_path(store, workspace_root), &bytes)?;
183    if let Some(group) = group {
184        write_atomic(&group_receipt_path(store, group, run), &bytes)?;
185    }
186    Ok(())
187}
188
189fn workspace_key(workspace_root: &Path) -> String {
190    CacheDigest::blake3(workspace_root.to_string_lossy().as_bytes()).hash
191}
192
193fn latest_receipt_path(store: &Path, workspace_root: &Path) -> PathBuf {
194    store
195        .join(BUILD_RECEIPTS_DIR)
196        .join("checkouts")
197        .join(format!("{}.json", workspace_key(workspace_root)))
198}
199
200fn group_receipt_path(store: &Path, group: &str, run: &str) -> PathBuf {
201    store
202        .join(BUILD_RECEIPTS_DIR)
203        .join("groups")
204        .join(group_key(group))
205        .join(format!("{run}.json"))
206}
207
208fn group_key(group: &str) -> String {
209    CacheDigest::blake3(group.as_bytes()).hash
210}
211
212fn read_build_receipt(path: &Path) -> Option<BuildReceipt> {
213    let bytes = std::fs::read(path).ok()?;
214    let receipt = serde_json::from_slice::<BuildReceipt>(&bytes).ok()?;
215    let valid = TaskActionManifest {
216        version: 1,
217        task: receipt.identity.clone(),
218        predictions: receipt.predictions.clone(),
219    }
220    .validate();
221    (receipt.version == BUILD_RECEIPT_VERSION && valid).then_some(receipt)
222}
223
224fn validate_export_group(group: &str) -> Result<()> {
225    if group.is_empty() || group.len() > 256 || group.chars().any(char::is_control) {
226        eyre::bail!("invalid build export group");
227    }
228    Ok(())
229}
230
231/// Export the complete local cache closure of this checkout's most recent build.
232pub fn export_checkout(
233    store: &Path,
234    workspace_root: &Path,
235    archive: &Path,
236) -> Result<TransferOutcome> {
237    let receipt = read_build_receipt(&latest_receipt_path(store, workspace_root))
238        .filter(|receipt| receipt.workspace_root == workspace_root)
239        .ok_or_else(|| {
240            eyre::eyre!(
241                "no completed mbx build is recorded for {}",
242                workspace_root.display()
243            )
244        })?;
245    export_receipts(store, vec![receipt], archive)
246}
247
248/// Export the union of every completed build recorded under one CI group.
249pub fn export_group(store: &Path, group: &str, archive: &Path) -> Result<TransferOutcome> {
250    validate_export_group(group)?;
251    let root = store
252        .join(BUILD_RECEIPTS_DIR)
253        .join("groups")
254        .join(group_key(group));
255    let receipts = walk_files(&root)?
256        .into_iter()
257        .filter_map(|entry| read_build_receipt(&entry.path).map(|receipt| (entry.path, receipt)))
258        .filter(|(_, receipt)| receipt.group.as_deref() == Some(group))
259        .collect::<Vec<_>>();
260    if receipts.is_empty() {
261        eyre::bail!("no completed mbx builds are recorded for export group {group:?}");
262    }
263    let outcome = export_receipts(
264        store,
265        receipts
266            .iter()
267            .map(|(_, receipt)| receipt.clone())
268            .collect(),
269        archive,
270    )?;
271    // A receipt is a pending-export root. Retire only the files this export
272    // consumed, and only after its complete archive has been published. A
273    // concurrent build can add another uniquely named receipt to the group
274    // without this cleanup deleting work the archive did not include.
275    for (path, _) in receipts {
276        std::fs::remove_file(path)?;
277    }
278    let _ = std::fs::remove_dir(root);
279    Ok(outcome)
280}
281
282fn export_receipts(
283    store: &Path,
284    mut receipts: Vec<BuildReceipt>,
285    archive: &Path,
286) -> Result<TransferOutcome> {
287    receipts.sort_by_key(|receipt| receipt.completed_nanos);
288    let mut actions = BTreeSet::new();
289    let mut tasks = BTreeMap::new();
290    for receipt in receipts {
291        actions.extend(
292            receipt
293                .predictions
294                .iter()
295                .map(|prediction| prediction.action.clone()),
296        );
297        // One task manifest can carry only one action per invocation. The
298        // newest run of the same task is the useful prediction set to import;
299        // every older action remains in `actions` and therefore in the bundle.
300        tasks.insert(
301            receipt.identity.clone(),
302            TaskActionManifest {
303                version: 1,
304                task: receipt.identity,
305                predictions: receipt.predictions,
306            },
307        );
308    }
309    let tasks = tasks.into_values().collect::<Vec<_>>();
310    let (objects, result_paths) = strict_closure(store, &actions)?;
311    let parent = archive
312        .parent()
313        .filter(|parent| !parent.as_os_str().is_empty())
314        .unwrap_or_else(|| Path::new("."));
315    std::fs::create_dir_all(parent)?;
316    let temporary = tempfile::Builder::new()
317        .prefix(".mbx-export-")
318        .tempfile_in(parent)?;
319    let mut builder = tar::Builder::new(temporary.reopen()?);
320    append_bytes(
321        &mut builder,
322        Path::new(EXPORT_MANIFEST),
323        &serde_json::to_vec(&ExportManifest {
324            version: EXPORT_VERSION,
325            tasks,
326            actions: actions.iter().cloned().collect(),
327        })?,
328    )?;
329    for path in objects.iter().chain(result_paths.iter()) {
330        append_file(&mut builder, store, path)?;
331    }
332    builder.finish()?;
333    drop(builder);
334    temporary
335        .persist(archive)
336        .map_err(|error| error.error)
337        .wrap_err_with(|| format!("failed to publish {}", archive.display()))?;
338    let bytes = std::fs::metadata(archive)?.len();
339    Ok(TransferOutcome {
340        actions: actions.len() as u64,
341        objects: objects.len() as u64,
342        bytes,
343    })
344}
345
346/// Validate a cache export in isolation, then publish its objects and actions.
347pub fn import_archive(store: &Path, archive: &Path) -> Result<TransferOutcome> {
348    let staging = tempfile::tempdir()?;
349    let file = std::fs::File::open(archive)
350        .wrap_err_with(|| format!("failed to open {}", archive.display()))?;
351    let mut bundle = tar::Archive::new(file);
352    let mut seen = BTreeSet::new();
353    for entry in bundle.entries()? {
354        let mut entry = entry?;
355        let entry_type = entry.header().entry_type();
356        if !entry_type.is_file() && !entry_type.is_gnu_sparse() {
357            eyre::bail!("cache export contains a non-file entry");
358        }
359        let path = entry.path()?.into_owned();
360        validate_archive_path(&path)?;
361        if !seen.insert(path.clone()) {
362            eyre::bail!("cache export contains duplicate entry {}", path.display());
363        }
364        let destination = staging.path().join(&path);
365        std::fs::create_dir_all(destination.parent().expect("entry has a parent"))?;
366        // `unpack` understands GNU sparse maps. A plain stream copy expands
367        // holes into physical zeroes, which is both slower and much larger for
368        // Rust artifacts containing sparse sections.
369        entry.unpack(&destination)?;
370    }
371    let manifest: ExportManifest =
372        serde_json::from_slice(&std::fs::read(staging.path().join(EXPORT_MANIFEST))?)?;
373    let actions = manifest.actions.iter().cloned().collect::<BTreeSet<_>>();
374    let task_identities = manifest
375        .tasks
376        .iter()
377        .map(|task| task.task.as_str())
378        .collect::<BTreeSet<_>>();
379    if manifest.version != EXPORT_VERSION
380        || manifest.tasks.is_empty()
381        || actions.len() != manifest.actions.len()
382        || task_identities.len() != manifest.tasks.len()
383        || manifest.tasks.iter().any(|task| !task.validate())
384        || manifest
385            .actions
386            .iter()
387            .any(|action| action.algorithm != "blake3" || action.validate().is_err())
388        || manifest.tasks.iter().any(|task| {
389            task.predictions
390                .iter()
391                .any(|prediction| !actions.contains(&prediction.action))
392        })
393    {
394        eyre::bail!("unsupported or invalid cache export manifest");
395    }
396    let (objects, result_paths) = strict_closure(staging.path(), &actions)
397        .wrap_err("cache export is incomplete or corrupt")?;
398
399    let cas = LocalCas::new(store);
400    for path in &objects {
401        let relative = path.strip_prefix(staging.path())?;
402        let source = staging.path().join(relative);
403        let digest = addressed_digest(staging.path(), &source, false)
404            .ok_or_else(|| eyre::eyre!("invalid cache object path {}", relative.display()))?;
405        cas.store_file(&digest, &source)?;
406    }
407    let action_cache = mbx_cache_core::LocalActionCache::new(store);
408    for path in &result_paths {
409        let result: RemoteActionResult = serde_json::from_slice(&std::fs::read(path)?)?;
410        action_cache.store(&result)?;
411    }
412    for task in manifest.tasks {
413        merge_imported_manifest(store, task)?;
414    }
415    Ok(TransferOutcome {
416        actions: actions.len() as u64,
417        objects: objects.len() as u64,
418        bytes: std::fs::metadata(archive)?.len(),
419    })
420}
421
422fn strict_closure(
423    store: &Path,
424    actions: &BTreeSet<CacheDigest>,
425) -> Result<(BTreeSet<PathBuf>, BTreeSet<PathBuf>)> {
426    let cas = LocalCas::new(store);
427    let action_cache = mbx_cache_core::LocalActionCache::new(store);
428    let mut objects = BTreeSet::new();
429    let mut results = BTreeSet::new();
430    let mut directories = BTreeSet::new();
431    for action in actions {
432        let result = action_cache
433            .find(action)?
434            .ok_or_else(|| eyre::eyre!("action result is missing for {}", action.hash))?;
435        results.insert(action_cache.path_for(action)?);
436        require_object(&cas, &mut objects, &result.action)?;
437        if let Some(metadata) = &result.metadata {
438            require_object(&cas, &mut objects, metadata)?;
439            let captured: CapturedOutput =
440                serde_json::from_slice(&std::fs::read(cas.path_for(metadata)?)?)
441                    .wrap_err("action metadata is invalid")?;
442            require_object(&cas, &mut objects, &captured.stdout)?;
443            require_object(&cas, &mut objects, &captured.stderr)?;
444        }
445        if let Some(root) = &result.output_root {
446            require_object(&cas, &mut objects, root)?;
447            let mut pending = vec![root.clone()];
448            while let Some(digest) = pending.pop() {
449                if !directories.insert(digest.clone()) {
450                    continue;
451                }
452                let directory: CacheDirectory =
453                    serde_json::from_slice(&std::fs::read(cas.path_for(&digest)?)?)
454                        .wrap_err("output directory is invalid")?;
455                for file in directory.files {
456                    require_object(&cas, &mut objects, &file.digest)?;
457                }
458                for child in directory.directories {
459                    require_object(&cas, &mut objects, &child.digest)?;
460                    pending.push(child.digest);
461                }
462            }
463        }
464    }
465    Ok((objects, results))
466}
467
468#[derive(Deserialize)]
469struct CapturedOutput {
470    stdout: CacheDigest,
471    stderr: CacheDigest,
472}
473
474fn merge_imported_manifest(destination: &Path, mut imported: TaskActionManifest) -> Result<()> {
475    let identity = imported.task.clone();
476    let destination_path = task_manifest_path(destination, &identity);
477    if let Ok(bytes) = std::fs::read(&destination_path)
478        && let Ok(existing) = serde_json::from_slice::<TaskActionManifest>(&bytes)
479        && existing.task == identity
480        && existing.validate()
481    {
482        let mut predictions = imported
483            .predictions
484            .into_iter()
485            .map(|prediction| (prediction.invocation.clone(), prediction))
486            .collect::<BTreeMap<_, _>>();
487        predictions.extend(
488            existing
489                .predictions
490                .iter()
491                .cloned()
492                .map(|prediction| (prediction.invocation.clone(), prediction)),
493        );
494        imported.predictions = predictions.into_values().collect();
495        if !imported.validate() {
496            return Ok(());
497        }
498    }
499    write_atomic(&destination_path, &serde_json::to_vec(&imported)?)
500}
501
502fn require_object(
503    cas: &LocalCas,
504    objects: &mut BTreeSet<PathBuf>,
505    digest: &CacheDigest,
506) -> Result<()> {
507    let path = cas
508        .find(digest)?
509        .ok_or_else(|| eyre::eyre!("cache object is missing for {}", digest.hash))?;
510    objects.insert(path);
511    Ok(())
512}
513
514fn task_manifest_path(store: &Path, identity: &str) -> PathBuf {
515    store
516        .join("task-manifests/v1")
517        .join(format!("{identity}.json"))
518}
519
520fn append_file(builder: &mut tar::Builder<std::fs::File>, root: &Path, path: &Path) -> Result<()> {
521    let name = path.strip_prefix(root)?;
522    builder.append_path_with_name(path, name)?;
523    Ok(())
524}
525
526fn append_bytes(
527    builder: &mut tar::Builder<std::fs::File>,
528    name: &Path,
529    bytes: &[u8],
530) -> Result<()> {
531    let mut header = tar::Header::new_gnu();
532    header.set_size(bytes.len() as u64);
533    header.set_mode(0o644);
534    header.set_cksum();
535    builder.append_data(&mut header, name, bytes)?;
536    Ok(())
537}
538
539fn validate_archive_path(path: &Path) -> Result<()> {
540    if path.as_os_str().is_empty()
541        || path.is_absolute()
542        || path
543            .components()
544            .any(|component| !matches!(component, Component::Normal(_)))
545    {
546        eyre::bail!("cache export contains unsafe path {}", path.display());
547    }
548    let text = path.to_string_lossy();
549    if text != EXPORT_MANIFEST
550        && !text.starts_with("cas/v1/")
551        && !text.starts_with("action-results/v1/")
552    {
553        eyre::bail!("cache export contains unexpected path {}", path.display());
554    }
555    Ok(())
556}
557
558/// One checkout's claim on the actions a build identity recorded.
559///
560/// The target directory is not read back by anything yet; it is recorded
561/// because it is the other half of what the shim mapped, and a record that
562/// names only half of it would have to be rewritten to answer where the
563/// artifacts went.
564#[derive(Debug, Clone, Serialize, Deserialize)]
565#[serde(deny_unknown_fields)]
566struct CheckoutRecord {
567    version: u8,
568    workspace_root: PathBuf,
569    target_dir: PathBuf,
570    updated_secs: u64,
571}
572
573/// Summarize what the store currently holds.
574pub fn stats(store: &Path) -> Result<StoreStats> {
575    let objects = walk_files(&store.join(CAS_DIR))?;
576    let results = walk_files(&store.join(ACTION_RESULTS_DIR))?;
577    let checkouts = scan_checkouts(store)?;
578    Ok(StoreStats {
579        objects: objects.len() as u64,
580        object_bytes: objects.iter().map(|entry| entry.size).sum(),
581        action_results: results.len() as u64,
582        action_result_bytes: results.iter().map(|entry| entry.size).sum(),
583        live_checkouts: checkouts.live_records,
584        stale_checkouts: checkouts.stale_records.len() as u64,
585    })
586}
587
588/// Size the target tree a checkout record names, if it names one.
589///
590/// A build with no target directory of its own -- one driven by make or CMake,
591/// which writes wherever it was told -- records the checkout in that field
592/// rather than inventing a directory. Walking it would report the whole source
593/// tree, `.git` included, as though it were build output.
594fn checkout_target_bytes(sizes: &mut BTreeMap<PathBuf, u64>, record: &CheckoutRecord) -> u64 {
595    if record.target_dir == record.workspace_root {
596        return 0;
597    }
598    cached_tree_bytes(sizes, &record.target_dir)
599}
600
601/// Attribute cache and target bytes to each recorded workspace.
602///
603/// Shared objects are counted for every workspace that can reach them. That
604/// makes each row answer "how much keeps this workspace warm" without
605/// pretending shared storage can be divided exactly between projects.
606pub fn projects(store: &Path) -> Result<Vec<ProjectUsage>> {
607    let mut projects: BTreeMap<PathBuf, (BTreeSet<String>, bool, u64)> = BTreeMap::new();
608    let mut target_sizes = BTreeMap::new();
609    let root = store.join(CHECKOUTS_DIR);
610    for identity_entry in read_dir_or_empty(&root)? {
611        let identity = identity_entry.file_name().to_string_lossy().into_owned();
612        if !is_task_identity(&identity) || !identity_entry.file_type()?.is_dir() {
613            continue;
614        }
615        for entry in walk_files(&identity_entry.path())? {
616            let Some(record) = read_checkout_record(&entry.path) else {
617                continue;
618            };
619            let project = projects.entry(record.workspace_root.clone()).or_default();
620            if claim_is_live(store, &record) {
621                project.0.insert(identity.clone());
622                project.1 = true;
623            }
624            project.2 = project
625                .2
626                .max(checkout_target_bytes(&mut target_sizes, &record));
627        }
628    }
629    let action_cache = mbx_cache_core::LocalActionCache::new(store);
630    let mut usages = Vec::new();
631    for (workspace_root, (identities, live, target_bytes)) in projects {
632        let mut paths = rooted_objects(store, &identities)?;
633        for identity in &identities {
634            for action in task_manifest_actions(store, identity).unwrap_or_default() {
635                if let Ok(path) = action_cache.path_for(&action) {
636                    paths.insert(path);
637                }
638            }
639        }
640        let action_bytes = paths
641            .iter()
642            .filter_map(|path| std::fs::metadata(path).ok())
643            .map(|metadata| metadata.len())
644            .sum();
645        usages.push(ProjectUsage {
646            workspace_root,
647            identities: identities.len() as u64,
648            action_bytes,
649            target_bytes,
650            live,
651        });
652    }
653    usages.sort_by(|left, right| {
654        right
655            .action_bytes
656            .saturating_add(right.target_bytes)
657            .cmp(&left.action_bytes.saturating_add(left.target_bytes))
658            .then_with(|| left.workspace_root.cmp(&right.workspace_root))
659    });
660    Ok(usages)
661}
662
663/// Return the largest blobs and action-result records in descending order.
664pub fn largest(store: &Path, limit: usize) -> Result<Vec<LargestEntry>> {
665    let mut entries = Vec::new();
666    for (kind, root, action_result) in [
667        ("object", store.join(CAS_DIR), false),
668        ("action result", store.join(ACTION_RESULTS_DIR), true),
669    ] {
670        entries.extend(
671            walk_files(&root)?
672                .into_iter()
673                .filter(|entry| addressed_digest(store, &entry.path, action_result).is_some())
674                .map(|entry| LargestEntry {
675                    kind,
676                    path: entry.path,
677                    bytes: entry.size,
678                }),
679        );
680    }
681    entries.sort_by(|left, right| {
682        right
683            .bytes
684            .cmp(&left.bytes)
685            .then_with(|| left.path.cmp(&right.path))
686    });
687    entries.truncate(limit);
688    Ok(entries)
689}
690
691/// Verify every addressable CAS object and action-result record.
692pub fn verify(store: &Path) -> Result<VerifyOutcome> {
693    let cas = LocalCas::new(store);
694    let action_cache = mbx_cache_core::LocalActionCache::new(store);
695    // Verification reports the store as it is on disk and removes nothing, so
696    // no object is pending removal the way one is mid-sweep.
697    let virtually_removed = HashSet::new();
698    let mut outcome = VerifyOutcome::default();
699    for entry in walk_files(&store.join(CAS_DIR))? {
700        let Some(digest) = addressed_digest(store, &entry.path, false) else {
701            continue;
702        };
703        outcome.checked_objects += 1;
704        if !matches!(cas.find(&digest), Ok(Some(_))) {
705            outcome.problems.push(entry.path);
706        }
707    }
708    for entry in walk_files(&store.join(ACTION_RESULTS_DIR))? {
709        let Some(digest) = addressed_digest(store, &entry.path, true) else {
710            continue;
711        };
712        outcome.checked_action_results += 1;
713        if !matches!(action_cache.find(&digest), Ok(Some(_)))
714            || action_result_is_dangling(&cas, &entry.path, &virtually_removed)?
715        {
716            outcome.problems.push(entry.path);
717        }
718    }
719    outcome.problems.sort();
720    Ok(outcome)
721}
722
723/// Remove checkout claims belonging to exactly one workspace.
724pub fn remove_project(store: &Path, workspace_root: &Path) -> Result<RemoveProjectOutcome> {
725    let mut outcome = RemoveProjectOutcome::default();
726    for entry in walk_files(&store.join(CHECKOUTS_DIR))? {
727        if read_checkout_record(&entry.path)
728            .is_some_and(|record| record.workspace_root == workspace_root)
729        {
730            std::fs::remove_file(&entry.path)?;
731            outcome.removed_checkout_records += 1;
732            if let Some(parent) = entry.path.parent() {
733                let _ = std::fs::remove_dir(parent);
734            }
735        }
736    }
737    match std::fs::remove_file(latest_receipt_path(store, workspace_root)) {
738        Ok(()) => {}
739        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
740        Err(error) => return Err(error.into()),
741    }
742    Ok(outcome)
743}
744
745/// Record that `workspace_root` built `identity`.
746///
747/// Identities are shared by every checkout of one dependency graph, so this is
748/// what later tells the collector whether anything still needs what a build
749/// cached: one file per checkout under the identity it built. Writing a whole
750/// file per checkout rather than merging a list into one keeps concurrent
751/// builds out of each other's way -- there is nothing to merge, so there is no
752/// lock and no lost update.
753///
754/// A build with no target directory of its own passes `workspace_root` for
755/// `target_dir`, which reads as "none" rather than as a tree to measure.
756pub fn record_checkout(
757    store: &Path,
758    identity: &str,
759    workspace_root: &Path,
760    target_dir: &Path,
761) -> Result<()> {
762    let record = CheckoutRecord {
763        version: CHECKOUT_RECORD_VERSION,
764        workspace_root: workspace_root.to_path_buf(),
765        target_dir: target_dir.to_path_buf(),
766        updated_secs: SystemTime::now()
767            .duration_since(UNIX_EPOCH)
768            .map(|since| since.as_secs())
769            .unwrap_or_default(),
770    };
771    let mut contents = serde_json::to_vec(&record)?;
772    contents.push(b'\n');
773    write_atomic(
774        &checkout_record_path(store, identity, workspace_root),
775        &contents,
776    )
777}
778
779fn checkout_record_path(store: &Path, identity: &str, workspace_root: &Path) -> PathBuf {
780    // The path is hashed rather than escaped: it is only ever compared against
781    // the same hash of the same path, and every filesystem in play disagrees
782    // about which characters a name may hold.
783    let key = CacheDigest::blake3(workspace_root.to_string_lossy().as_bytes()).hash;
784    store
785        .join(CHECKOUTS_DIR)
786        .join(identity)
787        .join(format!("{key}.json"))
788}
789
790/// Evict objects until the store fits within `max_bytes`.
791///
792/// Objects no live checkout can reach are evicted first, so deleting a worktree
793/// releases what only that worktree needed. Within each class eviction is
794/// least-recently-used by access time where the filesystem records it, falling
795/// back to modification time. A restore verifies the blob it serves, which
796/// means it reads the whole file and the access time is real -- but only where
797/// the mount records one, so the ordering is an approximation either way.
798/// Evicting a live object costs a recompile, never correctness.
799pub fn gc(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
800    gc_with_mode(store, max_bytes, false)
801}
802
803/// Describe the collection `gc` would perform without changing the store.
804pub fn gc_dry_run(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
805    gc_with_mode(store, max_bytes, true)
806}
807
808fn gc_with_mode(store: &Path, max_bytes: u64, dry_run: bool) -> Result<GcOutcome> {
809    let mut objects = walk_files(&store.join(CAS_DIR))?;
810    let results = walk_files(&store.join(ACTION_RESULTS_DIR))?;
811    let mut live_bytes = objects
812        .iter()
813        .chain(results.iter())
814        .map(|entry| entry.size)
815        .sum::<u64>();
816
817    let mut outcome = GcOutcome::default();
818
819    // Prune before deciding what is rooted, so a checkout deleted since the
820    // last sweep stops protecting its artifacts during this one.
821    let checkouts = scan_checkouts(store)?;
822    for path in &checkouts.stale_records {
823        if dry_run || matches!(remove(path)?, Removal::Removed) {
824            outcome.removed_checkout_records += 1;
825        }
826        // The identity directory is left behind empty otherwise. It may hold
827        // other checkouts, in which case this fails and that is the answer.
828        if !dry_run && let Some(parent) = path.parent() {
829            let _ = std::fs::remove_dir(parent);
830        }
831    }
832
833    let mut virtually_removed = HashSet::new();
834    if live_bytes > max_bytes {
835        // Reachability costs a read per action result and per output tree, so
836        // it is computed only when something is actually about to be evicted.
837        // Under budget a sweep stays a directory walk and nothing more.
838        let mut rooted = rooted_objects(store, &checkouts.live_identities)?;
839        // A grouped CI export deliberately retains the exact actions from
840        // every command in the job. Later commands can replace the current
841        // task-manifest prediction for the same invocation, but the earlier
842        // receipt still has to remain exportable until the post-job step runs.
843        // Receipt bookkeeping is not part of the cache budget, so consult it
844        // only when a sweep is already necessary and root its object closure.
845        rooted.extend(rooted_action_objects(
846            store,
847            grouped_receipt_actions(store)?,
848        ));
849        // `false` sorts first, so unrooted objects go before rooted ones and
850        // each class goes oldest-first. A store with no records at all roots
851        // nothing and this is exactly the LRU it was before.
852        objects.sort_by_cached_key(|entry| (rooted.contains(&entry.path), entry.used));
853        let evicted = evict_objects(&objects, &rooted, live_bytes, max_bytes, |path| {
854            if dry_run {
855                virtually_removed.insert(path.to_path_buf());
856                Ok(Removal::Removed)
857            } else {
858                remove(path)
859            }
860        })?;
861        live_bytes = evicted.remaining_bytes;
862        outcome.removed_objects += evicted.removed_objects;
863        outcome.removed_bytes += evicted.removed_bytes;
864    }
865
866    // An action result whose objects are gone can only produce a miss, so drop
867    // it rather than leave the index pointing at nothing. This runs whether or
868    // not this call evicted anything, since another process may have, and a
869    // store that is over budget on results alone still needs the sweep.
870    let cas = LocalCas::new(store);
871    for entry in &results {
872        if action_result_is_dangling(&cas, &entry.path, &virtually_removed)? {
873            match if dry_run {
874                Removal::Removed
875            } else {
876                remove(&entry.path)?
877            } {
878                Removal::Removed => {
879                    live_bytes = live_bytes.saturating_sub(entry.size);
880                    outcome.removed_action_results += 1;
881                    outcome.removed_bytes += entry.size;
882                }
883                Removal::Missing => live_bytes = live_bytes.saturating_sub(entry.size),
884                Removal::Blocked => {}
885            }
886        }
887    }
888
889    let sessions = prune_sessions(store, dry_run)?;
890    outcome.removed_session_streams += sessions.removed_streams;
891    outcome.removed_bytes += sessions.removed_bytes;
892
893    outcome.remaining_bytes = live_bytes;
894    Ok(outcome)
895}
896
897#[derive(Debug, Default, PartialEq, Eq)]
898struct SessionPrune {
899    removed_streams: u64,
900    removed_bytes: u64,
901}
902
903fn prune_sessions(store: &Path, dry_run: bool) -> Result<SessionPrune> {
904    let mut prune = SessionPrune::default();
905    let ids = events::session_ids(store);
906    let surplus = ids.len().saturating_sub(MAX_SESSIONS);
907    for (index, id) in ids.iter().enumerate() {
908        let paths = events::session_paths(store, id);
909        let metadata = match std::fs::metadata(&paths.events) {
910            Ok(metadata) => metadata,
911            Err(_) => continue,
912        };
913        let stale = metadata
914            .modified()
915            .ok()
916            .and_then(|at| SystemTime::now().duration_since(at).ok())
917            .is_some_and(|age| age > SESSION_RETENTION);
918        if index >= surplus && !stale {
919            continue;
920        }
921        if events::session_is_live(store, id) {
922            continue;
923        }
924        if dry_run {
925            prune.removed_streams += 1;
926            prune.removed_bytes += metadata.len();
927            continue;
928        }
929        if matches!(remove(&paths.events)?, Removal::Removed) {
930            prune.removed_streams += 1;
931            prune.removed_bytes += metadata.len();
932            let _ = std::fs::remove_file(&paths.lock);
933        }
934    }
935    if !dry_run {
936        for lock in events::orphaned_locks(store) {
937            let _ = std::fs::remove_file(lock);
938        }
939    }
940    Ok(prune)
941}
942
943/// Sweep the store if `interval` has passed since the last attempt.
944///
945/// The stamp is written before the sweep, not after. Two builds finishing
946/// together should cost one sweep between them, and a sweep that dies partway
947/// should wait its turn like any other rather than retry on every build.
948pub fn sweep_if_due(store: &Path, max_bytes: u64, interval: Duration) -> Result<Option<GcOutcome>> {
949    if !claim_sweep(store, interval)? {
950        return Ok(None);
951    }
952    gc(store, max_bytes).map(Some)
953}
954
955/// Atomically stamp a due sweep before its callers perform coordinated GC.
956pub fn claim_sweep(store: &Path, interval: Duration) -> Result<bool> {
957    let lock_path = store.join(SWEEP_LOCK);
958    std::fs::create_dir_all(lock_path.parent().expect("sweep lock has a parent"))?;
959    let mut lock = fslock::LockFile::open(&lock_path)?;
960    lock.lock()?;
961
962    let stamp = store.join(SWEEP_STAMP);
963    if let Ok(metadata) = std::fs::metadata(&stamp)
964        && let Ok(modified) = metadata.modified()
965        && let Ok(since) = modified.elapsed()
966        && since < interval
967    {
968        return Ok(false);
969    }
970    write_atomic(&stamp, b"")?;
971    Ok(true)
972}
973
974/// What the checkout registry says about the identities in this store.
975struct CheckoutScan {
976    live_identities: BTreeSet<String>,
977    live_records: u64,
978    stale_records: Vec<PathBuf>,
979}
980
981fn scan_checkouts(store: &Path) -> Result<CheckoutScan> {
982    let root = store.join(CHECKOUTS_DIR);
983    let mut scan = CheckoutScan {
984        live_identities: BTreeSet::new(),
985        live_records: 0,
986        stale_records: Vec::new(),
987    };
988    let listing = match std::fs::read_dir(&root) {
989        Ok(listing) => listing,
990        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(scan),
991        Err(error) => {
992            return Err(error).wrap_err_with(|| format!("failed to read {}", root.display()));
993        }
994    };
995    for entry in listing {
996        let entry = entry?;
997        let name = entry.file_name().to_string_lossy().into_owned();
998        // Anything that is not a directory named for an identity was not written
999        // by mbx. Both halves matter: `walk_files` tolerates only a missing
1000        // directory, so a plain file whose name happens to look like an identity
1001        // would fail the whole scan -- and that scan runs inside every sweep,
1002        // including the automatic one after a build.
1003        if !is_task_identity(&name) || !entry.file_type().is_ok_and(|kind| kind.is_dir()) {
1004            continue;
1005        }
1006        for record in walk_files(&entry.path())? {
1007            match read_checkout_record(&record.path) {
1008                Some(checkout) if claim_is_live(store, &checkout) => {
1009                    scan.live_identities.insert(name.clone());
1010                    scan.live_records += 1;
1011                }
1012                Some(_) => scan.stale_records.push(record.path),
1013                // A record this build cannot read claims nothing, but it is not
1014                // evidence the checkout is gone either, so leave it alone.
1015                None => {}
1016            }
1017        }
1018    }
1019    Ok(scan)
1020}
1021
1022fn read_checkout_record(path: &Path) -> Option<CheckoutRecord> {
1023    let bytes = std::fs::read(path).ok()?;
1024    let record = serde_json::from_slice::<CheckoutRecord>(&bytes).ok()?;
1025    (record.version == CHECKOUT_RECORD_VERSION).then_some(record)
1026}
1027
1028/// Actions whose successful grouped export has not retired its receipts yet.
1029fn grouped_receipt_actions(store: &Path) -> Result<BTreeSet<CacheDigest>> {
1030    let root = store.join(BUILD_RECEIPTS_DIR).join("groups");
1031    Ok(walk_files(&root)?
1032        .into_iter()
1033        .filter_map(|entry| read_build_receipt(&entry.path))
1034        .flat_map(|receipt| receipt.predictions.into_iter())
1035        .map(|prediction| prediction.action)
1036        .collect())
1037}
1038
1039/// Whether a recorded claim still speaks for a checkout that is using this
1040/// store.
1041fn claim_is_live(store: &Path, record: &CheckoutRecord) -> bool {
1042    checkout_is_live_on(store, &record.workspace_root) && !claim_has_expired(record)
1043}
1044
1045fn claim_has_expired(record: &CheckoutRecord) -> bool {
1046    let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) else {
1047        return false;
1048    };
1049    now.as_secs()
1050        .saturating_sub(record.updated_secs)
1051        .gt(&CHECKOUT_RETENTION.as_secs())
1052}
1053
1054/// Whether the checkout a record names is still on disk.
1055///
1056/// Absence is believed only when the checkout is definitely absent and its
1057/// nearest existing ancestor is on the same filesystem as the store. Walking
1058/// all the way to that ancestor matters for worktree managers and temporary
1059/// directories, which commonly remove a checkout together with one or more of
1060/// its otherwise-empty parents.
1061///
1062/// A different filesystem can be a mount whose contents are temporarily
1063/// unavailable, so uncertainty there remains live. Errors are treated the same
1064/// way: being wrong in that direction only delays collection; the other way
1065/// round throws away a warm cache someone is still using.
1066pub fn checkout_is_live_on(store: &Path, workspace_root: &Path) -> bool {
1067    if !matches!(workspace_root.try_exists(), Ok(false)) {
1068        return true;
1069    }
1070
1071    let Ok(store_metadata) = std::fs::metadata(store) else {
1072        return true;
1073    };
1074    for ancestor in workspace_root.ancestors().skip(1) {
1075        match std::fs::metadata(ancestor) {
1076            Ok(ancestor_metadata) => {
1077                return !same_filesystem(store, &store_metadata, ancestor, &ancestor_metadata);
1078            }
1079            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
1080            Err(_) => return true,
1081        }
1082    }
1083    true
1084}
1085
1086/// Whether a checkout is still on disk, corroborating absence through its
1087/// immediate parent.
1088///
1089/// Prefer [`checkout_is_live_on`] when an existing store or managed-target
1090/// root is available to distinguish deletion from an unavailable filesystem.
1091pub fn checkout_is_live(workspace_root: &Path) -> bool {
1092    let Some(parent) = workspace_root.parent() else {
1093        return true;
1094    };
1095    checkout_is_live_on(parent, workspace_root)
1096}
1097
1098#[cfg(unix)]
1099fn same_filesystem(
1100    _a_path: &Path,
1101    a: &std::fs::Metadata,
1102    _b_path: &Path,
1103    b: &std::fs::Metadata,
1104) -> bool {
1105    use std::os::unix::fs::MetadataExt as _;
1106    a.dev() == b.dev()
1107}
1108
1109#[cfg(windows)]
1110fn same_filesystem(
1111    a_path: &Path,
1112    _a: &std::fs::Metadata,
1113    b_path: &Path,
1114    _b: &std::fs::Metadata,
1115) -> bool {
1116    windows_volume_name(a_path).is_some_and(|volume| windows_volume_name(b_path) == Some(volume))
1117}
1118
1119#[cfg(windows)]
1120fn windows_volume_name(path: &Path) -> Option<Vec<u16>> {
1121    use std::os::windows::ffi::OsStrExt as _;
1122    use windows_sys::Win32::Storage::FileSystem::{
1123        GetVolumeNameForVolumeMountPointW, GetVolumePathNameW,
1124    };
1125
1126    let path = path
1127        .as_os_str()
1128        .encode_wide()
1129        .chain(Some(0))
1130        .collect::<Vec<_>>();
1131    // Both APIs document MAX_PATH-sized output buffers for volume paths and
1132    // names. Failure is uncertainty, which deliberately keeps the checkout
1133    // live.
1134    let mut mount = vec![0_u16; 261];
1135    // SAFETY: `path` is nul-terminated and `mount` is writable for the stated
1136    // number of UTF-16 code units.
1137    if unsafe { GetVolumePathNameW(path.as_ptr(), mount.as_mut_ptr(), mount.len() as u32) } == 0 {
1138        return None;
1139    }
1140    let mut volume = vec![0_u16; 261];
1141    // SAFETY: the successful call above left `mount` nul-terminated, and
1142    // `volume` is writable for the stated number of UTF-16 code units.
1143    if unsafe {
1144        GetVolumeNameForVolumeMountPointW(mount.as_ptr(), volume.as_mut_ptr(), volume.len() as u32)
1145    } == 0
1146    {
1147        return None;
1148    }
1149    volume.truncate(volume.iter().position(|unit| *unit == 0)?);
1150    Some(volume)
1151}
1152
1153#[cfg(not(any(unix, windows)))]
1154fn same_filesystem(
1155    _a_path: &Path,
1156    _a: &std::fs::Metadata,
1157    _b_path: &Path,
1158    _b: &std::fs::Metadata,
1159) -> bool {
1160    false
1161}
1162
1163/// CAS paths reachable from the builds that live checkouts still depend on.
1164///
1165/// Recursing into output trees is the point: the descriptor blobs are tiny and
1166/// the leaf artifacts are where the bytes are, so rooting that stopped at the
1167/// top level would protect nothing worth protecting. Every read here is
1168/// tolerant -- a blob that has already been evicted, or that no longer parses,
1169/// simply roots less, and rooting less can only mean collecting sooner.
1170fn rooted_objects(store: &Path, identities: &BTreeSet<String>) -> Result<HashSet<PathBuf>> {
1171    let mut actions = BTreeSet::new();
1172    for identity in identities {
1173        // One manifest this build cannot read is not worth abandoning a sweep
1174        // over. It roots nothing, which can only mean collecting sooner.
1175        let manifest_actions = match task_manifest_actions(store, identity) {
1176            Ok(actions) => actions,
1177            Err(error) => {
1178                log::debug!("could not read the manifest for {identity}: {error}");
1179                continue;
1180            }
1181        };
1182        actions.extend(manifest_actions);
1183    }
1184    Ok(rooted_action_objects(store, actions))
1185}
1186
1187/// Return every local CAS path needed to restore the supplied actions.
1188fn rooted_action_objects(
1189    store: &Path,
1190    actions: impl IntoIterator<Item = CacheDigest>,
1191) -> HashSet<PathBuf> {
1192    let cas = LocalCas::new(store);
1193    let mut rooted = HashSet::new();
1194    let mut visited = BTreeSet::new();
1195    for action in actions {
1196        let Some(result) = read_action_result(store, &action) else {
1197            continue;
1198        };
1199        root_digest(&cas, &mut rooted, &result.action);
1200        if let Some(metadata) = &result.metadata {
1201            root_digest(&cas, &mut rooted, metadata);
1202            // The metadata blob is a descriptor too: it names the captured
1203            // stdout and stderr, and a restore reads both. Stopping at the
1204            // descriptor would leave the diagnostics of every action
1205            // unrooted -- including the one empty blob that every silent
1206            // compilation shares, which is enough on its own to turn a
1207            // rooted hit back into a miss.
1208            if let Some(rustc) = read_rustc_metadata(&cas, metadata) {
1209                root_digest(&cas, &mut rooted, &rustc.stdout);
1210                root_digest(&cas, &mut rooted, &rustc.stderr);
1211            }
1212        }
1213        if let Some(output_root) = &result.output_root {
1214            root_digest(&cas, &mut rooted, output_root);
1215            root_tree(&cas, output_root, &mut rooted, &mut visited);
1216        }
1217    }
1218    rooted
1219}
1220
1221fn read_action_result(store: &Path, action: &CacheDigest) -> Option<RemoteActionResult> {
1222    let path = mbx_cache_core::LocalActionCache::new(store)
1223        .path_for(action)
1224        .ok()?;
1225    let bytes = std::fs::read(path).ok()?;
1226    serde_json::from_slice(&bytes).ok()
1227}
1228
1229fn root_tree(
1230    cas: &LocalCas,
1231    output_root: &CacheDigest,
1232    rooted: &mut HashSet<PathBuf>,
1233    visited: &mut BTreeSet<CacheDigest>,
1234) {
1235    let mut pending = vec![output_root.clone()];
1236    while let Some(digest) = pending.pop() {
1237        if !visited.insert(digest.clone()) {
1238            continue;
1239        }
1240        let Some(directory) = read_directory(cas, &digest) else {
1241            continue;
1242        };
1243        for file in &directory.files {
1244            root_digest(cas, rooted, &file.digest);
1245        }
1246        for child in &directory.directories {
1247            root_digest(cas, rooted, &child.digest);
1248            pending.push(child.digest.clone());
1249        }
1250    }
1251}
1252
1253fn read_rustc_metadata(cas: &LocalCas, digest: &CacheDigest) -> Option<RustcMetadata> {
1254    let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1255    serde_json::from_slice(&bytes).ok()
1256}
1257
1258fn read_directory(cas: &LocalCas, digest: &CacheDigest) -> Option<CacheDirectory> {
1259    let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1260    serde_json::from_slice(&bytes).ok()
1261}
1262
1263fn root_digest(cas: &LocalCas, rooted: &mut HashSet<PathBuf>, digest: &CacheDigest) {
1264    if let Ok(path) = cas.path_for(digest) {
1265        rooted.insert(path);
1266    }
1267}
1268
1269#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1270enum Removal {
1271    Removed,
1272    Missing,
1273    Blocked,
1274}
1275
1276fn remove(path: &Path) -> Result<Removal> {
1277    match std::fs::remove_file(path) {
1278        Ok(()) => Ok(Removal::Removed),
1279        // A concurrent build may have evicted or replaced it already.
1280        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Removal::Missing),
1281        // Windows refuses to unlink a file another process holds open, which is
1282        // what a concurrent build reading this blob looks like. Skipping it
1283        // leaves the store over budget until the next sweep; failing would
1284        // abandon the sweep and leave it over budget for longer.
1285        Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
1286            log::debug!("could not evict {}: {error}", path.display());
1287            Ok(Removal::Blocked)
1288        }
1289        Err(error) => Err(error).wrap_err_with(|| format!("failed to evict {}", path.display())),
1290    }
1291}
1292
1293#[derive(Debug, Default, PartialEq, Eq)]
1294struct ObjectEvictions {
1295    removed_objects: u64,
1296    removed_bytes: u64,
1297    remaining_bytes: u64,
1298}
1299
1300/// Evict sorted objects without crossing the checkout-protection boundary
1301/// when an unrooted object cannot be removed.
1302fn evict_objects(
1303    objects: &[Entry],
1304    rooted: &HashSet<PathBuf>,
1305    mut live_bytes: u64,
1306    max_bytes: u64,
1307    mut remove_file: impl FnMut(&Path) -> Result<Removal>,
1308) -> Result<ObjectEvictions> {
1309    let mut outcome = ObjectEvictions::default();
1310    let mut blocked_unrooted = false;
1311    for entry in objects {
1312        if live_bytes <= max_bytes {
1313            break;
1314        }
1315        let is_rooted = rooted.contains(&entry.path);
1316        if is_rooted && blocked_unrooted {
1317            // A locked unrooted blob is temporarily part of the irreducible
1318            // store. Deleting a live checkout's artifacts in its place would
1319            // invert the protection ordering this collector promises.
1320            break;
1321        }
1322        match remove_file(&entry.path)? {
1323            Removal::Removed => {
1324                live_bytes = live_bytes.saturating_sub(entry.size);
1325                outcome.removed_objects += 1;
1326                outcome.removed_bytes += entry.size;
1327            }
1328            Removal::Missing => live_bytes = live_bytes.saturating_sub(entry.size),
1329            Removal::Blocked if !is_rooted => blocked_unrooted = true,
1330            Removal::Blocked => {}
1331        }
1332    }
1333    outcome.remaining_bytes = live_bytes;
1334    Ok(outcome)
1335}
1336
1337/// Whether an action result references an object the store no longer has.
1338///
1339/// The digests checked here are exactly the ones `LocalActionCache::store`
1340/// requires. If the sweep checked fewer, eviction could leave an entry that
1341/// cannot be republished -- `store` would reject the identical result for a
1342/// missing blob while the index still claimed to hold it.
1343///
1344/// Presence is all that is checked, not content. Verifying would re-hash a
1345/// large fraction of the store on every sweep, which a chore could afford and
1346/// an automatic sweep cannot; it would also refresh the access time of every
1347/// descriptor blob while leaving the artifacts alone, biasing the very
1348/// ordering this module depends on. A blob that is present but corrupt now
1349/// keeps its result and turns into a miss on restore, and both CAS write paths
1350/// republish over a corrupt blob rather than trusting it.
1351///
1352/// Only the top-level objects are checked; a result whose output tree lost a
1353/// nested object still restores as a miss, which is safe.
1354fn action_result_is_dangling(
1355    cas: &LocalCas,
1356    path: &Path,
1357    virtually_removed: &HashSet<PathBuf>,
1358) -> Result<bool> {
1359    let bytes = match std::fs::read(path) {
1360        Ok(bytes) => bytes,
1361        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
1362        Err(error) => {
1363            return Err(error).wrap_err_with(|| format!("failed to read {}", path.display()));
1364        }
1365    };
1366    // An unparseable result is already useless; the cache rejects it on read.
1367    let Ok(result) = serde_json::from_slice::<RemoteActionResult>(&bytes) else {
1368        return Ok(false);
1369    };
1370    for digest in [
1371        Some(&result.action),
1372        result.metadata.as_ref(),
1373        result.output_root.as_ref(),
1374    ]
1375    .into_iter()
1376    .flatten()
1377    {
1378        match cas.path_for(digest) {
1379            Ok(path) => {
1380                if virtually_removed.contains(&path) || !path.exists() {
1381                    return Ok(true);
1382                }
1383            }
1384            // A digest the CAS cannot even address is not one it can hold.
1385            Err(_) => return Ok(true),
1386        }
1387    }
1388    Ok(false)
1389}
1390
1391struct Entry {
1392    path: PathBuf,
1393    size: u64,
1394    used: SystemTime,
1395}
1396
1397fn walk_files(root: &Path) -> Result<Vec<Entry>> {
1398    let mut entries = Vec::new();
1399    let mut pending = vec![root.to_path_buf()];
1400    while let Some(directory) = pending.pop() {
1401        let listing = match std::fs::read_dir(&directory) {
1402            Ok(listing) => listing,
1403            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1404            Err(error) => {
1405                return Err(error)
1406                    .wrap_err_with(|| format!("failed to read {}", directory.display()));
1407            }
1408        };
1409        for entry in listing {
1410            let entry = entry?;
1411            let metadata = match entry.metadata() {
1412                Ok(metadata) => metadata,
1413                // A concurrent build may be publishing into the store.
1414                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1415                Err(error) => return Err(error.into()),
1416            };
1417            if metadata.is_dir() {
1418                pending.push(entry.path());
1419            } else if metadata.is_file() {
1420                let used = metadata
1421                    .accessed()
1422                    .or_else(|_| metadata.modified())
1423                    .unwrap_or(SystemTime::UNIX_EPOCH);
1424                entries.push(Entry {
1425                    path: entry.path(),
1426                    size: metadata.len(),
1427                    used,
1428                });
1429            }
1430        }
1431    }
1432    Ok(entries)
1433}
1434
1435fn read_dir_or_empty(root: &Path) -> Result<Vec<std::fs::DirEntry>> {
1436    match std::fs::read_dir(root) {
1437        Ok(entries) => entries
1438            .collect::<std::io::Result<Vec<_>>>()
1439            .map_err(Into::into),
1440        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
1441        Err(error) => Err(error).wrap_err_with(|| format!("failed to read {}", root.display())),
1442    }
1443}
1444
1445fn digest_from_path(path: &Path, action_result: bool) -> Option<CacheDigest> {
1446    let name = path.file_name()?.to_str()?;
1447    let name = if action_result {
1448        name.strip_suffix(".json")?
1449    } else {
1450        name
1451    };
1452    let (hash, size) = name.rsplit_once('-')?;
1453    let algorithm = path.parent()?.parent()?.file_name()?.to_str()?.to_string();
1454    let digest = CacheDigest {
1455        algorithm,
1456        hash: hash.to_string(),
1457        size: size.parse().ok()?,
1458    };
1459    digest.validate().ok()?;
1460    Some(digest)
1461}
1462
1463fn addressed_digest(store: &Path, path: &Path, action_result: bool) -> Option<CacheDigest> {
1464    let digest = digest_from_path(path, action_result)?;
1465    let canonical = if action_result {
1466        mbx_cache_core::LocalActionCache::new(store)
1467            .path_for(&digest)
1468            .ok()?
1469    } else {
1470        LocalCas::new(store).path_for(&digest).ok()?
1471    };
1472    (canonical == path).then_some(digest)
1473}
1474
1475fn cached_tree_bytes(cache: &mut BTreeMap<PathBuf, u64>, root: &Path) -> u64 {
1476    *cache
1477        .entry(root.to_path_buf())
1478        .or_insert_with(|| tree_bytes(root))
1479}
1480
1481fn tree_bytes(root: &Path) -> u64 {
1482    let Ok(root_metadata) = std::fs::metadata(root) else {
1483        return 0;
1484    };
1485    if root_metadata.is_file() {
1486        return root_metadata.len();
1487    }
1488    let Ok(entries) = std::fs::read_dir(root) else {
1489        return 0;
1490    };
1491    let mut bytes = 0_u64;
1492    let mut pending = entries
1493        .filter_map(std::result::Result::ok)
1494        .map(|entry| entry.path())
1495        .collect::<Vec<_>>();
1496    while let Some(path) = pending.pop() {
1497        let Ok(metadata) = std::fs::symlink_metadata(&path) else {
1498            continue;
1499        };
1500        if metadata.is_file() {
1501            bytes = bytes.saturating_add(metadata.len());
1502        } else if metadata.is_dir()
1503            && let Ok(entries) = std::fs::read_dir(path)
1504        {
1505            pending.extend(
1506                entries
1507                    .filter_map(std::result::Result::ok)
1508                    .map(|entry| entry.path()),
1509            );
1510        }
1511    }
1512    bytes
1513}
1514
1515#[cfg(test)]
1516#[path = "store_tests.rs"]
1517mod tests;
1518
1519fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> {
1520    let parent = path
1521        .parent()
1522        .filter(|parent| !parent.as_os_str().is_empty())
1523        .ok_or_else(|| eyre::eyre!("path has no parent: {}", path.display()))?;
1524    std::fs::create_dir_all(parent)
1525        .wrap_err_with(|| format!("failed to create {}", parent.display()))?;
1526    let mut temporary = tempfile::Builder::new()
1527        .prefix(".mbx-")
1528        .tempfile_in(parent)?;
1529    use std::io::Write as _;
1530    temporary.write_all(contents)?;
1531    temporary.as_file_mut().sync_all()?;
1532    temporary
1533        .persist(path)
1534        .map_err(|error| error.error)
1535        .wrap_err_with(|| format!("failed atomic write: {}", path.display()))?;
1536    Ok(())
1537}