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
591impl std::fmt::Display for ArtifactCacheTimings {
592 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
593 write!(
594 formatter,
595 "total={:?} coordination_lock={:?} content_lock={:?} namespace_lock={:?} inputs={:?} lookup={:?} build={:?} validation={:?} publication={:?} materialization={:?} maintenance={:?}",
596 self.total,
597 self.coordination_lock_wait,
598 self.content_lock_wait,
599 self.namespace_lock_wait,
600 self.input_capture,
601 self.cache_lookup,
602 self.caller_build,
603 self.output_validation,
604 self.publication,
605 self.materialization,
606 self.maintenance,
607 )
608 }
609}
610
611impl std::fmt::Display for ArtifactCacheOutcome {
612 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
613 let state = if self.is_reused() { "reused" } else { "built" };
614 write!(
615 formatter,
616 "{state} key={} artifacts={} {}",
617 self.record().key,
618 self.record().artifacts.len(),
619 self.record().timings,
620 )
621 }
622}
623
624impl ArtifactBuildTransaction {
625 #[must_use]
630 pub fn staging_directory(&self) -> &Path {
631 &self.staging_directory
632 }
633
634 pub fn output_path(&self, name: &str) -> Result<PathBuf, ArtifactCacheError> {
636 self.output_index(name)
637 .map(|index| staged_output_path(&self.staging_directory, index))
638 .ok_or_else(|| ArtifactCacheError::UnknownOutput {
639 name: name.to_owned(),
640 })
641 }
642
643 pub fn import_output(&self, name: &str, source: &Path) -> Result<(), ArtifactCacheError> {
645 let destination = self.output_path(name)?;
646 copy_file_atomic(source, &destination).map_err(|source_error| ArtifactCacheError::Io {
647 operation: "import artifact output into staging",
648 path: destination,
649 source: source_error,
650 })?;
651 Ok(())
652 }
653
654 pub fn commit(mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
656 let result = self.commit_inner();
657 match result {
658 Ok(outcome) => Ok(outcome),
659 Err(transaction_error) if self.staging_armed => {
660 let path = self.staging_directory.clone();
661 match remove_path_if_present(&path) {
662 Ok(()) => {
663 self.staging_armed = false;
664 Err(transaction_error)
665 }
666 Err(source) => Err(ArtifactCacheError::FailedTransactionCleanup {
667 transaction_error: Box::new(transaction_error),
668 path,
669 source,
670 }),
671 }
672 }
673 Err(transaction_error) => Err(transaction_error),
674 }
675 }
676
677 pub fn abort(mut self) -> Result<(), ArtifactCacheError> {
679 remove_path_if_present(&self.staging_directory).map_err(|source| {
680 ArtifactCacheError::Io {
681 operation: "abort artifact cache transaction",
682 path: self.staging_directory.clone(),
683 source,
684 }
685 })?;
686 self.staging_armed = false;
687 Ok(())
688 }
689
690 fn output_index(&self, name: &str) -> Option<usize> {
691 self.spec
692 .outputs
693 .iter()
694 .position(|output| output.name == name)
695 }
696
697 fn commit_inner(&mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
698 self.timings.caller_build = Some(self.caller_build_started.elapsed());
699
700 let validation_started = Instant::now();
701 let output_info = inspect_complete_output_set(&self.spec, &self.staging_directory)?;
702 self.timings.output_validation = validation_started.elapsed();
703
704 let capture_started = Instant::now();
705 revalidate_cargo_build_input_fingerprints(&self.spec)?;
706 let verified = resolve_key(&self.spec)?;
707 self.timings.input_capture = self
708 .timings
709 .input_capture
710 .saturating_add(capture_started.elapsed());
711 if verified.input_digest != self.resolved.input_digest {
712 return Err(ArtifactCacheError::InputsChangedDuringBuild {
713 before: self.resolved.input_digest,
714 after: verified.input_digest,
715 });
716 }
717
718 let publication_started = Instant::now();
719 let manifest = manifest_contents(self.resolved.key, &self.spec, &output_info);
720 write_atomic(
721 &self.staging_directory.join(MANIFEST_FILE),
722 manifest.as_bytes(),
723 )
724 .map_err(|source| ArtifactCacheError::Io {
725 operation: "write artifact cache manifest",
726 path: self.staging_directory.join(MANIFEST_FILE),
727 source,
728 })?;
729 let namespace_lock_path = namespace_lock_path(&self.spec);
730 let (_namespace_lock, namespace_wait) =
731 lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
732 self.timings.namespace_lock_wait = self
733 .timings
734 .namespace_lock_wait
735 .saturating_add(namespace_wait);
736 remove_path_if_present(&self.entry_directory).map_err(|source| ArtifactCacheError::Io {
737 operation: "remove conflicting artifact cache entry",
738 path: self.entry_directory.clone(),
739 source,
740 })?;
741 fs::rename(&self.staging_directory, &self.entry_directory).map_err(|source| {
742 ArtifactCacheError::Io {
743 operation: "publish artifact cache entry",
744 path: self.entry_directory.clone(),
745 source,
746 }
747 })?;
748 self.staging_armed = false;
749 self.timings.publication = publication_started.elapsed();
750
751 let materialization_started = Instant::now();
752 materialize_outputs(&self.spec, &self.entry_directory)?;
753 self.timings.materialization = materialization_started.elapsed();
754 record_cache_entry_use(&self.entry_directory).map_err(artifact_cache_fs_error)?;
755 let (maintenance, maintenance_timing) = perform_maintenance_locked(
756 &self.spec,
757 &self.namespace_directory,
758 &self.entry_directory,
759 );
760 self.timings.maintenance = maintenance_timing;
761 self.timings.total = self.total_started.elapsed();
762
763 Ok(ArtifactCacheOutcome::Built(cache_record(
764 &self.spec,
765 self.resolved,
766 self.timings,
767 maintenance,
768 )))
769 }
770}
771
772impl Drop for ArtifactBuildTransaction {
773 fn drop(&mut self) {
774 if self.staging_armed {
775 let _ = remove_path_if_present(&self.staging_directory);
776 }
777 }
778}
779
780pub fn prepare_artifact_cache(
782 spec: &ArtifactCacheSpec,
783) -> Result<ArtifactCachePreparation, ArtifactCacheError> {
784 let total_started = Instant::now();
785 validate_spec(spec)?;
786 validate_filesystem_boundaries(spec)?;
787 let namespace_directory = initialize_cache(spec)?;
788
789 let coordination_lock_path = coordination_lock_path(spec);
790 let (coordination_lock, coordination_wait) =
791 lock_cache_file(&coordination_lock_path).map_err(artifact_cache_fs_error)?;
792 let mut timings = ArtifactCacheTimings {
793 coordination_lock_wait: coordination_wait,
794 ..ArtifactCacheTimings::default()
795 };
796 let initial_cargo_started = Instant::now();
797 revalidate_cargo_build_input_fingerprints(spec)?;
798 timings.input_capture = initial_cargo_started.elapsed();
799 let mut last_change = None;
800
801 for _ in 0..MAX_PREPARATION_RETRIES {
802 let capture_started = Instant::now();
803 let resolved = resolve_key(spec)?;
804 timings.input_capture = timings
805 .input_capture
806 .saturating_add(capture_started.elapsed());
807
808 let content_lock_path = content_lock_path(spec, resolved.key);
809 let (content_lock, content_wait) =
810 lock_cache_file(&content_lock_path).map_err(artifact_cache_fs_error)?;
811 timings.content_lock_wait = timings.content_lock_wait.saturating_add(content_wait);
812
813 let verification_started = Instant::now();
814 let verified = resolve_key(spec)?;
815 timings.input_capture = timings
816 .input_capture
817 .saturating_add(verification_started.elapsed());
818 if resolved.input_digest != verified.input_digest {
819 last_change = Some((resolved.input_digest, verified.input_digest));
820 drop(content_lock);
821 continue;
822 }
823
824 let entry_directory = entry_directory(&namespace_directory, resolved.key);
825 let namespace_lock_path = namespace_lock_path(spec);
826 let (namespace_lock, namespace_wait) =
827 lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
828 timings.namespace_lock_wait = timings.namespace_lock_wait.saturating_add(namespace_wait);
829 let lookup_started = Instant::now();
830 let reusable = cache_entry_is_valid(spec, resolved.key, &entry_directory)?;
831 timings.cache_lookup = timings
832 .cache_lookup
833 .saturating_add(lookup_started.elapsed());
834
835 if reusable {
836 let materialization_started = Instant::now();
837 materialize_outputs(spec, &entry_directory)?;
838 timings.materialization = timings
839 .materialization
840 .saturating_add(materialization_started.elapsed());
841 let after_started = Instant::now();
842 revalidate_cargo_build_input_fingerprints(spec)?;
843 let after = resolve_key(spec)?;
844 timings.input_capture = timings
845 .input_capture
846 .saturating_add(after_started.elapsed());
847 if after.input_digest != resolved.input_digest {
848 last_change = Some((resolved.input_digest, after.input_digest));
849 drop(namespace_lock);
850 drop(content_lock);
851 continue;
852 }
853 record_cache_entry_use(&entry_directory).map_err(artifact_cache_fs_error)?;
854 let (maintenance, maintenance_timing) =
855 perform_maintenance_locked(spec, &namespace_directory, &entry_directory);
856 timings.maintenance = maintenance_timing;
857 timings.total = total_started.elapsed();
858 return Ok(ArtifactCachePreparation::Reused(cache_record(
859 spec,
860 resolved,
861 timings,
862 maintenance,
863 )));
864 }
865
866 remove_path_if_present(&entry_directory).map_err(|source| ArtifactCacheError::Io {
867 operation: "remove invalid artifact cache entry",
868 path: entry_directory.clone(),
869 source,
870 })?;
871 let staging_directory = create_staging_directory(&namespace_directory, resolved.key)?;
872 drop(namespace_lock);
873 return Ok(ArtifactCachePreparation::Build(ArtifactBuildTransaction {
874 spec: Box::new(spec.clone()),
875 resolved,
876 staging_directory,
877 entry_directory,
878 namespace_directory,
879 _coordination_lock: coordination_lock,
880 _content_lock: content_lock,
881 timings,
882 total_started,
883 caller_build_started: Instant::now(),
884 staging_armed: true,
885 }));
886 }
887
888 let (before, after) = last_change.expect("preparation retries require a recorded input change");
889 Err(ArtifactCacheError::InputsChangedDuringPreparation { before, after })
890}
891
892fn revalidate_cargo_build_input_fingerprints(
893 spec: &ArtifactCacheSpec,
894) -> Result<(), ArtifactCacheError> {
895 for cargo_input in &spec.cargo_build_inputs {
896 let current = resolve_cargo_build_inputs(&cargo_input.build_spec).map_err(|source| {
897 ArtifactCacheError::CargoBuildInputRevalidation {
898 label: cargo_input.label.clone(),
899 source,
900 }
901 })?;
902 let before = cargo_input.resolved.fingerprint();
903 let after = current.fingerprint();
904 if after != before {
905 return Err(ArtifactCacheError::CargoBuildInputsChanged {
906 label: cargo_input.label.clone(),
907 before,
908 after,
909 });
910 }
911 }
912 Ok(())
913}
914
915pub fn prune_artifact_cache(
923 cache_root: &Path,
924 namespace: &str,
925 policy: ArtifactCachePrunePolicy,
926) -> Result<ArtifactCachePruneReport, ArtifactCacheError> {
927 validate_identifier("namespace", namespace)?;
928 if cache_root.as_os_str().is_empty() {
929 return invalid_spec("cache root must not be empty");
930 }
931 ensure_cache_directory_tag(cache_root).map_err(artifact_cache_fs_error)?;
932 let namespace_directory = namespace_directory_for(cache_root, namespace);
933 let lock_path = namespace_lock_path_for(cache_root, namespace);
934 let (_lock, _) = lock_cache_file(&lock_path).map_err(artifact_cache_fs_error)?;
935 prune_artifact_namespace_locked(cache_root, &namespace_directory, policy, None)
936}
937
938fn initialize_cache(spec: &ArtifactCacheSpec) -> Result<PathBuf, ArtifactCacheError> {
939 ensure_cache_directory_tag(&spec.cache_root).map_err(artifact_cache_fs_error)?;
940 let namespace = namespace_directory(spec);
941 fs::create_dir_all(entries_directory(&namespace)).map_err(|source| ArtifactCacheError::Io {
942 operation: "create artifact cache namespace",
943 path: namespace.clone(),
944 source,
945 })?;
946 Ok(namespace)
947}
948
949fn validate_spec(spec: &ArtifactCacheSpec) -> Result<(), ArtifactCacheError> {
950 validate_identifier("namespace", &spec.namespace)?;
951 validate_identifier("recipe identity", &spec.recipe_id)?;
952 validate_identifier("coordination scope", &spec.coordination_scope)?;
953 if spec.cache_root.as_os_str().is_empty() {
954 return invalid_spec("cache root must not be empty");
955 }
956 if spec.outputs.is_empty() {
957 return invalid_spec("at least one artifact output is required");
958 }
959
960 let mut labels = BTreeSet::new();
961 for input in &spec.inputs {
962 validate_path_label("input", &input.label)?;
963 if !labels.insert(format!("input/{}", input.label)) {
964 return invalid_spec(&format!("duplicate input label `{}`", input.label));
965 }
966 }
967 for tool in &spec.tools {
968 validate_path_label("tool", &tool.label)?;
969 if !labels.insert(format!("tool/{}", tool.label)) {
970 return invalid_spec(&format!("duplicate tool label `{}`", tool.label));
971 }
972 }
973 let mut identity_labels = BTreeSet::new();
974 for identity in &spec.identities {
975 validate_label("identity", &identity.label)?;
976 if !identity_labels.insert(&identity.label) {
977 return invalid_spec(&format!("duplicate identity label `{}`", identity.label));
978 }
979 }
980 let mut cargo_input_labels = BTreeSet::new();
981 for cargo_inputs in &spec.cargo_build_inputs {
982 validate_label("Cargo build input", &cargo_inputs.label)?;
983 if !cargo_input_labels.insert(&cargo_inputs.label) {
984 return invalid_spec(&format!(
985 "duplicate Cargo build input label `{}`",
986 cargo_inputs.label
987 ));
988 }
989 }
990 if spec
991 .environment
992 .keys()
993 .any(|name| name.as_os_str().is_empty())
994 {
995 return invalid_spec("environment names must not be empty");
996 }
997 let mut output_names = BTreeSet::new();
998 let mut destinations = BTreeSet::new();
999 for output in &spec.outputs {
1000 validate_output_name(&output.name)?;
1001 if !output_names.insert(&output.name) {
1002 return invalid_spec(&format!("duplicate output name `{}`", output.name));
1003 }
1004 if output.destination.as_os_str().is_empty() {
1005 return invalid_spec(&format!(
1006 "output `{}` destination must not be empty",
1007 output.name
1008 ));
1009 }
1010 if !destinations.insert(&output.destination) {
1011 return invalid_spec(&format!(
1012 "output `{}` shares a destination with another output",
1013 output.name
1014 ));
1015 }
1016 }
1017 Ok(())
1018}
1019
1020fn validate_filesystem_boundaries(spec: &ArtifactCacheSpec) -> Result<(), ArtifactCacheError> {
1021 let cache_root =
1022 canonicalize_allow_missing(&spec.cache_root).map_err(|source| ArtifactCacheError::Io {
1023 operation: "resolve artifact cache root",
1024 path: spec.cache_root.clone(),
1025 source,
1026 })?;
1027 let mut declared_paths = Vec::with_capacity(spec.inputs.len() + spec.tools.len());
1028 for (kind, labeled_paths) in [("input", &spec.inputs), ("tool", &spec.tools)] {
1029 for labeled in labeled_paths {
1030 let canonical =
1031 canonicalize_path(&labeled.path, "canonicalize declared artifact cache path")?;
1032 if canonical.starts_with(&cache_root) {
1033 return invalid_spec(&format!(
1034 "{kind} `{}` must not be located inside the artifact cache root",
1035 labeled.label
1036 ));
1037 }
1038 let is_directory = fs::metadata(&canonical)
1039 .map_err(|source| ArtifactCacheError::Io {
1040 operation: "inspect declared artifact cache path",
1041 path: canonical.clone(),
1042 source,
1043 })?
1044 .is_dir();
1045 declared_paths.push((kind, labeled.label.as_str(), canonical, is_directory));
1046 }
1047 }
1048 for cargo_inputs in &spec.cargo_build_inputs {
1049 if resolved_cargo_inputs_watch_path(&cargo_inputs.resolved, &cache_root)? {
1050 return invalid_spec(&format!(
1051 "artifact cache root must be outside resolved Cargo build inputs `{}` or inside one of their generated-state exclusions",
1052 cargo_inputs.label
1053 ));
1054 }
1055 }
1056
1057 let mut destinations = BTreeSet::new();
1058 for output in &spec.outputs {
1059 let destination = canonicalize_allow_missing(&output.destination).map_err(|source| {
1060 ArtifactCacheError::Io {
1061 operation: "resolve artifact output destination",
1062 path: output.destination.clone(),
1063 source,
1064 }
1065 })?;
1066 if destination.starts_with(&cache_root) {
1067 return invalid_spec(&format!(
1068 "output `{}` destination must be outside the artifact cache root",
1069 output.name
1070 ));
1071 }
1072 if !destinations.insert(destination.clone()) {
1073 return invalid_spec(&format!(
1074 "output `{}` resolves to the same destination as another output",
1075 output.name
1076 ));
1077 }
1078 for (kind, label, declared, is_directory) in &declared_paths {
1079 if destination == *declared || (*is_directory && destination.starts_with(declared)) {
1080 return invalid_spec(&format!(
1081 "output `{}` destination overlaps declared {kind} `{label}`",
1082 output.name
1083 ));
1084 }
1085 }
1086 for cargo_inputs in &spec.cargo_build_inputs {
1087 if resolved_cargo_inputs_watch_path(&cargo_inputs.resolved, &destination)? {
1088 return invalid_spec(&format!(
1089 "output `{}` destination must be outside resolved Cargo build inputs `{}` or inside one of their generated-state exclusions",
1090 output.name, cargo_inputs.label
1091 ));
1092 }
1093 }
1094 match fs::metadata(&destination) {
1095 Ok(metadata) if metadata.is_dir() => {
1096 return invalid_spec(&format!(
1097 "output `{}` destination must not be an existing directory",
1098 output.name
1099 ));
1100 }
1101 Ok(_) => {}
1102 Err(error) if error.kind() == io::ErrorKind::NotFound => {}
1103 Err(source) => {
1104 return Err(ArtifactCacheError::Io {
1105 operation: "inspect artifact output destination",
1106 path: destination,
1107 source,
1108 });
1109 }
1110 }
1111 }
1112 Ok(())
1113}
1114
1115fn resolved_cargo_inputs_watch_path(
1116 resolved: &ResolvedCargoBuildInputs,
1117 candidate: &Path,
1118) -> Result<bool, ArtifactCacheError> {
1119 for exclusion in resolved.exclusions() {
1120 let exclusion =
1121 canonicalize_allow_missing(exclusion).map_err(|source| ArtifactCacheError::Io {
1122 operation: "resolve Cargo build input exclusion",
1123 path: exclusion.clone(),
1124 source,
1125 })?;
1126 if candidate.starts_with(exclusion) {
1127 return Ok(false);
1128 }
1129 }
1130
1131 for input in resolved.inputs() {
1132 let path = canonicalize_path(input.path(), "canonicalize resolved Cargo build input")?;
1133 let is_directory = fs::metadata(&path)
1134 .map_err(|source| ArtifactCacheError::Io {
1135 operation: "inspect resolved Cargo build input",
1136 path: path.clone(),
1137 source,
1138 })?
1139 .is_dir();
1140 if candidate == path || (is_directory && candidate.starts_with(path)) {
1141 return Ok(true);
1142 }
1143 }
1144 Ok(false)
1145}
1146
1147fn canonicalize_path(path: &Path, operation: &'static str) -> Result<PathBuf, ArtifactCacheError> {
1148 path.canonicalize()
1149 .map_err(|source| ArtifactCacheError::Io {
1150 operation,
1151 path: path.to_owned(),
1152 source,
1153 })
1154}
1155
1156fn canonicalize_allow_missing(path: &Path) -> io::Result<PathBuf> {
1157 let absolute = if path.is_absolute() {
1158 path.to_owned()
1159 } else {
1160 std::env::current_dir()?.join(path)
1161 };
1162 let mut resolved = PathBuf::new();
1163 let mut missing_depth = 0_usize;
1164 for component in absolute.components() {
1165 match component {
1166 Component::Prefix(_) | Component::RootDir | Component::Normal(_) => {
1167 let candidate = resolved.join(component.as_os_str());
1168 if missing_depth == 0 && matches!(component, Component::Normal(_)) {
1169 match candidate.canonicalize() {
1170 Ok(canonical) => resolved = canonical,
1171 Err(error) if error.kind() == io::ErrorKind::NotFound => {
1172 resolved = candidate;
1173 missing_depth = 1;
1174 }
1175 Err(error) => return Err(error),
1176 }
1177 } else {
1178 resolved = candidate;
1179 if matches!(component, Component::Normal(_)) && missing_depth > 0 {
1180 missing_depth += 1;
1181 }
1182 }
1183 }
1184 Component::CurDir => {}
1185 Component::ParentDir => {
1186 resolved.pop();
1187 missing_depth = missing_depth.saturating_sub(1);
1188 }
1189 }
1190 }
1191 Ok(resolved)
1192}
1193
1194fn validate_identifier(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1195 if value.is_empty() {
1196 return invalid_spec(&format!("{kind} must not be empty"));
1197 }
1198 if value.len() > 256 {
1199 return invalid_spec(&format!("{kind} must not exceed 256 bytes"));
1200 }
1201 Ok(())
1202}
1203
1204fn validate_label(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1205 if value.is_empty() {
1206 return invalid_spec(&format!("{kind} label must not be empty"));
1207 }
1208 if value.len() > 256 {
1209 return invalid_spec(&format!("{kind} label must not exceed 256 bytes"));
1210 }
1211 Ok(())
1212}
1213
1214fn validate_path_label(kind: &str, value: &str) -> Result<(), ArtifactCacheError> {
1215 validate_label(kind, value)?;
1216 if value
1217 .bytes()
1218 .any(|byte| !(byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'/')))
1219 || value
1220 .split('/')
1221 .any(|component| component.is_empty() || matches!(component, "." | ".."))
1222 {
1223 return invalid_spec(&format!(
1224 "{kind} label `{value}` must be a portable relative logical path"
1225 ));
1226 }
1227 Ok(())
1228}
1229
1230fn validate_output_name(name: &str) -> Result<(), ArtifactCacheError> {
1231 if name.is_empty() || name.len() > 128 {
1232 return invalid_spec("output names must contain 1 to 128 bytes");
1233 }
1234 if !name
1235 .bytes()
1236 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
1237 {
1238 return invalid_spec(&format!(
1239 "output name `{name}` must use only ASCII letters, digits, dot, dash, or underscore"
1240 ));
1241 }
1242 if matches!(name, "." | "..") {
1243 return invalid_spec("output names must not be dot path components");
1244 }
1245 Ok(())
1246}
1247
1248fn invalid_spec<T>(message: &str) -> Result<T, ArtifactCacheError> {
1249 Err(ArtifactCacheError::InvalidSpec {
1250 message: message.to_owned(),
1251 })
1252}
1253
1254fn resolve_key(spec: &ArtifactCacheSpec) -> Result<ResolvedKey, ArtifactCacheError> {
1255 let paths = spec
1256 .inputs
1257 .iter()
1258 .map(|input| {
1259 (
1260 PathBuf::from("input").join(&input.label),
1261 input.path.clone(),
1262 )
1263 })
1264 .chain(
1265 spec.tools
1266 .iter()
1267 .map(|tool| (PathBuf::from("tool").join(&tool.label), tool.path.clone())),
1268 )
1269 .collect::<Vec<_>>();
1270 let declared_input_digest = digest_labeled_paths(
1271 "artifact-set-inputs-v1",
1272 &paths,
1273 std::slice::from_ref(&spec.cache_root),
1274 )
1275 .map_err(|source| ArtifactCacheError::Io {
1276 operation: "hash artifact cache inputs",
1277 path: spec.cache_root.clone(),
1278 source,
1279 })?;
1280 let input_digest = if spec.cargo_build_inputs.is_empty() {
1281 declared_input_digest
1282 } else {
1283 let mut cargo_inputs = spec.cargo_build_inputs.iter().collect::<Vec<_>>();
1284 cargo_inputs.sort_by(|left, right| left.label.cmp(&right.label));
1285 let mut inputs = InputHasher::new("artifact-set-inputs-with-cargo-v1");
1286 inputs.field("declared-input-digest", declared_input_digest.as_bytes());
1287 for cargo_input in cargo_inputs {
1288 let current = cargo_input
1289 .resolved
1290 .current_input_digest()
1291 .map_err(|source| ArtifactCacheError::CargoBuildInputRevalidation {
1292 label: cargo_input.label.clone(),
1293 source,
1294 })?;
1295 let before = cargo_input.resolved.input_digest();
1296 if current != before {
1297 return Err(ArtifactCacheError::CargoBuildInputsChanged {
1298 label: cargo_input.label.clone(),
1299 before,
1300 after: current,
1301 });
1302 }
1303 inputs.field("cargo-input-label", cargo_input.label.as_bytes());
1304 inputs.field(
1305 "cargo-build-fingerprint",
1306 cargo_input.resolved.fingerprint().as_bytes(),
1307 );
1308 inputs.field("cargo-input-digest", current.as_bytes());
1309 }
1310 inputs.finish()
1311 };
1312
1313 let mut hasher = InputHasher::new(ARTIFACT_CACHE_FORMAT);
1314 hasher.field("namespace", spec.namespace.as_bytes());
1315 hasher.field("recipe-id", spec.recipe_id.as_bytes());
1316 hasher.field("input-digest", input_digest.as_bytes());
1317 for argument in &spec.arguments {
1318 hasher.field("argument", &os_bytes(argument));
1319 }
1320 for (name, value) in &spec.environment {
1321 hasher.field("environment-name", &os_bytes(name));
1322 match value {
1323 Some(value) => hasher.field("environment-value", &os_bytes(value)),
1324 None => hasher.field("environment-unset", b""),
1325 }
1326 }
1327 let mut identities = spec.identities.iter().collect::<Vec<_>>();
1328 identities.sort_by(|left, right| left.label.cmp(&right.label));
1329 for identity in identities {
1330 hasher.field("identity-label", identity.label.as_bytes());
1331 hasher.field("identity-value", &identity.value);
1332 }
1333 for output in &spec.outputs {
1334 hasher.field("output-name", output.name.as_bytes());
1335 hasher.field(
1336 "output-validation",
1337 output.validation.cache_token().as_bytes(),
1338 );
1339 }
1340 Ok(ResolvedKey {
1341 key: hasher.finish(),
1342 input_digest,
1343 })
1344}
1345
1346fn cache_entry_is_valid(
1347 spec: &ArtifactCacheSpec,
1348 key: InputDigest,
1349 entry: &Path,
1350) -> Result<bool, ArtifactCacheError> {
1351 let entry_metadata = match fs::symlink_metadata(entry) {
1352 Ok(metadata) => metadata,
1353 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
1354 Err(source) => {
1355 return Err(ArtifactCacheError::Io {
1356 operation: "inspect artifact cache entry",
1357 path: entry.to_owned(),
1358 source,
1359 });
1360 }
1361 };
1362 if !entry_metadata.file_type().is_dir() || !cache_entry_root_is_valid(entry)? {
1363 return Ok(false);
1364 }
1365 let manifest_path = entry.join(MANIFEST_FILE);
1366 let manifest_metadata = match fs::symlink_metadata(&manifest_path) {
1367 Ok(metadata) => metadata,
1368 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
1369 Err(source) => {
1370 return Err(ArtifactCacheError::Io {
1371 operation: "inspect artifact cache manifest",
1372 path: manifest_path,
1373 source,
1374 });
1375 }
1376 };
1377 if !manifest_metadata.file_type().is_file() {
1378 return Ok(false);
1379 }
1380 let manifest = fs::read(&manifest_path).map_err(|source| ArtifactCacheError::Io {
1381 operation: "read artifact cache manifest",
1382 path: manifest_path,
1383 source,
1384 })?;
1385 let Some(output_info) = inspect_cached_output_set(spec, entry)? else {
1386 return Ok(false);
1387 };
1388 Ok(manifest == manifest_contents(key, spec, &output_info).as_bytes())
1389}
1390
1391fn inspect_complete_output_set(
1392 spec: &ArtifactCacheSpec,
1393 root: &Path,
1394) -> Result<Vec<ArtifactInfo>, ArtifactCacheError> {
1395 let mut info = Vec::new();
1396 let mut invalid = Vec::new();
1397 let output_directory = root.join("outputs");
1398 if !is_plain_directory(
1399 &output_directory,
1400 "inspect artifact staging output directory",
1401 )? {
1402 invalid.push(("<outputs>".to_owned(), output_directory));
1403 return Err(ArtifactCacheError::InvalidOutputs { outputs: invalid });
1404 }
1405 let outputs = &spec.outputs;
1406 for (index, output) in outputs.iter().enumerate() {
1407 let path = staged_output_path(root, index);
1408 match inspect_artifact(&path, output.validation) {
1409 Ok(Some(artifact)) => info.push(artifact),
1410 Ok(None) => invalid.push((output.name.clone(), path)),
1411 Err(source) => {
1412 return Err(ArtifactCacheError::Io {
1413 operation: "inspect staged artifact output",
1414 path,
1415 source,
1416 });
1417 }
1418 }
1419 }
1420 invalid.extend(
1421 undeclared_output_paths(root, outputs.len())?
1422 .into_iter()
1423 .map(|path| ("<undeclared>".to_owned(), path)),
1424 );
1425 invalid.extend(
1426 undeclared_child_paths(
1427 root,
1428 &BTreeSet::from([OsString::from("outputs")]),
1429 "read artifact staging directory",
1430 )?
1431 .into_iter()
1432 .map(|path| ("<undeclared>".to_owned(), path)),
1433 );
1434 if invalid.is_empty() {
1435 Ok(info)
1436 } else {
1437 Err(ArtifactCacheError::InvalidOutputs { outputs: invalid })
1438 }
1439}
1440
1441fn inspect_cached_output_set(
1442 spec: &ArtifactCacheSpec,
1443 root: &Path,
1444) -> Result<Option<Vec<ArtifactInfo>>, ArtifactCacheError> {
1445 let mut info = Vec::new();
1446 let outputs = &spec.outputs;
1447 for (index, output) in outputs.iter().enumerate() {
1448 let path = staged_output_path(root, index);
1449 match inspect_artifact(&path, output.validation) {
1450 Ok(Some(artifact)) => info.push(artifact),
1451 Ok(None) => return Ok(None),
1452 Err(source) => {
1453 return Err(ArtifactCacheError::Io {
1454 operation: "inspect cached artifact output",
1455 path,
1456 source,
1457 });
1458 }
1459 }
1460 }
1461 if !undeclared_output_paths(root, outputs.len())?.is_empty() {
1462 return Ok(None);
1463 }
1464 Ok(Some(info))
1465}
1466
1467fn undeclared_output_paths(
1468 root: &Path,
1469 output_count: usize,
1470) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1471 let output_directory = root.join("outputs");
1472 let expected = (0..output_count)
1473 .map(format_output_index)
1474 .map(OsString::from)
1475 .collect::<BTreeSet<_>>();
1476 undeclared_child_paths(
1477 &output_directory,
1478 &expected,
1479 "read artifact output directory",
1480 )
1481}
1482
1483fn undeclared_child_paths(
1484 directory: &Path,
1485 expected: &BTreeSet<OsString>,
1486 operation: &'static str,
1487) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1488 let entries = fs::read_dir(directory).map_err(|source| ArtifactCacheError::Io {
1489 operation,
1490 path: directory.to_owned(),
1491 source,
1492 })?;
1493 let mut undeclared = Vec::new();
1494 for entry in entries {
1495 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1496 operation,
1497 path: directory.to_owned(),
1498 source,
1499 })?;
1500 if !expected.contains(&entry.file_name()) {
1501 undeclared.push(entry.path());
1502 }
1503 }
1504 Ok(undeclared)
1505}
1506
1507fn cache_entry_root_is_valid(root: &Path) -> Result<bool, ArtifactCacheError> {
1508 let expected = BTreeSet::from([
1509 OsString::from("outputs"),
1510 OsString::from(MANIFEST_FILE),
1511 OsString::from(LAST_USED_FILE),
1512 ]);
1513 if !undeclared_child_paths(root, &expected, "read artifact cache entry")?.is_empty() {
1514 return Ok(false);
1515 }
1516 if !is_plain_directory(
1517 &root.join("outputs"),
1518 "inspect artifact cache output directory",
1519 )? {
1520 return Ok(false);
1521 }
1522 let last_used = root.join(LAST_USED_FILE);
1523 match fs::symlink_metadata(&last_used) {
1524 Ok(metadata) => Ok(metadata.file_type().is_file()),
1525 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(true),
1526 Err(source) => Err(ArtifactCacheError::Io {
1527 operation: "inspect artifact cache use marker",
1528 path: last_used,
1529 source,
1530 }),
1531 }
1532}
1533
1534fn is_plain_directory(path: &Path, operation: &'static str) -> Result<bool, ArtifactCacheError> {
1535 match fs::symlink_metadata(path) {
1536 Ok(metadata) => Ok(metadata.file_type().is_dir()),
1537 Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
1538 Err(source) => Err(ArtifactCacheError::Io {
1539 operation,
1540 path: path.to_owned(),
1541 source,
1542 }),
1543 }
1544}
1545
1546fn inspect_artifact(
1547 path: &Path,
1548 validation: ArtifactOutputValidation,
1549) -> io::Result<Option<ArtifactInfo>> {
1550 let metadata = match fs::symlink_metadata(path) {
1551 Ok(metadata) => metadata,
1552 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1553 Err(error) => return Err(error),
1554 };
1555 if !metadata.file_type().is_file() {
1556 return Ok(None);
1557 }
1558 if validation == ArtifactOutputValidation::NonEmptyFile && metadata.len() == 0 {
1559 return Ok(None);
1560 }
1561 let (bytes, digest) = digest_file("artifact-set-output-v1", path)?;
1562 Ok(Some(ArtifactInfo { bytes, digest }))
1563}
1564
1565fn manifest_contents(
1566 key: InputDigest,
1567 spec: &ArtifactCacheSpec,
1568 output_info: &[ArtifactInfo],
1569) -> String {
1570 let mut manifest = format!("{ARTIFACT_CACHE_FORMAT}\nkey:{key}\n");
1571 for ((index, output), info) in spec.outputs.iter().enumerate().zip(output_info) {
1572 use std::fmt::Write as _;
1573 writeln!(
1574 manifest,
1575 "output:{index}:{}:{}:{}:{}",
1576 output.name,
1577 output.validation.cache_token(),
1578 info.bytes,
1579 info.digest,
1580 )
1581 .expect("writing an artifact manifest to a String cannot fail");
1582 }
1583 manifest
1584}
1585
1586fn materialize_outputs(spec: &ArtifactCacheSpec, entry: &Path) -> Result<(), ArtifactCacheError> {
1587 for (index, output) in spec.outputs.iter().enumerate() {
1588 let cached = staged_output_path(entry, index);
1589 copy_file_atomic(&cached, &output.destination).map_err(|source| {
1590 ArtifactCacheError::Io {
1591 operation: "materialize artifact output",
1592 path: output.destination.clone(),
1593 source,
1594 }
1595 })?;
1596 }
1597 Ok(())
1598}
1599
1600fn perform_maintenance_locked(
1601 spec: &ArtifactCacheSpec,
1602 namespace: &Path,
1603 protected_entry: &Path,
1604) -> (Option<ArtifactCacheMaintenance>, Option<Duration>) {
1605 spec.prune_policy.map_or((None, None), |policy| {
1606 let identity = policy.maintenance_identity();
1607 perform_scheduled_cache_maintenance(namespace, spec.prune_interval, &identity, || {
1608 prune_artifact_namespace_locked(
1609 &spec.cache_root,
1610 namespace,
1611 policy,
1612 Some(protected_entry),
1613 )
1614 .map_err(|error| error.to_string())
1615 })
1616 })
1617}
1618
1619fn prune_artifact_namespace_locked(
1620 cache_root: &Path,
1621 namespace: &Path,
1622 policy: ArtifactCachePrunePolicy,
1623 protected_entry: Option<&Path>,
1624) -> Result<ArtifactCachePruneReport, ArtifactCacheError> {
1625 let mut report = prune_direct_child_directories(
1626 &entries_directory(namespace),
1627 policy,
1628 protected_entry,
1629 is_sha256_directory,
1630 )
1631 .map_err(artifact_cache_fs_error)?;
1632 remove_abandoned_staging(cache_root, namespace, &mut report)?;
1633 Ok(report)
1634}
1635
1636fn remove_abandoned_staging(
1637 cache_root: &Path,
1638 namespace: &Path,
1639 report: &mut ArtifactCachePruneReport,
1640) -> Result<(), ArtifactCacheError> {
1641 let staging_root = namespace.join("staging");
1642 let entries = match fs::read_dir(&staging_root) {
1643 Ok(entries) => entries,
1644 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
1645 Err(source) => {
1646 return Err(ArtifactCacheError::Io {
1647 operation: "read artifact staging root during pruning",
1648 path: staging_root,
1649 source,
1650 });
1651 }
1652 };
1653 for entry in entries {
1654 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1655 operation: "read artifact staging entry during pruning",
1656 path: staging_root.clone(),
1657 source,
1658 })?;
1659 let file_type = entry.file_type().map_err(|source| ArtifactCacheError::Io {
1660 operation: "inspect artifact staging entry during pruning",
1661 path: entry.path(),
1662 source,
1663 })?;
1664 if !file_type.is_dir() {
1665 continue;
1666 }
1667 let file_name = entry.file_name();
1668 let Some(key) = staging_content_key(&file_name) else {
1669 continue;
1670 };
1671 let lock_path = content_lock_path_for_key(cache_root, key);
1672 let Some(_content_lock) =
1673 try_lock_cache_file(&lock_path).map_err(artifact_cache_fs_error)?
1674 else {
1675 continue;
1676 };
1677 let path = entry.path();
1678 let bytes = directory_logical_size(&path).map_err(|source| ArtifactCacheError::Io {
1679 operation: "measure abandoned artifact staging directory",
1680 path: path.clone(),
1681 source,
1682 })?;
1683 remove_path_if_present(&path).map_err(|source| ArtifactCacheError::Io {
1684 operation: "remove abandoned artifact staging directory during pruning",
1685 path,
1686 source,
1687 })?;
1688 report.record_uncommitted_removal(bytes);
1689 }
1690 Ok(())
1691}
1692
1693fn staging_content_key(name: &OsStr) -> Option<&str> {
1694 let name = name.to_str()?;
1695 let (key, suffix) = name.split_once('-')?;
1696 (!suffix.is_empty() && key.len() == 64 && key.as_bytes().iter().all(u8::is_ascii_hexdigit))
1697 .then_some(key)
1698}
1699
1700fn cache_record(
1701 spec: &ArtifactCacheSpec,
1702 resolved: ResolvedKey,
1703 timings: ArtifactCacheTimings,
1704 maintenance: Option<ArtifactCacheMaintenance>,
1705) -> ArtifactCacheRecord {
1706 ArtifactCacheRecord {
1707 key: resolved.key,
1708 input_digest: resolved.input_digest,
1709 artifacts: spec
1710 .outputs
1711 .iter()
1712 .map(|output| ArtifactCacheArtifact {
1713 name: output.name.clone(),
1714 path: output.destination.clone(),
1715 })
1716 .collect(),
1717 timings,
1718 maintenance,
1719 }
1720}
1721
1722fn create_staging_directory(
1723 namespace: &Path,
1724 key: InputDigest,
1725) -> Result<PathBuf, ArtifactCacheError> {
1726 let staging_root = namespace.join("staging");
1727 fs::create_dir_all(&staging_root).map_err(|source| ArtifactCacheError::Io {
1728 operation: "create artifact staging root",
1729 path: staging_root.clone(),
1730 source,
1731 })?;
1732 remove_same_key_staging(&staging_root, key)?;
1733 let sequence = STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed);
1734 let staging = staging_root.join(format!("{key}-{}-{sequence}", std::process::id()));
1735 fs::create_dir_all(staging.join("outputs")).map_err(|source| ArtifactCacheError::Io {
1736 operation: "create artifact transaction staging directory",
1737 path: staging.clone(),
1738 source,
1739 })?;
1740 Ok(staging)
1741}
1742
1743fn remove_same_key_staging(root: &Path, key: InputDigest) -> Result<(), ArtifactCacheError> {
1744 let prefix = format!("{key}-");
1745 let entries = fs::read_dir(root).map_err(|source| ArtifactCacheError::Io {
1746 operation: "read artifact staging root",
1747 path: root.to_owned(),
1748 source,
1749 })?;
1750 for entry in entries {
1751 let entry = entry.map_err(|source| ArtifactCacheError::Io {
1752 operation: "read artifact staging entry",
1753 path: root.to_owned(),
1754 source,
1755 })?;
1756 if entry.file_name().to_string_lossy().starts_with(&prefix) {
1757 remove_path_if_present(&entry.path()).map_err(|source| ArtifactCacheError::Io {
1758 operation: "remove abandoned artifact staging directory",
1759 path: entry.path(),
1760 source,
1761 })?;
1762 }
1763 }
1764 Ok(())
1765}
1766
1767fn staged_output_path(root: &Path, index: usize) -> PathBuf {
1768 root.join("outputs").join(format_output_index(index))
1769}
1770
1771fn format_output_index(index: usize) -> String {
1772 format!("{index:04}.artifact")
1773}
1774
1775fn namespace_directory(spec: &ArtifactCacheSpec) -> PathBuf {
1776 namespace_directory_for(&spec.cache_root, &spec.namespace)
1777}
1778
1779fn namespace_directory_for(cache_root: &Path, namespace: &str) -> PathBuf {
1780 cache_root
1781 .join(".ic-testkit/artifact-sets/namespaces")
1782 .join(identifier_digest("artifact-cache-namespace-v1", namespace))
1783}
1784
1785fn entries_directory(namespace: &Path) -> PathBuf {
1786 namespace.join("entries")
1787}
1788
1789fn entry_directory(namespace: &Path, key: InputDigest) -> PathBuf {
1790 entries_directory(namespace).join(key.to_hex())
1791}
1792
1793fn coordination_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1794 spec.cache_root
1795 .join(".ic-testkit/artifact-sets/locks/coordination")
1796 .join(format!(
1797 "{}.lock",
1798 identifier_digest("artifact-cache-coordination-v1", &spec.coordination_scope,)
1799 ))
1800}
1801
1802fn content_lock_path(spec: &ArtifactCacheSpec, key: InputDigest) -> PathBuf {
1803 content_lock_path_for_key(&spec.cache_root, &key.to_hex())
1804}
1805
1806fn content_lock_path_for_key(cache_root: &Path, key: &str) -> PathBuf {
1807 cache_root
1808 .join(".ic-testkit/artifact-sets/locks/content")
1809 .join(format!("{key}.lock"))
1810}
1811
1812fn namespace_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1813 namespace_lock_path_for(&spec.cache_root, &spec.namespace)
1814}
1815
1816fn namespace_lock_path_for(cache_root: &Path, namespace: &str) -> PathBuf {
1817 cache_root
1818 .join(".ic-testkit/artifact-sets/locks/namespaces")
1819 .join(format!(
1820 "{}.lock",
1821 identifier_digest("artifact-cache-namespace-v1", namespace)
1822 ))
1823}
1824
1825fn identifier_digest(domain: &str, identifier: &str) -> String {
1826 digest_bytes(domain, identifier.as_bytes()).to_hex()
1827}
1828
1829fn artifact_cache_fs_error(error: CacheFsError) -> ArtifactCacheError {
1830 ArtifactCacheError::Io {
1831 operation: error.operation,
1832 path: error.path,
1833 source: error.source,
1834 }
1835}
1836
1837impl ArtifactOutputValidation {
1838 const fn cache_token(self) -> &'static str {
1839 match self {
1840 Self::RegularFile => "regular-file",
1841 Self::NonEmptyFile => "nonempty-file",
1842 }
1843 }
1844}
1845
1846impl std::fmt::Display for ArtifactCacheError {
1847 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1848 match self {
1849 Self::InvalidSpec { message } => {
1850 write!(formatter, "invalid artifact cache spec: {message}")
1851 }
1852 Self::Io {
1853 operation,
1854 path,
1855 source,
1856 } => write!(
1857 formatter,
1858 "failed to {operation} at {}: {source}",
1859 path.display()
1860 ),
1861 Self::InputsChangedDuringPreparation { before, after } => write!(
1862 formatter,
1863 "artifact inputs repeatedly changed during cache preparation: {before} -> {after}",
1864 ),
1865 Self::InputsChangedDuringBuild { before, after } => write!(
1866 formatter,
1867 "artifact inputs changed while the caller was building: {before} -> {after}",
1868 ),
1869 Self::CargoBuildInputsChanged {
1870 label,
1871 before,
1872 after,
1873 } => write!(
1874 formatter,
1875 "resolved Cargo build inputs `{label}` changed: {before} -> {after}",
1876 ),
1877 Self::CargoBuildInputRevalidation { label, source } => write!(
1878 formatter,
1879 "failed to revalidate resolved Cargo build inputs `{label}`: {source}",
1880 ),
1881 Self::InvalidOutputs { outputs } => write!(
1882 formatter,
1883 "artifact transaction has missing or invalid outputs: {}",
1884 outputs
1885 .iter()
1886 .map(|(name, path)| format!("{name} ({})", path.display()))
1887 .collect::<Vec<_>>()
1888 .join(", "),
1889 ),
1890 Self::UnknownOutput { name } => {
1891 write!(
1892 formatter,
1893 "artifact transaction has no output named `{name}`"
1894 )
1895 }
1896 Self::FailedTransactionCleanup {
1897 transaction_error,
1898 path,
1899 source,
1900 } => write!(
1901 formatter,
1902 "artifact transaction failed ({transaction_error}) and staging cleanup at {} also failed: {source}",
1903 path.display(),
1904 ),
1905 }
1906 }
1907}
1908
1909impl std::error::Error for ArtifactCacheError {
1910 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1911 match self {
1912 Self::Io { source, .. } | Self::FailedTransactionCleanup { source, .. } => Some(source),
1913 Self::CargoBuildInputRevalidation { source, .. } => Some(source),
1914 _ => None,
1915 }
1916 }
1917}
1918
1919#[cfg(test)]
1920#[path = "transaction/tests.rs"]
1921mod tests;