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