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))
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    export_receipts(store, receipts, archive)
264}
265
266fn export_receipts(
267    store: &Path,
268    mut receipts: Vec<BuildReceipt>,
269    archive: &Path,
270) -> Result<TransferOutcome> {
271    receipts.sort_by_key(|receipt| receipt.completed_nanos);
272    let mut actions = BTreeSet::new();
273    let mut tasks = BTreeMap::new();
274    for receipt in receipts {
275        actions.extend(
276            receipt
277                .predictions
278                .iter()
279                .map(|prediction| prediction.action.clone()),
280        );
281        // One task manifest can carry only one action per invocation. The
282        // newest run of the same task is the useful prediction set to import;
283        // every older action remains in `actions` and therefore in the bundle.
284        tasks.insert(
285            receipt.identity.clone(),
286            TaskActionManifest {
287                version: 1,
288                task: receipt.identity,
289                predictions: receipt.predictions,
290            },
291        );
292    }
293    let tasks = tasks.into_values().collect::<Vec<_>>();
294    let (objects, result_paths) = strict_closure(store, &actions)?;
295    let parent = archive
296        .parent()
297        .filter(|parent| !parent.as_os_str().is_empty())
298        .unwrap_or_else(|| Path::new("."));
299    std::fs::create_dir_all(parent)?;
300    let temporary = tempfile::Builder::new()
301        .prefix(".mbx-export-")
302        .tempfile_in(parent)?;
303    let mut builder = tar::Builder::new(temporary.reopen()?);
304    append_bytes(
305        &mut builder,
306        Path::new(EXPORT_MANIFEST),
307        &serde_json::to_vec(&ExportManifest {
308            version: EXPORT_VERSION,
309            tasks,
310            actions: actions.iter().cloned().collect(),
311        })?,
312    )?;
313    for path in objects.iter().chain(result_paths.iter()) {
314        append_file(&mut builder, store, path)?;
315    }
316    builder.finish()?;
317    drop(builder);
318    temporary
319        .persist(archive)
320        .map_err(|error| error.error)
321        .wrap_err_with(|| format!("failed to publish {}", archive.display()))?;
322    let bytes = std::fs::metadata(archive)?.len();
323    Ok(TransferOutcome {
324        actions: actions.len() as u64,
325        objects: objects.len() as u64,
326        bytes,
327    })
328}
329
330/// Validate a cache export in isolation, then publish its objects and actions.
331pub fn import_archive(store: &Path, archive: &Path) -> Result<TransferOutcome> {
332    let staging = tempfile::tempdir()?;
333    let file = std::fs::File::open(archive)
334        .wrap_err_with(|| format!("failed to open {}", archive.display()))?;
335    let mut bundle = tar::Archive::new(file);
336    let mut seen = BTreeSet::new();
337    for entry in bundle.entries()? {
338        let mut entry = entry?;
339        let entry_type = entry.header().entry_type();
340        if !entry_type.is_file() && !entry_type.is_gnu_sparse() {
341            eyre::bail!("cache export contains a non-file entry");
342        }
343        let path = entry.path()?.into_owned();
344        validate_archive_path(&path)?;
345        if !seen.insert(path.clone()) {
346            eyre::bail!("cache export contains duplicate entry {}", path.display());
347        }
348        let destination = staging.path().join(&path);
349        std::fs::create_dir_all(destination.parent().expect("entry has a parent"))?;
350        // `unpack` understands GNU sparse maps. A plain stream copy expands
351        // holes into physical zeroes, which is both slower and much larger for
352        // Rust artifacts containing sparse sections.
353        entry.unpack(&destination)?;
354    }
355    let manifest: ExportManifest =
356        serde_json::from_slice(&std::fs::read(staging.path().join(EXPORT_MANIFEST))?)?;
357    let actions = manifest.actions.iter().cloned().collect::<BTreeSet<_>>();
358    let task_identities = manifest
359        .tasks
360        .iter()
361        .map(|task| task.task.as_str())
362        .collect::<BTreeSet<_>>();
363    if manifest.version != EXPORT_VERSION
364        || manifest.tasks.is_empty()
365        || actions.len() != manifest.actions.len()
366        || task_identities.len() != manifest.tasks.len()
367        || manifest.tasks.iter().any(|task| !task.validate())
368        || manifest
369            .actions
370            .iter()
371            .any(|action| action.algorithm != "blake3" || action.validate().is_err())
372        || manifest.tasks.iter().any(|task| {
373            task.predictions
374                .iter()
375                .any(|prediction| !actions.contains(&prediction.action))
376        })
377    {
378        eyre::bail!("unsupported or invalid cache export manifest");
379    }
380    let (objects, result_paths) = strict_closure(staging.path(), &actions)
381        .wrap_err("cache export is incomplete or corrupt")?;
382
383    let cas = LocalCas::new(store);
384    for path in &objects {
385        let relative = path.strip_prefix(staging.path())?;
386        let source = staging.path().join(relative);
387        let digest = addressed_digest(staging.path(), &source, false)
388            .ok_or_else(|| eyre::eyre!("invalid cache object path {}", relative.display()))?;
389        cas.store_file(&digest, &source)?;
390    }
391    let action_cache = mbx_cache_core::LocalActionCache::new(store);
392    for path in &result_paths {
393        let result: RemoteActionResult = serde_json::from_slice(&std::fs::read(path)?)?;
394        action_cache.store(&result)?;
395    }
396    for task in manifest.tasks {
397        merge_imported_manifest(store, task)?;
398    }
399    Ok(TransferOutcome {
400        actions: actions.len() as u64,
401        objects: objects.len() as u64,
402        bytes: std::fs::metadata(archive)?.len(),
403    })
404}
405
406fn strict_closure(
407    store: &Path,
408    actions: &BTreeSet<CacheDigest>,
409) -> Result<(BTreeSet<PathBuf>, BTreeSet<PathBuf>)> {
410    let cas = LocalCas::new(store);
411    let action_cache = mbx_cache_core::LocalActionCache::new(store);
412    let mut objects = BTreeSet::new();
413    let mut results = BTreeSet::new();
414    let mut directories = BTreeSet::new();
415    for action in actions {
416        let result = action_cache
417            .find(action)?
418            .ok_or_else(|| eyre::eyre!("action result is missing for {}", action.hash))?;
419        results.insert(action_cache.path_for(action)?);
420        require_object(&cas, &mut objects, &result.action)?;
421        if let Some(metadata) = &result.metadata {
422            require_object(&cas, &mut objects, metadata)?;
423            let captured: CapturedOutput =
424                serde_json::from_slice(&std::fs::read(cas.path_for(metadata)?)?)
425                    .wrap_err("action metadata is invalid")?;
426            require_object(&cas, &mut objects, &captured.stdout)?;
427            require_object(&cas, &mut objects, &captured.stderr)?;
428        }
429        if let Some(root) = &result.output_root {
430            require_object(&cas, &mut objects, root)?;
431            let mut pending = vec![root.clone()];
432            while let Some(digest) = pending.pop() {
433                if !directories.insert(digest.clone()) {
434                    continue;
435                }
436                let directory: CacheDirectory =
437                    serde_json::from_slice(&std::fs::read(cas.path_for(&digest)?)?)
438                        .wrap_err("output directory is invalid")?;
439                for file in directory.files {
440                    require_object(&cas, &mut objects, &file.digest)?;
441                }
442                for child in directory.directories {
443                    require_object(&cas, &mut objects, &child.digest)?;
444                    pending.push(child.digest);
445                }
446            }
447        }
448    }
449    Ok((objects, results))
450}
451
452#[derive(Deserialize)]
453struct CapturedOutput {
454    stdout: CacheDigest,
455    stderr: CacheDigest,
456}
457
458fn merge_imported_manifest(destination: &Path, mut imported: TaskActionManifest) -> Result<()> {
459    let identity = imported.task.clone();
460    let destination_path = task_manifest_path(destination, &identity);
461    if let Ok(bytes) = std::fs::read(&destination_path)
462        && let Ok(existing) = serde_json::from_slice::<TaskActionManifest>(&bytes)
463        && existing.task == identity
464        && existing.validate()
465    {
466        let mut predictions = imported
467            .predictions
468            .into_iter()
469            .map(|prediction| (prediction.invocation.clone(), prediction))
470            .collect::<BTreeMap<_, _>>();
471        predictions.extend(
472            existing
473                .predictions
474                .iter()
475                .cloned()
476                .map(|prediction| (prediction.invocation.clone(), prediction)),
477        );
478        imported.predictions = predictions.into_values().collect();
479        if !imported.validate() {
480            return Ok(());
481        }
482    }
483    write_atomic(&destination_path, &serde_json::to_vec(&imported)?)
484}
485
486fn require_object(
487    cas: &LocalCas,
488    objects: &mut BTreeSet<PathBuf>,
489    digest: &CacheDigest,
490) -> Result<()> {
491    let path = cas
492        .find(digest)?
493        .ok_or_else(|| eyre::eyre!("cache object is missing for {}", digest.hash))?;
494    objects.insert(path);
495    Ok(())
496}
497
498fn task_manifest_path(store: &Path, identity: &str) -> PathBuf {
499    store
500        .join("task-manifests/v1")
501        .join(format!("{identity}.json"))
502}
503
504fn append_file(builder: &mut tar::Builder<std::fs::File>, root: &Path, path: &Path) -> Result<()> {
505    let name = path.strip_prefix(root)?;
506    builder.append_path_with_name(path, name)?;
507    Ok(())
508}
509
510fn append_bytes(
511    builder: &mut tar::Builder<std::fs::File>,
512    name: &Path,
513    bytes: &[u8],
514) -> Result<()> {
515    let mut header = tar::Header::new_gnu();
516    header.set_size(bytes.len() as u64);
517    header.set_mode(0o644);
518    header.set_cksum();
519    builder.append_data(&mut header, name, bytes)?;
520    Ok(())
521}
522
523fn validate_archive_path(path: &Path) -> Result<()> {
524    if path.as_os_str().is_empty()
525        || path.is_absolute()
526        || path
527            .components()
528            .any(|component| !matches!(component, Component::Normal(_)))
529    {
530        eyre::bail!("cache export contains unsafe path {}", path.display());
531    }
532    let text = path.to_string_lossy();
533    if text != EXPORT_MANIFEST
534        && !text.starts_with("cas/v1/")
535        && !text.starts_with("action-results/v1/")
536    {
537        eyre::bail!("cache export contains unexpected path {}", path.display());
538    }
539    Ok(())
540}
541
542/// One checkout's claim on the actions a build identity recorded.
543///
544/// The target directory is not read back by anything yet; it is recorded
545/// because it is the other half of what the shim mapped, and a record that
546/// names only half of it would have to be rewritten to answer where the
547/// artifacts went.
548#[derive(Debug, Clone, Serialize, Deserialize)]
549#[serde(deny_unknown_fields)]
550struct CheckoutRecord {
551    version: u8,
552    workspace_root: PathBuf,
553    target_dir: PathBuf,
554    updated_secs: u64,
555}
556
557/// Summarize what the store currently holds.
558pub fn stats(store: &Path) -> Result<StoreStats> {
559    let objects = walk_files(&store.join(CAS_DIR))?;
560    let results = walk_files(&store.join(ACTION_RESULTS_DIR))?;
561    let checkouts = scan_checkouts(store)?;
562    Ok(StoreStats {
563        objects: objects.len() as u64,
564        object_bytes: objects.iter().map(|entry| entry.size).sum(),
565        action_results: results.len() as u64,
566        action_result_bytes: results.iter().map(|entry| entry.size).sum(),
567        live_checkouts: checkouts.live_records,
568        stale_checkouts: checkouts.stale_records.len() as u64,
569    })
570}
571
572/// Size the target tree a checkout record names, if it names one.
573///
574/// A build with no target directory of its own -- one driven by make or CMake,
575/// which writes wherever it was told -- records the checkout in that field
576/// rather than inventing a directory. Walking it would report the whole source
577/// tree, `.git` included, as though it were build output.
578fn checkout_target_bytes(sizes: &mut BTreeMap<PathBuf, u64>, record: &CheckoutRecord) -> u64 {
579    if record.target_dir == record.workspace_root {
580        return 0;
581    }
582    cached_tree_bytes(sizes, &record.target_dir)
583}
584
585/// Attribute cache and target bytes to each recorded workspace.
586///
587/// Shared objects are counted for every workspace that can reach them. That
588/// makes each row answer "how much keeps this workspace warm" without
589/// pretending shared storage can be divided exactly between projects.
590pub fn projects(store: &Path) -> Result<Vec<ProjectUsage>> {
591    let mut projects: BTreeMap<PathBuf, (BTreeSet<String>, bool, u64)> = BTreeMap::new();
592    let mut target_sizes = BTreeMap::new();
593    let root = store.join(CHECKOUTS_DIR);
594    for identity_entry in read_dir_or_empty(&root)? {
595        let identity = identity_entry.file_name().to_string_lossy().into_owned();
596        if !is_task_identity(&identity) || !identity_entry.file_type()?.is_dir() {
597            continue;
598        }
599        for entry in walk_files(&identity_entry.path())? {
600            let Some(record) = read_checkout_record(&entry.path) else {
601                continue;
602            };
603            let project = projects.entry(record.workspace_root.clone()).or_default();
604            if claim_is_live(&record) {
605                project.0.insert(identity.clone());
606                project.1 = true;
607            }
608            project.2 = project
609                .2
610                .max(checkout_target_bytes(&mut target_sizes, &record));
611        }
612    }
613    let action_cache = mbx_cache_core::LocalActionCache::new(store);
614    let mut usages = Vec::new();
615    for (workspace_root, (identities, live, target_bytes)) in projects {
616        let mut paths = rooted_objects(store, &identities)?;
617        for identity in &identities {
618            for action in task_manifest_actions(store, identity).unwrap_or_default() {
619                if let Ok(path) = action_cache.path_for(&action) {
620                    paths.insert(path);
621                }
622            }
623        }
624        let action_bytes = paths
625            .iter()
626            .filter_map(|path| std::fs::metadata(path).ok())
627            .map(|metadata| metadata.len())
628            .sum();
629        usages.push(ProjectUsage {
630            workspace_root,
631            identities: identities.len() as u64,
632            action_bytes,
633            target_bytes,
634            live,
635        });
636    }
637    usages.sort_by(|left, right| {
638        right
639            .action_bytes
640            .saturating_add(right.target_bytes)
641            .cmp(&left.action_bytes.saturating_add(left.target_bytes))
642            .then_with(|| left.workspace_root.cmp(&right.workspace_root))
643    });
644    Ok(usages)
645}
646
647/// Return the largest blobs and action-result records in descending order.
648pub fn largest(store: &Path, limit: usize) -> Result<Vec<LargestEntry>> {
649    let mut entries = Vec::new();
650    for (kind, root, action_result) in [
651        ("object", store.join(CAS_DIR), false),
652        ("action result", store.join(ACTION_RESULTS_DIR), true),
653    ] {
654        entries.extend(
655            walk_files(&root)?
656                .into_iter()
657                .filter(|entry| addressed_digest(store, &entry.path, action_result).is_some())
658                .map(|entry| LargestEntry {
659                    kind,
660                    path: entry.path,
661                    bytes: entry.size,
662                }),
663        );
664    }
665    entries.sort_by(|left, right| {
666        right
667            .bytes
668            .cmp(&left.bytes)
669            .then_with(|| left.path.cmp(&right.path))
670    });
671    entries.truncate(limit);
672    Ok(entries)
673}
674
675/// Verify every addressable CAS object and action-result record.
676pub fn verify(store: &Path) -> Result<VerifyOutcome> {
677    let cas = LocalCas::new(store);
678    let action_cache = mbx_cache_core::LocalActionCache::new(store);
679    // Verification reports the store as it is on disk and removes nothing, so
680    // no object is pending removal the way one is mid-sweep.
681    let virtually_removed = HashSet::new();
682    let mut outcome = VerifyOutcome::default();
683    for entry in walk_files(&store.join(CAS_DIR))? {
684        let Some(digest) = addressed_digest(store, &entry.path, false) else {
685            continue;
686        };
687        outcome.checked_objects += 1;
688        if !matches!(cas.find(&digest), Ok(Some(_))) {
689            outcome.problems.push(entry.path);
690        }
691    }
692    for entry in walk_files(&store.join(ACTION_RESULTS_DIR))? {
693        let Some(digest) = addressed_digest(store, &entry.path, true) else {
694            continue;
695        };
696        outcome.checked_action_results += 1;
697        if !matches!(action_cache.find(&digest), Ok(Some(_)))
698            || action_result_is_dangling(&cas, &entry.path, &virtually_removed)?
699        {
700            outcome.problems.push(entry.path);
701        }
702    }
703    outcome.problems.sort();
704    Ok(outcome)
705}
706
707/// Remove checkout claims belonging to exactly one workspace.
708pub fn remove_project(store: &Path, workspace_root: &Path) -> Result<RemoveProjectOutcome> {
709    let mut outcome = RemoveProjectOutcome::default();
710    for entry in walk_files(&store.join(CHECKOUTS_DIR))? {
711        if read_checkout_record(&entry.path)
712            .is_some_and(|record| record.workspace_root == workspace_root)
713        {
714            std::fs::remove_file(&entry.path)?;
715            outcome.removed_checkout_records += 1;
716            if let Some(parent) = entry.path.parent() {
717                let _ = std::fs::remove_dir(parent);
718            }
719        }
720    }
721    match std::fs::remove_file(latest_receipt_path(store, workspace_root)) {
722        Ok(()) => {}
723        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
724        Err(error) => return Err(error.into()),
725    }
726    Ok(outcome)
727}
728
729/// Record that `workspace_root` built `identity`.
730///
731/// Identities are shared by every checkout of one dependency graph, so this is
732/// what later tells the collector whether anything still needs what a build
733/// cached: one file per checkout under the identity it built. Writing a whole
734/// file per checkout rather than merging a list into one keeps concurrent
735/// builds out of each other's way -- there is nothing to merge, so there is no
736/// lock and no lost update.
737///
738/// A build with no target directory of its own passes `workspace_root` for
739/// `target_dir`, which reads as "none" rather than as a tree to measure.
740pub fn record_checkout(
741    store: &Path,
742    identity: &str,
743    workspace_root: &Path,
744    target_dir: &Path,
745) -> Result<()> {
746    let record = CheckoutRecord {
747        version: CHECKOUT_RECORD_VERSION,
748        workspace_root: workspace_root.to_path_buf(),
749        target_dir: target_dir.to_path_buf(),
750        updated_secs: SystemTime::now()
751            .duration_since(UNIX_EPOCH)
752            .map(|since| since.as_secs())
753            .unwrap_or_default(),
754    };
755    let mut contents = serde_json::to_vec(&record)?;
756    contents.push(b'\n');
757    write_atomic(
758        &checkout_record_path(store, identity, workspace_root),
759        &contents,
760    )
761}
762
763fn checkout_record_path(store: &Path, identity: &str, workspace_root: &Path) -> PathBuf {
764    // The path is hashed rather than escaped: it is only ever compared against
765    // the same hash of the same path, and every filesystem in play disagrees
766    // about which characters a name may hold.
767    let key = CacheDigest::blake3(workspace_root.to_string_lossy().as_bytes()).hash;
768    store
769        .join(CHECKOUTS_DIR)
770        .join(identity)
771        .join(format!("{key}.json"))
772}
773
774/// Evict objects until the store fits within `max_bytes`.
775///
776/// Objects no live checkout can reach are evicted first, so deleting a worktree
777/// releases what only that worktree needed. Within each class eviction is
778/// least-recently-used by access time where the filesystem records it, falling
779/// back to modification time. A restore verifies the blob it serves, which
780/// means it reads the whole file and the access time is real -- but only where
781/// the mount records one, so the ordering is an approximation either way.
782/// Evicting a live object costs a recompile, never correctness.
783pub fn gc(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
784    gc_with_mode(store, max_bytes, false)
785}
786
787/// Describe the collection `gc` would perform without changing the store.
788pub fn gc_dry_run(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
789    gc_with_mode(store, max_bytes, true)
790}
791
792fn gc_with_mode(store: &Path, max_bytes: u64, dry_run: bool) -> Result<GcOutcome> {
793    let mut objects = walk_files(&store.join(CAS_DIR))?;
794    let results = walk_files(&store.join(ACTION_RESULTS_DIR))?;
795    let mut live_bytes = objects
796        .iter()
797        .chain(results.iter())
798        .map(|entry| entry.size)
799        .sum::<u64>();
800
801    let mut outcome = GcOutcome::default();
802
803    // Prune before deciding what is rooted, so a checkout deleted since the
804    // last sweep stops protecting its artifacts during this one.
805    let checkouts = scan_checkouts(store)?;
806    for path in &checkouts.stale_records {
807        if dry_run || matches!(remove(path)?, Removal::Removed) {
808            outcome.removed_checkout_records += 1;
809        }
810        // The identity directory is left behind empty otherwise. It may hold
811        // other checkouts, in which case this fails and that is the answer.
812        if !dry_run && let Some(parent) = path.parent() {
813            let _ = std::fs::remove_dir(parent);
814        }
815    }
816
817    let mut virtually_removed = HashSet::new();
818    if live_bytes > max_bytes {
819        // Reachability costs a read per action result and per output tree, so
820        // it is computed only when something is actually about to be evicted.
821        // Under budget a sweep stays a directory walk and nothing more.
822        let rooted = rooted_objects(store, &checkouts.live_identities)?;
823        // `false` sorts first, so unrooted objects go before rooted ones and
824        // each class goes oldest-first. A store with no records at all roots
825        // nothing and this is exactly the LRU it was before.
826        objects.sort_by_cached_key(|entry| (rooted.contains(&entry.path), entry.used));
827        let evicted = evict_objects(&objects, &rooted, live_bytes, max_bytes, |path| {
828            if dry_run {
829                virtually_removed.insert(path.to_path_buf());
830                Ok(Removal::Removed)
831            } else {
832                remove(path)
833            }
834        })?;
835        live_bytes = evicted.remaining_bytes;
836        outcome.removed_objects += evicted.removed_objects;
837        outcome.removed_bytes += evicted.removed_bytes;
838    }
839
840    // An action result whose objects are gone can only produce a miss, so drop
841    // it rather than leave the index pointing at nothing. This runs whether or
842    // not this call evicted anything, since another process may have, and a
843    // store that is over budget on results alone still needs the sweep.
844    let cas = LocalCas::new(store);
845    for entry in &results {
846        if action_result_is_dangling(&cas, &entry.path, &virtually_removed)? {
847            match if dry_run {
848                Removal::Removed
849            } else {
850                remove(&entry.path)?
851            } {
852                Removal::Removed => {
853                    live_bytes = live_bytes.saturating_sub(entry.size);
854                    outcome.removed_action_results += 1;
855                    outcome.removed_bytes += entry.size;
856                }
857                Removal::Missing => live_bytes = live_bytes.saturating_sub(entry.size),
858                Removal::Blocked => {}
859            }
860        }
861    }
862
863    let sessions = prune_sessions(store, dry_run)?;
864    outcome.removed_session_streams += sessions.removed_streams;
865    outcome.removed_bytes += sessions.removed_bytes;
866
867    outcome.remaining_bytes = live_bytes;
868    Ok(outcome)
869}
870
871#[derive(Debug, Default, PartialEq, Eq)]
872struct SessionPrune {
873    removed_streams: u64,
874    removed_bytes: u64,
875}
876
877fn prune_sessions(store: &Path, dry_run: bool) -> Result<SessionPrune> {
878    let mut prune = SessionPrune::default();
879    let ids = events::session_ids(store);
880    let surplus = ids.len().saturating_sub(MAX_SESSIONS);
881    for (index, id) in ids.iter().enumerate() {
882        let paths = events::session_paths(store, id);
883        let metadata = match std::fs::metadata(&paths.events) {
884            Ok(metadata) => metadata,
885            Err(_) => continue,
886        };
887        let stale = metadata
888            .modified()
889            .ok()
890            .and_then(|at| SystemTime::now().duration_since(at).ok())
891            .is_some_and(|age| age > SESSION_RETENTION);
892        if index >= surplus && !stale {
893            continue;
894        }
895        if events::session_is_live(store, id) {
896            continue;
897        }
898        if dry_run {
899            prune.removed_streams += 1;
900            prune.removed_bytes += metadata.len();
901            continue;
902        }
903        if matches!(remove(&paths.events)?, Removal::Removed) {
904            prune.removed_streams += 1;
905            prune.removed_bytes += metadata.len();
906            let _ = std::fs::remove_file(&paths.lock);
907        }
908    }
909    if !dry_run {
910        for lock in events::orphaned_locks(store) {
911            let _ = std::fs::remove_file(lock);
912        }
913    }
914    Ok(prune)
915}
916
917/// Sweep the store if `interval` has passed since the last attempt.
918///
919/// The stamp is written before the sweep, not after. Two builds finishing
920/// together should cost one sweep between them, and a sweep that dies partway
921/// should wait its turn like any other rather than retry on every build.
922pub fn sweep_if_due(store: &Path, max_bytes: u64, interval: Duration) -> Result<Option<GcOutcome>> {
923    if !claim_sweep(store, interval)? {
924        return Ok(None);
925    }
926    gc(store, max_bytes).map(Some)
927}
928
929/// Atomically stamp a due sweep before its callers perform coordinated GC.
930pub fn claim_sweep(store: &Path, interval: Duration) -> Result<bool> {
931    let lock_path = store.join(SWEEP_LOCK);
932    std::fs::create_dir_all(lock_path.parent().expect("sweep lock has a parent"))?;
933    let mut lock = fslock::LockFile::open(&lock_path)?;
934    lock.lock()?;
935
936    let stamp = store.join(SWEEP_STAMP);
937    if let Ok(metadata) = std::fs::metadata(&stamp)
938        && let Ok(modified) = metadata.modified()
939        && let Ok(since) = modified.elapsed()
940        && since < interval
941    {
942        return Ok(false);
943    }
944    write_atomic(&stamp, b"")?;
945    Ok(true)
946}
947
948/// What the checkout registry says about the identities in this store.
949struct CheckoutScan {
950    live_identities: BTreeSet<String>,
951    live_records: u64,
952    stale_records: Vec<PathBuf>,
953}
954
955fn scan_checkouts(store: &Path) -> Result<CheckoutScan> {
956    let root = store.join(CHECKOUTS_DIR);
957    let mut scan = CheckoutScan {
958        live_identities: BTreeSet::new(),
959        live_records: 0,
960        stale_records: Vec::new(),
961    };
962    let listing = match std::fs::read_dir(&root) {
963        Ok(listing) => listing,
964        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(scan),
965        Err(error) => {
966            return Err(error).wrap_err_with(|| format!("failed to read {}", root.display()));
967        }
968    };
969    for entry in listing {
970        let entry = entry?;
971        let name = entry.file_name().to_string_lossy().into_owned();
972        // Anything that is not a directory named for an identity was not written
973        // by mbx. Both halves matter: `walk_files` tolerates only a missing
974        // directory, so a plain file whose name happens to look like an identity
975        // would fail the whole scan -- and that scan runs inside every sweep,
976        // including the automatic one after a build.
977        if !is_task_identity(&name) || !entry.file_type().is_ok_and(|kind| kind.is_dir()) {
978            continue;
979        }
980        for record in walk_files(&entry.path())? {
981            match read_checkout_record(&record.path) {
982                Some(checkout) if claim_is_live(&checkout) => {
983                    scan.live_identities.insert(name.clone());
984                    scan.live_records += 1;
985                }
986                Some(_) => scan.stale_records.push(record.path),
987                // A record this build cannot read claims nothing, but it is not
988                // evidence the checkout is gone either, so leave it alone.
989                None => {}
990            }
991        }
992    }
993    Ok(scan)
994}
995
996fn read_checkout_record(path: &Path) -> Option<CheckoutRecord> {
997    let bytes = std::fs::read(path).ok()?;
998    let record = serde_json::from_slice::<CheckoutRecord>(&bytes).ok()?;
999    (record.version == CHECKOUT_RECORD_VERSION).then_some(record)
1000}
1001
1002/// Whether a recorded claim still speaks for a checkout that is using this
1003/// store.
1004fn claim_is_live(record: &CheckoutRecord) -> bool {
1005    checkout_is_live(&record.workspace_root) && !claim_has_expired(record)
1006}
1007
1008fn claim_has_expired(record: &CheckoutRecord) -> bool {
1009    let Ok(now) = SystemTime::now().duration_since(UNIX_EPOCH) else {
1010        return false;
1011    };
1012    now.as_secs()
1013        .saturating_sub(record.updated_secs)
1014        .gt(&CHECKOUT_RETENTION.as_secs())
1015}
1016
1017/// Whether the checkout a record names is still on disk.
1018///
1019/// Absence is believed only when the checkout is definitely absent *and* its
1020/// parent directory is definitely still there. Deleting a worktree, or a whole
1021/// project directory, leaves the parent behind; a volume being ejected or a
1022/// network mount going away takes several levels with it, and that is the case
1023/// worth being careful about -- a temporarily absent mount must not un-root a
1024/// checkout that is really there.
1025///
1026/// So both questions are asked the same way: only a definite answer counts, and
1027/// an error at either step means live. Being wrong in that direction only
1028/// delays collection; the other way round throws away a warm cache someone is
1029/// still using.
1030pub fn checkout_is_live(workspace_root: &Path) -> bool {
1031    if !matches!(workspace_root.try_exists(), Ok(false)) {
1032        return true;
1033    }
1034    match workspace_root.parent() {
1035        Some(parent) => !matches!(parent.try_exists(), Ok(true)),
1036        // A root directory has no parent to corroborate anything with.
1037        None => true,
1038    }
1039}
1040
1041/// CAS paths reachable from the builds that live checkouts still depend on.
1042///
1043/// Recursing into output trees is the point: the descriptor blobs are tiny and
1044/// the leaf artifacts are where the bytes are, so rooting that stopped at the
1045/// top level would protect nothing worth protecting. Every read here is
1046/// tolerant -- a blob that has already been evicted, or that no longer parses,
1047/// simply roots less, and rooting less can only mean collecting sooner.
1048fn rooted_objects(store: &Path, identities: &BTreeSet<String>) -> Result<HashSet<PathBuf>> {
1049    let cas = LocalCas::new(store);
1050    let mut rooted = HashSet::new();
1051    let mut visited = BTreeSet::new();
1052    for identity in identities {
1053        // One manifest this build cannot read is not worth abandoning a sweep
1054        // over. It roots nothing, which can only mean collecting sooner.
1055        let actions = match task_manifest_actions(store, identity) {
1056            Ok(actions) => actions,
1057            Err(error) => {
1058                log::debug!("could not read the manifest for {identity}: {error}");
1059                continue;
1060            }
1061        };
1062        for action in actions {
1063            let Some(result) = read_action_result(store, &action) else {
1064                continue;
1065            };
1066            root_digest(&cas, &mut rooted, &result.action);
1067            if let Some(metadata) = &result.metadata {
1068                root_digest(&cas, &mut rooted, metadata);
1069                // The metadata blob is a descriptor too: it names the captured
1070                // stdout and stderr, and a restore reads both. Stopping at the
1071                // descriptor would leave the diagnostics of every action
1072                // unrooted -- including the one empty blob that every silent
1073                // compilation shares, which is enough on its own to turn a
1074                // rooted hit back into a miss.
1075                if let Some(rustc) = read_rustc_metadata(&cas, metadata) {
1076                    root_digest(&cas, &mut rooted, &rustc.stdout);
1077                    root_digest(&cas, &mut rooted, &rustc.stderr);
1078                }
1079            }
1080            if let Some(output_root) = &result.output_root {
1081                root_digest(&cas, &mut rooted, output_root);
1082                root_tree(&cas, output_root, &mut rooted, &mut visited);
1083            }
1084        }
1085    }
1086    Ok(rooted)
1087}
1088
1089fn read_action_result(store: &Path, action: &CacheDigest) -> Option<RemoteActionResult> {
1090    let path = mbx_cache_core::LocalActionCache::new(store)
1091        .path_for(action)
1092        .ok()?;
1093    let bytes = std::fs::read(path).ok()?;
1094    serde_json::from_slice(&bytes).ok()
1095}
1096
1097fn root_tree(
1098    cas: &LocalCas,
1099    output_root: &CacheDigest,
1100    rooted: &mut HashSet<PathBuf>,
1101    visited: &mut BTreeSet<CacheDigest>,
1102) {
1103    let mut pending = vec![output_root.clone()];
1104    while let Some(digest) = pending.pop() {
1105        if !visited.insert(digest.clone()) {
1106            continue;
1107        }
1108        let Some(directory) = read_directory(cas, &digest) else {
1109            continue;
1110        };
1111        for file in &directory.files {
1112            root_digest(cas, rooted, &file.digest);
1113        }
1114        for child in &directory.directories {
1115            root_digest(cas, rooted, &child.digest);
1116            pending.push(child.digest.clone());
1117        }
1118    }
1119}
1120
1121fn read_rustc_metadata(cas: &LocalCas, digest: &CacheDigest) -> Option<RustcMetadata> {
1122    let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1123    serde_json::from_slice(&bytes).ok()
1124}
1125
1126fn read_directory(cas: &LocalCas, digest: &CacheDigest) -> Option<CacheDirectory> {
1127    let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1128    serde_json::from_slice(&bytes).ok()
1129}
1130
1131fn root_digest(cas: &LocalCas, rooted: &mut HashSet<PathBuf>, digest: &CacheDigest) {
1132    if let Ok(path) = cas.path_for(digest) {
1133        rooted.insert(path);
1134    }
1135}
1136
1137#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1138enum Removal {
1139    Removed,
1140    Missing,
1141    Blocked,
1142}
1143
1144fn remove(path: &Path) -> Result<Removal> {
1145    match std::fs::remove_file(path) {
1146        Ok(()) => Ok(Removal::Removed),
1147        // A concurrent build may have evicted or replaced it already.
1148        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Removal::Missing),
1149        // Windows refuses to unlink a file another process holds open, which is
1150        // what a concurrent build reading this blob looks like. Skipping it
1151        // leaves the store over budget until the next sweep; failing would
1152        // abandon the sweep and leave it over budget for longer.
1153        Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
1154            log::debug!("could not evict {}: {error}", path.display());
1155            Ok(Removal::Blocked)
1156        }
1157        Err(error) => Err(error).wrap_err_with(|| format!("failed to evict {}", path.display())),
1158    }
1159}
1160
1161#[derive(Debug, Default, PartialEq, Eq)]
1162struct ObjectEvictions {
1163    removed_objects: u64,
1164    removed_bytes: u64,
1165    remaining_bytes: u64,
1166}
1167
1168/// Evict sorted objects without crossing the checkout-protection boundary
1169/// when an unrooted object cannot be removed.
1170fn evict_objects(
1171    objects: &[Entry],
1172    rooted: &HashSet<PathBuf>,
1173    mut live_bytes: u64,
1174    max_bytes: u64,
1175    mut remove_file: impl FnMut(&Path) -> Result<Removal>,
1176) -> Result<ObjectEvictions> {
1177    let mut outcome = ObjectEvictions::default();
1178    let mut blocked_unrooted = false;
1179    for entry in objects {
1180        if live_bytes <= max_bytes {
1181            break;
1182        }
1183        let is_rooted = rooted.contains(&entry.path);
1184        if is_rooted && blocked_unrooted {
1185            // A locked unrooted blob is temporarily part of the irreducible
1186            // store. Deleting a live checkout's artifacts in its place would
1187            // invert the protection ordering this collector promises.
1188            break;
1189        }
1190        match remove_file(&entry.path)? {
1191            Removal::Removed => {
1192                live_bytes = live_bytes.saturating_sub(entry.size);
1193                outcome.removed_objects += 1;
1194                outcome.removed_bytes += entry.size;
1195            }
1196            Removal::Missing => live_bytes = live_bytes.saturating_sub(entry.size),
1197            Removal::Blocked if !is_rooted => blocked_unrooted = true,
1198            Removal::Blocked => {}
1199        }
1200    }
1201    outcome.remaining_bytes = live_bytes;
1202    Ok(outcome)
1203}
1204
1205/// Whether an action result references an object the store no longer has.
1206///
1207/// The digests checked here are exactly the ones `LocalActionCache::store`
1208/// requires. If the sweep checked fewer, eviction could leave an entry that
1209/// cannot be republished -- `store` would reject the identical result for a
1210/// missing blob while the index still claimed to hold it.
1211///
1212/// Presence is all that is checked, not content. Verifying would re-hash a
1213/// large fraction of the store on every sweep, which a chore could afford and
1214/// an automatic sweep cannot; it would also refresh the access time of every
1215/// descriptor blob while leaving the artifacts alone, biasing the very
1216/// ordering this module depends on. A blob that is present but corrupt now
1217/// keeps its result and turns into a miss on restore, and both CAS write paths
1218/// republish over a corrupt blob rather than trusting it.
1219///
1220/// Only the top-level objects are checked; a result whose output tree lost a
1221/// nested object still restores as a miss, which is safe.
1222fn action_result_is_dangling(
1223    cas: &LocalCas,
1224    path: &Path,
1225    virtually_removed: &HashSet<PathBuf>,
1226) -> Result<bool> {
1227    let bytes = match std::fs::read(path) {
1228        Ok(bytes) => bytes,
1229        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
1230        Err(error) => {
1231            return Err(error).wrap_err_with(|| format!("failed to read {}", path.display()));
1232        }
1233    };
1234    // An unparseable result is already useless; the cache rejects it on read.
1235    let Ok(result) = serde_json::from_slice::<RemoteActionResult>(&bytes) else {
1236        return Ok(false);
1237    };
1238    for digest in [
1239        Some(&result.action),
1240        result.metadata.as_ref(),
1241        result.output_root.as_ref(),
1242    ]
1243    .into_iter()
1244    .flatten()
1245    {
1246        match cas.path_for(digest) {
1247            Ok(path) => {
1248                if virtually_removed.contains(&path) || !path.exists() {
1249                    return Ok(true);
1250                }
1251            }
1252            // A digest the CAS cannot even address is not one it can hold.
1253            Err(_) => return Ok(true),
1254        }
1255    }
1256    Ok(false)
1257}
1258
1259struct Entry {
1260    path: PathBuf,
1261    size: u64,
1262    used: SystemTime,
1263}
1264
1265fn walk_files(root: &Path) -> Result<Vec<Entry>> {
1266    let mut entries = Vec::new();
1267    let mut pending = vec![root.to_path_buf()];
1268    while let Some(directory) = pending.pop() {
1269        let listing = match std::fs::read_dir(&directory) {
1270            Ok(listing) => listing,
1271            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1272            Err(error) => {
1273                return Err(error)
1274                    .wrap_err_with(|| format!("failed to read {}", directory.display()));
1275            }
1276        };
1277        for entry in listing {
1278            let entry = entry?;
1279            let metadata = match entry.metadata() {
1280                Ok(metadata) => metadata,
1281                // A concurrent build may be publishing into the store.
1282                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1283                Err(error) => return Err(error.into()),
1284            };
1285            if metadata.is_dir() {
1286                pending.push(entry.path());
1287            } else if metadata.is_file() {
1288                let used = metadata
1289                    .accessed()
1290                    .or_else(|_| metadata.modified())
1291                    .unwrap_or(SystemTime::UNIX_EPOCH);
1292                entries.push(Entry {
1293                    path: entry.path(),
1294                    size: metadata.len(),
1295                    used,
1296                });
1297            }
1298        }
1299    }
1300    Ok(entries)
1301}
1302
1303fn read_dir_or_empty(root: &Path) -> Result<Vec<std::fs::DirEntry>> {
1304    match std::fs::read_dir(root) {
1305        Ok(entries) => entries
1306            .collect::<std::io::Result<Vec<_>>>()
1307            .map_err(Into::into),
1308        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
1309        Err(error) => Err(error).wrap_err_with(|| format!("failed to read {}", root.display())),
1310    }
1311}
1312
1313fn digest_from_path(path: &Path, action_result: bool) -> Option<CacheDigest> {
1314    let name = path.file_name()?.to_str()?;
1315    let name = if action_result {
1316        name.strip_suffix(".json")?
1317    } else {
1318        name
1319    };
1320    let (hash, size) = name.rsplit_once('-')?;
1321    let algorithm = path.parent()?.parent()?.file_name()?.to_str()?.to_string();
1322    let digest = CacheDigest {
1323        algorithm,
1324        hash: hash.to_string(),
1325        size: size.parse().ok()?,
1326    };
1327    digest.validate().ok()?;
1328    Some(digest)
1329}
1330
1331fn addressed_digest(store: &Path, path: &Path, action_result: bool) -> Option<CacheDigest> {
1332    let digest = digest_from_path(path, action_result)?;
1333    let canonical = if action_result {
1334        mbx_cache_core::LocalActionCache::new(store)
1335            .path_for(&digest)
1336            .ok()?
1337    } else {
1338        LocalCas::new(store).path_for(&digest).ok()?
1339    };
1340    (canonical == path).then_some(digest)
1341}
1342
1343fn cached_tree_bytes(cache: &mut BTreeMap<PathBuf, u64>, root: &Path) -> u64 {
1344    *cache
1345        .entry(root.to_path_buf())
1346        .or_insert_with(|| tree_bytes(root))
1347}
1348
1349fn tree_bytes(root: &Path) -> u64 {
1350    let Ok(root_metadata) = std::fs::metadata(root) else {
1351        return 0;
1352    };
1353    if root_metadata.is_file() {
1354        return root_metadata.len();
1355    }
1356    let Ok(entries) = std::fs::read_dir(root) else {
1357        return 0;
1358    };
1359    let mut bytes = 0_u64;
1360    let mut pending = entries
1361        .filter_map(std::result::Result::ok)
1362        .map(|entry| entry.path())
1363        .collect::<Vec<_>>();
1364    while let Some(path) = pending.pop() {
1365        let Ok(metadata) = std::fs::symlink_metadata(&path) else {
1366            continue;
1367        };
1368        if metadata.is_file() {
1369            bytes = bytes.saturating_add(metadata.len());
1370        } else if metadata.is_dir()
1371            && let Ok(entries) = std::fs::read_dir(path)
1372        {
1373            pending.extend(
1374                entries
1375                    .filter_map(std::result::Result::ok)
1376                    .map(|entry| entry.path()),
1377            );
1378        }
1379    }
1380    bytes
1381}
1382
1383#[cfg(test)]
1384#[path = "store_tests.rs"]
1385mod tests;
1386
1387fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> {
1388    let parent = path
1389        .parent()
1390        .filter(|parent| !parent.as_os_str().is_empty())
1391        .ok_or_else(|| eyre::eyre!("path has no parent: {}", path.display()))?;
1392    std::fs::create_dir_all(parent)
1393        .wrap_err_with(|| format!("failed to create {}", parent.display()))?;
1394    let mut temporary = tempfile::Builder::new()
1395        .prefix(".mbx-")
1396        .tempfile_in(parent)?;
1397    use std::io::Write as _;
1398    temporary.write_all(contents)?;
1399    temporary.as_file_mut().sync_all()?;
1400    temporary
1401        .persist(path)
1402        .map_err(|error| error.error)
1403        .wrap_err_with(|| format!("failed atomic write: {}", path.display()))?;
1404    Ok(())
1405}