1mod 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
38const 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 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 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#[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
137pub 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
231pub 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
248pub 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 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 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
346pub 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 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#[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
573pub 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
588fn 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
601pub 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
663pub 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
691pub fn verify(store: &Path) -> Result<VerifyOutcome> {
693 let cas = LocalCas::new(store);
694 let action_cache = mbx_cache_core::LocalActionCache::new(store);
695 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
723pub 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
745pub 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 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
790pub fn gc(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
800 gc_with_mode(store, max_bytes, false)
801}
802
803pub 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 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 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 let mut rooted = rooted_objects(store, &checkouts.live_identities)?;
839 rooted.extend(rooted_action_objects(
846 store,
847 grouped_receipt_actions(store)?,
848 ));
849 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 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
943pub 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
955pub 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
974struct 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 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 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
1028fn 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
1039fn 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
1054pub 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
1086pub 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 let mut mount = vec![0_u16; 261];
1135 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 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
1163fn rooted_objects(store: &Path, identities: &BTreeSet<String>) -> Result<HashSet<PathBuf>> {
1171 let mut actions = BTreeSet::new();
1172 for identity in identities {
1173 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
1187fn 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 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 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Removal::Missing),
1281 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
1300fn 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 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
1337fn 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 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 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 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}