1use std::{
2 collections::{BTreeMap, BTreeSet},
3 ffi::{OsStr, OsString},
4 fmt::Write as _,
5 fs::{self, File},
6 io,
7 path::{Path, PathBuf},
8 sync::atomic::{AtomicU64, Ordering},
9 time::{Duration, Instant},
10};
11
12use crate::timing::saturating_add_optional_duration;
13
14use super::{
15 cache_fs::{
16 ArtifactCacheMaintenance, ArtifactCachePrunePolicy, ArtifactCachePruneReport, CacheFsError,
17 LAST_USED_FILE, RetainedCacheEntry, canonicalize_allow_missing, directory_logical_size,
18 ensure_cache_directory_tag, is_sha256_directory, lock_cache_file,
19 perform_scheduled_cache_maintenance, prune_direct_child_directories,
20 record_cache_entry_use, remove_path_if_present, remove_unretained_entry,
21 try_lock_cache_file,
22 },
23 digest::{
24 FileDigest, InputDigest, InputHasher, copy_file_atomic, destination_matches_digest,
25 digest_bytes, digest_file, digest_labeled_paths, os_bytes, read_file_with_limit,
26 },
27 wasm_cache::{
28 ResolvedCargoBuildInputs, WasmBuildError, WasmBuildSpec, resolve_cargo_build_inputs,
29 },
30};
31
32const ARTIFACT_CACHE_FORMAT: &str = "ic-testkit-artifact-set-v1";
33const MANIFEST_FILE: &str = "manifest.ic-testkit";
34const OUTPUT_FILE_SUFFIX: &str = ".artifact";
35const MAX_PREPARATION_RETRIES: usize = 3;
36static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0);
37
38#[derive(Clone, Copy, Debug, Default, Eq, Ord, PartialEq, PartialOrd)]
40pub enum ArtifactOutputValidation {
41 RegularFile,
43 #[default]
45 NonEmptyFile,
46}
47
48#[derive(Clone, Eq, PartialEq)]
56pub struct ArtifactCacheSpec {
57 cache_root: PathBuf,
58 namespace: String,
59 recipe_id: String,
60 coordination_scope: String,
61 inputs: Vec<LabeledPath>,
62 tools: Vec<LabeledPath>,
63 arguments: Vec<OsString>,
64 environment: BTreeMap<OsString, Option<OsString>>,
65 identities: Vec<LabeledIdentity>,
67 cargo_build_inputs: Vec<CargoBuildInputSet>,
68 outputs: Vec<OutputSpec>,
69 prune_policy: Option<ArtifactCachePrunePolicy>,
70 prune_interval: Option<Duration>,
71}
72
73pub enum ArtifactCachePreparation {
75 Reused(ArtifactCacheRecord),
77 Build(ArtifactBuildTransaction),
79}
80
81#[derive(Clone, Debug, Eq, PartialEq)]
83pub enum ArtifactCacheOutcome {
84 Built(ArtifactCacheRecord),
86 Reused(ArtifactCacheRecord),
88}
89
90#[derive(Clone, Debug, Eq, PartialEq)]
92pub struct ArtifactCacheRecord {
93 key: InputDigest,
94 input_digest: InputDigest,
95 artifacts: Vec<ArtifactCacheArtifact>,
96 _retention: RetainedCacheEntry,
97 timings: ArtifactCacheTimings,
98 maintenance: Option<ArtifactCacheMaintenance>,
99}
100
101#[derive(Clone, Debug, Eq, PartialEq)]
103pub struct ArtifactCacheArtifact {
104 name: String,
105 path: PathBuf,
106}
107
108#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
110pub struct ArtifactCacheTimings {
111 coordination_lock_wait: Duration,
112 content_lock_wait: Duration,
113 namespace_lock_wait: Duration,
114 input_capture: Duration,
115 cache_lookup: Duration,
116 caller_build: Option<Duration>,
117 output_validation: Duration,
118 publication: Duration,
119 materialization: Duration,
120 maintenance: Option<Duration>,
121 total: Duration,
122}
123
124pub struct ArtifactBuildTransaction {
126 spec: Box<ArtifactCacheSpec>,
127 resolved: ResolvedKey,
128 staging_directory: PathBuf,
129 entry_directory: PathBuf,
130 namespace_directory: PathBuf,
131 _coordination_lock: File,
132 _content_lock: File,
133 timings: ArtifactCacheTimings,
134 total_started: Instant,
135 caller_build_started: Instant,
136 staging_armed: bool,
137}
138
139#[non_exhaustive]
141#[derive(Debug)]
142pub enum ArtifactCacheError {
143 InvalidSpec {
145 message: String,
147 },
148 Io {
150 operation: &'static str,
152 path: PathBuf,
154 source: io::Error,
156 },
157 InputsChangedDuringPreparation {
159 before: InputDigest,
161 after: InputDigest,
163 },
164 InputsChangedDuringAcquisition {
166 before: InputDigest,
168 after: InputDigest,
170 },
171 CargoBuildInputsChanged {
173 label: String,
175 before: InputDigest,
177 after: InputDigest,
179 },
180 CargoBuildInputRevalidation {
182 label: String,
184 source: WasmBuildError,
186 },
187 InvalidOutputs {
189 outputs: Vec<(String, PathBuf)>,
191 },
192 UnknownOutput {
194 name: String,
196 },
197 FailedTransactionCleanup {
199 transaction_error: Box<Self>,
201 path: PathBuf,
203 source: io::Error,
205 },
206}
207
208#[derive(Clone, Debug, Eq, PartialEq)]
209struct LabeledPath {
210 label: String,
211 path: PathBuf,
212}
213
214#[derive(Clone, Eq, PartialEq)]
215struct LabeledIdentity {
216 label: String,
217 value: Vec<u8>,
218}
219
220#[derive(Clone, Eq, PartialEq)]
221struct CargoBuildInputSet {
222 label: String,
223 build_spec: WasmBuildSpec,
224 resolved: ResolvedCargoBuildInputs,
225}
226
227#[derive(Clone, Debug, Eq, PartialEq)]
228struct OutputSpec {
229 name: String,
230 destination: PathBuf,
231 validation: ArtifactOutputValidation,
232}
233
234#[derive(Clone, Copy)]
235struct ResolvedKey {
236 key: InputDigest,
237 input_digest: InputDigest,
238}
239
240impl ArtifactCacheSpec {
241 #[must_use]
247 pub fn new(cache_root: &Path, namespace: &str, recipe_id: &str) -> Self {
248 Self {
249 cache_root: cache_root.to_owned(),
250 namespace: namespace.to_owned(),
251 recipe_id: recipe_id.to_owned(),
252 coordination_scope: namespace.to_owned(),
253 inputs: Vec::new(),
254 tools: Vec::new(),
255 arguments: Vec::new(),
256 environment: BTreeMap::new(),
257 identities: Vec::new(),
258 cargo_build_inputs: Vec::new(),
259 outputs: Vec::new(),
260 prune_policy: None,
261 prune_interval: None,
262 }
263 }
264
265 #[must_use]
267 pub fn with_coordination_scope(mut self, coordination_scope: &str) -> Self {
268 coordination_scope.clone_into(&mut self.coordination_scope);
269 self
270 }
271
272 #[must_use]
274 pub fn with_input(mut self, label: &str, path: &Path) -> Self {
275 self.inputs.push(LabeledPath {
276 label: label.to_owned(),
277 path: path.to_owned(),
278 });
279 self
280 }
281
282 #[must_use]
284 pub fn with_tool(mut self, label: &str, path: &Path) -> Self {
285 self.tools.push(LabeledPath {
286 label: label.to_owned(),
287 path: path.to_owned(),
288 });
289 self
290 }
291
292 #[must_use]
294 pub fn with_arguments<I, S>(mut self, arguments: I) -> Self
295 where
296 I: IntoIterator<Item = S>,
297 S: AsRef<OsStr>,
298 {
299 self.arguments = arguments
300 .into_iter()
301 .map(|argument| argument.as_ref().to_owned())
302 .collect();
303 self
304 }
305
306 #[must_use]
311 pub fn with_environment<I, K, V>(mut self, environment: I) -> Self
312 where
313 I: IntoIterator<Item = (K, V)>,
314 K: Into<OsString>,
315 V: Into<OsString>,
316 {
317 self.environment.extend(
318 environment
319 .into_iter()
320 .map(|(name, value)| (name.into(), Some(value.into()))),
321 );
322 self
323 }
324
325 #[must_use]
330 pub fn with_unset_environment<I, S>(mut self, names: I) -> Self
331 where
332 I: IntoIterator<Item = S>,
333 S: Into<OsString>,
334 {
335 self.environment
336 .extend(names.into_iter().map(|name| (name.into(), None)));
337 self
338 }
339
340 #[must_use]
344 pub fn with_identity_bytes(mut self, label: &str, value: &[u8]) -> Self {
345 self.identities.push(LabeledIdentity {
346 label: label.to_owned(),
347 value: value.to_vec(),
348 });
349 self.identities
350 .sort_by(|left, right| left.label.cmp(&right.label));
351 self
352 }
353
354 #[must_use]
362 pub fn with_cargo_build_inputs(
363 mut self,
364 label: &str,
365 build_spec: &WasmBuildSpec,
366 resolved: &ResolvedCargoBuildInputs,
367 ) -> Self {
368 self.cargo_build_inputs.push(CargoBuildInputSet {
369 label: label.to_owned(),
370 build_spec: build_spec.clone(),
371 resolved: resolved.clone(),
372 });
373 self.cargo_build_inputs
374 .sort_by(|left, right| left.label.cmp(&right.label));
375 self
376 }
377
378 #[must_use]
380 pub fn with_output(self, name: &str, destination: &Path) -> Self {
381 self.with_output_validation(name, destination, ArtifactOutputValidation::NonEmptyFile)
382 }
383
384 #[must_use]
386 pub fn with_output_validation(
387 mut self,
388 name: &str,
389 destination: &Path,
390 validation: ArtifactOutputValidation,
391 ) -> Self {
392 self.outputs.push(OutputSpec {
393 name: name.to_owned(),
394 destination: destination.to_owned(),
395 validation,
396 });
397 self.outputs
398 .sort_by(|left, right| left.name.cmp(&right.name));
399 self
400 }
401
402 #[must_use]
404 pub const fn with_prune_policy(mut self, policy: ArtifactCachePrunePolicy) -> Self {
405 self.prune_policy = Some(policy);
406 self.prune_interval = None;
407 self
408 }
409
410 #[must_use]
418 pub const fn with_prune_policy_at_most_every(
419 mut self,
420 policy: ArtifactCachePrunePolicy,
421 minimum_interval: Duration,
422 ) -> Self {
423 self.prune_policy = Some(policy);
424 self.prune_interval = Some(minimum_interval);
425 self
426 }
427
428 #[must_use]
430 pub fn cache_root(&self) -> &Path {
431 &self.cache_root
432 }
433
434 #[must_use]
436 pub fn namespace(&self) -> &str {
437 &self.namespace
438 }
439
440 #[must_use]
442 pub fn recipe_id(&self) -> &str {
443 &self.recipe_id
444 }
445}
446
447impl std::fmt::Debug for ArtifactCacheSpec {
448 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
449 formatter
450 .debug_struct("ArtifactCacheSpec")
451 .field("cache_root", &self.cache_root)
452 .field("namespace", &self.namespace)
453 .field("recipe_id", &self.recipe_id)
454 .field("coordination_scope", &self.coordination_scope)
455 .field("inputs", &self.inputs)
456 .field("tools", &self.tools)
457 .field("argument_count", &self.arguments.len())
458 .field(
459 "environment_names",
460 &self.environment.keys().collect::<Vec<_>>(),
461 )
462 .field(
463 "identity_labels",
464 &self
465 .identities
466 .iter()
467 .map(|identity| identity.label.as_str())
468 .collect::<Vec<_>>(),
469 )
470 .field(
471 "cargo_build_input_labels",
472 &self
473 .cargo_build_inputs
474 .iter()
475 .map(|input| input.label.as_str())
476 .collect::<Vec<_>>(),
477 )
478 .field("outputs", &self.outputs)
479 .field("prune_policy", &self.prune_policy)
480 .field("prune_interval", &self.prune_interval)
481 .finish()
482 }
483}
484
485impl ArtifactCachePreparation {
486 #[must_use]
488 pub const fn reused_record(&self) -> Option<&ArtifactCacheRecord> {
489 match self {
490 Self::Reused(record) => Some(record),
491 Self::Build(_) => None,
492 }
493 }
494}
495
496impl ArtifactCacheOutcome {
497 #[must_use]
499 pub const fn record(&self) -> &ArtifactCacheRecord {
500 match self {
501 Self::Built(record) | Self::Reused(record) => record,
502 }
503 }
504
505 #[must_use]
507 pub const fn is_reused(&self) -> bool {
508 matches!(self, Self::Reused(_))
509 }
510}
511
512impl ArtifactCacheRecord {
513 #[must_use]
515 pub const fn key(&self) -> InputDigest {
516 self.key
517 }
518
519 #[must_use]
521 pub const fn input_digest(&self) -> InputDigest {
522 self.input_digest
523 }
524
525 #[must_use]
531 pub fn artifacts(&self) -> &[ArtifactCacheArtifact] {
532 &self.artifacts
533 }
534
535 #[must_use]
537 pub const fn timings(&self) -> ArtifactCacheTimings {
538 self.timings
539 }
540
541 #[must_use]
543 pub const fn maintenance(&self) -> Option<&ArtifactCacheMaintenance> {
544 self.maintenance.as_ref()
545 }
546}
547
548impl ArtifactCacheArtifact {
549 #[must_use]
551 pub fn name(&self) -> &str {
552 &self.name
553 }
554
555 #[must_use]
557 pub fn path(&self) -> &Path {
558 &self.path
559 }
560}
561
562impl ArtifactCacheTimings {
563 #[must_use]
565 pub const fn coordination_lock_wait(self) -> Duration {
566 self.coordination_lock_wait
567 }
568
569 #[must_use]
571 pub const fn content_lock_wait(self) -> Duration {
572 self.content_lock_wait
573 }
574
575 #[must_use]
577 pub const fn namespace_lock_wait(self) -> Duration {
578 self.namespace_lock_wait
579 }
580
581 #[must_use]
583 pub const fn input_capture(self) -> Duration {
584 self.input_capture
585 }
586
587 #[must_use]
589 pub const fn cache_lookup(self) -> Duration {
590 self.cache_lookup
591 }
592
593 #[must_use]
595 pub const fn caller_build(self) -> Option<Duration> {
596 self.caller_build
597 }
598
599 #[must_use]
601 pub const fn output_validation(self) -> Duration {
602 self.output_validation
603 }
604
605 #[must_use]
607 pub const fn publication(self) -> Duration {
608 self.publication
609 }
610
611 #[must_use]
613 pub const fn materialization(self) -> Duration {
614 self.materialization
615 }
616
617 #[must_use]
619 pub const fn maintenance(self) -> Option<Duration> {
620 self.maintenance
621 }
622
623 #[must_use]
625 pub const fn total(self) -> Duration {
626 self.total
627 }
628
629 pub(super) const fn saturating_add(self, other: Self) -> Self {
630 Self {
631 coordination_lock_wait: self
632 .coordination_lock_wait
633 .saturating_add(other.coordination_lock_wait),
634 content_lock_wait: self
635 .content_lock_wait
636 .saturating_add(other.content_lock_wait),
637 namespace_lock_wait: self
638 .namespace_lock_wait
639 .saturating_add(other.namespace_lock_wait),
640 input_capture: self.input_capture.saturating_add(other.input_capture),
641 cache_lookup: self.cache_lookup.saturating_add(other.cache_lookup),
642 caller_build: saturating_add_optional_duration(self.caller_build, other.caller_build),
643 output_validation: self
644 .output_validation
645 .saturating_add(other.output_validation),
646 publication: self.publication.saturating_add(other.publication),
647 materialization: self.materialization.saturating_add(other.materialization),
648 maintenance: saturating_add_optional_duration(self.maintenance, other.maintenance),
649 total: self.total.saturating_add(other.total),
650 }
651 }
652}
653
654impl std::fmt::Display for ArtifactCacheTimings {
655 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
656 write!(
657 formatter,
658 "total={:?} coordination_lock={:?} content_lock={:?} namespace_lock={:?} inputs={:?} lookup={:?} build={:?} validation={:?} publication={:?} materialization={:?} maintenance={:?}",
659 self.total,
660 self.coordination_lock_wait,
661 self.content_lock_wait,
662 self.namespace_lock_wait,
663 self.input_capture,
664 self.cache_lookup,
665 self.caller_build,
666 self.output_validation,
667 self.publication,
668 self.materialization,
669 self.maintenance,
670 )
671 }
672}
673
674impl std::fmt::Display for ArtifactCacheOutcome {
675 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
676 let state = if self.is_reused() { "reused" } else { "built" };
677 write!(
678 formatter,
679 "{state} key={} artifacts={} {}",
680 self.record().key,
681 self.record().artifacts.len(),
682 self.record().timings,
683 )
684 }
685}
686
687impl ArtifactBuildTransaction {
688 #[must_use]
693 pub fn staging_directory(&self) -> &Path {
694 &self.staging_directory
695 }
696
697 pub fn output_path(&self, name: &str) -> Result<PathBuf, ArtifactCacheError> {
702 self.output_index(name)
703 .map(|index| staged_output_path(&self.staging_directory, index))
704 .ok_or_else(|| ArtifactCacheError::UnknownOutput {
705 name: name.to_owned(),
706 })
707 }
708
709 pub fn import_output(&self, name: &str, source: &Path) -> Result<(), ArtifactCacheError> {
714 let destination = self.output_path(name)?;
715 copy_file_atomic(source, &destination).map_err(|source_error| ArtifactCacheError::Io {
716 operation: "import artifact output into staging",
717 path: destination,
718 source: source_error,
719 })?;
720 Ok(())
721 }
722
723 pub fn commit(mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
729 let result = self.commit_inner();
730 match result {
731 Ok(outcome) => Ok(outcome),
732 Err(transaction_error) if self.staging_armed => {
733 let path = self.staging_directory.clone();
734 match remove_path_if_present(&path) {
735 Ok(()) => {
736 self.staging_armed = false;
737 Err(transaction_error)
738 }
739 Err(source) => Err(ArtifactCacheError::FailedTransactionCleanup {
740 transaction_error: Box::new(transaction_error),
741 path,
742 source,
743 }),
744 }
745 }
746 Err(transaction_error) => Err(transaction_error),
747 }
748 }
749
750 pub fn abort(mut self) -> Result<(), ArtifactCacheError> {
755 remove_path_if_present(&self.staging_directory).map_err(|source| {
756 ArtifactCacheError::Io {
757 operation: "abort artifact cache transaction",
758 path: self.staging_directory.clone(),
759 source,
760 }
761 })?;
762 self.staging_armed = false;
763 Ok(())
764 }
765
766 fn output_index(&self, name: &str) -> Option<usize> {
767 self.spec
768 .outputs
769 .iter()
770 .position(|output| output.name == name)
771 }
772
773 fn commit_inner(&mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
774 self.timings.caller_build = Some(self.caller_build_started.elapsed());
775
776 let validation_started = Instant::now();
777 let output_info = inspect_complete_output_set(&self.spec, &self.staging_directory)?;
778 self.timings.output_validation = validation_started.elapsed();
779
780 let capture_started = Instant::now();
781 revalidate_cargo_build_input_fingerprints(&self.spec)?;
782 let verified = resolve_key(&self.spec)?;
783 self.timings.input_capture = self
784 .timings
785 .input_capture
786 .saturating_add(capture_started.elapsed());
787 if verified.input_digest != self.resolved.input_digest {
788 return Err(ArtifactCacheError::InputsChangedDuringAcquisition {
789 before: self.resolved.input_digest,
790 after: verified.input_digest,
791 });
792 }
793
794 let publication_started = Instant::now();
795 let manifest =
796 manifest_contents(self.resolved.key, &self.spec, output_info.iter().copied());
797 ic_host_fs::durable::write_bytes(
798 &self.staging_directory.join(MANIFEST_FILE),
799 manifest.as_bytes(),
800 )
801 .map_err(|source| ArtifactCacheError::Io {
802 operation: "write artifact cache manifest",
803 path: self.staging_directory.join(MANIFEST_FILE),
804 source: io::Error::other(source),
805 })?;
806 let namespace_lock_path = namespace_lock_path(&self.spec);
807 let (_namespace_lock, namespace_wait) =
808 lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
809 self.timings.namespace_lock_wait = self
810 .timings
811 .namespace_lock_wait
812 .saturating_add(namespace_wait);
813 remove_unretained_entry(&self.entry_directory).map_err(artifact_cache_fs_error)?;
814 fs::rename(&self.staging_directory, &self.entry_directory).map_err(|source| {
815 ArtifactCacheError::Io {
816 operation: "publish artifact cache entry",
817 path: self.entry_directory.clone(),
818 source,
819 }
820 })?;
821 self.staging_armed = false;
822 self.timings.publication = publication_started.elapsed();
823
824 let materialization_started = Instant::now();
825 materialize_outputs(&self.spec, &self.entry_directory, None)?;
826 self.timings.materialization = materialization_started.elapsed();
827 record_cache_entry_use(&self.entry_directory).map_err(artifact_cache_fs_error)?;
828 let (maintenance, maintenance_timing) = perform_maintenance_locked(
829 &self.spec,
830 &self.namespace_directory,
831 &self.entry_directory,
832 );
833 self.timings.maintenance = maintenance_timing;
834 self.timings.total = self.total_started.elapsed();
835
836 Ok(ArtifactCacheOutcome::Built(cache_record(
837 &self.spec,
838 self.resolved,
839 self.timings,
840 maintenance,
841 )?))
842 }
843}
844
845impl Drop for ArtifactBuildTransaction {
846 fn drop(&mut self) {
847 if self.staging_armed {
848 let _ = remove_path_if_present(&self.staging_directory);
849 }
850 }
851}
852
853pub fn prepare_artifact_cache(
872 spec: &ArtifactCacheSpec,
873) -> Result<ArtifactCachePreparation, ArtifactCacheError> {
874 let total_started = Instant::now();
875 validate_spec(spec)?;
876 validate_filesystem_boundaries(spec)?;
877 let namespace_directory = initialize_cache(spec)?;
878
879 let coordination_lock_path = coordination_lock_path(spec);
880 let (coordination_lock, coordination_wait) =
881 lock_cache_file(&coordination_lock_path).map_err(artifact_cache_fs_error)?;
882 let mut timings = ArtifactCacheTimings {
883 coordination_lock_wait: coordination_wait,
884 ..ArtifactCacheTimings::default()
885 };
886 let initial_cargo_started = Instant::now();
887 revalidate_cargo_build_input_fingerprints(spec)?;
888 timings.input_capture = initial_cargo_started.elapsed();
889 let mut last_change = None;
890
891 for _ in 0..MAX_PREPARATION_RETRIES {
892 let capture_started = Instant::now();
893 let resolved = resolve_key(spec)?;
894 timings.input_capture = timings
895 .input_capture
896 .saturating_add(capture_started.elapsed());
897
898 let content_lock_path = content_lock_path(spec, resolved.key);
899 let (content_lock, content_wait) =
900 lock_cache_file(&content_lock_path).map_err(artifact_cache_fs_error)?;
901 timings.content_lock_wait = timings.content_lock_wait.saturating_add(content_wait);
902
903 let verification_started = Instant::now();
904 let verified = resolve_key(spec)?;
905 timings.input_capture = timings
906 .input_capture
907 .saturating_add(verification_started.elapsed());
908 if resolved.input_digest != verified.input_digest {
909 last_change = Some((resolved.input_digest, verified.input_digest));
910 drop(content_lock);
911 continue;
912 }
913
914 let entry_directory = entry_directory(&namespace_directory, resolved.key);
915 let namespace_lock_path = namespace_lock_path(spec);
916 let (namespace_lock, namespace_wait) =
917 lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
918 timings.namespace_lock_wait = timings.namespace_lock_wait.saturating_add(namespace_wait);
919 let lookup_started = Instant::now();
920 let verified_outputs = validated_cache_outputs(spec, resolved.key, &entry_directory)?;
921 timings.cache_lookup = timings
922 .cache_lookup
923 .saturating_add(lookup_started.elapsed());
924
925 if let Some(output_info) = verified_outputs {
926 let materialization_started = Instant::now();
927 materialize_outputs(spec, &entry_directory, Some(&output_info))?;
928 timings.materialization = timings
929 .materialization
930 .saturating_add(materialization_started.elapsed());
931 let after_started = Instant::now();
932 revalidate_cargo_build_input_fingerprints(spec)?;
933 let after = resolve_key(spec)?;
934 timings.input_capture = timings
935 .input_capture
936 .saturating_add(after_started.elapsed());
937 if after.input_digest != resolved.input_digest {
938 last_change = Some((resolved.input_digest, after.input_digest));
939 drop(namespace_lock);
940 drop(content_lock);
941 continue;
942 }
943 record_cache_entry_use(&entry_directory).map_err(artifact_cache_fs_error)?;
944 let (maintenance, maintenance_timing) =
945 perform_maintenance_locked(spec, &namespace_directory, &entry_directory);
946 timings.maintenance = maintenance_timing;
947 timings.total = total_started.elapsed();
948 return Ok(ArtifactCachePreparation::Reused(cache_record(
949 spec,
950 resolved,
951 timings,
952 maintenance,
953 )?));
954 }
955
956 remove_unretained_entry(&entry_directory).map_err(artifact_cache_fs_error)?;
957 let staging_directory = create_staging_directory(&namespace_directory, resolved.key)?;
958 drop(namespace_lock);
959 return Ok(ArtifactCachePreparation::Build(ArtifactBuildTransaction {
960 spec: Box::new(spec.clone()),
961 resolved,
962 staging_directory,
963 entry_directory,
964 namespace_directory,
965 _coordination_lock: coordination_lock,
966 _content_lock: content_lock,
967 timings,
968 total_started,
969 caller_build_started: Instant::now(),
970 staging_armed: true,
971 }));
972 }
973
974 let (before, after) = last_change.expect("preparation retries require a recorded input change");
975 Err(ArtifactCacheError::InputsChangedDuringPreparation { before, after })
976}
977
978fn revalidate_cargo_build_input_fingerprints(
979 spec: &ArtifactCacheSpec,
980) -> Result<(), ArtifactCacheError> {
981 for cargo_input in &spec.cargo_build_inputs {
982 let current = resolve_cargo_build_inputs(&cargo_input.build_spec).map_err(|source| {
983 ArtifactCacheError::CargoBuildInputRevalidation {
984 label: cargo_input.label.clone(),
985 source,
986 }
987 })?;
988 let before_validation = cargo_input.resolved.validation_digest();
989 let after_validation = current.validation_digest();
990 if after_validation != before_validation {
991 return Err(ArtifactCacheError::CargoBuildInputsChanged {
992 label: cargo_input.label.clone(),
993 before: before_validation,
994 after: after_validation,
995 });
996 }
997 let before = cargo_input.resolved.fingerprint();
998 let after = current.fingerprint();
999 if after != before {
1000 return Err(ArtifactCacheError::CargoBuildInputsChanged {
1001 label: cargo_input.label.clone(),
1002 before,
1003 after,
1004 });
1005 }
1006 }
1007 Ok(())
1008}
1009
1010pub fn prune_artifact_cache(
1022 cache_root: &Path,
1023 namespace: &str,
1024 policy: ArtifactCachePrunePolicy,
1025) -> Result<ArtifactCachePruneReport, ArtifactCacheError> {
1026 validate_identifier("namespace", namespace)?;
1027 if cache_root.as_os_str().is_empty() {
1028 return invalid_spec("cache root must not be empty");
1029 }
1030 ensure_cache_directory_tag(cache_root).map_err(artifact_cache_fs_error)?;
1031 let namespace_directory = namespace_directory_for(cache_root, namespace);
1032 let lock_path = namespace_lock_path_for(cache_root, namespace);
1033 let (_lock, _) = lock_cache_file(&lock_path).map_err(artifact_cache_fs_error)?;
1034 prune_artifact_namespace_locked(cache_root, &namespace_directory, policy, None)
1035}
1036
1037fn initialize_cache(spec: &ArtifactCacheSpec) -> Result<PathBuf, ArtifactCacheError> {
1038 ensure_cache_directory_tag(&spec.cache_root).map_err(artifact_cache_fs_error)?;
1039 let namespace = namespace_directory(spec);
1040 fs::create_dir_all(entries_directory(&namespace)).map_err(|source| ArtifactCacheError::Io {
1041 operation: "create artifact cache namespace",
1042 path: namespace.clone(),
1043 source,
1044 })?;
1045 Ok(namespace)
1046}
1047
1048fn validate_spec(spec: &ArtifactCacheSpec) -> Result<(), ArtifactCacheError> {
1049 validate_identifier("namespace", &spec.namespace)?;
1050 validate_identifier("recipe identity", &spec.recipe_id)?;
1051 validate_identifier("coordination scope", &spec.coordination_scope)?;
1052 if spec.cache_root.as_os_str().is_empty() {
1053 return invalid_spec("cache root must not be empty");
1054 }
1055 if spec.outputs.is_empty() {
1056 return invalid_spec("at least one artifact output is required");
1057 }
1058
1059 let mut labels = BTreeSet::new();
1060 for (kind, paths) in [("input", &spec.inputs), ("tool", &spec.tools)] {
1061 for path in paths {
1062 validate_path_label(kind, &path.label)?;
1063 if !labels.insert((kind, path.label.as_str())) {
1064 return invalid_spec(&format!("duplicate {kind} label `{}`", path.label));
1065 }
1066 }
1067 }
1068 let mut identity_labels = BTreeSet::new();
1069 for identity in &spec.identities {
1070 validate_label("identity", &identity.label)?;
1071 if !identity_labels.insert(&identity.label) {
1072 return invalid_spec(&format!("duplicate identity label `{}`", identity.label));
1073 }
1074 }
1075 let mut cargo_input_labels = BTreeSet::new();
1076 for cargo_inputs in &spec.cargo_build_inputs {
1077 validate_label("Cargo build input", &cargo_inputs.label)?;
1078 if !cargo_input_labels.insert(&cargo_inputs.label) {
1079 return invalid_spec(&format!(
1080 "duplicate Cargo build input label `{}`",
1081 cargo_inputs.label
1082 ));
1083 }
1084 }
1085 if spec
1086 .environment
1087 .keys()
1088 .any(|name| name.as_os_str().is_empty())
1089 {
1090 return invalid_spec("environment names must not be empty");
1091 }
1092 let mut output_names = BTreeSet::new();
1093 let mut destinations = BTreeSet::new();
1094 for output in &spec.outputs {
1095 validate_output_name(&output.name)?;
1096 if !output_names.insert(&output.name) {
1097 return invalid_spec(&format!("duplicate output name `{}`", output.name));
1098 }
1099 if output.destination.as_os_str().is_empty() {
1100 return invalid_spec(&format!(
1101 "output `{}` destination must not be empty",
1102 output.name
1103 ));
1104 }
1105 if !destinations.insert(&output.destination) {
1106 return invalid_spec(&format!(
1107 "output `{}` shares a destination with another output",
1108 output.name
1109 ));
1110 }
1111 }
1112 Ok(())
1113}
1114
1115fn validate_filesystem_boundaries(spec: &ArtifactCacheSpec) -> Result<(), ArtifactCacheError> {
1116 let cache_root =
1117 canonicalize_allow_missing(&spec.cache_root).map_err(|source| ArtifactCacheError::Io {
1118 operation: "resolve artifact cache root",
1119 path: spec.cache_root.clone(),
1120 source,
1121 })?;
1122 let mut declared_paths = Vec::with_capacity(spec.inputs.len() + spec.tools.len());
1123 for (kind, labeled_paths) in [("input", &spec.inputs), ("tool", &spec.tools)] {
1124 for labeled in labeled_paths {
1125 let canonical =
1126 canonicalize_path(&labeled.path, "canonicalize declared artifact cache path")?;
1127 if canonical.starts_with(&cache_root) {
1128 return invalid_spec(&format!(
1129 "{kind} `{}` must not be located inside the artifact cache root",
1130 labeled.label
1131 ));
1132 }
1133 let is_directory = fs::metadata(&canonical)
1134 .map_err(|source| ArtifactCacheError::Io {
1135 operation: "inspect declared artifact cache path",
1136 path: canonical.clone(),
1137 source,
1138 })?
1139 .is_dir();
1140 let entry_path =
1141 resolve_file_entry(&labeled.path, "resolve declared artifact cache path entry")?;
1142 if entry_path != canonical {
1143 declared_paths.push((kind, labeled.label.as_str(), entry_path, false));
1144 }
1145 declared_paths.push((kind, labeled.label.as_str(), canonical, is_directory));
1146 }
1147 }
1148 for cargo_inputs in &spec.cargo_build_inputs {
1149 if resolved_cargo_inputs_watch_path(&cargo_inputs.resolved, &cache_root)? {
1150 return invalid_spec(&format!(
1151 "artifact cache root must be outside resolved Cargo build inputs `{}` or inside one of their generated-state exclusions",
1152 cargo_inputs.label
1153 ));
1154 }
1155 }
1156
1157 let mut destinations = BTreeSet::new();
1158 for output in &spec.outputs {
1159 let destination =
1160 resolve_file_entry(&output.destination, "resolve artifact output destination")?;
1161 if destination.starts_with(&cache_root) {
1162 return invalid_spec(&format!(
1163 "output `{}` destination must be outside the artifact cache root",
1164 output.name
1165 ));
1166 }
1167 if destinations.iter().any(|other: &PathBuf| {
1168 destination.starts_with(other) || other.starts_with(&destination)
1169 }) {
1170 return invalid_spec(&format!(
1171 "output `{}` destination overlaps another output",
1172 output.name
1173 ));
1174 }
1175 destinations.insert(destination.clone());
1176 for (kind, label, declared, is_directory) in &declared_paths {
1177 if destination == *declared || (*is_directory && destination.starts_with(declared)) {
1178 return invalid_spec(&format!(
1179 "output `{}` destination overlaps declared {kind} `{label}`",
1180 output.name
1181 ));
1182 }
1183 }
1184 for cargo_inputs in &spec.cargo_build_inputs {
1185 if resolved_cargo_inputs_watch_path(&cargo_inputs.resolved, &destination)? {
1186 return invalid_spec(&format!(
1187 "output `{}` destination must be outside resolved Cargo build inputs `{}` or inside one of their generated-state exclusions",
1188 output.name, cargo_inputs.label
1189 ));
1190 }
1191 }
1192 match fs::symlink_metadata(&destination) {
1196 Ok(metadata)
1197 if metadata.is_dir()
1198 || (metadata.file_type().is_symlink()
1199 && fs::metadata(&destination).is_ok_and(|target| target.is_dir())) =>
1200 {
1201 return invalid_spec(&format!(
1202 "output `{}` destination must not be an existing directory",
1203 output.name
1204 ));
1205 }
1206 Ok(_) => {}
1207 Err(error) if error.kind() == io::ErrorKind::NotFound => {}
1208 Err(source) => {
1209 return Err(ArtifactCacheError::Io {
1210 operation: "inspect artifact output destination",
1211 path: destination,
1212 source,
1213 });
1214 }
1215 }
1216 }
1217 Ok(())
1218}
1219
1220fn resolve_file_entry(path: &Path, operation: &'static str) -> Result<PathBuf, ArtifactCacheError> {
1223 let resolved = match (path.parent(), path.file_name()) {
1224 (Some(parent), Some(name)) => {
1225 canonicalize_allow_missing(parent).map(|parent| parent.join(name))
1226 }
1227 _ => canonicalize_allow_missing(path),
1228 };
1229 resolved.map_err(|source| ArtifactCacheError::Io {
1230 operation,
1231 path: path.to_owned(),
1232 source,
1233 })
1234}
1235
1236fn resolved_cargo_inputs_watch_path(
1237 resolved: &ResolvedCargoBuildInputs,
1238 candidate: &Path,
1239) -> Result<bool, ArtifactCacheError> {
1240 for exclusion in resolved.exclusions() {
1241 let exclusion =
1242 canonicalize_allow_missing(exclusion).map_err(|source| ArtifactCacheError::Io {
1243 operation: "resolve Cargo build input exclusion",
1244 path: exclusion.clone(),
1245 source,
1246 })?;
1247 if candidate.starts_with(exclusion) {
1248 return Ok(false);
1249 }
1250 }
1251
1252 for input in resolved.inputs() {
1253 let path = canonicalize_path(input.path(), "canonicalize resolved Cargo build input")?;
1254 let is_directory = fs::metadata(&path)
1255 .map_err(|source| ArtifactCacheError::Io {
1256 operation: "inspect resolved Cargo build input",
1257 path: path.clone(),
1258 source,
1259 })?
1260 .is_dir();
1261 if candidate == path || (is_directory && candidate.starts_with(path)) {
1262 return Ok(true);
1263 }
1264 }
1265 Ok(false)
1266}
1267
1268fn canonicalize_path(path: &Path, operation: &'static str) -> Result<PathBuf, ArtifactCacheError> {
1269 path.canonicalize()
1270 .map_err(|source| ArtifactCacheError::Io {
1271 operation,
1272 path: path.to_owned(),
1273 source,
1274 })
1275}
1276
1277fn validate_identifier(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1278 if value.is_empty() {
1279 return invalid_spec(&format!("{kind} must not be empty"));
1280 }
1281 if value.len() > 256 {
1282 return invalid_spec(&format!("{kind} must not exceed 256 bytes"));
1283 }
1284 Ok(())
1285}
1286
1287fn validate_label(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1288 if value.is_empty() {
1289 return invalid_spec(&format!("{kind} label must not be empty"));
1290 }
1291 if value.len() > 256 {
1292 return invalid_spec(&format!("{kind} label must not exceed 256 bytes"));
1293 }
1294 Ok(())
1295}
1296
1297fn validate_path_label(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1298 validate_label(kind, value)?;
1299 if value
1300 .bytes()
1301 .any(|byte| !(byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'/')))
1302 || value
1303 .split('/')
1304 .any(|component| component.is_empty() || matches!(component, "." | ".."))
1305 {
1306 return invalid_spec(&format!(
1307 "{kind} label `{value}` must be a portable relative logical path"
1308 ));
1309 }
1310 Ok(())
1311}
1312
1313fn validate_output_name(name: &str) -> Result<(), ArtifactCacheError> {
1314 if name.is_empty() || name.len() > 128 {
1315 return invalid_spec("output names must contain 1 to 128 bytes");
1316 }
1317 if !name
1318 .bytes()
1319 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
1320 {
1321 return invalid_spec(&format!(
1322 "output name `{name}` must use only ASCII letters, digits, dot, dash, or underscore"
1323 ));
1324 }
1325 if matches!(name, "." | "..") {
1326 return invalid_spec("output names must not be dot path components");
1327 }
1328 Ok(())
1329}
1330
1331fn invalid_spec<T>(message: &str) -> Result<T, ArtifactCacheError> {
1332 Err(ArtifactCacheError::InvalidSpec {
1333 message: message.to_owned(),
1334 })
1335}
1336
1337fn resolve_key(spec: &ArtifactCacheSpec) -> Result<ResolvedKey, ArtifactCacheError> {
1338 let paths = spec
1339 .inputs
1340 .iter()
1341 .map(|input| (PathBuf::from("input").join(&input.label), &input.path))
1342 .chain(
1343 spec.tools
1344 .iter()
1345 .map(|tool| (PathBuf::from("tool").join(&tool.label), &tool.path)),
1346 );
1347 let declared_input_digest = digest_labeled_paths(
1348 "artifact-set-inputs-v1",
1349 paths,
1350 std::slice::from_ref(&spec.cache_root),
1351 )
1352 .map_err(|source| ArtifactCacheError::Io {
1353 operation: "hash artifact cache inputs",
1354 path: spec.cache_root.clone(),
1355 source,
1356 })?;
1357 let input_digest = if spec.cargo_build_inputs.is_empty() {
1358 declared_input_digest
1359 } else {
1360 let mut inputs = InputHasher::new("artifact-set-inputs-with-cargo-v1");
1361 inputs.field("declared-input-digest", declared_input_digest.as_bytes());
1362 for cargo_input in &spec.cargo_build_inputs {
1363 let current = cargo_input
1364 .resolved
1365 .current_validation_digest()
1366 .map_err(|source| ArtifactCacheError::CargoBuildInputRevalidation {
1367 label: cargo_input.label.clone(),
1368 source,
1369 })?;
1370 let before = cargo_input.resolved.validation_digest();
1371 if current != before {
1372 return Err(ArtifactCacheError::CargoBuildInputsChanged {
1373 label: cargo_input.label.clone(),
1374 before,
1375 after: current,
1376 });
1377 }
1378 inputs.field("cargo-input-label", cargo_input.label.as_bytes());
1379 inputs.field(
1380 "cargo-build-fingerprint",
1381 cargo_input.resolved.fingerprint().as_bytes(),
1382 );
1383 inputs.field(
1384 "cargo-input-digest",
1385 cargo_input.resolved.input_digest().as_bytes(),
1386 );
1387 }
1388 inputs.finish()
1389 };
1390
1391 let mut hasher = InputHasher::new(ARTIFACT_CACHE_FORMAT);
1392 hasher.field("namespace", spec.namespace.as_bytes());
1393 hasher.field("recipe-id", spec.recipe_id.as_bytes());
1394 hasher.field("input-digest", input_digest.as_bytes());
1395 for argument in &spec.arguments {
1396 hasher.field("argument", &os_bytes(argument));
1397 }
1398 for (name, value) in &spec.environment {
1399 hasher.field("environment-name", &os_bytes(name));
1400 match value {
1401 Some(value) => hasher.field("environment-value", &os_bytes(value)),
1402 None => hasher.field("environment-unset", b""),
1403 }
1404 }
1405 for identity in &spec.identities {
1406 hasher.field("identity-label", identity.label.as_bytes());
1407 hasher.field("identity-value", &identity.value);
1408 }
1409 for output in &spec.outputs {
1410 hasher.field("output-name", output.name.as_bytes());
1411 hasher.field(
1412 "output-validation",
1413 output.validation.cache_token().as_bytes(),
1414 );
1415 }
1416 Ok(ResolvedKey {
1417 key: hasher.finish(),
1418 input_digest,
1419 })
1420}
1421
1422fn validated_cache_outputs(
1423 spec: &ArtifactCacheSpec,
1424 key: InputDigest,
1425 entry: &Path,
1426) -> Result<Option<Vec<FileDigest>>, ArtifactCacheError> {
1427 let entry_metadata = match fs::symlink_metadata(entry) {
1428 Ok(metadata) => metadata,
1429 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1430 Err(source) => {
1431 return Err(ArtifactCacheError::Io {
1432 operation: "inspect artifact cache entry",
1433 path: entry.to_owned(),
1434 source,
1435 });
1436 }
1437 };
1438 if !entry_metadata.file_type().is_dir() || !cache_entry_root_is_valid(entry)? {
1439 return Ok(None);
1440 }
1441 let manifest_path = entry.join(MANIFEST_FILE);
1442 let manifest_metadata = match fs::symlink_metadata(&manifest_path) {
1443 Ok(metadata) => metadata,
1444 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1445 Err(source) => {
1446 return Err(ArtifactCacheError::Io {
1447 operation: "inspect artifact cache manifest",
1448 path: manifest_path,
1449 source,
1450 });
1451 }
1452 };
1453 if !manifest_metadata.file_type().is_file() {
1454 return Ok(None);
1455 }
1456 let maximum_len = manifest_contents(
1460 key,
1461 spec,
1462 std::iter::repeat(FileDigest {
1463 bytes: u64::MAX,
1464 digest: key,
1465 }),
1466 )
1467 .len();
1468 let Some(manifest) = read_file_with_limit(&manifest_path, maximum_len).map_err(|source| {
1469 ArtifactCacheError::Io {
1470 operation: "read artifact cache manifest",
1471 path: manifest_path,
1472 source,
1473 }
1474 })?
1475 else {
1476 return Ok(None);
1477 };
1478 if !manifest.starts_with(manifest_header(key).as_bytes()) {
1479 return Ok(None);
1480 }
1481 let Some(output_info) = inspect_cached_output_set(spec, entry)? else {
1482 return Ok(None);
1483 };
1484 Ok(
1485 (manifest == manifest_contents(key, spec, output_info.iter().copied()).as_bytes())
1486 .then_some(output_info),
1487 )
1488}
1489
1490fn inspect_complete_output_set(
1491 spec: &ArtifactCacheSpec,
1492 root: &Path,
1493) -> Result<Vec<FileDigest>, ArtifactCacheError> {
1494 let mut info = Vec::new();
1495 let mut invalid = Vec::new();
1496 let output_directory = root.join("outputs");
1497 if !is_plain_directory(
1498 &output_directory,
1499 "inspect artifact staging output directory",
1500 )? {
1501 invalid.push(("<outputs>".to_owned(), output_directory));
1502 return Err(ArtifactCacheError::InvalidOutputs { outputs: invalid });
1503 }
1504 let outputs = &spec.outputs;
1505 for (index, output) in outputs.iter().enumerate() {
1506 let path = staged_output_path(root, index);
1507 match inspect_artifact(&path, output.validation) {
1508 Ok(Some(artifact)) => info.push(artifact),
1509 Ok(None) => invalid.push((output.name.clone(), path)),
1510 Err(source) => {
1511 return Err(ArtifactCacheError::Io {
1512 operation: "inspect staged artifact output",
1513 path,
1514 source,
1515 });
1516 }
1517 }
1518 }
1519 invalid.extend(
1520 undeclared_output_paths(root, outputs.len())?
1521 .into_iter()
1522 .map(|path| ("<undeclared>".to_owned(), path)),
1523 );
1524 invalid.extend(
1525 undeclared_child_paths(
1526 root,
1527 |name| name == "outputs",
1528 "read artifact staging directory",
1529 )?
1530 .into_iter()
1531 .map(|path| ("<undeclared>".to_owned(), path)),
1532 );
1533 if invalid.is_empty() {
1534 Ok(info)
1535 } else {
1536 Err(ArtifactCacheError::InvalidOutputs { outputs: invalid })
1537 }
1538}
1539
1540fn inspect_cached_output_set(
1541 spec: &ArtifactCacheSpec,
1542 root: &Path,
1543) -> Result<Option<Vec<FileDigest>>, ArtifactCacheError> {
1544 let mut info = Vec::new();
1545 let outputs = &spec.outputs;
1546 for (index, output) in outputs.iter().enumerate() {
1547 let path = staged_output_path(root, index);
1548 match inspect_artifact(&path, output.validation) {
1549 Ok(Some(artifact)) => info.push(artifact),
1550 Ok(None) => return Ok(None),
1551 Err(source) => {
1552 return Err(ArtifactCacheError::Io {
1553 operation: "inspect cached artifact output",
1554 path,
1555 source,
1556 });
1557 }
1558 }
1559 }
1560 if !undeclared_output_paths(root, outputs.len())?.is_empty() {
1561 return Ok(None);
1562 }
1563 Ok(Some(info))
1564}
1565
1566fn undeclared_output_paths(
1567 root: &Path,
1568 output_count: usize,
1569) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1570 let output_directory = root.join("outputs");
1571 undeclared_child_paths(
1572 &output_directory,
1573 |name| {
1574 let Some(name) = name.to_str() else {
1575 return false;
1576 };
1577 name.strip_suffix(OUTPUT_FILE_SUFFIX)
1578 .and_then(|index| index.parse::<usize>().ok())
1579 .is_some_and(|index| index < output_count && name == format_output_index(index))
1580 },
1581 "read artifact output directory",
1582 )
1583}
1584
1585fn undeclared_child_paths(
1586 directory: &Path,
1587 is_expected: impl Fn(&OsStr) -> bool,
1588 operation: &'static str,
1589) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1590 let entries = fs::read_dir(directory).map_err(|source| ArtifactCacheError::Io {
1591 operation,
1592 path: directory.to_owned(),
1593 source,
1594 })?;
1595 let mut undeclared = Vec::new();
1596 for entry in entries {
1597 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1598 operation,
1599 path: directory.to_owned(),
1600 source,
1601 })?;
1602 if !is_expected(&entry.file_name()) {
1603 undeclared.push(entry.path());
1604 }
1605 }
1606 Ok(undeclared)
1607}
1608
1609fn cache_entry_root_is_valid(root: &Path) -> Result<bool, ArtifactCacheError> {
1610 let expected = [
1611 "outputs",
1612 MANIFEST_FILE,
1613 LAST_USED_FILE,
1614 super::cache_fs::RETENTION_LOCK_FILE,
1615 ];
1616 if !undeclared_child_paths(
1617 root,
1618 |name| expected.iter().any(|expected| name == *expected),
1619 "read artifact cache entry",
1620 )?
1621 .is_empty()
1622 {
1623 return Ok(false);
1624 }
1625 if !is_plain_directory(
1626 &root.join("outputs"),
1627 "inspect artifact cache output directory",
1628 )? {
1629 return Ok(false);
1630 }
1631 let last_used = root.join(LAST_USED_FILE);
1632 match fs::symlink_metadata(&last_used) {
1633 Ok(metadata) => Ok(metadata.file_type().is_file()),
1634 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(true),
1635 Err(source) => Err(ArtifactCacheError::Io {
1636 operation: "inspect artifact cache use marker",
1637 path: last_used,
1638 source,
1639 }),
1640 }
1641}
1642
1643fn is_plain_directory(path: &Path, operation: &'static str) -> Result<bool, ArtifactCacheError> {
1644 match fs::symlink_metadata(path) {
1645 Ok(metadata) => Ok(metadata.file_type().is_dir()),
1646 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
1647 Err(source) => Err(ArtifactCacheError::Io {
1648 operation,
1649 path: path.to_owned(),
1650 source,
1651 }),
1652 }
1653}
1654
1655fn inspect_artifact(
1656 path: &Path,
1657 validation: ArtifactOutputValidation,
1658) -> io::Result<Option<FileDigest>> {
1659 let metadata = match fs::symlink_metadata(path) {
1660 Ok(metadata) => metadata,
1661 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1662 Err(error) => return Err(error),
1663 };
1664 if !metadata.file_type().is_file() {
1665 return Ok(None);
1666 }
1667 if validation == ArtifactOutputValidation::NonEmptyFile && metadata.len() == 0 {
1668 return Ok(None);
1669 }
1670 digest_file("artifact-set-output-v1", path).map(Some)
1671}
1672
1673fn manifest_header(key: InputDigest) -> String {
1674 format!("{ARTIFACT_CACHE_FORMAT}\nkey:{key}\n")
1675}
1676
1677fn manifest_contents(
1678 key: InputDigest,
1679 spec: &ArtifactCacheSpec,
1680 output_info: impl IntoIterator<Item = FileDigest>,
1681) -> String {
1682 let mut manifest = manifest_header(key);
1683 for ((index, output), info) in spec.outputs.iter().enumerate().zip(output_info) {
1684 writeln!(
1685 manifest,
1686 "output:{index}:{}:{}:{}:{}",
1687 output.name,
1688 output.validation.cache_token(),
1689 info.bytes,
1690 info.digest,
1691 )
1692 .expect("writing an artifact manifest to a String cannot fail");
1693 }
1694 manifest
1695}
1696
1697fn materialize_outputs(
1698 spec: &ArtifactCacheSpec,
1699 entry: &Path,
1700 verified_outputs: Option<&[FileDigest]>,
1701) -> Result<(), ArtifactCacheError> {
1702 for (index, output) in spec.outputs.iter().enumerate() {
1703 if verified_outputs.is_some_and(|info| {
1706 destination_matches_digest("artifact-set-output-v1", &output.destination, &info[index])
1707 }) {
1708 continue;
1709 }
1710 let cached = staged_output_path(entry, index);
1711 copy_file_atomic(&cached, &output.destination).map_err(|source| {
1712 ArtifactCacheError::Io {
1713 operation: "materialize artifact output",
1714 path: output.destination.clone(),
1715 source,
1716 }
1717 })?;
1718 }
1719 Ok(())
1720}
1721
1722fn perform_maintenance_locked(
1723 spec: &ArtifactCacheSpec,
1724 namespace: &Path,
1725 protected_entry: &Path,
1726) -> (Option<ArtifactCacheMaintenance>, Option<Duration>) {
1727 spec.prune_policy.map_or((None, None), |policy| {
1728 let identity = policy.maintenance_identity();
1729 perform_scheduled_cache_maintenance(namespace, spec.prune_interval, &identity, || {
1730 prune_artifact_namespace_locked(
1731 &spec.cache_root,
1732 namespace,
1733 policy,
1734 Some(protected_entry),
1735 )
1736 .map_err(|error| error.to_string())
1737 })
1738 })
1739}
1740
1741fn prune_artifact_namespace_locked(
1742 cache_root: &Path,
1743 namespace: &Path,
1744 policy: ArtifactCachePrunePolicy,
1745 protected_entry: Option<&Path>,
1746) -> Result<ArtifactCachePruneReport, ArtifactCacheError> {
1747 let mut report = prune_direct_child_directories(
1748 &entries_directory(namespace),
1749 policy,
1750 protected_entry,
1751 is_sha256_directory,
1752 )
1753 .map_err(artifact_cache_fs_error)?;
1754 remove_abandoned_staging(cache_root, namespace, &mut report)?;
1755 Ok(report)
1756}
1757
1758fn remove_abandoned_staging(
1759 cache_root: &Path,
1760 namespace: &Path,
1761 report: &mut ArtifactCachePruneReport,
1762) -> Result<(), ArtifactCacheError> {
1763 let staging_root = namespace.join("staging");
1764 let entries = match fs::read_dir(&staging_root) {
1765 Ok(entries) => entries,
1766 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
1767 Err(source) => {
1768 return Err(ArtifactCacheError::Io {
1769 operation: "read artifact staging root during pruning",
1770 path: staging_root,
1771 source,
1772 });
1773 }
1774 };
1775 for entry in entries {
1776 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1777 operation: "read artifact staging entry during pruning",
1778 path: staging_root.clone(),
1779 source,
1780 })?;
1781 let file_type = entry.file_type().map_err(|source| ArtifactCacheError::Io {
1782 operation: "inspect artifact staging entry during pruning",
1783 path: entry.path(),
1784 source,
1785 })?;
1786 if !file_type.is_dir() {
1787 continue;
1788 }
1789 let file_name = entry.file_name();
1790 let Some(key) = staging_content_key(&file_name) else {
1791 continue;
1792 };
1793 let lock_path = content_lock_path_for_key(cache_root, key);
1794 let Some(_content_lock) =
1795 try_lock_cache_file(&lock_path).map_err(artifact_cache_fs_error)?
1796 else {
1797 continue;
1798 };
1799 let path = entry.path();
1800 let bytes = directory_logical_size(&path).map_err(|source| ArtifactCacheError::Io {
1801 operation: "measure abandoned artifact staging directory",
1802 path: path.clone(),
1803 source,
1804 })?;
1805 remove_path_if_present(&path).map_err(|source| ArtifactCacheError::Io {
1806 operation: "remove abandoned artifact staging directory during pruning",
1807 path,
1808 source,
1809 })?;
1810 report.record_uncommitted_removal(bytes);
1811 }
1812 Ok(())
1813}
1814
1815fn staging_content_key(name: &OsStr) -> Option<&str> {
1816 let name = name.to_str()?;
1817 let (key, suffix) = name.split_once('-')?;
1818 (!suffix.is_empty() && key.len() == 64 && key.as_bytes().iter().all(u8::is_ascii_hexdigit))
1819 .then_some(key)
1820}
1821
1822fn cache_record(
1823 spec: &ArtifactCacheSpec,
1824 resolved: ResolvedKey,
1825 timings: ArtifactCacheTimings,
1826 maintenance: Option<ArtifactCacheMaintenance>,
1827) -> Result<ArtifactCacheRecord, ArtifactCacheError> {
1828 let entry = entry_directory(&namespace_directory(spec), resolved.key);
1829 let retention = RetainedCacheEntry::acquire(&entry).map_err(artifact_cache_fs_error)?;
1830 Ok(ArtifactCacheRecord {
1831 key: resolved.key,
1832 input_digest: resolved.input_digest,
1833 artifacts: spec
1834 .outputs
1835 .iter()
1836 .enumerate()
1837 .map(|(index, output)| ArtifactCacheArtifact {
1838 name: output.name.clone(),
1839 path: staged_output_path(&entry, index),
1840 })
1841 .collect(),
1842 timings,
1843 maintenance,
1844 _retention: retention,
1845 })
1846}
1847
1848fn create_staging_directory(
1849 namespace: &Path,
1850 key: InputDigest,
1851) -> Result<PathBuf, ArtifactCacheError> {
1852 let staging_root = namespace.join("staging");
1853 fs::create_dir_all(&staging_root).map_err(|source| ArtifactCacheError::Io {
1854 operation: "create artifact staging root",
1855 path: staging_root.clone(),
1856 source,
1857 })?;
1858 remove_same_key_staging(&staging_root, key)?;
1859 let sequence = STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed);
1860 let staging = staging_root.join(format!("{key}-{}-{sequence}", std::process::id()));
1861 fs::create_dir_all(staging.join("outputs")).map_err(|source| ArtifactCacheError::Io {
1862 operation: "create artifact transaction staging directory",
1863 path: staging.clone(),
1864 source,
1865 })?;
1866 Ok(staging)
1867}
1868
1869fn remove_same_key_staging(root: &Path, key: InputDigest) -> Result<(), ArtifactCacheError> {
1870 let prefix = format!("{key}-");
1871 let entries = fs::read_dir(root).map_err(|source| ArtifactCacheError::Io {
1872 operation: "read artifact staging root",
1873 path: root.to_owned(),
1874 source,
1875 })?;
1876 for entry in entries {
1877 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1878 operation: "read artifact staging entry",
1879 path: root.to_owned(),
1880 source,
1881 })?;
1882 if entry.file_name().to_string_lossy().starts_with(&prefix) {
1883 remove_path_if_present(&entry.path()).map_err(|source| ArtifactCacheError::Io {
1884 operation: "remove abandoned artifact staging directory",
1885 path: entry.path(),
1886 source,
1887 })?;
1888 }
1889 }
1890 Ok(())
1891}
1892
1893fn staged_output_path(root: &Path, index: usize) -> PathBuf {
1894 root.join("outputs").join(format_output_index(index))
1895}
1896
1897fn format_output_index(index: usize) -> String {
1898 format!("{index:04}{OUTPUT_FILE_SUFFIX}")
1899}
1900
1901fn namespace_directory(spec: &ArtifactCacheSpec) -> PathBuf {
1902 namespace_directory_for(&spec.cache_root, &spec.namespace)
1903}
1904
1905fn namespace_directory_for(cache_root: &Path, namespace: &str) -> PathBuf {
1906 cache_root
1907 .join(".ic-testkit/artifact-sets/namespaces")
1908 .join(identifier_digest("artifact-cache-namespace-v1", namespace))
1909}
1910
1911fn entries_directory(namespace: &Path) -> PathBuf {
1912 namespace.join("entries")
1913}
1914
1915fn entry_directory(namespace: &Path, key: InputDigest) -> PathBuf {
1916 entries_directory(namespace).join(key.to_hex())
1917}
1918
1919fn coordination_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1920 spec.cache_root
1921 .join(".ic-testkit/artifact-sets/locks/coordination")
1922 .join(format!(
1923 "{}.lock",
1924 identifier_digest("artifact-cache-coordination-v1", &spec.coordination_scope,)
1925 ))
1926}
1927
1928fn content_lock_path(spec: &ArtifactCacheSpec, key: InputDigest) -> PathBuf {
1929 content_lock_path_for_key(&spec.cache_root, &key.to_hex())
1930}
1931
1932fn content_lock_path_for_key(cache_root: &Path, key: &str) -> PathBuf {
1933 cache_root
1934 .join(".ic-testkit/artifact-sets/locks/content")
1935 .join(format!("{key}.lock"))
1936}
1937
1938fn namespace_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1939 namespace_lock_path_for(&spec.cache_root, &spec.namespace)
1940}
1941
1942fn namespace_lock_path_for(cache_root: &Path, namespace: &str) -> PathBuf {
1943 cache_root
1944 .join(".ic-testkit/artifact-sets/locks/namespaces")
1945 .join(format!(
1946 "{}.lock",
1947 identifier_digest("artifact-cache-namespace-v1", namespace)
1948 ))
1949}
1950
1951fn identifier_digest(domain: &str, identifier: &str) -> String {
1952 digest_bytes(domain, identifier.as_bytes()).to_hex()
1953}
1954
1955fn artifact_cache_fs_error(error: CacheFsError) -> ArtifactCacheError {
1956 ArtifactCacheError::Io {
1957 operation: error.operation,
1958 path: error.path,
1959 source: error.source,
1960 }
1961}
1962
1963impl ArtifactOutputValidation {
1964 const fn cache_token(self) -> &'static str {
1965 match self {
1966 Self::RegularFile => "regular-file",
1967 Self::NonEmptyFile => "nonempty-file",
1968 }
1969 }
1970}
1971
1972impl std::fmt::Display for ArtifactCacheError {
1973 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1974 match self {
1975 Self::InvalidSpec { message } => {
1976 write!(formatter, "invalid artifact cache spec: {message}")
1977 }
1978 Self::Io {
1979 operation,
1980 path,
1981 source,
1982 } => write!(
1983 formatter,
1984 "failed to {operation} at {}: {source}",
1985 path.display()
1986 ),
1987 Self::InputsChangedDuringPreparation { before, after } => write!(
1988 formatter,
1989 "artifact inputs repeatedly changed during cache preparation: {before} -> {after}",
1990 ),
1991 Self::InputsChangedDuringAcquisition { before, after } => write!(
1992 formatter,
1993 "artifact inputs changed while the caller was building: {before} -> {after}",
1994 ),
1995 Self::CargoBuildInputsChanged {
1996 label,
1997 before,
1998 after,
1999 } => write!(
2000 formatter,
2001 "resolved Cargo build inputs `{label}` changed: {before} -> {after}",
2002 ),
2003 Self::CargoBuildInputRevalidation { label, source } => write!(
2004 formatter,
2005 "failed to revalidate resolved Cargo build inputs `{label}`: {source}",
2006 ),
2007 Self::InvalidOutputs { outputs } => write!(
2008 formatter,
2009 "artifact transaction has missing or invalid outputs: {}",
2010 outputs
2011 .iter()
2012 .map(|(name, path)| format!("{name} ({})", path.display()))
2013 .collect::<Vec<_>>()
2014 .join(", "),
2015 ),
2016 Self::UnknownOutput { name } => {
2017 write!(
2018 formatter,
2019 "artifact transaction has no output named `{name}`"
2020 )
2021 }
2022 Self::FailedTransactionCleanup {
2023 transaction_error,
2024 path,
2025 source,
2026 } => write!(
2027 formatter,
2028 "artifact transaction failed ({transaction_error}) and staging cleanup at {} also failed: {source}",
2029 path.display(),
2030 ),
2031 }
2032 }
2033}
2034
2035impl std::error::Error for ArtifactCacheError {
2036 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
2037 match self {
2038 Self::Io { source, .. } | Self::FailedTransactionCleanup { source, .. } => Some(source),
2039 Self::CargoBuildInputRevalidation { source, .. } => Some(source),
2040 _ => None,
2041 }
2042 }
2043}
2044
2045#[cfg(test)]
2046mod tests;