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