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 OS-native ordered command arguments that contribute to the content key.
271    #[must_use]
272    pub fn with_arguments<I, S>(mut self, arguments: I) -> Self
273    where
274        I: IntoIterator<Item = S>,
275        S: AsRef<OsStr>,
276    {
277        self.arguments = arguments
278            .into_iter()
279            .map(|argument| argument.as_ref().to_owned())
280            .collect();
281        self
282    }
283
284    /// Set OS-native environment values that contribute to the content key.
285    #[must_use]
286    pub fn with_environment<I, K, V>(mut self, environment: I) -> Self
287    where
288        I: IntoIterator<Item = (K, V)>,
289        K: Into<OsString>,
290        V: Into<OsString>,
291    {
292        self.environment.extend(
293            environment
294                .into_iter()
295                .map(|(name, value)| (name.into(), Some(value.into()))),
296        );
297        self
298    }
299
300    /// Record OS-native environment names whose unset state contributes to the content key.
301    #[must_use]
302    pub fn with_unset_environment<I, S>(mut self, names: I) -> Self
303    where
304        I: IntoIterator<Item = S>,
305        S: Into<OsString>,
306    {
307        self.environment
308            .extend(names.into_iter().map(|name| (name.into(), None)));
309        self
310    }
311
312    /// Add opaque non-secret identity bytes under a stable logical label.
313    #[must_use]
314    pub fn with_identity_bytes(mut self, label: &str, value: &[u8]) -> Self {
315        self.identities.push(LabeledIdentity {
316            label: label.to_owned(),
317            value: value.to_vec(),
318        });
319        self
320    }
321
322    /// Add one exact Cargo input snapshot as transactional cache identity and guard.
323    ///
324    /// The resolved fingerprint keys toolchain, arguments, environment and the
325    /// dependency closure. Its Cargo-aware source/configuration paths and
326    /// exclusions are rehashed automatically during preparation, cache-hit
327    /// materialization and transaction commit.
328    #[must_use]
329    pub fn with_cargo_build_inputs(
330        mut self,
331        label: &str,
332        build_spec: &WasmBuildSpec,
333        resolved: &ResolvedCargoBuildInputs,
334    ) -> Self {
335        self.cargo_build_inputs.push(CargoBuildInputSet {
336            label: label.to_owned(),
337            build_spec: build_spec.clone(),
338            resolved: resolved.clone(),
339        });
340        self
341    }
342
343    /// Declare one nonempty regular-file output and its nonoverlapping public destination.
344    #[must_use]
345    pub fn with_output(self, name: &str, destination: &Path) -> Self {
346        self.with_output_validation(name, destination, ArtifactOutputValidation::NonEmptyFile)
347    }
348
349    /// Declare one output, nonoverlapping public destination, and validation policy.
350    #[must_use]
351    pub fn with_output_validation(
352        mut self,
353        name: &str,
354        destination: &Path,
355        validation: ArtifactOutputValidation,
356    ) -> Self {
357        self.outputs.push(OutputSpec {
358            name: name.to_owned(),
359            destination: destination.to_owned(),
360            validation,
361        });
362        self.outputs
363            .sort_by(|left, right| left.name.cmp(&right.name));
364        self
365    }
366
367    /// Apply best-effort retention while protecting the acquired entry.
368    #[must_use]
369    pub const fn with_prune_policy(mut self, policy: ArtifactCachePrunePolicy) -> Self {
370        self.prune_policy = Some(policy);
371        self.prune_interval = None;
372        self
373    }
374
375    /// Apply retention at most once per `minimum_interval` for this namespace.
376    ///
377    /// The small due-marker check still occurs under the namespace lock.
378    /// Successful acquisitions skip the directory scan until the interval has
379    /// elapsed. The interval applies to attempted maintenance, including a
380    /// nonfatal failed attempt. A zero interval is equivalent to
381    /// [`Self::with_prune_policy`].
382    #[must_use]
383    pub const fn with_prune_policy_at_most_every(
384        mut self,
385        policy: ArtifactCachePrunePolicy,
386        minimum_interval: Duration,
387    ) -> Self {
388        self.prune_policy = Some(policy);
389        self.prune_interval = Some(minimum_interval);
390        self
391    }
392
393    /// Caller-selected root containing cache data and coordination locks.
394    #[must_use]
395    pub fn cache_root(&self) -> &Path {
396        &self.cache_root
397    }
398
399    /// Stable caller-owned content-store namespace.
400    #[must_use]
401    pub fn namespace(&self) -> &str {
402        &self.namespace
403    }
404
405    /// Stable caller-owned external recipe identity.
406    #[must_use]
407    pub fn recipe_id(&self) -> &str {
408        &self.recipe_id
409    }
410}
411
412impl std::fmt::Debug for ArtifactCacheSpec {
413    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        formatter
415            .debug_struct("ArtifactCacheSpec")
416            .field("cache_root", &self.cache_root)
417            .field("namespace", &self.namespace)
418            .field("recipe_id", &self.recipe_id)
419            .field("coordination_scope", &self.coordination_scope)
420            .field("inputs", &self.inputs)
421            .field("tools", &self.tools)
422            .field("argument_count", &self.arguments.len())
423            .field(
424                "environment_names",
425                &self.environment.keys().collect::<Vec<_>>(),
426            )
427            .field(
428                "identity_labels",
429                &self
430                    .identities
431                    .iter()
432                    .map(|identity| identity.label.as_str())
433                    .collect::<Vec<_>>(),
434            )
435            .field(
436                "cargo_build_input_labels",
437                &self
438                    .cargo_build_inputs
439                    .iter()
440                    .map(|input| input.label.as_str())
441                    .collect::<Vec<_>>(),
442            )
443            .field("outputs", &self.outputs)
444            .field("prune_policy", &self.prune_policy)
445            .field("prune_interval", &self.prune_interval)
446            .finish()
447    }
448}
449
450impl ArtifactCachePreparation {
451    /// Return the already materialized record, or `None` for a cache miss transaction.
452    #[must_use]
453    pub const fn reused_record(&self) -> Option<&ArtifactCacheRecord> {
454        match self {
455            Self::Reused(record) => Some(record),
456            Self::Build(_) => None,
457        }
458    }
459}
460
461impl ArtifactCacheOutcome {
462    /// Read the common acquisition record.
463    #[must_use]
464    pub const fn record(&self) -> &ArtifactCacheRecord {
465        match self {
466            Self::Built(record) | Self::Reused(record) => record,
467        }
468    }
469
470    /// Report whether a complete matching artifact set was reused.
471    #[must_use]
472    pub const fn is_reused(&self) -> bool {
473        matches!(self, Self::Reused(_))
474    }
475}
476
477impl ArtifactCacheRecord {
478    /// Exact content key selecting the immutable cache entry.
479    #[must_use]
480    pub const fn key(&self) -> InputDigest {
481        self.key
482    }
483
484    /// Exact digest of declared input and tool paths.
485    #[must_use]
486    pub const fn input_digest(&self) -> InputDigest {
487        self.input_digest
488    }
489
490    /// Logical artifacts and their materialized caller destinations.
491    #[must_use]
492    pub fn artifacts(&self) -> &[ArtifactCacheArtifact] {
493        &self.artifacts
494    }
495
496    /// Phase timings for this acquisition.
497    #[must_use]
498    pub const fn timings(&self) -> ArtifactCacheTimings {
499        self.timings
500    }
501
502    /// Best-effort retention attempted during this acquisition, when configured.
503    #[must_use]
504    pub const fn maintenance(&self) -> Option<&ArtifactCacheMaintenance> {
505        self.maintenance.as_ref()
506    }
507}
508
509impl ArtifactCacheArtifact {
510    /// Stable logical output name.
511    #[must_use]
512    pub fn name(&self) -> &str {
513        &self.name
514    }
515
516    /// Caller-selected materialized destination.
517    #[must_use]
518    pub fn path(&self) -> &Path {
519        &self.path
520    }
521}
522
523impl ArtifactCacheTimings {
524    /// Time waiting for the caller-selected coordination scope.
525    #[must_use]
526    pub const fn coordination_lock_wait(self) -> Duration {
527        self.coordination_lock_wait
528    }
529
530    /// Time waiting for exact content-key ownership.
531    #[must_use]
532    pub const fn content_lock_wait(self) -> Duration {
533        self.content_lock_wait
534    }
535
536    /// Time waiting for short namespace mutations.
537    #[must_use]
538    pub const fn namespace_lock_wait(self) -> Duration {
539        self.namespace_lock_wait
540    }
541
542    /// Time spent hashing and verifying declared inputs.
543    #[must_use]
544    pub const fn input_capture(self) -> Duration {
545        self.input_capture
546    }
547
548    /// Time spent validating or rejecting a committed cache entry.
549    #[must_use]
550    pub const fn cache_lookup(self) -> Duration {
551        self.cache_lookup
552    }
553
554    /// Time the caller held a miss transaction before committing it.
555    #[must_use]
556    pub const fn caller_build(self) -> Option<Duration> {
557        self.caller_build
558    }
559
560    /// Time spent validating staged output files.
561    #[must_use]
562    pub const fn output_validation(self) -> Duration {
563        self.output_validation
564    }
565
566    /// Time spent publishing an immutable content entry.
567    #[must_use]
568    pub const fn publication(self) -> Duration {
569        self.publication
570    }
571
572    /// Time spent materializing caller-facing output destinations.
573    #[must_use]
574    pub const fn materialization(self) -> Duration {
575        self.materialization
576    }
577
578    /// Time spent on configured best-effort retention.
579    #[must_use]
580    pub const fn maintenance(self) -> Option<Duration> {
581        self.maintenance
582    }
583
584    /// Complete acquisition duration.
585    #[must_use]
586    pub const fn total(self) -> Duration {
587        self.total
588    }
589
590    pub(super) const fn saturating_add(self, other: Self) -> Self {
591        Self {
592            coordination_lock_wait: self
593                .coordination_lock_wait
594                .saturating_add(other.coordination_lock_wait),
595            content_lock_wait: self
596                .content_lock_wait
597                .saturating_add(other.content_lock_wait),
598            namespace_lock_wait: self
599                .namespace_lock_wait
600                .saturating_add(other.namespace_lock_wait),
601            input_capture: self.input_capture.saturating_add(other.input_capture),
602            cache_lookup: self.cache_lookup.saturating_add(other.cache_lookup),
603            caller_build: add_optional_duration(self.caller_build, other.caller_build),
604            output_validation: self
605                .output_validation
606                .saturating_add(other.output_validation),
607            publication: self.publication.saturating_add(other.publication),
608            materialization: self.materialization.saturating_add(other.materialization),
609            maintenance: add_optional_duration(self.maintenance, other.maintenance),
610            total: self.total.saturating_add(other.total),
611        }
612    }
613}
614
615const fn add_optional_duration(
616    left: Option<Duration>,
617    right: Option<Duration>,
618) -> Option<Duration> {
619    match (left, right) {
620        (None, None) => None,
621        (Some(duration), None) | (None, Some(duration)) => Some(duration),
622        (Some(left), Some(right)) => Some(left.saturating_add(right)),
623    }
624}
625
626impl std::fmt::Display for ArtifactCacheTimings {
627    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
628        write!(
629            formatter,
630            "total={:?} coordination_lock={:?} content_lock={:?} namespace_lock={:?} inputs={:?} lookup={:?} build={:?} validation={:?} publication={:?} materialization={:?} maintenance={:?}",
631            self.total,
632            self.coordination_lock_wait,
633            self.content_lock_wait,
634            self.namespace_lock_wait,
635            self.input_capture,
636            self.cache_lookup,
637            self.caller_build,
638            self.output_validation,
639            self.publication,
640            self.materialization,
641            self.maintenance,
642        )
643    }
644}
645
646impl std::fmt::Display for ArtifactCacheOutcome {
647    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
648        let state = if self.is_reused() { "reused" } else { "built" };
649        write!(
650            formatter,
651            "{state} key={} artifacts={} {}",
652            self.record().key,
653            self.record().artifacts.len(),
654            self.record().timings,
655        )
656    }
657}
658
659impl ArtifactBuildTransaction {
660    /// Transaction-owned directory in which the caller may run its build.
661    ///
662    /// Commit accepts only the cache-created `outputs` child at this root.
663    /// Callers must remove or relocate logs and other temporary root children.
664    #[must_use]
665    pub fn staging_directory(&self) -> &Path {
666        &self.staging_directory
667    }
668
669    /// Checked staging destination for one declared logical output.
670    pub fn output_path(&self, name: &str) -> Result<PathBuf, ArtifactCacheError> {
671        self.output_index(name)
672            .map(|index| staged_output_path(&self.staging_directory, index))
673            .ok_or_else(|| ArtifactCacheError::UnknownOutput {
674                name: name.to_owned(),
675            })
676    }
677
678    /// Copy a fixed external output into its transaction-owned staging path.
679    pub fn import_output(&self, name: &str, source: &Path) -> Result<(), ArtifactCacheError> {
680        let destination = self.output_path(name)?;
681        copy_file_atomic(source, &destination).map_err(|source_error| ArtifactCacheError::Io {
682            operation: "import artifact output into staging",
683            path: destination,
684            source: source_error,
685        })?;
686        Ok(())
687    }
688
689    /// Validate the exact staging schema and atomically publish the complete output set.
690    pub fn commit(mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
691        let result = self.commit_inner();
692        match result {
693            Ok(outcome) => Ok(outcome),
694            Err(transaction_error) if self.staging_armed => {
695                let path = self.staging_directory.clone();
696                match remove_path_if_present(&path) {
697                    Ok(()) => {
698                        self.staging_armed = false;
699                        Err(transaction_error)
700                    }
701                    Err(source) => Err(ArtifactCacheError::FailedTransactionCleanup {
702                        transaction_error: Box::new(transaction_error),
703                        path,
704                        source,
705                    }),
706                }
707            }
708            Err(transaction_error) => Err(transaction_error),
709        }
710    }
711
712    /// Abandon the transaction and synchronously remove its staging directory.
713    pub fn abort(mut self) -> Result<(), ArtifactCacheError> {
714        remove_path_if_present(&self.staging_directory).map_err(|source| {
715            ArtifactCacheError::Io {
716                operation: "abort artifact cache transaction",
717                path: self.staging_directory.clone(),
718                source,
719            }
720        })?;
721        self.staging_armed = false;
722        Ok(())
723    }
724
725    fn output_index(&self, name: &str) -> Option<usize> {
726        self.spec
727            .outputs
728            .iter()
729            .position(|output| output.name == name)
730    }
731
732    fn commit_inner(&mut self) -> Result<ArtifactCacheOutcome, ArtifactCacheError> {
733        self.timings.caller_build = Some(self.caller_build_started.elapsed());
734
735        let validation_started = Instant::now();
736        let output_info = inspect_complete_output_set(&self.spec, &self.staging_directory)?;
737        self.timings.output_validation = validation_started.elapsed();
738
739        let capture_started = Instant::now();
740        revalidate_cargo_build_input_fingerprints(&self.spec)?;
741        let verified = resolve_key(&self.spec)?;
742        self.timings.input_capture = self
743            .timings
744            .input_capture
745            .saturating_add(capture_started.elapsed());
746        if verified.input_digest != self.resolved.input_digest {
747            return Err(ArtifactCacheError::InputsChangedDuringBuild {
748                before: self.resolved.input_digest,
749                after: verified.input_digest,
750            });
751        }
752
753        let publication_started = Instant::now();
754        let manifest = manifest_contents(self.resolved.key, &self.spec, &output_info);
755        write_atomic(
756            &self.staging_directory.join(MANIFEST_FILE),
757            manifest.as_bytes(),
758        )
759        .map_err(|source| ArtifactCacheError::Io {
760            operation: "write artifact cache manifest",
761            path: self.staging_directory.join(MANIFEST_FILE),
762            source,
763        })?;
764        let namespace_lock_path = namespace_lock_path(&self.spec);
765        let (_namespace_lock, namespace_wait) =
766            lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
767        self.timings.namespace_lock_wait = self
768            .timings
769            .namespace_lock_wait
770            .saturating_add(namespace_wait);
771        remove_path_if_present(&self.entry_directory).map_err(|source| ArtifactCacheError::Io {
772            operation: "remove conflicting artifact cache entry",
773            path: self.entry_directory.clone(),
774            source,
775        })?;
776        fs::rename(&self.staging_directory, &self.entry_directory).map_err(|source| {
777            ArtifactCacheError::Io {
778                operation: "publish artifact cache entry",
779                path: self.entry_directory.clone(),
780                source,
781            }
782        })?;
783        self.staging_armed = false;
784        self.timings.publication = publication_started.elapsed();
785
786        let materialization_started = Instant::now();
787        materialize_outputs(&self.spec, &self.entry_directory)?;
788        self.timings.materialization = materialization_started.elapsed();
789        record_cache_entry_use(&self.entry_directory).map_err(artifact_cache_fs_error)?;
790        let (maintenance, maintenance_timing) = perform_maintenance_locked(
791            &self.spec,
792            &self.namespace_directory,
793            &self.entry_directory,
794        );
795        self.timings.maintenance = maintenance_timing;
796        self.timings.total = self.total_started.elapsed();
797
798        Ok(ArtifactCacheOutcome::Built(cache_record(
799            &self.spec,
800            self.resolved,
801            self.timings,
802            maintenance,
803        )))
804    }
805}
806
807impl Drop for ArtifactBuildTransaction {
808    fn drop(&mut self) {
809        if self.staging_armed {
810            let _ = remove_path_if_present(&self.staging_directory);
811        }
812    }
813}
814
815/// Verify or begin one exact external artifact-set transaction.
816pub fn prepare_artifact_cache(
817    spec: &ArtifactCacheSpec,
818) -> Result<ArtifactCachePreparation, ArtifactCacheError> {
819    let total_started = Instant::now();
820    validate_spec(spec)?;
821    validate_filesystem_boundaries(spec)?;
822    let namespace_directory = initialize_cache(spec)?;
823
824    let coordination_lock_path = coordination_lock_path(spec);
825    let (coordination_lock, coordination_wait) =
826        lock_cache_file(&coordination_lock_path).map_err(artifact_cache_fs_error)?;
827    let mut timings = ArtifactCacheTimings {
828        coordination_lock_wait: coordination_wait,
829        ..ArtifactCacheTimings::default()
830    };
831    let initial_cargo_started = Instant::now();
832    revalidate_cargo_build_input_fingerprints(spec)?;
833    timings.input_capture = initial_cargo_started.elapsed();
834    let mut last_change = None;
835
836    for _ in 0..MAX_PREPARATION_RETRIES {
837        let capture_started = Instant::now();
838        let resolved = resolve_key(spec)?;
839        timings.input_capture = timings
840            .input_capture
841            .saturating_add(capture_started.elapsed());
842
843        let content_lock_path = content_lock_path(spec, resolved.key);
844        let (content_lock, content_wait) =
845            lock_cache_file(&content_lock_path).map_err(artifact_cache_fs_error)?;
846        timings.content_lock_wait = timings.content_lock_wait.saturating_add(content_wait);
847
848        let verification_started = Instant::now();
849        let verified = resolve_key(spec)?;
850        timings.input_capture = timings
851            .input_capture
852            .saturating_add(verification_started.elapsed());
853        if resolved.input_digest != verified.input_digest {
854            last_change = Some((resolved.input_digest, verified.input_digest));
855            drop(content_lock);
856            continue;
857        }
858
859        let entry_directory = entry_directory(&namespace_directory, resolved.key);
860        let namespace_lock_path = namespace_lock_path(spec);
861        let (namespace_lock, namespace_wait) =
862            lock_cache_file(&namespace_lock_path).map_err(artifact_cache_fs_error)?;
863        timings.namespace_lock_wait = timings.namespace_lock_wait.saturating_add(namespace_wait);
864        let lookup_started = Instant::now();
865        let reusable = cache_entry_is_valid(spec, resolved.key, &entry_directory)?;
866        timings.cache_lookup = timings
867            .cache_lookup
868            .saturating_add(lookup_started.elapsed());
869
870        if reusable {
871            let materialization_started = Instant::now();
872            materialize_outputs(spec, &entry_directory)?;
873            timings.materialization = timings
874                .materialization
875                .saturating_add(materialization_started.elapsed());
876            let after_started = Instant::now();
877            revalidate_cargo_build_input_fingerprints(spec)?;
878            let after = resolve_key(spec)?;
879            timings.input_capture = timings
880                .input_capture
881                .saturating_add(after_started.elapsed());
882            if after.input_digest != resolved.input_digest {
883                last_change = Some((resolved.input_digest, after.input_digest));
884                drop(namespace_lock);
885                drop(content_lock);
886                continue;
887            }
888            record_cache_entry_use(&entry_directory).map_err(artifact_cache_fs_error)?;
889            let (maintenance, maintenance_timing) =
890                perform_maintenance_locked(spec, &namespace_directory, &entry_directory);
891            timings.maintenance = maintenance_timing;
892            timings.total = total_started.elapsed();
893            return Ok(ArtifactCachePreparation::Reused(cache_record(
894                spec,
895                resolved,
896                timings,
897                maintenance,
898            )));
899        }
900
901        remove_path_if_present(&entry_directory).map_err(|source| ArtifactCacheError::Io {
902            operation: "remove invalid artifact cache entry",
903            path: entry_directory.clone(),
904            source,
905        })?;
906        let staging_directory = create_staging_directory(&namespace_directory, resolved.key)?;
907        drop(namespace_lock);
908        return Ok(ArtifactCachePreparation::Build(ArtifactBuildTransaction {
909            spec: Box::new(spec.clone()),
910            resolved,
911            staging_directory,
912            entry_directory,
913            namespace_directory,
914            _coordination_lock: coordination_lock,
915            _content_lock: content_lock,
916            timings,
917            total_started,
918            caller_build_started: Instant::now(),
919            staging_armed: true,
920        }));
921    }
922
923    let (before, after) = last_change.expect("preparation retries require a recorded input change");
924    Err(ArtifactCacheError::InputsChangedDuringPreparation { before, after })
925}
926
927fn revalidate_cargo_build_input_fingerprints(
928    spec: &ArtifactCacheSpec,
929) -> Result<(), ArtifactCacheError> {
930    for cargo_input in &spec.cargo_build_inputs {
931        let current = resolve_cargo_build_inputs(&cargo_input.build_spec).map_err(|source| {
932            ArtifactCacheError::CargoBuildInputRevalidation {
933                label: cargo_input.label.clone(),
934                source,
935            }
936        })?;
937        let before = cargo_input.resolved.fingerprint();
938        let after = current.fingerprint();
939        if after != before {
940            return Err(ArtifactCacheError::CargoBuildInputsChanged {
941                label: cargo_input.label.clone(),
942                before,
943                after,
944            });
945        }
946    }
947    Ok(())
948}
949
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_input_digest()
1326                .map_err(|source| ArtifactCacheError::CargoBuildInputRevalidation {
1327                    label: cargo_input.label.clone(),
1328                    source,
1329                })?;
1330            let before = cargo_input.resolved.input_digest();
1331            if current != before {
1332                return Err(ArtifactCacheError::CargoBuildInputsChanged {
1333                    label: cargo_input.label.clone(),
1334                    before,
1335                    after: current,
1336                });
1337            }
1338            inputs.field("cargo-input-label", cargo_input.label.as_bytes());
1339            inputs.field(
1340                "cargo-build-fingerprint",
1341                cargo_input.resolved.fingerprint().as_bytes(),
1342            );
1343            inputs.field("cargo-input-digest", current.as_bytes());
1344        }
1345        inputs.finish()
1346    };
1347
1348    let mut hasher = InputHasher::new(ARTIFACT_CACHE_FORMAT);
1349    hasher.field("namespace", spec.namespace.as_bytes());
1350    hasher.field("recipe-id", spec.recipe_id.as_bytes());
1351    hasher.field("input-digest", input_digest.as_bytes());
1352    for argument in &spec.arguments {
1353        hasher.field("argument", &os_bytes(argument));
1354    }
1355    for (name, value) in &spec.environment {
1356        hasher.field("environment-name", &os_bytes(name));
1357        match value {
1358            Some(value) => hasher.field("environment-value", &os_bytes(value)),
1359            None => hasher.field("environment-unset", b""),
1360        }
1361    }
1362    let mut identities = spec.identities.iter().collect::<Vec<_>>();
1363    identities.sort_by(|left, right| left.label.cmp(&right.label));
1364    for identity in identities {
1365        hasher.field("identity-label", identity.label.as_bytes());
1366        hasher.field("identity-value", &identity.value);
1367    }
1368    for output in &spec.outputs {
1369        hasher.field("output-name", output.name.as_bytes());
1370        hasher.field(
1371            "output-validation",
1372            output.validation.cache_token().as_bytes(),
1373        );
1374    }
1375    Ok(ResolvedKey {
1376        key: hasher.finish(),
1377        input_digest,
1378    })
1379}
1380
1381fn cache_entry_is_valid(
1382    spec: &ArtifactCacheSpec,
1383    key: InputDigest,
1384    entry: &Path,
1385) -> Result<bool, ArtifactCacheError> {
1386    let entry_metadata = match fs::symlink_metadata(entry) {
1387        Ok(metadata) => metadata,
1388        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
1389        Err(source) => {
1390            return Err(ArtifactCacheError::Io {
1391                operation: "inspect artifact cache entry",
1392                path: entry.to_owned(),
1393                source,
1394            });
1395        }
1396    };
1397    if !entry_metadata.file_type().is_dir() || !cache_entry_root_is_valid(entry)? {
1398        return Ok(false);
1399    }
1400    let manifest_path = entry.join(MANIFEST_FILE);
1401    let manifest_metadata = match fs::symlink_metadata(&manifest_path) {
1402        Ok(metadata) => metadata,
1403        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
1404        Err(source) => {
1405            return Err(ArtifactCacheError::Io {
1406                operation: "inspect artifact cache manifest",
1407                path: manifest_path,
1408                source,
1409            });
1410        }
1411    };
1412    if !manifest_metadata.file_type().is_file() {
1413        return Ok(false);
1414    }
1415    let manifest = fs::read(&manifest_path).map_err(|source| ArtifactCacheError::Io {
1416        operation: "read artifact cache manifest",
1417        path: manifest_path,
1418        source,
1419    })?;
1420    let Some(output_info) = inspect_cached_output_set(spec, entry)? else {
1421        return Ok(false);
1422    };
1423    Ok(manifest == manifest_contents(key, spec, &output_info).as_bytes())
1424}
1425
1426fn inspect_complete_output_set(
1427    spec: &ArtifactCacheSpec,
1428    root: &Path,
1429) -> Result<Vec<ArtifactInfo>, ArtifactCacheError> {
1430    let mut info = Vec::new();
1431    let mut invalid = Vec::new();
1432    let output_directory = root.join("outputs");
1433    if !is_plain_directory(
1434        &output_directory,
1435        "inspect artifact staging output directory",
1436    )? {
1437        invalid.push(("<outputs>".to_owned(), output_directory));
1438        return Err(ArtifactCacheError::InvalidOutputs { outputs: invalid });
1439    }
1440    let outputs = &spec.outputs;
1441    for (index, output) in outputs.iter().enumerate() {
1442        let path = staged_output_path(root, index);
1443        match inspect_artifact(&path, output.validation) {
1444            Ok(Some(artifact)) => info.push(artifact),
1445            Ok(None) => invalid.push((output.name.clone(), path)),
1446            Err(source) => {
1447                return Err(ArtifactCacheError::Io {
1448                    operation: "inspect staged artifact output",
1449                    path,
1450                    source,
1451                });
1452            }
1453        }
1454    }
1455    invalid.extend(
1456        undeclared_output_paths(root, outputs.len())?
1457            .into_iter()
1458            .map(|path| ("<undeclared>".to_owned(), path)),
1459    );
1460    invalid.extend(
1461        undeclared_child_paths(
1462            root,
1463            &BTreeSet::from([OsString::from("outputs")]),
1464            "read artifact staging directory",
1465        )?
1466        .into_iter()
1467        .map(|path| ("<undeclared>".to_owned(), path)),
1468    );
1469    if invalid.is_empty() {
1470        Ok(info)
1471    } else {
1472        Err(ArtifactCacheError::InvalidOutputs { outputs: invalid })
1473    }
1474}
1475
1476fn inspect_cached_output_set(
1477    spec: &ArtifactCacheSpec,
1478    root: &Path,
1479) -> Result<Option<Vec<ArtifactInfo>>, ArtifactCacheError> {
1480    let mut info = Vec::new();
1481    let outputs = &spec.outputs;
1482    for (index, output) in outputs.iter().enumerate() {
1483        let path = staged_output_path(root, index);
1484        match inspect_artifact(&path, output.validation) {
1485            Ok(Some(artifact)) => info.push(artifact),
1486            Ok(None) => return Ok(None),
1487            Err(source) => {
1488                return Err(ArtifactCacheError::Io {
1489                    operation: "inspect cached artifact output",
1490                    path,
1491                    source,
1492                });
1493            }
1494        }
1495    }
1496    if !undeclared_output_paths(root, outputs.len())?.is_empty() {
1497        return Ok(None);
1498    }
1499    Ok(Some(info))
1500}
1501
1502fn undeclared_output_paths(
1503    root: &Path,
1504    output_count: usize,
1505) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1506    let output_directory = root.join("outputs");
1507    let expected = (0..output_count)
1508        .map(format_output_index)
1509        .map(OsString::from)
1510        .collect::<BTreeSet<_>>();
1511    undeclared_child_paths(
1512        &output_directory,
1513        &expected,
1514        "read artifact output directory",
1515    )
1516}
1517
1518fn undeclared_child_paths(
1519    directory: &Path,
1520    expected: &BTreeSet<OsString>,
1521    operation: &'static str,
1522) -> Result<Vec<PathBuf>, ArtifactCacheError> {
1523    let entries = fs::read_dir(directory).map_err(|source| ArtifactCacheError::Io {
1524        operation,
1525        path: directory.to_owned(),
1526        source,
1527    })?;
1528    let mut undeclared = Vec::new();
1529    for entry in entries {
1530        let entry = entry.map_err(|source| ArtifactCacheError::Io {
1531            operation,
1532            path: directory.to_owned(),
1533            source,
1534        })?;
1535        if !expected.contains(&entry.file_name()) {
1536            undeclared.push(entry.path());
1537        }
1538    }
1539    Ok(undeclared)
1540}
1541
1542fn cache_entry_root_is_valid(root: &Path) -> Result<bool, ArtifactCacheError> {
1543    let expected = BTreeSet::from([
1544        OsString::from("outputs"),
1545        OsString::from(MANIFEST_FILE),
1546        OsString::from(LAST_USED_FILE),
1547    ]);
1548    if !undeclared_child_paths(root, &expected, "read artifact cache entry")?.is_empty() {
1549        return Ok(false);
1550    }
1551    if !is_plain_directory(
1552        &root.join("outputs"),
1553        "inspect artifact cache output directory",
1554    )? {
1555        return Ok(false);
1556    }
1557    let last_used = root.join(LAST_USED_FILE);
1558    match fs::symlink_metadata(&last_used) {
1559        Ok(metadata) => Ok(metadata.file_type().is_file()),
1560        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(true),
1561        Err(source) => Err(ArtifactCacheError::Io {
1562            operation: "inspect artifact cache use marker",
1563            path: last_used,
1564            source,
1565        }),
1566    }
1567}
1568
1569fn is_plain_directory(path: &Path, operation: &'static str) -> Result<bool, ArtifactCacheError> {
1570    match fs::symlink_metadata(path) {
1571        Ok(metadata) => Ok(metadata.file_type().is_dir()),
1572        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
1573        Err(source) => Err(ArtifactCacheError::Io {
1574            operation,
1575            path: path.to_owned(),
1576            source,
1577        }),
1578    }
1579}
1580
1581fn inspect_artifact(
1582    path: &Path,
1583    validation: ArtifactOutputValidation,
1584) -> io::Result<Option<ArtifactInfo>> {
1585    let metadata = match fs::symlink_metadata(path) {
1586        Ok(metadata) => metadata,
1587        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1588        Err(error) => return Err(error),
1589    };
1590    if !metadata.file_type().is_file() {
1591        return Ok(None);
1592    }
1593    if validation == ArtifactOutputValidation::NonEmptyFile && metadata.len() == 0 {
1594        return Ok(None);
1595    }
1596    let (bytes, digest) = digest_file("artifact-set-output-v1", path)?;
1597    Ok(Some(ArtifactInfo { bytes, digest }))
1598}
1599
1600fn manifest_contents(
1601    key: InputDigest,
1602    spec: &ArtifactCacheSpec,
1603    output_info: &[ArtifactInfo],
1604) -> String {
1605    let mut manifest = format!("{ARTIFACT_CACHE_FORMAT}\nkey:{key}\n");
1606    for ((index, output), info) in spec.outputs.iter().enumerate().zip(output_info) {
1607        use std::fmt::Write as _;
1608        writeln!(
1609            manifest,
1610            "output:{index}:{}:{}:{}:{}",
1611            output.name,
1612            output.validation.cache_token(),
1613            info.bytes,
1614            info.digest,
1615        )
1616        .expect("writing an artifact manifest to a String cannot fail");
1617    }
1618    manifest
1619}
1620
1621fn materialize_outputs(spec: &ArtifactCacheSpec, entry: &Path) -> Result<(), ArtifactCacheError> {
1622    for (index, output) in spec.outputs.iter().enumerate() {
1623        let cached = staged_output_path(entry, index);
1624        copy_file_atomic(&cached, &output.destination).map_err(|source| {
1625            ArtifactCacheError::Io {
1626                operation: "materialize artifact output",
1627                path: output.destination.clone(),
1628                source,
1629            }
1630        })?;
1631    }
1632    Ok(())
1633}
1634
1635fn perform_maintenance_locked(
1636    spec: &ArtifactCacheSpec,
1637    namespace: &Path,
1638    protected_entry: &Path,
1639) -> (Option<ArtifactCacheMaintenance>, Option<Duration>) {
1640    spec.prune_policy.map_or((None, None), |policy| {
1641        let identity = policy.maintenance_identity();
1642        perform_scheduled_cache_maintenance(namespace, spec.prune_interval, &identity, || {
1643            prune_artifact_namespace_locked(
1644                &spec.cache_root,
1645                namespace,
1646                policy,
1647                Some(protected_entry),
1648            )
1649            .map_err(|error| error.to_string())
1650        })
1651    })
1652}
1653
1654fn prune_artifact_namespace_locked(
1655    cache_root: &Path,
1656    namespace: &Path,
1657    policy: ArtifactCachePrunePolicy,
1658    protected_entry: Option<&Path>,
1659) -> Result<ArtifactCachePruneReport, ArtifactCacheError> {
1660    let mut report = prune_direct_child_directories(
1661        &entries_directory(namespace),
1662        policy,
1663        protected_entry,
1664        is_sha256_directory,
1665    )
1666    .map_err(artifact_cache_fs_error)?;
1667    remove_abandoned_staging(cache_root, namespace, &mut report)?;
1668    Ok(report)
1669}
1670
1671fn remove_abandoned_staging(
1672    cache_root: &Path,
1673    namespace: &Path,
1674    report: &mut ArtifactCachePruneReport,
1675) -> Result<(), ArtifactCacheError> {
1676    let staging_root = namespace.join("staging");
1677    let entries = match fs::read_dir(&staging_root) {
1678        Ok(entries) => entries,
1679        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
1680        Err(source) => {
1681            return Err(ArtifactCacheError::Io {
1682                operation: "read artifact staging root during pruning",
1683                path: staging_root,
1684                source,
1685            });
1686        }
1687    };
1688    for entry in entries {
1689        let entry = entry.map_err(|source| ArtifactCacheError::Io {
1690            operation: "read artifact staging entry during pruning",
1691            path: staging_root.clone(),
1692            source,
1693        })?;
1694        let file_type = entry.file_type().map_err(|source| ArtifactCacheError::Io {
1695            operation: "inspect artifact staging entry during pruning",
1696            path: entry.path(),
1697            source,
1698        })?;
1699        if !file_type.is_dir() {
1700            continue;
1701        }
1702        let file_name = entry.file_name();
1703        let Some(key) = staging_content_key(&file_name) else {
1704            continue;
1705        };
1706        let lock_path = content_lock_path_for_key(cache_root, key);
1707        let Some(_content_lock) =
1708            try_lock_cache_file(&lock_path).map_err(artifact_cache_fs_error)?
1709        else {
1710            continue;
1711        };
1712        let path = entry.path();
1713        let bytes = directory_logical_size(&path).map_err(|source| ArtifactCacheError::Io {
1714            operation: "measure abandoned artifact staging directory",
1715            path: path.clone(),
1716            source,
1717        })?;
1718        remove_path_if_present(&path).map_err(|source| ArtifactCacheError::Io {
1719            operation: "remove abandoned artifact staging directory during pruning",
1720            path,
1721            source,
1722        })?;
1723        report.record_uncommitted_removal(bytes);
1724    }
1725    Ok(())
1726}
1727
1728fn staging_content_key(name: &OsStr) -> Option<&str> {
1729    let name = name.to_str()?;
1730    let (key, suffix) = name.split_once('-')?;
1731    (!suffix.is_empty() && key.len() == 64 && key.as_bytes().iter().all(u8::is_ascii_hexdigit))
1732        .then_some(key)
1733}
1734
1735fn cache_record(
1736    spec: &ArtifactCacheSpec,
1737    resolved: ResolvedKey,
1738    timings: ArtifactCacheTimings,
1739    maintenance: Option<ArtifactCacheMaintenance>,
1740) -> ArtifactCacheRecord {
1741    ArtifactCacheRecord {
1742        key: resolved.key,
1743        input_digest: resolved.input_digest,
1744        artifacts: spec
1745            .outputs
1746            .iter()
1747            .map(|output| ArtifactCacheArtifact {
1748                name: output.name.clone(),
1749                path: output.destination.clone(),
1750            })
1751            .collect(),
1752        timings,
1753        maintenance,
1754    }
1755}
1756
1757fn create_staging_directory(
1758    namespace: &Path,
1759    key: InputDigest,
1760) -> Result<PathBuf, ArtifactCacheError> {
1761    let staging_root = namespace.join("staging");
1762    fs::create_dir_all(&staging_root).map_err(|source| ArtifactCacheError::Io {
1763        operation: "create artifact staging root",
1764        path: staging_root.clone(),
1765        source,
1766    })?;
1767    remove_same_key_staging(&staging_root, key)?;
1768    let sequence = STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed);
1769    let staging = staging_root.join(format!("{key}-{}-{sequence}", std::process::id()));
1770    fs::create_dir_all(staging.join("outputs")).map_err(|source| ArtifactCacheError::Io {
1771        operation: "create artifact transaction staging directory",
1772        path: staging.clone(),
1773        source,
1774    })?;
1775    Ok(staging)
1776}
1777
1778fn remove_same_key_staging(root: &Path, key: InputDigest) -> Result<(), ArtifactCacheError> {
1779    let prefix = format!("{key}-");
1780    let entries = fs::read_dir(root).map_err(|source| ArtifactCacheError::Io {
1781        operation: "read artifact staging root",
1782        path: root.to_owned(),
1783        source,
1784    })?;
1785    for entry in entries {
1786        let entry = entry.map_err(|source| ArtifactCacheError::Io {
1787            operation: "read artifact staging entry",
1788            path: root.to_owned(),
1789            source,
1790        })?;
1791        if entry.file_name().to_string_lossy().starts_with(&prefix) {
1792            remove_path_if_present(&entry.path()).map_err(|source| ArtifactCacheError::Io {
1793                operation: "remove abandoned artifact staging directory",
1794                path: entry.path(),
1795                source,
1796            })?;
1797        }
1798    }
1799    Ok(())
1800}
1801
1802fn staged_output_path(root: &Path, index: usize) -> PathBuf {
1803    root.join("outputs").join(format_output_index(index))
1804}
1805
1806fn format_output_index(index: usize) -> String {
1807    format!("{index:04}.artifact")
1808}
1809
1810fn namespace_directory(spec: &ArtifactCacheSpec) -> PathBuf {
1811    namespace_directory_for(&spec.cache_root, &spec.namespace)
1812}
1813
1814fn namespace_directory_for(cache_root: &Path, namespace: &str) -> PathBuf {
1815    cache_root
1816        .join(".ic-testkit/artifact-sets/namespaces")
1817        .join(identifier_digest("artifact-cache-namespace-v1", namespace))
1818}
1819
1820fn entries_directory(namespace: &Path) -> PathBuf {
1821    namespace.join("entries")
1822}
1823
1824fn entry_directory(namespace: &Path, key: InputDigest) -> PathBuf {
1825    entries_directory(namespace).join(key.to_hex())
1826}
1827
1828fn coordination_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1829    spec.cache_root
1830        .join(".ic-testkit/artifact-sets/locks/coordination")
1831        .join(format!(
1832            "{}.lock",
1833            identifier_digest("artifact-cache-coordination-v1", &spec.coordination_scope,)
1834        ))
1835}
1836
1837fn content_lock_path(spec: &ArtifactCacheSpec, key: InputDigest) -> PathBuf {
1838    content_lock_path_for_key(&spec.cache_root, &key.to_hex())
1839}
1840
1841fn content_lock_path_for_key(cache_root: &Path, key: &str) -> PathBuf {
1842    cache_root
1843        .join(".ic-testkit/artifact-sets/locks/content")
1844        .join(format!("{key}.lock"))
1845}
1846
1847fn namespace_lock_path(spec: &ArtifactCacheSpec) -> PathBuf {
1848    namespace_lock_path_for(&spec.cache_root, &spec.namespace)
1849}
1850
1851fn namespace_lock_path_for(cache_root: &Path, namespace: &str) -> PathBuf {
1852    cache_root
1853        .join(".ic-testkit/artifact-sets/locks/namespaces")
1854        .join(format!(
1855            "{}.lock",
1856            identifier_digest("artifact-cache-namespace-v1", namespace)
1857        ))
1858}
1859
1860fn identifier_digest(domain: &str, identifier: &str) -> String {
1861    digest_bytes(domain, identifier.as_bytes()).to_hex()
1862}
1863
1864fn artifact_cache_fs_error(error: CacheFsError) -> ArtifactCacheError {
1865    ArtifactCacheError::Io {
1866        operation: error.operation,
1867        path: error.path,
1868        source: error.source,
1869    }
1870}
1871
1872impl ArtifactOutputValidation {
1873    const fn cache_token(self) -> &'static str {
1874        match self {
1875            Self::RegularFile => "regular-file",
1876            Self::NonEmptyFile => "nonempty-file",
1877        }
1878    }
1879}
1880
1881impl std::fmt::Display for ArtifactCacheError {
1882    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1883        match self {
1884            Self::InvalidSpec { message } => {
1885                write!(formatter, "invalid artifact cache spec: {message}")
1886            }
1887            Self::Io {
1888                operation,
1889                path,
1890                source,
1891            } => write!(
1892                formatter,
1893                "failed to {operation} at {}: {source}",
1894                path.display()
1895            ),
1896            Self::InputsChangedDuringPreparation { before, after } => write!(
1897                formatter,
1898                "artifact inputs repeatedly changed during cache preparation: {before} -> {after}",
1899            ),
1900            Self::InputsChangedDuringBuild { before, after } => write!(
1901                formatter,
1902                "artifact inputs changed while the caller was building: {before} -> {after}",
1903            ),
1904            Self::CargoBuildInputsChanged {
1905                label,
1906                before,
1907                after,
1908            } => write!(
1909                formatter,
1910                "resolved Cargo build inputs `{label}` changed: {before} -> {after}",
1911            ),
1912            Self::CargoBuildInputRevalidation { label, source } => write!(
1913                formatter,
1914                "failed to revalidate resolved Cargo build inputs `{label}`: {source}",
1915            ),
1916            Self::InvalidOutputs { outputs } => write!(
1917                formatter,
1918                "artifact transaction has missing or invalid outputs: {}",
1919                outputs
1920                    .iter()
1921                    .map(|(name, path)| format!("{name} ({})", path.display()))
1922                    .collect::<Vec<_>>()
1923                    .join(", "),
1924            ),
1925            Self::UnknownOutput { name } => {
1926                write!(
1927                    formatter,
1928                    "artifact transaction has no output named `{name}`"
1929                )
1930            }
1931            Self::FailedTransactionCleanup {
1932                transaction_error,
1933                path,
1934                source,
1935            } => write!(
1936                formatter,
1937                "artifact transaction failed ({transaction_error}) and staging cleanup at {} also failed: {source}",
1938                path.display(),
1939            ),
1940        }
1941    }
1942}
1943
1944impl std::error::Error for ArtifactCacheError {
1945    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1946        match self {
1947            Self::Io { source, .. } | Self::FailedTransactionCleanup { source, .. } => Some(source),
1948            Self::CargoBuildInputRevalidation { source, .. } => Some(source),
1949            _ => None,
1950        }
1951    }
1952}
1953
1954#[cfg(test)]
1955#[path = "transaction/tests.rs"]
1956mod tests;