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(&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(&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(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
1017pub 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 None => true,
1038 }
1039}
1040
1041fn 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 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 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 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Removal::Missing),
1149 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
1168fn 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 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
1205fn 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 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 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 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}