ic_testkit/artifacts/
cache_fs.rs

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

ic_testkit/artifacts/
cache_fs.rs
