ic_testkit/artifacts/
cache_fs.rs

1use fs2::FileExt as _;
2use std::{
3    fs::{self, File, OpenOptions},
4    io,
5    path::{Path, PathBuf},
6    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
7};
8
9use super::digest::write_atomic;
10
11const CACHE_DIRECTORY_TAG: &str = "Signature: 8a477f597d28d172789f06886806bc55\n\
12# This file is a cache directory tag created by ic-testkit.\n\
13# For information about cache directory tags see https://bford.info/cachedir/\n";
14pub(super) const CACHE_DIRECTORY_TAG_SIGNATURE: &str =
15    "Signature: 8a477f597d28d172789f06886806bc55\n";
16pub(super) const LAST_USED_FILE: &str = ".ic-testkit-last-used";
17const LAST_MAINTENANCE_FILE: &str = ".ic-testkit-last-maintenance";
18
19/// Caller-selected retention limits for content-addressed artifact entries.
20///
21/// Age pruning runs before size pruning. A policy without either limit scans
22/// the selected cache namespace and updates its cache metadata without
23/// removing entries.
24#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
25pub struct ArtifactCachePrunePolicy {
26    max_age: Option<Duration>,
27    max_size_bytes: Option<u64>,
28}
29
30/// Summary of one lock-coordinated artifact-cache pruning pass.
31#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
32pub struct ArtifactCachePruneReport {
33    entries_scanned: usize,
34    entries_removed: usize,
35    bytes_before: u64,
36    bytes_removed: u64,
37    uncommitted_directories_removed: usize,
38    uncommitted_bytes_removed: u64,
39}
40
41/// Nonfatal retention attempted as part of a successful cache acquisition.
42#[non_exhaustive]
43#[derive(Clone, Debug, Eq, PartialEq)]
44pub enum ArtifactCacheMaintenance {
45    /// Configured retention completed under the cache lock.
46    Pruned(ArtifactCachePruneReport),
47    /// Configured retention failed after the requested artifacts were ready.
48    PruneFailed {
49        /// Cache error rendered without invalidating the successful acquisition.
50        message: String,
51    },
52}
53
54impl ArtifactCachePrunePolicy {
55    /// Create a policy that records cache metadata without removing entries.
56    #[must_use]
57    pub const fn new() -> Self {
58        Self {
59            max_age: None,
60            max_size_bytes: None,
61        }
62    }
63
64    /// Remove entries older than `max_age` before applying the size limit.
65    #[must_use]
66    pub const fn with_max_age(mut self, max_age: Duration) -> Self {
67        self.max_age = Some(max_age);
68        self
69    }
70
71    /// Remove least-recently-used entries until retained logical size is at most `bytes`.
72    #[must_use]
73    pub const fn with_max_size_bytes(mut self, bytes: u64) -> Self {
74        self.max_size_bytes = Some(bytes);
75        self
76    }
77
78    /// Configured maximum entry age, if any.
79    #[must_use]
80    pub const fn max_age(self) -> Option<Duration> {
81        self.max_age
82    }
83
84    /// Configured maximum logical cache size in bytes, if any.
85    #[must_use]
86    pub const fn max_size_bytes(self) -> Option<u64> {
87        self.max_size_bytes
88    }
89
90    pub(super) fn maintenance_identity(self) -> String {
91        format!(
92            "age={:?};size={:?}",
93            self.max_age.map(|duration| duration.as_nanos()),
94            self.max_size_bytes
95        )
96    }
97}
98
99impl ArtifactCachePruneReport {
100    /// Number of content-addressed directories considered for pruning.
101    #[must_use]
102    pub const fn entries_scanned(self) -> usize {
103        self.entries_scanned
104    }
105
106    /// Number of content-addressed directories removed.
107    #[must_use]
108    pub const fn entries_removed(self) -> usize {
109        self.entries_removed
110    }
111
112    /// Number of content-addressed directories retained.
113    #[must_use]
114    pub const fn entries_retained(self) -> usize {
115        self.entries_scanned.saturating_sub(self.entries_removed)
116    }
117
118    /// Logical bytes occupied by scanned entries before pruning.
119    #[must_use]
120    pub const fn bytes_before(self) -> u64 {
121        self.bytes_before
122    }
123
124    /// Logical bytes removed by pruning.
125    #[must_use]
126    pub const fn bytes_removed(self) -> u64 {
127        self.bytes_removed
128    }
129
130    /// Logical bytes occupied by retained entries after pruning.
131    #[must_use]
132    pub const fn bytes_retained(self) -> u64 {
133        self.bytes_before.saturating_sub(self.bytes_removed)
134    }
135
136    /// Abandoned transaction directories removed outside the committed-entry totals.
137    #[must_use]
138    pub const fn uncommitted_directories_removed(self) -> usize {
139        self.uncommitted_directories_removed
140    }
141
142    /// Logical bytes removed from abandoned transaction directories.
143    #[must_use]
144    pub const fn uncommitted_bytes_removed(self) -> u64 {
145        self.uncommitted_bytes_removed
146    }
147
148    pub(super) const fn record_uncommitted_removal(&mut self, bytes: u64) {
149        self.uncommitted_directories_removed += 1;
150        self.uncommitted_bytes_removed = self.uncommitted_bytes_removed.saturating_add(bytes);
151    }
152}
153
154impl ArtifactCacheMaintenance {
155    /// Successful pruning report, or `None` when maintenance failed.
156    #[must_use]
157    pub const fn prune_report(&self) -> Option<ArtifactCachePruneReport> {
158        match self {
159            Self::Pruned(report) => Some(*report),
160            Self::PruneFailed { .. } => None,
161        }
162    }
163
164    /// Rendered maintenance failure, or `None` when pruning succeeded.
165    #[must_use]
166    pub fn failure_message(&self) -> Option<&str> {
167        match self {
168            Self::Pruned(_) => None,
169            Self::PruneFailed { message } => Some(message),
170        }
171    }
172}
173
174#[derive(Debug)]
175pub(super) struct CacheFsError {
176    pub(super) operation: &'static str,
177    pub(super) path: PathBuf,
178    pub(super) source: io::Error,
179}
180
181pub(super) fn ensure_cache_directory_tag(cache_root: &Path) -> Result<(), CacheFsError> {
182    let path = cache_root.join("CACHEDIR.TAG");
183    if fs::read_to_string(&path)
184        .is_ok_and(|contents| contents.starts_with(CACHE_DIRECTORY_TAG_SIGNATURE))
185    {
186        return Ok(());
187    }
188    write_atomic(&path, CACHE_DIRECTORY_TAG.as_bytes()).map_err(|source| CacheFsError {
189        operation: "write cache directory tag",
190        path,
191        source,
192    })
193}
194
195pub(super) fn lock_cache_file(path: &Path) -> Result<(File, Duration), CacheFsError> {
196    let file = open_cache_lock_file(path)?;
197    let started = Instant::now();
198    file.lock_exclusive().map_err(|source| CacheFsError {
199        operation: "lock cache",
200        path: path.to_owned(),
201        source,
202    })?;
203    Ok((file, started.elapsed()))
204}
205
206pub(super) fn try_lock_cache_file(path: &Path) -> Result<Option<File>, CacheFsError> {
207    let file = open_cache_lock_file(path)?;
208    match file.try_lock_exclusive() {
209        Ok(()) => Ok(Some(file)),
210        Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(None),
211        Err(source) => Err(CacheFsError {
212            operation: "try lock cache",
213            path: path.to_owned(),
214            source,
215        }),
216    }
217}
218
219fn open_cache_lock_file(path: &Path) -> Result<File, CacheFsError> {
220    if let Some(parent) = path.parent() {
221        fs::create_dir_all(parent).map_err(|source| CacheFsError {
222            operation: "create cache lock directory",
223            path: parent.to_owned(),
224            source,
225        })?;
226    }
227    OpenOptions::new()
228        .create(true)
229        .read(true)
230        .write(true)
231        .truncate(false)
232        .open(path)
233        .map_err(|source| CacheFsError {
234            operation: "open cache lock",
235            path: path.to_owned(),
236            source,
237        })
238}
239
240pub(super) fn record_cache_entry_use(path: &Path) -> Result<(), CacheFsError> {
241    write_last_used(path, SystemTime::now())
242}
243
244fn cache_maintenance_due(
245    path: &Path,
246    minimum_interval: Option<Duration>,
247    maintenance_identity: &str,
248) -> Result<bool, CacheFsError> {
249    let Some(minimum_interval) = minimum_interval else {
250        return Ok(true);
251    };
252    let marker = path.join(LAST_MAINTENANCE_FILE);
253    let contents = match fs::read_to_string(&marker) {
254        Ok(contents) => contents,
255        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(true),
256        Err(source) => {
257            return Err(CacheFsError {
258                operation: "read cache maintenance time",
259                path: marker,
260                source,
261            });
262        }
263    };
264    let mut lines = contents.lines();
265    let Some(last_maintenance) = lines.next().and_then(decode_system_time) else {
266        return Ok(true);
267    };
268    if lines.next() != Some(maintenance_identity) {
269        return Ok(true);
270    }
271    Ok(match SystemTime::now().duration_since(last_maintenance) {
272        Ok(elapsed) => elapsed >= minimum_interval,
273        Err(_) => true,
274    })
275}
276
277fn record_cache_maintenance(path: &Path, maintenance_identity: &str) -> Result<(), CacheFsError> {
278    fs::create_dir_all(path).map_err(|source| CacheFsError {
279        operation: "create cache maintenance directory",
280        path: path.to_owned(),
281        source,
282    })?;
283    let marker = path.join(LAST_MAINTENANCE_FILE);
284    let elapsed = encode_system_time(&marker, SystemTime::now())?;
285    let contents = format!("{}\n{maintenance_identity}\n", elapsed.as_nanos());
286    write_atomic(&marker, contents.as_bytes()).map_err(|source| CacheFsError {
287        operation: "record cache maintenance time",
288        path: marker,
289        source,
290    })
291}
292
293pub(super) fn perform_scheduled_cache_maintenance(
294    path: &Path,
295    minimum_interval: Option<Duration>,
296    maintenance_identity: &str,
297    maintenance: impl FnOnce() -> Result<ArtifactCachePruneReport, String>,
298) -> (Option<ArtifactCacheMaintenance>, Option<Duration>) {
299    let started = Instant::now();
300    match cache_maintenance_due(path, minimum_interval, maintenance_identity) {
301        Ok(false) => return (None, Some(started.elapsed())),
302        Ok(true) => {}
303        Err(error) => {
304            return (
305                Some(ArtifactCacheMaintenance::PruneFailed {
306                    message: error.to_string(),
307                }),
308                Some(started.elapsed()),
309            );
310        }
311    }
312
313    let result = maintenance();
314    let marker = record_cache_maintenance(path, maintenance_identity);
315    let outcome = match (result, marker) {
316        (Ok(report), Ok(())) => ArtifactCacheMaintenance::Pruned(report),
317        (Err(message), Ok(())) => ArtifactCacheMaintenance::PruneFailed { message },
318        (Ok(_), Err(error)) => ArtifactCacheMaintenance::PruneFailed {
319            message: error.to_string(),
320        },
321        (Err(message), Err(marker)) => ArtifactCacheMaintenance::PruneFailed {
322            message: format!(
323                "{message}; additionally failed to record the maintenance attempt: {marker}"
324            ),
325        },
326    };
327    (Some(outcome), Some(started.elapsed()))
328}
329
330pub(super) fn write_last_used(path: &Path, last_used: SystemTime) -> Result<(), CacheFsError> {
331    let marker = path.join(LAST_USED_FILE);
332    write_system_time(&marker, last_used, "record cache use time")
333}
334
335fn write_system_time(
336    path: &Path,
337    timestamp: SystemTime,
338    operation: &'static str,
339) -> Result<(), CacheFsError> {
340    let elapsed = encode_system_time(path, timestamp)?;
341    write_atomic(path, elapsed.as_nanos().to_string().as_bytes()).map_err(|source| CacheFsError {
342        operation,
343        path: path.to_owned(),
344        source,
345    })
346}
347
348fn encode_system_time(path: &Path, timestamp: SystemTime) -> Result<Duration, CacheFsError> {
349    timestamp
350        .duration_since(UNIX_EPOCH)
351        .map_err(|source| CacheFsError {
352            operation: "encode cache time",
353            path: path.to_owned(),
354            source: io::Error::new(io::ErrorKind::InvalidInput, source),
355        })
356}
357
358fn decode_system_time(contents: &str) -> Option<SystemTime> {
359    let nanoseconds = contents.parse::<u128>().ok()?;
360    let seconds = u64::try_from(nanoseconds / 1_000_000_000).ok()?;
361    let subsecond_nanos = (nanoseconds % 1_000_000_000) as u32;
362    UNIX_EPOCH.checked_add(Duration::new(seconds, subsecond_nanos))
363}
364
365pub(super) fn prune_direct_child_directories(
366    cache_root: &Path,
367    policy: ArtifactCachePrunePolicy,
368    protected_entry: Option<&Path>,
369    is_eligible: impl Fn(&Path) -> bool,
370) -> Result<ArtifactCachePruneReport, CacheFsError> {
371    let mut entries = cache_entries(cache_root, is_eligible)?;
372    let bytes_before = entries
373        .iter()
374        .fold(0_u64, |total, entry| total.saturating_add(entry.bytes));
375    let mut report = ArtifactCachePruneReport {
376        entries_scanned: entries.len(),
377        entries_removed: 0,
378        bytes_before,
379        bytes_removed: 0,
380        uncommitted_directories_removed: 0,
381        uncommitted_bytes_removed: 0,
382    };
383    let now = SystemTime::now();
384
385    if let Some(max_age) = policy.max_age() {
386        for entry in &mut entries {
387            let age = now.duration_since(entry.last_used).unwrap_or_default();
388            if protected_entry != Some(entry.path.as_path()) && age > max_age {
389                remove_cache_entry(entry, &mut report)?;
390            }
391        }
392    }
393
394    if let Some(max_size_bytes) = policy.max_size_bytes() {
395        entries.sort_by(|left, right| {
396            left.last_used
397                .cmp(&right.last_used)
398                .then_with(|| left.path.cmp(&right.path))
399        });
400        for entry in &mut entries {
401            if report.bytes_retained() <= max_size_bytes {
402                break;
403            }
404            if protected_entry == Some(entry.path.as_path()) {
405                continue;
406            }
407            remove_cache_entry(entry, &mut report)?;
408        }
409    }
410
411    Ok(report)
412}
413
414pub(super) fn directory_logical_size(path: &Path) -> io::Result<u64> {
415    let mut total = 0_u64;
416    let mut pending = vec![path.to_owned()];
417    while let Some(current) = pending.pop() {
418        let metadata = fs::symlink_metadata(&current)?;
419        if metadata.is_dir() {
420            for entry in fs::read_dir(&current)? {
421                pending.push(entry?.path());
422            }
423        } else {
424            total = total.saturating_add(metadata.len());
425        }
426    }
427    Ok(total)
428}
429
430pub(super) fn is_sha256_directory(path: &Path) -> bool {
431    path.file_name().is_some_and(|name| {
432        let bytes = name.as_encoded_bytes();
433        bytes.len() == 64 && bytes.iter().all(u8::is_ascii_hexdigit)
434    })
435}
436
437pub(super) fn remove_path_if_present(path: &Path) -> io::Result<()> {
438    let metadata = match fs::symlink_metadata(path) {
439        Ok(metadata) => metadata,
440        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
441        Err(error) => return Err(error),
442    };
443    if metadata.file_type().is_dir() {
444        fs::remove_dir_all(path)
445    } else {
446        fs::remove_file(path)
447    }
448}
449
450struct CacheEntry {
451    path: PathBuf,
452    bytes: u64,
453    last_used: SystemTime,
454    removed: bool,
455}
456
457impl std::fmt::Display for CacheFsError {
458    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
459        write!(
460            formatter,
461            "failed to {} at {}: {}",
462            self.operation,
463            self.path.display(),
464            self.source
465        )
466    }
467}
468
469impl std::error::Error for CacheFsError {
470    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
471        Some(&self.source)
472    }
473}
474
475fn cache_entries(
476    cache_root: &Path,
477    is_eligible: impl Fn(&Path) -> bool,
478) -> Result<Vec<CacheEntry>, CacheFsError> {
479    let read_dir = match fs::read_dir(cache_root) {
480        Ok(read_dir) => read_dir,
481        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
482        Err(source) => {
483            return Err(CacheFsError {
484                operation: "read cache directory",
485                path: cache_root.to_owned(),
486                source,
487            });
488        }
489    };
490    let mut entries = Vec::new();
491    for directory_entry in read_dir {
492        let directory_entry = directory_entry.map_err(|source| CacheFsError {
493            operation: "read cache entry",
494            path: cache_root.to_owned(),
495            source,
496        })?;
497        let path = directory_entry.path();
498        let file_type = directory_entry.file_type().map_err(|source| CacheFsError {
499            operation: "inspect cache entry",
500            path: path.clone(),
501            source,
502        })?;
503        if !file_type.is_dir() || !is_eligible(&path) {
504            continue;
505        }
506        let bytes = directory_logical_size(&path).map_err(|source| CacheFsError {
507            operation: "measure cache entry",
508            path: path.clone(),
509            source,
510        })?;
511        let last_used = cache_entry_last_used(&path).map_err(|source| CacheFsError {
512            operation: "read cache use time",
513            path: path.clone(),
514            source,
515        })?;
516        entries.push(CacheEntry {
517            path,
518            bytes,
519            last_used,
520            removed: false,
521        });
522    }
523    Ok(entries)
524}
525
526pub(super) fn cache_entry_last_used(path: &Path) -> io::Result<SystemTime> {
527    let marker = path.join(LAST_USED_FILE);
528    if let Ok(contents) = fs::read_to_string(&marker)
529        && let Some(timestamp) = decode_system_time(&contents)
530    {
531        return Ok(timestamp);
532    }
533    fs::metadata(path)?.modified()
534}
535
536fn remove_cache_entry(
537    entry: &mut CacheEntry,
538    report: &mut ArtifactCachePruneReport,
539) -> Result<(), CacheFsError> {
540    if entry.removed {
541        return Ok(());
542    }
543    remove_path_if_present(&entry.path).map_err(|source| CacheFsError {
544        operation: "prune cache entry",
545        path: entry.path.clone(),
546        source,
547    })?;
548    entry.removed = true;
549    report.entries_removed += 1;
550    report.bytes_removed = report.bytes_removed.saturating_add(entry.bytes);
551    Ok(())
552}
ic_testkit/artifacts/cache_fs.rs

ic_testkit/artifacts/
cache_fs.rs
