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