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