Skip to main content

ic_testkit/artifacts/
transaction.rs

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