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))
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 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
330pub 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 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#[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
557pub 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
572fn 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
585pub 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(store, &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
647pub 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
675pub fn verify(store: &Path) -> Result<VerifyOutcome> {
677 let cas = LocalCas::new(store);
678 let action_cache = mbx_cache_core::LocalActionCache::new(store);
679 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
707pub 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
729pub 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 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
774pub fn gc(store: &Path, max_bytes: u64) -> Result<GcOutcome> {
784 gc_with_mode(store, max_bytes, false)
785}
786
787pub 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 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 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 let rooted = rooted_objects(store, &checkouts.live_identities)?;
823 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 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
917pub 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
929pub 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
948struct 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 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(store, &checkout) => {
983 scan.live_identities.insert(name.clone());
984 scan.live_records += 1;
985 }
986 Some(_) => scan.stale_records.push(record.path),
987 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
1002fn claim_is_live(store: &Path, record: &CheckoutRecord) -> bool {
1005 checkout_is_live_on(store, &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
1017pub fn checkout_is_live_on(store: &Path, workspace_root: &Path) -> bool {
1030 if !matches!(workspace_root.try_exists(), Ok(false)) {
1031 return true;
1032 }
1033
1034 let Ok(store_metadata) = std::fs::metadata(store) else {
1035 return true;
1036 };
1037 for ancestor in workspace_root.ancestors().skip(1) {
1038 match std::fs::metadata(ancestor) {
1039 Ok(ancestor_metadata) => {
1040 return !same_filesystem(store, &store_metadata, ancestor, &ancestor_metadata);
1041 }
1042 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
1043 Err(_) => return true,
1044 }
1045 }
1046 true
1047}
1048
1049pub fn checkout_is_live(workspace_root: &Path) -> bool {
1055 let Some(parent) = workspace_root.parent() else {
1056 return true;
1057 };
1058 checkout_is_live_on(parent, workspace_root)
1059}
1060
1061#[cfg(unix)]
1062fn same_filesystem(
1063 _a_path: &Path,
1064 a: &std::fs::Metadata,
1065 _b_path: &Path,
1066 b: &std::fs::Metadata,
1067) -> bool {
1068 use std::os::unix::fs::MetadataExt as _;
1069 a.dev() == b.dev()
1070}
1071
1072#[cfg(windows)]
1073fn same_filesystem(
1074 a_path: &Path,
1075 _a: &std::fs::Metadata,
1076 b_path: &Path,
1077 _b: &std::fs::Metadata,
1078) -> bool {
1079 windows_volume_name(a_path).is_some_and(|volume| windows_volume_name(b_path) == Some(volume))
1080}
1081
1082#[cfg(windows)]
1083fn windows_volume_name(path: &Path) -> Option<Vec<u16>> {
1084 use std::os::windows::ffi::OsStrExt as _;
1085 use windows_sys::Win32::Storage::FileSystem::{
1086 GetVolumeNameForVolumeMountPointW, GetVolumePathNameW,
1087 };
1088
1089 let path = path
1090 .as_os_str()
1091 .encode_wide()
1092 .chain(Some(0))
1093 .collect::<Vec<_>>();
1094 let mut mount = vec![0_u16; 261];
1098 if unsafe { GetVolumePathNameW(path.as_ptr(), mount.as_mut_ptr(), mount.len() as u32) } == 0 {
1101 return None;
1102 }
1103 let mut volume = vec![0_u16; 261];
1104 if unsafe {
1107 GetVolumeNameForVolumeMountPointW(mount.as_ptr(), volume.as_mut_ptr(), volume.len() as u32)
1108 } == 0
1109 {
1110 return None;
1111 }
1112 volume.truncate(volume.iter().position(|unit| *unit == 0)?);
1113 Some(volume)
1114}
1115
1116#[cfg(not(any(unix, windows)))]
1117fn same_filesystem(
1118 _a_path: &Path,
1119 _a: &std::fs::Metadata,
1120 _b_path: &Path,
1121 _b: &std::fs::Metadata,
1122) -> bool {
1123 false
1124}
1125
1126fn rooted_objects(store: &Path, identities: &BTreeSet<String>) -> Result<HashSet<PathBuf>> {
1134 let cas = LocalCas::new(store);
1135 let mut rooted = HashSet::new();
1136 let mut visited = BTreeSet::new();
1137 for identity in identities {
1138 let actions = match task_manifest_actions(store, identity) {
1141 Ok(actions) => actions,
1142 Err(error) => {
1143 log::debug!("could not read the manifest for {identity}: {error}");
1144 continue;
1145 }
1146 };
1147 for action in actions {
1148 let Some(result) = read_action_result(store, &action) else {
1149 continue;
1150 };
1151 root_digest(&cas, &mut rooted, &result.action);
1152 if let Some(metadata) = &result.metadata {
1153 root_digest(&cas, &mut rooted, metadata);
1154 if let Some(rustc) = read_rustc_metadata(&cas, metadata) {
1161 root_digest(&cas, &mut rooted, &rustc.stdout);
1162 root_digest(&cas, &mut rooted, &rustc.stderr);
1163 }
1164 }
1165 if let Some(output_root) = &result.output_root {
1166 root_digest(&cas, &mut rooted, output_root);
1167 root_tree(&cas, output_root, &mut rooted, &mut visited);
1168 }
1169 }
1170 }
1171 Ok(rooted)
1172}
1173
1174fn read_action_result(store: &Path, action: &CacheDigest) -> Option<RemoteActionResult> {
1175 let path = mbx_cache_core::LocalActionCache::new(store)
1176 .path_for(action)
1177 .ok()?;
1178 let bytes = std::fs::read(path).ok()?;
1179 serde_json::from_slice(&bytes).ok()
1180}
1181
1182fn root_tree(
1183 cas: &LocalCas,
1184 output_root: &CacheDigest,
1185 rooted: &mut HashSet<PathBuf>,
1186 visited: &mut BTreeSet<CacheDigest>,
1187) {
1188 let mut pending = vec![output_root.clone()];
1189 while let Some(digest) = pending.pop() {
1190 if !visited.insert(digest.clone()) {
1191 continue;
1192 }
1193 let Some(directory) = read_directory(cas, &digest) else {
1194 continue;
1195 };
1196 for file in &directory.files {
1197 root_digest(cas, rooted, &file.digest);
1198 }
1199 for child in &directory.directories {
1200 root_digest(cas, rooted, &child.digest);
1201 pending.push(child.digest.clone());
1202 }
1203 }
1204}
1205
1206fn read_rustc_metadata(cas: &LocalCas, digest: &CacheDigest) -> Option<RustcMetadata> {
1207 let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1208 serde_json::from_slice(&bytes).ok()
1209}
1210
1211fn read_directory(cas: &LocalCas, digest: &CacheDigest) -> Option<CacheDirectory> {
1212 let bytes = std::fs::read(cas.path_for(digest).ok()?).ok()?;
1213 serde_json::from_slice(&bytes).ok()
1214}
1215
1216fn root_digest(cas: &LocalCas, rooted: &mut HashSet<PathBuf>, digest: &CacheDigest) {
1217 if let Ok(path) = cas.path_for(digest) {
1218 rooted.insert(path);
1219 }
1220}
1221
1222#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1223enum Removal {
1224 Removed,
1225 Missing,
1226 Blocked,
1227}
1228
1229fn remove(path: &Path) -> Result<Removal> {
1230 match std::fs::remove_file(path) {
1231 Ok(()) => Ok(Removal::Removed),
1232 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Removal::Missing),
1234 Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
1239 log::debug!("could not evict {}: {error}", path.display());
1240 Ok(Removal::Blocked)
1241 }
1242 Err(error) => Err(error).wrap_err_with(|| format!("failed to evict {}", path.display())),
1243 }
1244}
1245
1246#[derive(Debug, Default, PartialEq, Eq)]
1247struct ObjectEvictions {
1248 removed_objects: u64,
1249 removed_bytes: u64,
1250 remaining_bytes: u64,
1251}
1252
1253fn evict_objects(
1256 objects: &[Entry],
1257 rooted: &HashSet<PathBuf>,
1258 mut live_bytes: u64,
1259 max_bytes: u64,
1260 mut remove_file: impl FnMut(&Path) -> Result<Removal>,
1261) -> Result<ObjectEvictions> {
1262 let mut outcome = ObjectEvictions::default();
1263 let mut blocked_unrooted = false;
1264 for entry in objects {
1265 if live_bytes <= max_bytes {
1266 break;
1267 }
1268 let is_rooted = rooted.contains(&entry.path);
1269 if is_rooted && blocked_unrooted {
1270 break;
1274 }
1275 match remove_file(&entry.path)? {
1276 Removal::Removed => {
1277 live_bytes = live_bytes.saturating_sub(entry.size);
1278 outcome.removed_objects += 1;
1279 outcome.removed_bytes += entry.size;
1280 }
1281 Removal::Missing => live_bytes = live_bytes.saturating_sub(entry.size),
1282 Removal::Blocked if !is_rooted => blocked_unrooted = true,
1283 Removal::Blocked => {}
1284 }
1285 }
1286 outcome.remaining_bytes = live_bytes;
1287 Ok(outcome)
1288}
1289
1290fn action_result_is_dangling(
1308 cas: &LocalCas,
1309 path: &Path,
1310 virtually_removed: &HashSet<PathBuf>,
1311) -> Result<bool> {
1312 let bytes = match std::fs::read(path) {
1313 Ok(bytes) => bytes,
1314 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
1315 Err(error) => {
1316 return Err(error).wrap_err_with(|| format!("failed to read {}", path.display()));
1317 }
1318 };
1319 let Ok(result) = serde_json::from_slice::<RemoteActionResult>(&bytes) else {
1321 return Ok(false);
1322 };
1323 for digest in [
1324 Some(&result.action),
1325 result.metadata.as_ref(),
1326 result.output_root.as_ref(),
1327 ]
1328 .into_iter()
1329 .flatten()
1330 {
1331 match cas.path_for(digest) {
1332 Ok(path) => {
1333 if virtually_removed.contains(&path) || !path.exists() {
1334 return Ok(true);
1335 }
1336 }
1337 Err(_) => return Ok(true),
1339 }
1340 }
1341 Ok(false)
1342}
1343
1344struct Entry {
1345 path: PathBuf,
1346 size: u64,
1347 used: SystemTime,
1348}
1349
1350fn walk_files(root: &Path) -> Result<Vec<Entry>> {
1351 let mut entries = Vec::new();
1352 let mut pending = vec![root.to_path_buf()];
1353 while let Some(directory) = pending.pop() {
1354 let listing = match std::fs::read_dir(&directory) {
1355 Ok(listing) => listing,
1356 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1357 Err(error) => {
1358 return Err(error)
1359 .wrap_err_with(|| format!("failed to read {}", directory.display()));
1360 }
1361 };
1362 for entry in listing {
1363 let entry = entry?;
1364 let metadata = match entry.metadata() {
1365 Ok(metadata) => metadata,
1366 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1368 Err(error) => return Err(error.into()),
1369 };
1370 if metadata.is_dir() {
1371 pending.push(entry.path());
1372 } else if metadata.is_file() {
1373 let used = metadata
1374 .accessed()
1375 .or_else(|_| metadata.modified())
1376 .unwrap_or(SystemTime::UNIX_EPOCH);
1377 entries.push(Entry {
1378 path: entry.path(),
1379 size: metadata.len(),
1380 used,
1381 });
1382 }
1383 }
1384 }
1385 Ok(entries)
1386}
1387
1388fn read_dir_or_empty(root: &Path) -> Result<Vec<std::fs::DirEntry>> {
1389 match std::fs::read_dir(root) {
1390 Ok(entries) => entries
1391 .collect::<std::io::Result<Vec<_>>>()
1392 .map_err(Into::into),
1393 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
1394 Err(error) => Err(error).wrap_err_with(|| format!("failed to read {}", root.display())),
1395 }
1396}
1397
1398fn digest_from_path(path: &Path, action_result: bool) -> Option<CacheDigest> {
1399 let name = path.file_name()?.to_str()?;
1400 let name = if action_result {
1401 name.strip_suffix(".json")?
1402 } else {
1403 name
1404 };
1405 let (hash, size) = name.rsplit_once('-')?;
1406 let algorithm = path.parent()?.parent()?.file_name()?.to_str()?.to_string();
1407 let digest = CacheDigest {
1408 algorithm,
1409 hash: hash.to_string(),
1410 size: size.parse().ok()?,
1411 };
1412 digest.validate().ok()?;
1413 Some(digest)
1414}
1415
1416fn addressed_digest(store: &Path, path: &Path, action_result: bool) -> Option<CacheDigest> {
1417 let digest = digest_from_path(path, action_result)?;
1418 let canonical = if action_result {
1419 mbx_cache_core::LocalActionCache::new(store)
1420 .path_for(&digest)
1421 .ok()?
1422 } else {
1423 LocalCas::new(store).path_for(&digest).ok()?
1424 };
1425 (canonical == path).then_some(digest)
1426}
1427
1428fn cached_tree_bytes(cache: &mut BTreeMap<PathBuf, u64>, root: &Path) -> u64 {
1429 *cache
1430 .entry(root.to_path_buf())
1431 .or_insert_with(|| tree_bytes(root))
1432}
1433
1434fn tree_bytes(root: &Path) -> u64 {
1435 let Ok(root_metadata) = std::fs::metadata(root) else {
1436 return 0;
1437 };
1438 if root_metadata.is_file() {
1439 return root_metadata.len();
1440 }
1441 let Ok(entries) = std::fs::read_dir(root) else {
1442 return 0;
1443 };
1444 let mut bytes = 0_u64;
1445 let mut pending = entries
1446 .filter_map(std::result::Result::ok)
1447 .map(|entry| entry.path())
1448 .collect::<Vec<_>>();
1449 while let Some(path) = pending.pop() {
1450 let Ok(metadata) = std::fs::symlink_metadata(&path) else {
1451 continue;
1452 };
1453 if metadata.is_file() {
1454 bytes = bytes.saturating_add(metadata.len());
1455 } else if metadata.is_dir()
1456 && let Ok(entries) = std::fs::read_dir(path)
1457 {
1458 pending.extend(
1459 entries
1460 .filter_map(std::result::Result::ok)
1461 .map(|entry| entry.path()),
1462 );
1463 }
1464 }
1465 bytes
1466}
1467
1468#[cfg(test)]
1469#[path = "store_tests.rs"]
1470mod tests;
1471
1472fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> {
1473 let parent = path
1474 .parent()
1475 .filter(|parent| !parent.as_os_str().is_empty())
1476 .ok_or_else(|| eyre::eyre!("path has no parent: {}", path.display()))?;
1477 std::fs::create_dir_all(parent)
1478 .wrap_err_with(|| format!("failed to create {}", parent.display()))?;
1479 let mut temporary = tempfile::Builder::new()
1480 .prefix(".mbx-")
1481 .tempfile_in(parent)?;
1482 use std::io::Write as _;
1483 temporary.write_all(contents)?;
1484 temporary.as_file_mut().sync_all()?;
1485 temporary
1486 .persist(path)
1487 .map_err(|error| error.error)
1488 .wrap_err_with(|| format!("failed atomic write: {}", path.display()))?;
1489 Ok(())
1490}