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