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