Skip to main content

cachekit/backend/
file.rs

1//! Local filesystem backend, byte-compatible with cachekit-py's File backend.
2//!
3//! On-disk layout (shared cross-SDK contract — a py and an rs process pointed
4//! at the same directory read each other's entries):
5//!
6//! - **Filename**: lowercase-hex Blake2b-128 (16-byte digest) of the UTF-8
7//!   cache key — filesystem-safe, no directory traversal.
8//! - **Header** (14 bytes): `b"CK"` magic `[0:2]`, format version `1` `[2:3]`,
9//!   reserved `0` `[3:4]`, flags `u16` BE `[4:6]`, expiry unix-seconds `u64`
10//!   BE `[6:14]` (`0` = never expires). Payload follows the header.
11//! - **Writes** are atomic: payload lands in a unique `{name}.tmp.{pid}.…`
12//!   file, fsynced, then renamed over the final path. Readers see the old or
13//!   the new entry, never a torn one.
14//! - **Expiry is lazy**: expired and corrupt entries are unlinked when a
15//!   read touches them, exactly like cachekit-py.
16//!
17//! ## Concurrency model (py parity)
18//!
19//! - **In-process**: every operation serializes on a backend-wide mutex —
20//!   the equivalent of cachekit-py's `threading.RLock`. This closes the
21//!   expire-unlink vs. concurrent-set lost-write race within a process.
22//!   (ponytail: one backend-wide lock, matching py; shard per-key if
23//!   contention ever matters.)
24//! - **Cross-process** (unix): advisory `flock` — shared on reads, exclusive
25//!   on the in-place expiry rewrite — matching py's `fcntl.flock` usage,
26//!   non-blocking with contention surfacing as a timeout error like py.
27//!   Unlinks are additionally **inode-validated**: an entry is only removed
28//!   if the path still names the inode the decision was made on. Combined
29//!   with the in-process lock this fully closes the stale-unlink race within
30//!   a process; cross-process it narrows the exposure to the stat-to-unlink
31//!   instruction window (a rename landing exactly there loses one fresh
32//!   entry — a cache miss, never corruption — the same class py accepts).
33//! - Remaining accepted exposure, identical to py: `refresh_ttl` racing a
34//!   concurrent replace in *another process* returns `false` (refresh what
35//!   was opened is no longer the live entry), and an in-place expiry rewrite
36//!   torn by power loss yields a wrong expiry, never a corrupt payload.
37//!
38//! Deliberate v1 divergences from cachekit-py (documented, not silent):
39//! LRU eviction / size caps / entry-count caps and the mmap zero-copy read
40//! path are not implemented — the directory grows until entries expire or
41//! the caller clears it.
42//!
43//! Requires a tokio runtime: file I/O runs on the blocking thread pool via
44//! `spawn_blocking` (except under the `unsync` feature, where it runs inline
45//! on the caller — the sync-in-async trade-off cachekit-py documents).
46
47use std::collections::HashMap;
48use std::fs;
49use std::io::{Read, Seek, SeekFrom, Write};
50use std::path::{Path, PathBuf};
51use std::sync::atomic::{AtomicU64, Ordering};
52use std::sync::Arc;
53use std::time::{Duration, SystemTime, UNIX_EPOCH};
54
55use async_trait::async_trait;
56
57use crate::backend::{run_blocking, Backend, HealthStatus, TtlInspectable};
58use crate::error::{BackendError, BackendErrorKind};
59
60// Header layout — byte-identical to cachekit-py's `backends/file/backend.py`.
61const MAGIC: &[u8; 2] = b"CK";
62const FORMAT_VERSION: u8 = 1;
63const HEADER_SIZE: usize = 14;
64const EXPIRY_OFFSET: u64 = 6;
65
66/// TTL ceiling (10 years), matching cachekit-py's `MAX_TTL_SECONDS` overflow guard.
67const MAX_TTL_SECS: u64 = 10 * 365 * 24 * 60 * 60;
68
69/// Orphaned temp files older than this are swept at build time (py parity).
70const TEMP_FILE_MAX_AGE: Duration = Duration::from_secs(60);
71
72/// Process-global sequence folded into temp-file names so two writes of the
73/// same key in the same nanosecond tick can never collide on the temp path.
74static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
75
76// ── Header encode/decode ─────────────────────────────────────────────────────
77
78fn now_secs() -> u64 {
79    // A pre-epoch clock degrades to "now = 0": nothing reads as expired.
80    SystemTime::now()
81        .duration_since(UNIX_EPOCH)
82        .map(|d| d.as_secs())
83        .unwrap_or(0)
84}
85
86fn build_header(expiry_unix_secs: u64) -> [u8; HEADER_SIZE] {
87    let mut h = [0u8; HEADER_SIZE];
88    h[0..2].copy_from_slice(MAGIC);
89    h[2] = FORMAT_VERSION;
90    // [3] reserved and [4..6] flags stay zero (py writes the same).
91    h[6..14].copy_from_slice(&expiry_unix_secs.to_be_bytes());
92    h
93}
94
95enum ParsedHeader {
96    /// Short file, wrong magic, or unknown version — unlink on sight.
97    Corrupt,
98    /// Past its expiry timestamp — unlink on sight.
99    Expired,
100    /// Readable entry; `expiry` is 0 for never-expires.
101    Live { expiry: u64 },
102}
103
104fn parse_header(bytes: &[u8]) -> ParsedHeader {
105    if bytes.len() < HEADER_SIZE || &bytes[0..2] != MAGIC || bytes[2] != FORMAT_VERSION {
106        return ParsedHeader::Corrupt;
107    }
108    let mut expiry_be = [0u8; 8];
109    expiry_be.copy_from_slice(&bytes[6..14]);
110    let expiry = u64::from_be_bytes(expiry_be);
111    if expiry > 0 && now_secs() > expiry {
112        return ParsedHeader::Expired;
113    }
114    ParsedHeader::Live { expiry }
115}
116
117/// Read exactly the 14-byte header (or as much as the file holds).
118fn fill_header(file: &mut fs::File) -> std::io::Result<([u8; HEADER_SIZE], usize)> {
119    let mut header = [0u8; HEADER_SIZE];
120    let mut filled = 0;
121    while filled < HEADER_SIZE {
122        match file.read(&mut header[filled..]) {
123            Ok(0) => break, // short file → parses as Corrupt
124            Ok(n) => filled += n,
125            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
126            Err(e) => return Err(e),
127        }
128    }
129    Ok((header, filled))
130}
131
132// ── Error mapping ─────────────────────────────────────────────────────────────
133
134/// Mirror of cachekit-py's `_classify_os_error` retry semantics.
135fn classify_io(e: &std::io::Error) -> BackendErrorKind {
136    use std::io::ErrorKind as K;
137    match e.kind() {
138        // Disk full may clear up; a busy/locked file may too.
139        K::StorageFull => BackendErrorKind::Transient,
140        K::PermissionDenied => BackendErrorKind::Transient,
141        K::ReadOnlyFilesystem => BackendErrorKind::Permanent,
142        K::TimedOut => BackendErrorKind::Timeout,
143        _ if is_symlink_loop(e) => BackendErrorKind::Permanent,
144        _ => BackendErrorKind::Transient,
145    }
146}
147
148/// ELOOP — `O_NOFOLLOW` refused a symlink. `io::ErrorKind::FilesystemLoop`
149/// is still unstable (io_error_more), so match the raw errno.
150fn is_symlink_loop(e: &std::io::Error) -> bool {
151    #[cfg(unix)]
152    {
153        e.raw_os_error() == Some(libc::ELOOP)
154    }
155    #[cfg(not(unix))]
156    {
157        let _ = e;
158        false
159    }
160}
161
162fn file_err(e: std::io::Error, what: &str) -> BackendError {
163    BackendError {
164        kind: classify_io(&e),
165        message: format!("{what}: {e}"),
166        source: Some(Box::new(e)),
167    }
168}
169
170/// `true` when an open failed because the entry is absent — or is a symlink
171/// rejected by `O_NOFOLLOW`, which py also treats as "not found" rather than
172/// following it (symlink-swap defence in shared tmp directories).
173fn is_missing(e: &std::io::Error) -> bool {
174    e.kind() == std::io::ErrorKind::NotFound || is_symlink_loop(e)
175}
176
177// ── Open / lock / unlink helpers ─────────────────────────────────────────────
178
179fn open_opts(read: bool, write: bool) -> fs::OpenOptions {
180    let mut opts = fs::OpenOptions::new();
181    opts.read(read).write(write);
182    #[cfg(unix)]
183    {
184        use std::os::unix::fs::OpenOptionsExt;
185        // Refuse to follow a symlink planted at the entry path (py parity).
186        opts.custom_flags(libc::O_NOFOLLOW);
187    }
188    opts
189}
190
191/// Advisory cross-process lock, matching py's non-blocking `fcntl.flock`:
192/// shared for reads, exclusive for the in-place expiry rewrite. Contention
193/// surfaces as a timeout error (py raises its TIMEOUT type). The lock is
194/// released when the file handle drops. No-op off unix (py falls back to
195/// msvcrt there; rs has no Windows locking yet — in-process serialization
196/// still applies).
197fn flock_nb(file: &fs::File, exclusive: bool) -> Result<(), BackendError> {
198    #[cfg(unix)]
199    {
200        use rustix::fs::{flock, FlockOperation};
201        let op = if exclusive {
202            FlockOperation::NonBlockingLockExclusive
203        } else {
204            FlockOperation::NonBlockingLockShared
205        };
206        flock(file, op).map_err(|errno| {
207            if errno == rustix::io::Errno::WOULDBLOCK || errno == rustix::io::Errno::AGAIN {
208                BackendError::timeout("file lock contended (held by another process)")
209            } else {
210                file_err(std::io::Error::from(errno), "failed to lock cache file")
211            }
212        })
213    }
214    #[cfg(not(unix))]
215    {
216        let _ = (file, exclusive);
217        Ok(())
218    }
219}
220
221fn unlink_quiet(path: &Path) {
222    let _ = fs::remove_file(path);
223}
224
225/// Remove `path` only if it still names the inode `opened` was read from.
226///
227/// A stale-decision unlink (entry looked expired/corrupt on the fd we hold)
228/// must never delete a *fresh* entry a concurrent writer just renamed over
229/// the path — validate device+inode first. Off unix there is no inode API;
230/// the in-process lock still serializes same-process races and we fall back
231/// to a plain unlink (matching py's Windows behaviour).
232fn unlink_if_same_inode(path: &Path, opened: &fs::File) {
233    if still_linked(path, opened) {
234        unlink_quiet(path);
235    }
236}
237
238/// `true` if `path` still names the inode `opened` was read from (unix).
239/// Off unix, assume yes (in-process lock covers same-process races).
240fn still_linked(path: &Path, opened: &fs::File) -> bool {
241    #[cfg(unix)]
242    {
243        use std::os::unix::fs::MetadataExt;
244        let (Ok(on_disk), Ok(held)) = (fs::symlink_metadata(path), opened.metadata()) else {
245            return false;
246        };
247        on_disk.dev() == held.dev() && on_disk.ino() == held.ino()
248    }
249    #[cfg(not(unix))]
250    {
251        let _ = (path, opened);
252        true
253    }
254}
255
256// ── Sync cores (run on the blocking pool, under the backend-wide lock) ───────
257
258fn read_entry(path: &Path) -> Result<Option<Vec<u8>>, BackendError> {
259    let mut file = match open_opts(true, false).open(path) {
260        Ok(f) => f,
261        Err(e) if is_missing(&e) => return Ok(None),
262        Err(e) => return Err(file_err(e, "failed to open cache file")),
263    };
264    flock_nb(&file, false)?;
265    let mut data = Vec::new();
266    file.read_to_end(&mut data)
267        .map_err(|e| file_err(e, "failed to read cache file"))?;
268
269    match parse_header(&data) {
270        ParsedHeader::Corrupt | ParsedHeader::Expired => {
271            unlink_if_same_inode(path, &file);
272            Ok(None)
273        }
274        ParsedHeader::Live { .. } => Ok(Some(data.split_off(HEADER_SIZE))),
275    }
276}
277
278/// Header-only probe used by `exists` and `ttl`. Corrupt/expired entries are
279/// unlinked (inode-validated) and reported as absent, mirroring py.
280fn probe_header(path: &Path) -> Result<Option<u64>, BackendError> {
281    let mut file = match open_opts(true, false).open(path) {
282        Ok(f) => f,
283        Err(e) if is_missing(&e) => return Ok(None),
284        Err(e) => return Err(file_err(e, "failed to open cache file")),
285    };
286    flock_nb(&file, false)?;
287    let (header, filled) =
288        fill_header(&mut file).map_err(|e| file_err(e, "failed to read cache file header"))?;
289    match parse_header(&header[..filled]) {
290        ParsedHeader::Corrupt | ParsedHeader::Expired => {
291            unlink_if_same_inode(path, &file);
292            Ok(None)
293        }
294        ParsedHeader::Live { expiry } => Ok(Some(expiry)),
295    }
296}
297
298fn write_entry(path: &Path, header: [u8; HEADER_SIZE], value: &[u8]) -> Result<(), BackendError> {
299    let temp = temp_path(path);
300
301    // Creation failure means we own nothing on disk — never unlink here, or
302    // an EEXIST loser would delete another writer's in-flight temp file.
303    let mut opts = open_opts(false, true);
304    opts.create_new(true); // O_CREAT | O_EXCL — never reuse a planted file
305    #[cfg(unix)]
306    {
307        use std::os::unix::fs::OpenOptionsExt;
308        opts.mode(0o600); // owner-only, py parity
309    }
310    let mut file = opts
311        .open(&temp)
312        .map_err(|e| file_err(e, "failed to create cache temp file"))?;
313
314    // From here the temp file is ours: clean it up on any failure.
315    let write_all = |file: &mut fs::File| -> std::io::Result<()> {
316        file.write_all(&header)?;
317        file.write_all(value)?;
318        // Data must be durable before the rename makes it visible.
319        file.sync_all()
320    };
321    let result = write_all(&mut file).and_then(|()| {
322        drop(file);
323        fs::rename(&temp, path)
324    });
325    result.map_err(|e| {
326        unlink_quiet(&temp);
327        file_err(e, "failed to write cache file")
328    })
329}
330
331fn delete_entry(path: &Path) -> Result<bool, BackendError> {
332    match fs::remove_file(path) {
333        Ok(()) => Ok(true),
334        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
335        Err(e) => Err(file_err(e, "failed to delete cache file")),
336    }
337}
338
339fn entry_ttl(path: &Path) -> Result<Option<Duration>, BackendError> {
340    match probe_header(path)? {
341        // Missing/expired/corrupt — or 0 = never expires: exists but has no
342        // TTL to report (py parity).
343        None | Some(0) => Ok(None),
344        Some(expiry) => Ok(Some(Duration::from_secs(expiry.saturating_sub(now_secs())))),
345    }
346}
347
348fn rewrite_expiry(path: &Path, new_expiry: u64) -> Result<bool, BackendError> {
349    let mut file = match open_opts(true, true).open(path) {
350        Ok(f) => f,
351        Err(e) if is_missing(&e) => return Ok(false),
352        Err(e) => return Err(file_err(e, "failed to open cache file")),
353    };
354    flock_nb(&file, true)?;
355
356    let (header, filled) =
357        fill_header(&mut file).map_err(|e| file_err(e, "failed to read cache file header"))?;
358    match parse_header(&header[..filled]) {
359        ParsedHeader::Corrupt | ParsedHeader::Expired => {
360            // An expired entry is absent, not refreshable (py parity).
361            unlink_if_same_inode(path, &file);
362            return Ok(false);
363        }
364        ParsedHeader::Live { .. } => {}
365    }
366
367    // If a concurrent writer replaced the entry since we opened it, our fd
368    // points at an orphaned inode — writing there would report success while
369    // the live entry keeps its old expiry (silent early eviction). Refuse:
370    // `false` honestly says "what you opened is no longer the live entry".
371    if !still_linked(path, &file) {
372        return Ok(false);
373    }
374
375    // Overwrite ONLY the 8-byte expiry field. Same trade-off py documents: a
376    // torn write on power loss yields a wrong expiry, never a corrupt
377    // payload — magic, version, and data are untouched.
378    let do_write = |f: &mut fs::File| -> std::io::Result<()> {
379        f.seek(SeekFrom::Start(EXPIRY_OFFSET))?;
380        f.write_all(&new_expiry.to_be_bytes())?;
381        f.sync_all()
382    };
383    do_write(&mut file).map_err(|e| file_err(e, "failed to refresh cache file TTL"))?;
384    Ok(true)
385}
386
387fn temp_path(target: &Path) -> PathBuf {
388    // `{name}.tmp.{pid}.{nanos}.{seq}` — contains the `.tmp.` marker py's
389    // startup sweep globs for, so a shared directory gets cross-SDK orphan
390    // cleanup. The process-global sequence makes the name collision-proof
391    // even for same-key writes landing in the same nanosecond tick.
392    let nanos = SystemTime::now()
393        .duration_since(UNIX_EPOCH)
394        .map(|d| d.as_nanos())
395        .unwrap_or(0);
396    let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
397    let name = target
398        .file_name()
399        .map(|n| n.to_string_lossy().into_owned())
400        .unwrap_or_default();
401    target.with_file_name(format!("{name}.tmp.{}.{nanos}.{seq}", std::process::id()))
402}
403
404/// Best-effort sweep of orphaned temp files older than [`TEMP_FILE_MAX_AGE`].
405fn cleanup_temp_files(cache_dir: &Path) {
406    let Ok(entries) = fs::read_dir(cache_dir) else {
407        return;
408    };
409    for entry in entries.flatten() {
410        let name = entry.file_name();
411        if !name.to_string_lossy().contains(".tmp.") {
412            continue;
413        }
414        // symlink_metadata never follows links — a planted symlink is skipped.
415        let Ok(meta) = entry.path().symlink_metadata() else {
416            continue;
417        };
418        if !meta.is_file() {
419            continue;
420        }
421        let stale = meta
422            .modified()
423            .ok()
424            .and_then(|m| m.elapsed().ok())
425            .is_some_and(|age| age > TEMP_FILE_MAX_AGE);
426        if stale {
427            unlink_quiet(&entry.path());
428        }
429    }
430}
431
432// ── FileBackend ───────────────────────────────────────────────────────────────
433
434/// Local filesystem cache backend.
435///
436/// See the [module docs](self) for the on-disk format, the concurrency
437/// model, and the deliberate v1 divergences from cachekit-py.
438#[derive(Debug, Clone)]
439pub struct FileBackend {
440    cache_dir: PathBuf,
441    /// Backend-wide op serialization — py `RLock` parity. Clones share it.
442    /// Async so contended waiters cost a queued future, not a parked
443    /// blocking-pool thread (a hot backend must not monopolize the pool).
444    lock: Arc<tokio::sync::Mutex<()>>,
445}
446
447impl FileBackend {
448    /// Start building a [`FileBackend`].
449    pub fn builder() -> FileBackendBuilder {
450        FileBackendBuilder::default()
451    }
452
453    /// The resolved directory this backend stores entries in.
454    #[must_use]
455    pub fn cache_dir(&self) -> &Path {
456        &self.cache_dir
457    }
458
459    /// Blake2b-128 hex of the key — the same digest `tracing` events carry as
460    /// `key_hash`, so a log line names the file it touched.
461    fn entry_path(&self, key: &str) -> PathBuf {
462        self.cache_dir.join(crate::metrics::key_hash(key))
463    }
464
465    /// Run `f` on the blocking pool holding the backend-wide lock.
466    ///
467    /// The lock is acquired in async context BEFORE the hop to the blocking
468    /// pool: N contended ops queue as futures instead of parking N blocking
469    /// threads on a mutex (which would starve every other `spawn_blocking`
470    /// user in the process). The owned guard rides into the closure and
471    /// releases when the I/O finishes.
472    async fn locked<T: Send + 'static>(
473        &self,
474        f: impl FnOnce() -> Result<T, BackendError> + Send + 'static,
475    ) -> Result<T, BackendError> {
476        let guard = Arc::clone(&self.lock).lock_owned().await;
477        run_blocking(move || {
478            let _guard = guard;
479            f()
480        })
481        .await
482    }
483}
484
485// ── Backend impl ──────────────────────────────────────────────────────────────
486
487#[cfg(not(target_arch = "wasm32"))]
488#[cfg_attr(not(feature = "unsync"), async_trait)]
489#[cfg_attr(feature = "unsync", async_trait(?Send))]
490impl Backend for FileBackend {
491    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
492        let path = self.entry_path(key);
493        self.locked(move || read_entry(&path)).await
494    }
495
496    async fn set(
497        &self,
498        key: &str,
499        value: Vec<u8>,
500        ttl: Option<Duration>,
501    ) -> Result<(), BackendError> {
502        // None = never expires (expiry field 0). Sub-second TTLs round up to
503        // 1s (Redis-backend parity); absurd TTLs are rejected like py.
504        let expiry = match ttl {
505            None => 0,
506            Some(d) => {
507                let secs = d.as_secs().max(1);
508                if secs > MAX_TTL_SECS {
509                    return Err(BackendError::permanent(format!(
510                        "TTL {secs}s out of range [1, {MAX_TTL_SECS}] (max 10 years)"
511                    )));
512                }
513                now_secs().saturating_add(secs)
514            }
515        };
516
517        let path = self.entry_path(key);
518        self.locked(move || write_entry(&path, build_header(expiry), &value))
519            .await
520    }
521
522    async fn delete(&self, key: &str) -> Result<bool, BackendError> {
523        let path = self.entry_path(key);
524        self.locked(move || delete_entry(&path)).await
525    }
526
527    async fn exists(&self, key: &str) -> Result<bool, BackendError> {
528        let path = self.entry_path(key);
529        self.locked(move || Ok(probe_header(&path)?.is_some()))
530            .await
531    }
532
533    async fn health(&self) -> Result<HealthStatus, BackendError> {
534        let start = std::time::Instant::now();
535
536        // Write/read/delete round-trip — same probe as cachekit-py.
537        let key = "__health_check__";
538        let probe = b"health_check_data".to_vec();
539        self.set(key, probe.clone(), Some(Duration::from_secs(60)))
540            .await?;
541        let read_back = self.get(key).await?;
542        self.delete(key).await?;
543
544        let round_trip_ok = read_back.as_deref() == Some(probe.as_slice());
545        let latency = start.elapsed();
546
547        let mut details = HashMap::new();
548        details.insert(
549            "cache_dir".to_string(),
550            self.cache_dir.display().to_string(),
551        );
552        if !round_trip_ok {
553            details.insert(
554                "error".to_string(),
555                "round-trip verification failed".to_string(),
556            );
557        }
558        Ok(HealthStatus {
559            is_healthy: round_trip_ok,
560            latency_ms: latency.as_secs_f64() * 1000.0,
561            backend_type: "file".to_string(),
562            details,
563        })
564    }
565}
566
567// ── TtlInspectable impl ───────────────────────────────────────────────────────
568
569#[cfg(not(target_arch = "wasm32"))]
570#[cfg_attr(not(feature = "unsync"), async_trait)]
571#[cfg_attr(feature = "unsync", async_trait(?Send))]
572impl TtlInspectable for FileBackend {
573    async fn ttl(&self, key: &str) -> Result<Option<Duration>, BackendError> {
574        let path = self.entry_path(key);
575        self.locked(move || entry_ttl(&path)).await
576    }
577
578    async fn refresh_ttl(&self, key: &str, ttl: Duration) -> Result<bool, BackendError> {
579        let secs = ttl.as_secs();
580        // Duration::ZERO is rejected, matching the CachekitIO backend — in
581        // this SDK "no expiry" is spelled `set(.., None)`, not a zero TTL
582        // (py's `refresh_ttl(key, 0)` making an entry permanent is py's
583        // spelling of the same contract).
584        if secs == 0 {
585            return Err(BackendError::permanent(
586                "refresh_ttl requires at least 1 second".to_string(),
587            ));
588        }
589        if secs > MAX_TTL_SECS {
590            return Err(BackendError::permanent(format!(
591                "TTL {secs}s out of range [1, {MAX_TTL_SECS}] (max 10 years)"
592            )));
593        }
594
595        let new_expiry = now_secs().saturating_add(secs);
596        let path = self.entry_path(key);
597        self.locked(move || rewrite_expiry(&path, new_expiry)).await
598    }
599}
600
601// ── Builder ───────────────────────────────────────────────────────────────────
602
603/// Builder for [`FileBackend`].
604#[derive(Default)]
605#[must_use]
606pub struct FileBackendBuilder {
607    cache_dir: Option<PathBuf>,
608}
609
610impl FileBackendBuilder {
611    /// Set the cache directory (default: `<system temp dir>/cachekit`, the
612    /// same default as cachekit-py).
613    pub fn cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
614        self.cache_dir = Some(dir.into());
615        self
616    }
617
618    /// Create the cache directory if needed and construct a [`FileBackend`].
619    ///
620    /// On unix, a **pre-existing** directory must be owned by the current
621    /// user and not group/other-writable — a default like `/tmp/cachekit`
622    /// pre-created by another local user would otherwise let them poison
623    /// plaintext cache entries (CWE-377). Also sweeps orphaned temp files
624    /// left by crashed writers (py parity). The one-time directory setup
625    /// does its I/O inline; per-operation I/O runs on the blocking pool.
626    ///
627    /// # Errors
628    ///
629    /// Returns a config error if the directory cannot be created or
630    /// resolved, is not owned by the current user, or is group/other-writable.
631    pub fn build(self) -> Result<FileBackend, crate::error::CachekitError> {
632        use crate::error::CachekitError;
633
634        let dir = self
635            .cache_dir
636            .unwrap_or_else(|| std::env::temp_dir().join("cachekit"));
637
638        let create = || -> std::io::Result<()> {
639            let mut builder = fs::DirBuilder::new();
640            builder.recursive(true);
641            #[cfg(unix)]
642            {
643                use std::os::unix::fs::DirBuilderExt;
644                builder.mode(0o700); // owner-only, py parity
645            }
646            builder.create(&dir)
647        };
648        create().map_err(|e| {
649            CachekitError::Config(format!(
650                "failed to create cache directory {}: {e}",
651                dir.display()
652            ))
653        })?;
654
655        // Resolve symlinks once up front (py resolves via realpath per key).
656        let cache_dir = fs::canonicalize(&dir).map_err(|e| {
657            CachekitError::Config(format!(
658                "failed to resolve cache directory {}: {e}",
659                dir.display()
660            ))
661        })?;
662
663        // Ownership/mode gate for pre-existing directories (unix).
664        #[cfg(unix)]
665        {
666            use std::os::unix::fs::MetadataExt;
667            let meta = fs::metadata(&cache_dir).map_err(|e| {
668                CachekitError::Config(format!(
669                    "failed to stat cache directory {}: {e}",
670                    cache_dir.display()
671                ))
672            })?;
673            let euid = rustix::process::geteuid().as_raw();
674            if meta.uid() != euid {
675                return Err(CachekitError::Config(format!(
676                    "cache directory {} is owned by uid {} (expected {euid}) — refusing a \
677                     directory another local user controls",
678                    cache_dir.display(),
679                    meta.uid()
680                )));
681            }
682            if meta.mode() & 0o022 != 0 {
683                return Err(CachekitError::Config(format!(
684                    "cache directory {} is group/other-writable (mode {:o}) — another local \
685                     user could plant or replace cache entries; chmod it to 0700",
686                    cache_dir.display(),
687                    meta.mode() & 0o777
688                )));
689            }
690        }
691
692        cleanup_temp_files(&cache_dir);
693
694        Ok(FileBackend {
695            cache_dir,
696            lock: Arc::new(tokio::sync::Mutex::new(())),
697        })
698    }
699}
700
701#[cfg(test)]
702#[allow(clippy::expect_used, clippy::unwrap_used)] // test-only: failures should panic loudly
703mod tests {
704    use super::*;
705
706    /// Compile-time proof that FileBackend implements TtlInspectable.
707    fn _assert_ttl_inspectable(_b: &dyn TtlInspectable) {}
708
709    #[test]
710    fn file_is_ttl_inspectable() {
711        fn _check(backend: &FileBackend) {
712            _assert_ttl_inspectable(backend);
713        }
714    }
715
716    fn backend_at(dir: &Path) -> FileBackend {
717        FileBackend {
718            cache_dir: dir.to_path_buf(),
719            lock: Arc::new(tokio::sync::Mutex::new(())),
720        }
721    }
722
723    #[test]
724    fn filename_hash_matches_py_blake2b16_hex() {
725        // cachekit-py: hashlib.blake2b(b"key", digest_size=16).hexdigest()
726        let backend = backend_at(Path::new("/cache"));
727        let path = backend.entry_path("ns:app:func:m.f:args:abc:v1");
728        assert_eq!(
729            path,
730            PathBuf::from("/cache/bcb35ae6f64fa65b2770ab3af631b1ce")
731        );
732    }
733
734    #[test]
735    fn header_layout_is_py_byte_compatible() {
736        let header = build_header(0x0102_0304_0506_0708);
737        assert_eq!(&header[0..2], b"CK");
738        assert_eq!(header[2], 1); // version
739        assert_eq!(header[3], 0); // reserved
740        assert_eq!(&header[4..6], &[0, 0]); // flags u16 BE
741        assert_eq!(&header[6..14], &[1, 2, 3, 4, 5, 6, 7, 8]); // expiry u64 BE
742    }
743
744    // ── On-disk format: vectors generated by cachekit-py's header builder ────
745    // (b"CK" + version + reserved + struct.pack(">H", 0) + struct.pack(">Q", expiry))
746
747    fn write_raw(dir: &Path, name: &str, hex_bytes: &str) -> PathBuf {
748        let path = dir.join(name);
749        fs::write(&path, hex::decode(hex_bytes).unwrap()).unwrap();
750        path
751    }
752
753    #[test]
754    fn py_written_never_expires_entry_reads_back() {
755        let dir = tempfile::tempdir().expect("tempdir");
756        let path = write_raw(
757            dir.path(),
758            "entry",
759            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
760        );
761        let payload = read_entry(&path).expect("read should not error");
762        assert_eq!(payload.as_deref(), Some(b"hello from python".as_slice()));
763        assert!(path.exists(), "live entry must not be unlinked");
764    }
765
766    #[test]
767    fn py_written_future_expiry_entry_reads_back() {
768        let dir = tempfile::tempdir().expect("tempdir");
769        // expiry 4102444800 = 2100-01-01
770        let path = write_raw(
771            dir.path(),
772            "entry",
773            "434b0100000000000000f48657006672657368",
774        );
775        let payload = read_entry(&path).expect("read should not error");
776        assert_eq!(payload.as_deref(), Some(b"fresh".as_slice()));
777
778        let remaining = entry_ttl(&path)
779            .expect("ttl should not error")
780            .expect("future expiry must report a TTL");
781        assert!(remaining > Duration::from_secs(3600), "TTL: {remaining:?}");
782    }
783
784    #[test]
785    fn py_written_expired_entry_is_unlinked_on_read() {
786        let dir = tempfile::tempdir().expect("tempdir");
787        // expiry 1 = 1970-01-01T00:00:01Z
788        let path = write_raw(
789            dir.path(),
790            "entry",
791            "434b0100000000000000000000017374616c65",
792        );
793        assert_eq!(read_entry(&path).expect("read should not error"), None);
794        assert!(!path.exists(), "expired entry must be unlinked on read");
795    }
796
797    #[test]
798    fn corrupt_and_short_entries_are_unlinked_on_read() {
799        let dir = tempfile::tempdir().expect("tempdir");
800
801        // Wrong magic ("XX"), otherwise well-formed.
802        let bad_magic = write_raw(dir.path(), "bad-magic", "585801000000000000000000000061");
803        assert_eq!(read_entry(&bad_magic).expect("no error"), None);
804        assert!(!bad_magic.exists());
805
806        // Unknown format version (2).
807        let bad_version = write_raw(dir.path(), "bad-version", "434b02000000000000000000000061");
808        assert_eq!(read_entry(&bad_version).expect("no error"), None);
809        assert!(!bad_version.exists());
810
811        // Shorter than the 14-byte header.
812        let short = write_raw(dir.path(), "short", "434b01");
813        assert_eq!(read_entry(&short).expect("no error"), None);
814        assert!(!short.exists());
815    }
816
817    #[test]
818    fn ttl_semantics_match_py() {
819        let dir = tempfile::tempdir().expect("tempdir");
820
821        // Never-expires: exists, but reports no TTL.
822        let never = write_raw(
823            dir.path(),
824            "never",
825            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
826        );
827        assert_eq!(entry_ttl(&never).expect("no error"), None);
828        assert!(never.exists());
829
830        // Expired: absent, unlinked.
831        let expired = write_raw(
832            dir.path(),
833            "expired",
834            "434b0100000000000000000000017374616c65",
835        );
836        assert_eq!(entry_ttl(&expired).expect("no error"), None);
837        assert!(!expired.exists());
838
839        // Missing: absent, no error.
840        assert_eq!(entry_ttl(&dir.path().join("nope")).expect("no error"), None);
841    }
842
843    #[test]
844    fn rewrite_expiry_semantics_match_py() {
845        let dir = tempfile::tempdir().expect("tempdir");
846
847        // Live entry: refreshed in place — payload untouched, expiry moved.
848        let live = write_raw(dir.path(), "live", "434b0100000000000000f48657006672657368");
849        let new_expiry = now_secs() + 42;
850        assert!(rewrite_expiry(&live, new_expiry).expect("no error"));
851        let remaining = entry_ttl(&live).expect("no error").expect("has TTL");
852        assert!(remaining <= Duration::from_secs(42));
853        assert_eq!(
854            read_entry(&live).expect("no error").as_deref(),
855            Some(b"fresh".as_slice()),
856            "payload must survive an in-place expiry rewrite"
857        );
858
859        // Expired entry: treated as absent and unlinked.
860        let expired = write_raw(
861            dir.path(),
862            "expired",
863            "434b0100000000000000000000017374616c65",
864        );
865        assert!(!rewrite_expiry(&expired, new_expiry).expect("no error"));
866        assert!(!expired.exists());
867
868        // Missing entry: false, no error.
869        assert!(!rewrite_expiry(&dir.path().join("nope"), new_expiry).expect("no error"));
870    }
871
872    #[test]
873    fn temp_file_sweep_removes_stale_keeps_fresh() {
874        let dir = tempfile::tempdir().expect("tempdir");
875
876        let stale = dir.path().join("abc.tmp.123.456");
877        fs::write(&stale, b"orphan").expect("write");
878        let two_minutes_ago = SystemTime::now() - Duration::from_secs(120);
879        fs::File::options()
880            .write(true)
881            .open(&stale)
882            .expect("open")
883            .set_modified(two_minutes_ago)
884            .expect("set mtime");
885
886        let fresh = dir.path().join("def.tmp.123.789");
887        fs::write(&fresh, b"in flight").expect("write");
888
889        let entry = dir.path().join("0123abcd");
890        fs::write(&entry, b"not a temp file").expect("write");
891
892        cleanup_temp_files(dir.path());
893
894        assert!(!stale.exists(), "stale temp file must be swept");
895        assert!(fresh.exists(), "fresh temp file must be kept");
896        assert!(entry.exists(), "cache entries must never be swept");
897    }
898
899    // ── Hardening ────────────────────────────────────────────────────────────
900
901    #[cfg(unix)]
902    #[test]
903    fn builder_rejects_group_or_other_writable_directory() {
904        use std::os::unix::fs::PermissionsExt;
905
906        // CWE-377: a pre-existing world/group-writable cache dir (e.g. a
907        // planted /tmp/cachekit) lets another local user poison plaintext
908        // entries — build() must refuse it.
909        let dir = tempfile::tempdir().expect("tempdir");
910        let target = dir.path().join("shared");
911        fs::create_dir(&target).expect("mkdir");
912        fs::set_permissions(&target, fs::Permissions::from_mode(0o770)).expect("chmod");
913
914        let err = FileBackend::builder()
915            .cache_dir(&target)
916            .build()
917            .expect_err("group-writable dir must be rejected");
918        assert!(err.to_string().contains("writable"), "{err}");
919
920        // Owner-only passes.
921        fs::set_permissions(&target, fs::Permissions::from_mode(0o700)).expect("chmod");
922        FileBackend::builder()
923            .cache_dir(&target)
924            .build()
925            .expect("0700 dir must be accepted");
926    }
927
928    #[test]
929    fn temp_paths_never_collide() {
930        let target = Path::new("/cache/abc");
931        let a = temp_path(target);
932        let b = temp_path(target);
933        assert_ne!(a, b, "same-key temp paths must be unique");
934        assert!(a.to_string_lossy().contains(".tmp."), "py sweep pattern");
935    }
936
937    #[cfg(unix)]
938    #[test]
939    fn stale_unlink_decision_spares_replaced_entry() {
940        // Stale-unlink race: a get() that decided "expired" must not delete
941        // the fresh entry a concurrent set() renamed over the path.
942        let dir = tempfile::tempdir().expect("tempdir");
943        let path = write_raw(
944            dir.path(),
945            "entry",
946            "434b0100000000000000000000017374616c65", // expired
947        );
948
949        // Open the expired inode (the stale decision's fd)…
950        let stale_fd = open_opts(true, false).open(&path).expect("open");
951
952        // …then a "concurrent writer" replaces the entry (new inode).
953        write_entry(&path, build_header(0), b"fresh").expect("replace");
954
955        // The stale unlink must be refused: path now names a different inode.
956        unlink_if_same_inode(&path, &stale_fd);
957        assert_eq!(
958            read_entry(&path).expect("no error").as_deref(),
959            Some(b"fresh".as_slice()),
960            "inode-validated unlink must spare the replaced entry"
961        );
962
963        // Same guard for refresh: the stale fd is no longer the live entry.
964        assert!(!still_linked(&path, &stale_fd));
965        let live_fd = open_opts(true, false).open(&path).expect("open");
966        assert!(still_linked(&path, &live_fd));
967    }
968
969    #[cfg(unix)]
970    #[test]
971    fn flock_contention_surfaces_as_timeout() {
972        use rustix::fs::{flock, FlockOperation};
973
974        let dir = tempfile::tempdir().expect("tempdir");
975        let path = write_raw(
976            dir.path(),
977            "entry",
978            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
979        );
980
981        // Another "process" holds an exclusive lock…
982        let holder = fs::File::open(&path).expect("open");
983        flock(&holder, FlockOperation::NonBlockingLockExclusive).expect("lock");
984
985        // …so our shared-lock read errors as a timeout (py parity), it does
986        // not block and does not misreport the entry as missing.
987        let err = read_entry(&path).expect_err("contended read must error");
988        assert_eq!(err.kind, BackendErrorKind::Timeout, "{err}");
989    }
990
991    #[tokio::test]
992    async fn expired_read_racing_fresh_set_never_loses_the_write() {
993        // The stale-unlink race, end to end through the public API: an expired
994        // entry is read (deciding to unlink) while a fresh set replaces it.
995        // Whatever the interleaving, the fresh value must survive.
996        let dir = tempfile::tempdir().expect("tempdir");
997        // tempfile dirs follow the umask (e.g. 0775) — the builder's
998        // ownership/mode gate would rightly reject that; make it 0700.
999        #[cfg(unix)]
1000        {
1001            use std::os::unix::fs::PermissionsExt;
1002            fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).expect("chmod");
1003        }
1004        let backend = FileBackend::builder()
1005            .cache_dir(dir.path())
1006            .build()
1007            .expect("build");
1008        let key = "raced";
1009        let path = backend.entry_path(key);
1010
1011        for round in 0..50 {
1012            // Plant an already-expired entry directly (py-format bytes).
1013            fs::write(
1014                &path,
1015                hex::decode("434b0100000000000000000000017374616c65").unwrap(),
1016            )
1017            .expect("plant expired entry");
1018
1019            let reader = {
1020                let b = backend.clone();
1021                tokio::spawn(async move { b.get(key).await })
1022            };
1023            let writer = {
1024                let b = backend.clone();
1025                tokio::spawn(async move { b.set(key, b"fresh".to_vec(), None).await })
1026            };
1027            reader.await.expect("join").expect("get");
1028            writer.await.expect("join").expect("set");
1029
1030            assert_eq!(
1031                backend.get(key).await.expect("get").as_deref(),
1032                Some(b"fresh".as_slice()),
1033                "round {round}: fresh write lost to a stale expired-unlink"
1034            );
1035        }
1036    }
1037}