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//! - **Orphan sweep on first write**: temp files older than 60 s, left by
17//!   crashed writers (py's or rs's — both use the `.tmp.` marker), are
18//!   removed by each backend instance's first `set`, once. cachekit-py sweeps
19//!   at startup instead; rs defers it because the sweep reads the whole
20//!   directory, and a short-lived reader of a large cache (a CLI hit) would
21//!   pay that on every run. Only writers leave orphans, so a reader-only
22//!   process never needs it.
23//!
24//! ## Concurrency model
25//!
26//! - **In-process**: operations on the same key serialize on one of
27//!   64 mutexes (`LOCK_STRIPES`), picked by the first byte of the entry's
28//!   Blake2b-128 filename. This closes the expire-unlink vs. concurrent-set
29//!   lost-write race within a process: that race only ever involves one entry
30//!   path, so one key always maps to one stripe. Keys on different stripes
31//!   do not wait on each other's fsync; two keys share a stripe about 1 time
32//!   in 64 and then still serialize (cachekit-py serializes everything on
33//!   one `threading.RLock`, because its RLock also guards LRU and size
34//!   bookkeeping; rs has none yet — porting eviction or size caps adds shared
35//!   cross-key state, and these stripes must be revisited then).
36//! - **Cross-process** (unix): advisory `flock` — shared on reads, exclusive
37//!   on the in-place expiry rewrite — matching py's `fcntl.flock` usage,
38//!   non-blocking with contention surfacing as a timeout error like py.
39//!   Unlinks are additionally **inode-validated**: an entry is only removed
40//!   if the path still names the inode the decision was made on. Combined
41//!   with the in-process lock this fully closes the stale-unlink race within
42//!   a process; cross-process it narrows the exposure to the stat-to-unlink
43//!   instruction window (a rename landing exactly there loses one fresh
44//!   entry — a cache miss, never corruption — the same class py accepts).
45//! - Remaining accepted exposure, identical to py: `refresh_ttl` racing a
46//!   concurrent replace in *another process* returns `false` (refresh what
47//!   was opened is no longer the live entry), and an in-place expiry rewrite
48//!   torn by power loss yields a wrong expiry, never a corrupt payload.
49//!
50//! Deliberate v1 divergences from cachekit-py (documented, not silent):
51//! LRU eviction / size caps / entry-count caps and the mmap zero-copy read
52//! path are not implemented — the directory grows until entries expire or
53//! the caller clears it.
54//!
55//! Requires a tokio runtime: file I/O runs on the blocking thread pool via
56//! `spawn_blocking` (except under the `unsync` feature, where it runs inline
57//! on the caller — the sync-in-async trade-off cachekit-py documents).
58
59use std::collections::HashMap;
60use std::fs;
61use std::io::{Read, Seek, SeekFrom, Write};
62use std::path::{Path, PathBuf};
63use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
64use std::sync::Arc;
65use std::time::{Duration, SystemTime, UNIX_EPOCH};
66
67use async_trait::async_trait;
68
69use crate::backend::{run_blocking, Backend, HealthStatus, TtlInspectable};
70use crate::error::{BackendError, BackendErrorKind};
71
72// Header layout — byte-identical to cachekit-py's `backends/file/backend.py`.
73const MAGIC: &[u8; 2] = b"CK";
74const FORMAT_VERSION: u8 = 1;
75const HEADER_SIZE: usize = 14;
76const EXPIRY_OFFSET: u64 = 6;
77
78/// TTL ceiling (10 years), matching cachekit-py's `MAX_TTL_SECONDS` overflow guard.
79const MAX_TTL_SECS: u64 = 10 * 365 * 24 * 60 * 60;
80
81/// Orphaned temp files older than this are swept by a backend's first `set`
82/// (py uses the same threshold, but sweeps at startup).
83const TEMP_FILE_MAX_AGE: Duration = Duration::from_secs(60);
84
85/// Process-global sequence folded into temp-file names so two writes of the
86/// same key in the same nanosecond tick can never collide on the temp path.
87static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
88
89// ── Header encode/decode ─────────────────────────────────────────────────────
90
91fn now_secs() -> u64 {
92    // A pre-epoch clock degrades to "now = 0": nothing reads as expired.
93    SystemTime::now()
94        .duration_since(UNIX_EPOCH)
95        .map(|d| d.as_secs())
96        .unwrap_or(0)
97}
98
99fn build_header(expiry_unix_secs: u64) -> [u8; HEADER_SIZE] {
100    let mut h = [0u8; HEADER_SIZE];
101    h[0..2].copy_from_slice(MAGIC);
102    h[2] = FORMAT_VERSION;
103    // [3] reserved and [4..6] flags stay zero (py writes the same).
104    h[6..14].copy_from_slice(&expiry_unix_secs.to_be_bytes());
105    h
106}
107
108enum ParsedHeader {
109    /// Short file, wrong magic, or unknown version — unlink on sight.
110    Corrupt,
111    /// Past its expiry timestamp — unlink on sight.
112    Expired,
113    /// Nonzero reserved byte or flags: a transform this reader does not
114    /// implement. A miss that MUST NOT delete, rewrite, or return the payload.
115    Unsupported,
116    /// Readable entry; `expiry` is 0 for never-expires.
117    Live { expiry: u64 },
118}
119
120/// `now` is the reader clock in Unix seconds (a parameter so tests can pin it).
121fn parse_header(bytes: &[u8], now: u64) -> ParsedHeader {
122    if bytes.len() < HEADER_SIZE || &bytes[0..2] != MAGIC || bytes[2] != FORMAT_VERSION {
123        return ParsedHeader::Corrupt;
124    }
125    if bytes[3] != 0 || bytes[4..6] != [0, 0] {
126        return ParsedHeader::Unsupported;
127    }
128    let mut expiry_be = [0u8; 8];
129    expiry_be.copy_from_slice(&bytes[6..14]);
130    let expiry = u64::from_be_bytes(expiry_be);
131    // Expired once the clock reaches the deadline (spec: `now >= expiry`).
132    if expiry > 0 && now >= expiry {
133        return ParsedHeader::Expired;
134    }
135    ParsedHeader::Live { expiry }
136}
137
138/// Read exactly the 14-byte header (or as much as the file holds).
139fn fill_header(file: &mut fs::File) -> std::io::Result<([u8; HEADER_SIZE], usize)> {
140    let mut header = [0u8; HEADER_SIZE];
141    let mut filled = 0;
142    while filled < HEADER_SIZE {
143        match file.read(&mut header[filled..]) {
144            Ok(0) => break, // short file → parses as Corrupt
145            Ok(n) => filled += n,
146            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
147            Err(e) => return Err(e),
148        }
149    }
150    Ok((header, filled))
151}
152
153// ── Error mapping ─────────────────────────────────────────────────────────────
154
155/// Mirror of cachekit-py's `_classify_os_error` retry semantics.
156fn classify_io(e: &std::io::Error) -> BackendErrorKind {
157    use std::io::ErrorKind as K;
158    match e.kind() {
159        // Disk full may clear up; a busy/locked file may too.
160        K::StorageFull => BackendErrorKind::Transient,
161        K::PermissionDenied => BackendErrorKind::Transient,
162        K::ReadOnlyFilesystem => BackendErrorKind::Permanent,
163        K::TimedOut => BackendErrorKind::Timeout,
164        _ if is_symlink_loop(e) => BackendErrorKind::Permanent,
165        _ => BackendErrorKind::Transient,
166    }
167}
168
169/// ELOOP — `O_NOFOLLOW` refused a symlink. `io::ErrorKind::FilesystemLoop`
170/// is still unstable (io_error_more), so match the raw errno.
171fn is_symlink_loop(e: &std::io::Error) -> bool {
172    #[cfg(unix)]
173    {
174        e.raw_os_error() == Some(libc::ELOOP)
175    }
176    #[cfg(not(unix))]
177    {
178        let _ = e;
179        false
180    }
181}
182
183fn file_err(e: std::io::Error, what: &str) -> BackendError {
184    BackendError {
185        kind: classify_io(&e),
186        message: format!("{what}: {e}"),
187        source: Some(Box::new(e)),
188    }
189}
190
191/// `true` when an open failed because the entry is absent — or is a symlink
192/// rejected by `O_NOFOLLOW`, which py also treats as "not found" rather than
193/// following it (symlink-swap defence in shared tmp directories).
194fn is_missing(e: &std::io::Error) -> bool {
195    e.kind() == std::io::ErrorKind::NotFound || is_symlink_loop(e)
196}
197
198// ── Open / lock / unlink helpers ─────────────────────────────────────────────
199
200fn open_opts(read: bool, write: bool) -> fs::OpenOptions {
201    let mut opts = fs::OpenOptions::new();
202    opts.read(read).write(write);
203    #[cfg(unix)]
204    {
205        use std::os::unix::fs::OpenOptionsExt;
206        // Refuse to follow a symlink planted at the entry path (py parity).
207        opts.custom_flags(libc::O_NOFOLLOW);
208    }
209    opts
210}
211
212/// Advisory cross-process lock, matching py's non-blocking `fcntl.flock`:
213/// shared for reads, exclusive for the in-place expiry rewrite. Contention
214/// surfaces as a timeout error (py raises its TIMEOUT type). The lock is
215/// released when the file handle drops. No-op off unix (py falls back to
216/// msvcrt there; rs has no Windows locking yet — in-process serialization
217/// still applies).
218fn flock_nb(file: &fs::File, exclusive: bool) -> Result<(), BackendError> {
219    #[cfg(unix)]
220    {
221        use rustix::fs::{flock, FlockOperation};
222        let op = if exclusive {
223            FlockOperation::NonBlockingLockExclusive
224        } else {
225            FlockOperation::NonBlockingLockShared
226        };
227        flock(file, op).map_err(|errno| {
228            if errno == rustix::io::Errno::WOULDBLOCK || errno == rustix::io::Errno::AGAIN {
229                BackendError::timeout("file lock contended (held by another process)")
230            } else {
231                file_err(std::io::Error::from(errno), "failed to lock cache file")
232            }
233        })
234    }
235    #[cfg(not(unix))]
236    {
237        let _ = (file, exclusive);
238        Ok(())
239    }
240}
241
242fn unlink_quiet(path: &Path) {
243    let _ = fs::remove_file(path);
244}
245
246/// Remove `path` only if it still names the inode `opened` was read from.
247///
248/// A stale-decision unlink (entry looked expired/corrupt on the fd we hold)
249/// must never delete a *fresh* entry a concurrent writer just renamed over
250/// the path — validate device+inode first. Off unix there is no inode API;
251/// the in-process lock still serializes same-process races and we fall back
252/// to a plain unlink (matching py's Windows behaviour).
253fn unlink_if_same_inode(path: &Path, opened: &fs::File) {
254    if still_linked(path, opened) {
255        unlink_quiet(path);
256    }
257}
258
259/// `true` if `path` still names the inode `opened` was read from (unix).
260/// Off unix, assume yes (in-process lock covers same-process races).
261fn still_linked(path: &Path, opened: &fs::File) -> bool {
262    #[cfg(unix)]
263    {
264        use std::os::unix::fs::MetadataExt;
265        let (Ok(on_disk), Ok(held)) = (fs::symlink_metadata(path), opened.metadata()) else {
266            return false;
267        };
268        on_disk.dev() == held.dev() && on_disk.ino() == held.ino()
269    }
270    #[cfg(not(unix))]
271    {
272        let _ = (path, opened);
273        true
274    }
275}
276
277// ── Sync cores (run on the blocking pool, under the key's stripe lock) ───────
278
279fn read_entry(path: &Path) -> Result<Option<Vec<u8>>, BackendError> {
280    let mut file = match open_opts(true, false).open(path) {
281        Ok(f) => f,
282        Err(e) if is_missing(&e) => return Ok(None),
283        Err(e) => return Err(file_err(e, "failed to open cache file")),
284    };
285    flock_nb(&file, false)?;
286    let mut data = Vec::new();
287    file.read_to_end(&mut data)
288        .map_err(|e| file_err(e, "failed to read cache file"))?;
289
290    match parse_header(&data, now_secs()) {
291        ParsedHeader::Corrupt | ParsedHeader::Expired => {
292            unlink_if_same_inode(path, &file);
293            Ok(None)
294        }
295        ParsedHeader::Unsupported => Ok(None),
296        ParsedHeader::Live { .. } => Ok(Some(data.split_off(HEADER_SIZE))),
297    }
298}
299
300/// Header-only probe used by `exists` and `ttl`. Corrupt/expired entries are
301/// unlinked (inode-validated) and reported as absent, mirroring py.
302fn probe_header(path: &Path) -> Result<Option<u64>, BackendError> {
303    let mut file = match open_opts(true, false).open(path) {
304        Ok(f) => f,
305        Err(e) if is_missing(&e) => return Ok(None),
306        Err(e) => return Err(file_err(e, "failed to open cache file")),
307    };
308    flock_nb(&file, false)?;
309    let (header, filled) =
310        fill_header(&mut file).map_err(|e| file_err(e, "failed to read cache file header"))?;
311    match parse_header(&header[..filled], now_secs()) {
312        ParsedHeader::Corrupt | ParsedHeader::Expired => {
313            unlink_if_same_inode(path, &file);
314            Ok(None)
315        }
316        ParsedHeader::Unsupported => Ok(None),
317        ParsedHeader::Live { expiry } => Ok(Some(expiry)),
318    }
319}
320
321fn write_entry(path: &Path, header: [u8; HEADER_SIZE], value: &[u8]) -> Result<(), BackendError> {
322    let temp = temp_path(path);
323
324    // Creation failure means we own nothing on disk — never unlink here, or
325    // an EEXIST loser would delete another writer's in-flight temp file.
326    let mut opts = open_opts(false, true);
327    opts.create_new(true); // O_CREAT | O_EXCL — never reuse a planted file
328    #[cfg(unix)]
329    {
330        use std::os::unix::fs::OpenOptionsExt;
331        opts.mode(0o600); // owner-only, py parity
332    }
333    let mut file = opts
334        .open(&temp)
335        .map_err(|e| file_err(e, "failed to create cache temp file"))?;
336
337    // From here the temp file is ours: clean it up on any failure.
338    let write_all = |file: &mut fs::File| -> std::io::Result<()> {
339        file.write_all(&header)?;
340        file.write_all(value)?;
341        // Data must be durable before the rename makes it visible.
342        file.sync_all()
343    };
344    let result = write_all(&mut file).and_then(|()| {
345        drop(file);
346        fs::rename(&temp, path)
347    });
348    result.map_err(|e| {
349        unlink_quiet(&temp);
350        file_err(e, "failed to write cache file")
351    })
352}
353
354fn delete_entry(path: &Path) -> Result<bool, BackendError> {
355    match fs::remove_file(path) {
356        Ok(()) => Ok(true),
357        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
358        Err(e) => Err(file_err(e, "failed to delete cache file")),
359    }
360}
361
362fn entry_ttl(path: &Path) -> Result<Option<Duration>, BackendError> {
363    match probe_header(path)? {
364        // Missing/expired/corrupt — or 0 = never expires: exists but has no
365        // TTL to report (py parity).
366        None | Some(0) => Ok(None),
367        Some(expiry) => Ok(Some(Duration::from_secs(expiry.saturating_sub(now_secs())))),
368    }
369}
370
371fn rewrite_expiry(path: &Path, new_expiry: u64) -> Result<bool, BackendError> {
372    let mut file = match open_opts(true, true).open(path) {
373        Ok(f) => f,
374        Err(e) if is_missing(&e) => return Ok(false),
375        Err(e) => return Err(file_err(e, "failed to open cache file")),
376    };
377    flock_nb(&file, true)?;
378
379    let (header, filled) =
380        fill_header(&mut file).map_err(|e| file_err(e, "failed to read cache file header"))?;
381    match parse_header(&header[..filled], now_secs()) {
382        ParsedHeader::Corrupt | ParsedHeader::Expired => {
383            // An expired entry is absent, not refreshable (py parity).
384            unlink_if_same_inode(path, &file);
385            return Ok(false);
386        }
387        // Never rewrite an entry we cannot interpret.
388        ParsedHeader::Unsupported => return Ok(false),
389        ParsedHeader::Live { .. } => {}
390    }
391
392    // If a concurrent writer replaced the entry since we opened it, our fd
393    // points at an orphaned inode — writing there would report success while
394    // the live entry keeps its old expiry (silent early eviction). Refuse:
395    // `false` honestly says "what you opened is no longer the live entry".
396    if !still_linked(path, &file) {
397        return Ok(false);
398    }
399
400    // Overwrite ONLY the 8-byte expiry field. Same trade-off py documents: a
401    // torn write on power loss yields a wrong expiry, never a corrupt
402    // payload — magic, version, and data are untouched.
403    let do_write = |f: &mut fs::File| -> std::io::Result<()> {
404        f.seek(SeekFrom::Start(EXPIRY_OFFSET))?;
405        f.write_all(&new_expiry.to_be_bytes())?;
406        f.sync_all()
407    };
408    do_write(&mut file).map_err(|e| file_err(e, "failed to refresh cache file TTL"))?;
409    Ok(true)
410}
411
412fn temp_path(target: &Path) -> PathBuf {
413    // `{name}.tmp.{pid}.{nanos}.{seq}` — contains the `.tmp.` marker py's
414    // startup sweep and rs's first-write sweep look for, so a shared
415    // directory gets cross-SDK orphan cleanup. The process-global sequence makes the name collision-proof
416    // even for same-key writes landing in the same nanosecond tick.
417    let nanos = SystemTime::now()
418        .duration_since(UNIX_EPOCH)
419        .map(|d| d.as_nanos())
420        .unwrap_or(0);
421    let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
422    let name = target
423        .file_name()
424        .map(|n| n.to_string_lossy().into_owned())
425        .unwrap_or_default();
426    target.with_file_name(format!("{name}.tmp.{}.{nanos}.{seq}", std::process::id()))
427}
428
429/// Best-effort sweep of orphaned temp files older than [`TEMP_FILE_MAX_AGE`].
430fn cleanup_temp_files(cache_dir: &Path) {
431    let Ok(entries) = fs::read_dir(cache_dir) else {
432        return;
433    };
434    for entry in entries.flatten() {
435        let name = entry.file_name();
436        if !name.to_string_lossy().contains(".tmp.") {
437            continue;
438        }
439        // symlink_metadata never follows links — a planted symlink is skipped.
440        let Ok(meta) = entry.path().symlink_metadata() else {
441            continue;
442        };
443        if !meta.is_file() {
444            continue;
445        }
446        let stale = meta
447            .modified()
448            .ok()
449            .and_then(|m| m.elapsed().ok())
450            .is_some_and(|age| age > TEMP_FILE_MAX_AGE);
451        if stale {
452            unlink_quiet(&entry.path());
453        }
454    }
455}
456
457// ── FileBackend ───────────────────────────────────────────────────────────────
458
459/// Local filesystem cache backend.
460///
461/// See the [module docs](self) for the on-disk format, the concurrency
462/// model, and the deliberate v1 divergences from cachekit-py.
463#[derive(Debug, Clone)]
464pub struct FileBackend {
465    cache_dir: PathBuf,
466    /// Per-key op serialization, striped by filename. Clones share them.
467    /// Async so contended waiters cost a queued future, not a parked
468    /// blocking-pool thread (a hot backend must not monopolize the pool).
469    stripes: Arc<[Arc<tokio::sync::Mutex<()>>; LOCK_STRIPES]>,
470    /// Whether this backend has swept orphaned temp files yet; the first
471    /// `set` does it. Clones share it, so an instance sweeps once.
472    swept: Arc<AtomicBool>,
473}
474
475/// Number of in-process lock stripes (a power of two, at most 256: the stripe
476/// is the low bits of the filename's first byte).
477const LOCK_STRIPES: usize = 64;
478
479fn new_stripes() -> Arc<[Arc<tokio::sync::Mutex<()>>; LOCK_STRIPES]> {
480    Arc::new(std::array::from_fn(|_| {
481        Arc::new(tokio::sync::Mutex::new(()))
482    }))
483}
484
485/// Stripe for an entry path: the first byte of its lowercase-hex Blake2b-128
486/// filename, so every op on one key takes the same stripe.
487fn stripe_of(path: &Path) -> usize {
488    path.file_name()
489        .and_then(|n| n.to_str())
490        .and_then(|n| n.get(..2))
491        .and_then(|b| u8::from_str_radix(b, 16).ok())
492        .map_or(0, |b| usize::from(b) % LOCK_STRIPES)
493}
494
495impl FileBackend {
496    /// Start building a [`FileBackend`].
497    pub fn builder() -> FileBackendBuilder {
498        FileBackendBuilder::default()
499    }
500
501    /// The resolved directory this backend stores entries in.
502    #[must_use]
503    pub fn cache_dir(&self) -> &Path {
504        &self.cache_dir
505    }
506
507    /// Blake2b-128 hex of the key — the same digest `tracing` events carry as
508    /// `key_hash`, so a log line names the file it touched.
509    fn entry_path(&self, key: &str) -> PathBuf {
510        self.cache_dir.join(crate::metrics::key_hash(key))
511    }
512
513    /// Run `f` on the blocking pool holding lock stripe `stripe` (see
514    /// [`stripe_of`]).
515    ///
516    /// The lock is acquired in async context BEFORE the hop to the blocking
517    /// pool: N contended ops queue as futures instead of parking N blocking
518    /// threads on a mutex (which would starve every other `spawn_blocking`
519    /// user in the process). The owned guard rides into the closure and
520    /// releases when the I/O finishes.
521    async fn locked<T: Send + 'static>(
522        &self,
523        stripe: usize,
524        f: impl FnOnce() -> Result<T, BackendError> + Send + 'static,
525    ) -> Result<T, BackendError> {
526        let guard = Arc::clone(&self.stripes[stripe]).lock_owned().await;
527        run_blocking(move || {
528            let _guard = guard;
529            f()
530        })
531        .await
532    }
533}
534
535// ── Backend impl ──────────────────────────────────────────────────────────────
536
537#[cfg(not(target_arch = "wasm32"))]
538#[cfg_attr(not(feature = "unsync"), async_trait)]
539#[cfg_attr(feature = "unsync", async_trait(?Send))]
540impl Backend for FileBackend {
541    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
542        let path = self.entry_path(key);
543        self.locked(stripe_of(&path), move || read_entry(&path))
544            .await
545    }
546
547    async fn set(
548        &self,
549        key: &str,
550        value: Vec<u8>,
551        ttl: Option<Duration>,
552    ) -> Result<(), BackendError> {
553        // None = never expires (expiry field 0). Sub-second TTLs round up to
554        // 1s (Redis-backend parity); absurd TTLs are rejected like py.
555        // The deadline is stored in whole seconds rounded down, as the spec,
556        // py and ts write it, so an N-second TTL lives N-1 to N seconds.
557        // Rounding up here would make rs entries outlive py/ts ones by 1s.
558        let expiry = match ttl {
559            None => 0,
560            Some(d) => {
561                let secs = d.as_secs().max(1);
562                if secs > MAX_TTL_SECS {
563                    return Err(BackendError::permanent(format!(
564                        "TTL {secs}s out of range [1, {MAX_TTL_SECS}] (max 10 years)"
565                    )));
566                }
567                now_secs().saturating_add(secs)
568            }
569        };
570
571        if !self.swept.load(Ordering::Relaxed) {
572            // No stripe: the sweep touches only temp files past the orphan
573            // threshold, never an entry, and one stripe could not fence the
574            // other stripes' writes anyway. Swapped inside the closure, so
575            // only a sweep that actually runs uses up the flag; once running,
576            // it finishes even if this set is cancelled.
577            let swept = Arc::clone(&self.swept);
578            let dir = self.cache_dir.clone();
579            run_blocking(move || {
580                if !swept.swap(true, Ordering::Relaxed) {
581                    cleanup_temp_files(&dir);
582                }
583                Ok(())
584            })
585            .await?;
586        }
587
588        let path = self.entry_path(key);
589        self.locked(stripe_of(&path), move || {
590            write_entry(&path, build_header(expiry), &value)
591        })
592        .await
593    }
594
595    async fn delete(&self, key: &str) -> Result<bool, BackendError> {
596        let path = self.entry_path(key);
597        self.locked(stripe_of(&path), move || delete_entry(&path))
598            .await
599    }
600
601    async fn exists(&self, key: &str) -> Result<bool, BackendError> {
602        let path = self.entry_path(key);
603        self.locked(stripe_of(&path), move || Ok(probe_header(&path)?.is_some()))
604            .await
605    }
606
607    async fn health(&self) -> Result<HealthStatus, BackendError> {
608        let start = std::time::Instant::now();
609
610        // Write/read/delete round-trip — same probe as cachekit-py.
611        let key = "__health_check__";
612        let probe = b"health_check_data".to_vec();
613        self.set(key, probe.clone(), Some(Duration::from_secs(60)))
614            .await?;
615        let read_back = self.get(key).await?;
616        self.delete(key).await?;
617
618        let round_trip_ok = read_back.as_deref() == Some(probe.as_slice());
619        let latency = start.elapsed();
620
621        let mut details = HashMap::new();
622        details.insert(
623            "cache_dir".to_string(),
624            self.cache_dir.display().to_string(),
625        );
626        if !round_trip_ok {
627            details.insert(
628                "error".to_string(),
629                "round-trip verification failed".to_string(),
630            );
631        }
632        Ok(HealthStatus {
633            is_healthy: round_trip_ok,
634            latency_ms: latency.as_secs_f64() * 1000.0,
635            backend_type: "file".to_string(),
636            details,
637        })
638    }
639}
640
641// ── TtlInspectable impl ───────────────────────────────────────────────────────
642
643#[cfg(not(target_arch = "wasm32"))]
644#[cfg_attr(not(feature = "unsync"), async_trait)]
645#[cfg_attr(feature = "unsync", async_trait(?Send))]
646impl TtlInspectable for FileBackend {
647    async fn ttl(&self, key: &str) -> Result<Option<Duration>, BackendError> {
648        let path = self.entry_path(key);
649        self.locked(stripe_of(&path), move || entry_ttl(&path))
650            .await
651    }
652
653    async fn refresh_ttl(&self, key: &str, ttl: Duration) -> Result<bool, BackendError> {
654        let secs = ttl.as_secs();
655        // Duration::ZERO is rejected, matching the CachekitIO backend — in
656        // this SDK "no expiry" is spelled `set(.., None)`, not a zero TTL
657        // (py's `refresh_ttl(key, 0)` making an entry permanent is py's
658        // spelling of the same contract).
659        if secs == 0 {
660            return Err(BackendError::permanent(
661                "refresh_ttl requires at least 1 second".to_string(),
662            ));
663        }
664        if secs > MAX_TTL_SECS {
665            return Err(BackendError::permanent(format!(
666                "TTL {secs}s out of range [1, {MAX_TTL_SECS}] (max 10 years)"
667            )));
668        }
669
670        let new_expiry = now_secs().saturating_add(secs);
671        let path = self.entry_path(key);
672        self.locked(stripe_of(&path), move || rewrite_expiry(&path, new_expiry))
673            .await
674    }
675}
676
677// ── Builder ───────────────────────────────────────────────────────────────────
678
679/// Builder for [`FileBackend`].
680#[derive(Default)]
681#[must_use]
682pub struct FileBackendBuilder {
683    cache_dir: Option<PathBuf>,
684}
685
686impl FileBackendBuilder {
687    /// Set the cache directory (default: `<system temp dir>/cachekit`, the
688    /// same default as cachekit-py).
689    pub fn cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
690        self.cache_dir = Some(dir.into());
691        self
692    }
693
694    /// Create the cache directory if needed and construct a [`FileBackend`].
695    ///
696    /// On unix, a **pre-existing** directory must be owned by the current
697    /// user and not group/other-writable — a default like `/tmp/cachekit`
698    /// pre-created by another local user would otherwise let them poison
699    /// plaintext cache entries (CWE-377). Does not scan the directory: the
700    /// sweep of orphaned temp files runs on the backend's first `set` (see the
701    /// [module docs](self)). The one-time directory setup does its I/O
702    /// inline; per-operation I/O runs on the blocking pool.
703    ///
704    /// # Errors
705    ///
706    /// Returns a config error if the directory cannot be created or
707    /// resolved, is not owned by the current user, or is group/other-writable.
708    pub fn build(self) -> Result<FileBackend, crate::error::CachekitError> {
709        use crate::error::CachekitError;
710
711        let dir = self
712            .cache_dir
713            .unwrap_or_else(|| std::env::temp_dir().join("cachekit"));
714
715        let create = || -> std::io::Result<()> {
716            let mut builder = fs::DirBuilder::new();
717            builder.recursive(true);
718            #[cfg(unix)]
719            {
720                use std::os::unix::fs::DirBuilderExt;
721                builder.mode(0o700); // owner-only, py parity
722            }
723            builder.create(&dir)
724        };
725        create().map_err(|e| {
726            CachekitError::Config(format!(
727                "failed to create cache directory {}: {e}",
728                dir.display()
729            ))
730        })?;
731
732        // Resolve symlinks once up front (py resolves via realpath per key).
733        let cache_dir = fs::canonicalize(&dir).map_err(|e| {
734            CachekitError::Config(format!(
735                "failed to resolve cache directory {}: {e}",
736                dir.display()
737            ))
738        })?;
739
740        // Ownership/mode gate for pre-existing directories (unix).
741        #[cfg(unix)]
742        {
743            use std::os::unix::fs::MetadataExt;
744            let meta = fs::metadata(&cache_dir).map_err(|e| {
745                CachekitError::Config(format!(
746                    "failed to stat cache directory {}: {e}",
747                    cache_dir.display()
748                ))
749            })?;
750            let euid = rustix::process::geteuid().as_raw();
751            if meta.uid() != euid {
752                return Err(CachekitError::Config(format!(
753                    "cache directory {} is owned by uid {} (expected {euid}) — refusing a \
754                     directory another local user controls",
755                    cache_dir.display(),
756                    meta.uid()
757                )));
758            }
759            if meta.mode() & 0o022 != 0 {
760                return Err(CachekitError::Config(format!(
761                    "cache directory {} is group/other-writable (mode {:o}) — another local \
762                     user could plant or replace cache entries; chmod it to 0700",
763                    cache_dir.display(),
764                    meta.mode() & 0o777
765                )));
766            }
767        }
768
769        Ok(FileBackend {
770            cache_dir,
771            stripes: new_stripes(),
772            swept: Arc::new(AtomicBool::new(false)),
773        })
774    }
775}
776
777#[cfg(test)]
778#[allow(clippy::expect_used, clippy::unwrap_used)] // test-only: failures should panic loudly
779mod tests {
780    use super::*;
781
782    /// Compile-time proof that FileBackend implements TtlInspectable.
783    fn _assert_ttl_inspectable(_b: &dyn TtlInspectable) {}
784
785    #[test]
786    fn file_is_ttl_inspectable() {
787        fn _check(backend: &FileBackend) {
788            _assert_ttl_inspectable(backend);
789        }
790    }
791
792    fn backend_at(dir: &Path) -> FileBackend {
793        FileBackend {
794            cache_dir: dir.to_path_buf(),
795            stripes: new_stripes(),
796            swept: Arc::new(AtomicBool::new(false)),
797        }
798    }
799
800    #[test]
801    fn filename_hash_matches_py_blake2b16_hex() {
802        // cachekit-py: hashlib.blake2b(b"key", digest_size=16).hexdigest()
803        let backend = backend_at(Path::new("/cache"));
804        let path = backend.entry_path("ns:app:func:m.f:args:abc:v1");
805        assert_eq!(
806            path,
807            PathBuf::from("/cache/bcb35ae6f64fa65b2770ab3af631b1ce")
808        );
809    }
810
811    #[test]
812    fn header_layout_is_py_byte_compatible() {
813        let header = build_header(0x0102_0304_0506_0708);
814        assert_eq!(&header[0..2], b"CK");
815        assert_eq!(header[2], 1); // version
816        assert_eq!(header[3], 0); // reserved
817        assert_eq!(&header[4..6], &[0, 0]); // flags u16 BE
818        assert_eq!(&header[6..14], &[1, 2, 3, 4, 5, 6, 7, 8]); // expiry u64 BE
819    }
820
821    // ── On-disk format: vectors generated by cachekit-py's header builder ────
822    // (b"CK" + version + reserved + struct.pack(">H", 0) + struct.pack(">Q", expiry))
823
824    fn write_raw(dir: &Path, name: &str, hex_bytes: &str) -> PathBuf {
825        let path = dir.join(name);
826        fs::write(&path, hex::decode(hex_bytes).unwrap()).unwrap();
827        path
828    }
829
830    #[test]
831    fn py_written_never_expires_entry_reads_back() {
832        let dir = tempfile::tempdir().expect("tempdir");
833        let path = write_raw(
834            dir.path(),
835            "entry",
836            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
837        );
838        let payload = read_entry(&path).expect("read should not error");
839        assert_eq!(payload.as_deref(), Some(b"hello from python".as_slice()));
840        assert!(path.exists(), "live entry must not be unlinked");
841    }
842
843    #[test]
844    fn py_written_future_expiry_entry_reads_back() {
845        let dir = tempfile::tempdir().expect("tempdir");
846        // expiry 4102444800 = 2100-01-01
847        let path = write_raw(
848            dir.path(),
849            "entry",
850            "434b0100000000000000f48657006672657368",
851        );
852        let payload = read_entry(&path).expect("read should not error");
853        assert_eq!(payload.as_deref(), Some(b"fresh".as_slice()));
854
855        let remaining = entry_ttl(&path)
856            .expect("ttl should not error")
857            .expect("future expiry must report a TTL");
858        assert!(remaining > Duration::from_secs(3600), "TTL: {remaining:?}");
859    }
860
861    #[test]
862    fn py_written_expired_entry_is_unlinked_on_read() {
863        let dir = tempfile::tempdir().expect("tempdir");
864        // expiry 1 = 1970-01-01T00:00:01Z
865        let path = write_raw(
866            dir.path(),
867            "entry",
868            "434b0100000000000000000000017374616c65",
869        );
870        assert_eq!(read_entry(&path).expect("read should not error"), None);
871        assert!(!path.exists(), "expired entry must be unlinked on read");
872    }
873
874    #[test]
875    fn corrupt_and_short_entries_are_unlinked_on_read() {
876        let dir = tempfile::tempdir().expect("tempdir");
877
878        // Wrong magic ("XX"), otherwise well-formed.
879        let bad_magic = write_raw(dir.path(), "bad-magic", "585801000000000000000000000061");
880        assert_eq!(read_entry(&bad_magic).expect("no error"), None);
881        assert!(!bad_magic.exists());
882
883        // Unknown format version (2).
884        let bad_version = write_raw(dir.path(), "bad-version", "434b02000000000000000000000061");
885        assert_eq!(read_entry(&bad_version).expect("no error"), None);
886        assert!(!bad_version.exists());
887
888        // Shorter than the 14-byte header.
889        let short = write_raw(dir.path(), "short", "434b01");
890        assert_eq!(read_entry(&short).expect("no error"), None);
891        assert!(!short.exists());
892    }
893
894    // ── Protocol vectors ─────────────────────────────────────────────────────
895    //
896    // `tests/vectors/file-backend.json`, vendored verbatim from
897    // cachekit-io/protocol `test-vectors/file-backend.json` v1.2.0
898    // (sha256 `b7b0c51935a3a00ab340005ae1e091797b14938f92eecae23346bdccf0d12af4`).
899    // Do not edit the JSON here; regenerate upstream and re-vendor. Lives in
900    // this module, not `tests/`, because `parse_header` is crate-private.
901
902    const FILE_BACKEND_VECTORS: &str = include_str!("../../tests/vectors/file-backend.json");
903
904    /// sha256 of the vendored file, pinned so a local edit cannot drift from the
905    /// protocol copy unnoticed.
906    const FILE_BACKEND_SHA256: &str =
907        "b7b0c51935a3a00ab340005ae1e091797b14938f92eecae23346bdccf0d12af4"; // pragma: allowlist secret
908
909    #[test]
910    fn vendored_fixture_matches_the_pinned_sha256() {
911        use sha2::{Digest, Sha256};
912        assert_eq!(
913            hex::encode(Sha256::digest(FILE_BACKEND_VECTORS.as_bytes())),
914            FILE_BACKEND_SHA256,
915            "tests/vectors/file-backend.json differs from the pinned protocol copy: \
916             re-vendor it from protocol and update FILE_BACKEND_SHA256"
917        );
918    }
919
920    fn file_backend_vectors() -> Vec<serde_json::Value> {
921        let doc: serde_json::Value =
922            serde_json::from_str(FILE_BACKEND_VECTORS).expect("vector file parses");
923        assert_eq!(
924            doc["version"], "1.2.0",
925            "re-vendored vectors: update the pin"
926        );
927        doc["vectors"].as_array().expect("vectors array").clone()
928    }
929
930    #[test]
931    #[allow(clippy::panic)] // test-only: an unknown reader_action must fail loudly
932    fn file_backend_vectors_read_as_specified() {
933        let vectors = file_backend_vectors();
934        assert_eq!(vectors.len(), 12, "vectors in file-backend.json");
935        for v in &vectors {
936            let name = v["name"].as_str().expect("name");
937            let file_hex = v["file_hex"].as_str().expect("file_hex");
938            let original = hex::decode(file_hex).expect("file_hex is hex");
939            let action = v["reader_action"].as_str().expect("reader_action");
940
941            for probe in ["read_entry", "probe_header"] {
942                let dir = tempfile::tempdir().expect("tempdir");
943                let path = write_raw(dir.path(), "entry", file_hex);
944                let found = match probe {
945                    "read_entry" => {
946                        let got = read_entry(&path).expect("read should not error");
947                        if action == "return_payload" {
948                            let want = hex::decode(v["payload_hex"].as_str().unwrap()).unwrap();
949                            assert_eq!(got.as_deref(), Some(want.as_slice()), "{name}/{probe}");
950                        }
951                        got.is_some()
952                    }
953                    _ => probe_header(&path)
954                        .expect("probe should not error")
955                        .is_some(),
956                };
957                match action {
958                    "return_payload" | "miss_preserve" => {
959                        assert_eq!(found, action == "return_payload", "{name}/{probe}");
960                        assert_eq!(
961                            fs::read(&path).expect("file must remain"),
962                            original,
963                            "{name}/{probe}: file must be byte-for-byte unchanged"
964                        );
965                    }
966                    // The spec lets a reader lazily remove an expired entry.
967                    "miss_expired" => assert!(!found, "{name}/{probe}"),
968                    other => panic!("{name}: unknown reader_action {other}"),
969                }
970            }
971
972            // Evaluate at the vector's exact clock, where one is given.
973            if let Some(now) = v["reader_now_unix_seconds"].as_u64() {
974                let parsed = parse_header(&original, now);
975                match action {
976                    "miss_expired" => {
977                        assert!(matches!(parsed, ParsedHeader::Expired), "{name} at {now}")
978                    }
979                    "miss_preserve" => {
980                        assert!(
981                            matches!(parsed, ParsedHeader::Unsupported),
982                            "{name} at {now}"
983                        )
984                    }
985                    _ => assert!(
986                        matches!(parsed, ParsedHeader::Live { .. }),
987                        "{name} at {now}"
988                    ),
989                }
990            }
991        }
992    }
993
994    /// The writer side of the entry layout: every vector a version-1 writer can
995    /// produce (no flag, reserved 0) is the file name and bytes this backend writes.
996    #[test]
997    fn writer_produces_every_writable_vector_byte_for_byte() {
998        let dir = tempfile::tempdir().expect("tempdir");
999        let backend = backend_at(dir.path());
1000        let mut written = 0;
1001        for v in file_backend_vectors() {
1002            if v["flags"] != 0 || v["reserved"].as_u64().unwrap_or(0) != 0 {
1003                continue;
1004            }
1005            let name = v["name"].as_str().unwrap();
1006            let path = backend.entry_path(v["key_utf8"].as_str().unwrap());
1007            assert_eq!(
1008                path.file_name().and_then(|f| f.to_str()),
1009                v["filename"].as_str(),
1010                "{name}: file name"
1011            );
1012            let expiry = v["expiry_unix_seconds"].as_u64().expect("expiry");
1013            let payload = hex::decode(v["payload_hex"].as_str().unwrap()).unwrap();
1014            write_entry(&path, build_header(expiry), &payload).expect("write");
1015            assert_eq!(
1016                hex::encode(fs::read(&path).expect("written file")),
1017                v["file_hex"].as_str().unwrap(),
1018                "{name}: file bytes"
1019            );
1020            written += 1;
1021        }
1022        assert_eq!(written, 5, "writable vectors in file-backend.json");
1023    }
1024
1025    #[test]
1026    fn unsupported_header_is_a_miss_on_every_path_and_never_touched() {
1027        for v in file_backend_vectors() {
1028            if v["reader_action"] != "miss_preserve" {
1029                continue;
1030            }
1031            let name = v["name"].as_str().unwrap();
1032            let file_hex = v["file_hex"].as_str().unwrap();
1033            let original = hex::decode(file_hex).unwrap();
1034            let dir = tempfile::tempdir().expect("tempdir");
1035            let path = write_raw(dir.path(), "entry", file_hex);
1036
1037            assert_eq!(read_entry(&path).expect("get"), None, "{name}: get");
1038            assert_eq!(probe_header(&path).expect("exists"), None, "{name}: exists");
1039            assert_eq!(entry_ttl(&path).expect("ttl"), None, "{name}: ttl");
1040            assert!(
1041                !rewrite_expiry(&path, now_secs() + 3600).expect("refresh_ttl"),
1042                "{name}: refresh_ttl"
1043            );
1044            assert_eq!(
1045                fs::read(&path).expect("file must remain"),
1046                original,
1047                "{name}"
1048            );
1049        }
1050    }
1051
1052    #[test]
1053    fn expiry_is_reached_at_the_deadline_second() {
1054        let header = build_header(1_000);
1055        assert!(matches!(
1056            parse_header(&header, 999),
1057            ParsedHeader::Live { expiry: 1_000 }
1058        ));
1059        assert!(matches!(
1060            parse_header(&header, 1_000),
1061            ParsedHeader::Expired
1062        ));
1063        assert!(matches!(
1064            parse_header(&header, 1_001),
1065            ParsedHeader::Expired
1066        ));
1067        // 0 = never expires, whatever the clock says.
1068        assert!(matches!(
1069            parse_header(&build_header(0), u64::MAX),
1070            ParsedHeader::Live { expiry: 0 }
1071        ));
1072    }
1073
1074    #[test]
1075    fn corrupt_header_wins_over_flags() {
1076        // Wrong version with a nonzero flag is still corrupt (unlinkable):
1077        // the flag check runs after magic and version.
1078        let mut header = build_header(0);
1079        header[2] = 2;
1080        header[5] = 1;
1081        assert!(matches!(parse_header(&header, 0), ParsedHeader::Corrupt));
1082    }
1083
1084    #[test]
1085    fn ttl_semantics_match_py() {
1086        let dir = tempfile::tempdir().expect("tempdir");
1087
1088        // Never-expires: exists, but reports no TTL.
1089        let never = write_raw(
1090            dir.path(),
1091            "never",
1092            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
1093        );
1094        assert_eq!(entry_ttl(&never).expect("no error"), None);
1095        assert!(never.exists());
1096
1097        // Expired: absent, unlinked.
1098        let expired = write_raw(
1099            dir.path(),
1100            "expired",
1101            "434b0100000000000000000000017374616c65",
1102        );
1103        assert_eq!(entry_ttl(&expired).expect("no error"), None);
1104        assert!(!expired.exists());
1105
1106        // Missing: absent, no error.
1107        assert_eq!(entry_ttl(&dir.path().join("nope")).expect("no error"), None);
1108    }
1109
1110    #[test]
1111    fn rewrite_expiry_semantics_match_py() {
1112        let dir = tempfile::tempdir().expect("tempdir");
1113
1114        // Live entry: refreshed in place — payload untouched, expiry moved.
1115        let live = write_raw(dir.path(), "live", "434b0100000000000000f48657006672657368");
1116        let new_expiry = now_secs() + 42;
1117        assert!(rewrite_expiry(&live, new_expiry).expect("no error"));
1118        let remaining = entry_ttl(&live).expect("no error").expect("has TTL");
1119        assert!(remaining <= Duration::from_secs(42));
1120        assert_eq!(
1121            read_entry(&live).expect("no error").as_deref(),
1122            Some(b"fresh".as_slice()),
1123            "payload must survive an in-place expiry rewrite"
1124        );
1125
1126        // Expired entry: treated as absent and unlinked.
1127        let expired = write_raw(
1128            dir.path(),
1129            "expired",
1130            "434b0100000000000000000000017374616c65",
1131        );
1132        assert!(!rewrite_expiry(&expired, new_expiry).expect("no error"));
1133        assert!(!expired.exists());
1134
1135        // Missing entry: false, no error.
1136        assert!(!rewrite_expiry(&dir.path().join("nope"), new_expiry).expect("no error"));
1137    }
1138
1139    /// A `.tmp.` orphan two minutes old: past the sweep's 60 s threshold.
1140    fn plant_stale_temp(dir: &Path) -> PathBuf {
1141        let stale = dir.join("abc.tmp.123.456");
1142        fs::write(&stale, b"orphan").expect("write");
1143        fs::File::options()
1144            .write(true)
1145            .open(&stale)
1146            .expect("open")
1147            .set_modified(SystemTime::now() - Duration::from_secs(120))
1148            .expect("set mtime");
1149        stale
1150    }
1151
1152    #[test]
1153    fn temp_file_sweep_removes_stale_keeps_fresh() {
1154        let dir = tempfile::tempdir().expect("tempdir");
1155
1156        let stale = plant_stale_temp(dir.path());
1157
1158        let fresh = dir.path().join("def.tmp.123.789");
1159        fs::write(&fresh, b"in flight").expect("write");
1160
1161        let entry = dir.path().join("0123abcd");
1162        fs::write(&entry, b"not a temp file").expect("write");
1163
1164        cleanup_temp_files(dir.path());
1165
1166        assert!(!stale.exists(), "stale temp file must be swept");
1167        assert!(fresh.exists(), "fresh temp file must be kept");
1168        assert!(entry.exists(), "cache entries must never be swept");
1169    }
1170
1171    #[tokio::test]
1172    async fn orphan_sweep_runs_on_first_set_not_on_build() {
1173        // A short-lived reader (a CLI hit) must not pay a whole-directory
1174        // scan: build and reads leave orphans alone, the first write sweeps.
1175        let dir = tempfile::tempdir().expect("tempdir");
1176        let cache_dir = dir.path().join("cache");
1177        let build = || {
1178            FileBackend::builder()
1179                .cache_dir(&cache_dir)
1180                .build()
1181                .expect("build")
1182        };
1183        drop(build()); // creates the directory owner-only
1184
1185        let stale = plant_stale_temp(&cache_dir);
1186
1187        let backend = build();
1188        assert!(stale.exists(), "build must not scan the directory");
1189
1190        assert!(backend.get("k").await.expect("get").is_none());
1191        assert!(backend.exists("k").await.is_ok());
1192        assert!(stale.exists(), "reads must not sweep");
1193
1194        // Clones share the once-per-instance flag: one sweep, not one per clone.
1195        let clone = backend.clone();
1196        clone.set("k", b"v".to_vec(), None).await.expect("set");
1197        assert!(!stale.exists(), "first set must sweep stale orphans");
1198        assert!(
1199            backend.swept.load(Ordering::Relaxed),
1200            "clones share the flag"
1201        );
1202    }
1203
1204    // A set parked on its stripe needs the blocking pool; `unsync` runs I/O inline.
1205    #[cfg(not(feature = "unsync"))]
1206    #[tokio::test]
1207    async fn first_set_sweeps_without_waiting_for_its_stripe() {
1208        // The sweep touches only temp files past the orphan threshold, never
1209        // an entry, so it must neither wait for nor hold the key's stripe.
1210        let (_dir, backend) = private_backend();
1211        let stale = plant_stale_temp(backend.cache_dir());
1212        let stripe = stripe_of(&backend.entry_path("k"));
1213        let held = Arc::clone(&backend.stripes[stripe]).lock_owned().await;
1214
1215        let set = {
1216            let b = backend.clone();
1217            tokio::spawn(async move { b.set("k", b"v".to_vec(), None).await })
1218        };
1219        tokio::time::timeout(Duration::from_secs(5), async {
1220            while stale.exists() {
1221                tokio::time::sleep(Duration::from_millis(5)).await;
1222            }
1223        })
1224        .await
1225        .expect("the sweep must not wait for the set's stripe");
1226        assert!(
1227            !set.is_finished(),
1228            "the write itself still waits for its stripe"
1229        );
1230
1231        drop(held);
1232        set.await.expect("join").expect("set");
1233    }
1234
1235    // ── Hardening ────────────────────────────────────────────────────────────
1236
1237    #[cfg(unix)]
1238    #[test]
1239    fn builder_rejects_group_or_other_writable_directory() {
1240        use std::os::unix::fs::PermissionsExt;
1241
1242        // CWE-377: a pre-existing world/group-writable cache dir (e.g. a
1243        // planted /tmp/cachekit) lets another local user poison plaintext
1244        // entries — build() must refuse it.
1245        let dir = tempfile::tempdir().expect("tempdir");
1246        let target = dir.path().join("shared");
1247        fs::create_dir(&target).expect("mkdir");
1248        fs::set_permissions(&target, fs::Permissions::from_mode(0o770)).expect("chmod");
1249
1250        let err = FileBackend::builder()
1251            .cache_dir(&target)
1252            .build()
1253            .expect_err("group-writable dir must be rejected");
1254        assert!(err.to_string().contains("writable"), "{err}");
1255
1256        // Owner-only passes.
1257        fs::set_permissions(&target, fs::Permissions::from_mode(0o700)).expect("chmod");
1258        FileBackend::builder()
1259            .cache_dir(&target)
1260            .build()
1261            .expect("0700 dir must be accepted");
1262    }
1263
1264    #[test]
1265    fn temp_paths_never_collide() {
1266        let target = Path::new("/cache/abc");
1267        let a = temp_path(target);
1268        let b = temp_path(target);
1269        assert_ne!(a, b, "same-key temp paths must be unique");
1270        assert!(a.to_string_lossy().contains(".tmp."), "py sweep pattern");
1271    }
1272
1273    #[cfg(unix)]
1274    #[test]
1275    fn stale_unlink_decision_spares_replaced_entry() {
1276        // Stale-unlink race: a get() that decided "expired" must not delete
1277        // the fresh entry a concurrent set() renamed over the path.
1278        let dir = tempfile::tempdir().expect("tempdir");
1279        let path = write_raw(
1280            dir.path(),
1281            "entry",
1282            "434b0100000000000000000000017374616c65", // expired
1283        );
1284
1285        // Open the expired inode (the stale decision's fd)…
1286        let stale_fd = open_opts(true, false).open(&path).expect("open");
1287
1288        // …then a "concurrent writer" replaces the entry (new inode).
1289        write_entry(&path, build_header(0), b"fresh").expect("replace");
1290
1291        // The stale unlink must be refused: path now names a different inode.
1292        unlink_if_same_inode(&path, &stale_fd);
1293        assert_eq!(
1294            read_entry(&path).expect("no error").as_deref(),
1295            Some(b"fresh".as_slice()),
1296            "inode-validated unlink must spare the replaced entry"
1297        );
1298
1299        // Same guard for refresh: the stale fd is no longer the live entry.
1300        assert!(!still_linked(&path, &stale_fd));
1301        let live_fd = open_opts(true, false).open(&path).expect("open");
1302        assert!(still_linked(&path, &live_fd));
1303    }
1304
1305    #[cfg(unix)]
1306    #[test]
1307    fn flock_contention_surfaces_as_timeout() {
1308        use rustix::fs::{flock, FlockOperation};
1309
1310        let dir = tempfile::tempdir().expect("tempdir");
1311        let path = write_raw(
1312            dir.path(),
1313            "entry",
1314            "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
1315        );
1316
1317        // Another "process" holds an exclusive lock…
1318        let holder = fs::File::open(&path).expect("open");
1319        flock(&holder, FlockOperation::NonBlockingLockExclusive).expect("lock");
1320
1321        // …so our shared-lock read errors as a timeout (py parity), it does
1322        // not block and does not misreport the entry as missing.
1323        let err = read_entry(&path).expect_err("contended read must error");
1324        assert_eq!(err.kind, BackendErrorKind::Timeout, "{err}");
1325    }
1326
1327    /// A 0700 tempdir and a backend built on it through the public builder.
1328    fn private_backend() -> (tempfile::TempDir, FileBackend) {
1329        let dir = tempfile::tempdir().expect("tempdir");
1330        #[cfg(unix)]
1331        {
1332            use std::os::unix::fs::PermissionsExt;
1333            fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).expect("chmod");
1334        }
1335        let backend = FileBackend::builder()
1336            .cache_dir(dir.path())
1337            .build()
1338            .expect("build");
1339        (dir, backend)
1340    }
1341
1342    #[test]
1343    fn stripe_is_the_filename_first_byte() {
1344        // Pinned digest from filename_hash_matches_py_blake2b16_hex: 0xbc… → 0xbc % 64.
1345        let backend = backend_at(Path::new("/cache"));
1346        assert_eq!(
1347            stripe_of(&backend.entry_path("ns:app:func:m.f:args:abc:v1")),
1348            0xbc % LOCK_STRIPES
1349        );
1350        for i in 0..1000 {
1351            let key = format!("k{i}");
1352            let name = crate::metrics::key_hash(&key);
1353            let first = u8::from_str_radix(&name[..2], 16).expect("hex");
1354            assert_eq!(
1355                stripe_of(&backend.entry_path(&key)),
1356                usize::from(first) % LOCK_STRIPES
1357            );
1358        }
1359    }
1360
1361    // A blocking op held mid-flight needs the blocking pool; `unsync` runs I/O inline.
1362    #[cfg(not(feature = "unsync"))]
1363    #[tokio::test]
1364    async fn an_in_flight_op_blocks_its_key_and_no_other() {
1365        let (_dir, backend) = private_backend();
1366        let held = "held";
1367        let held_stripe = stripe_of(&backend.entry_path(held));
1368        let other = (0..)
1369            .map(|i| format!("other{i}"))
1370            .find(|k| stripe_of(&backend.entry_path(k)) != held_stripe)
1371            .expect("a key on another stripe");
1372        backend.set(held, b"h".to_vec(), None).await.expect("set");
1373        backend.set(&other, b"o".to_vec(), None).await.expect("set");
1374
1375        // Stand in for a write to `held` stuck in its fsync: an op that takes
1376        // whatever lock the backend uses for `held` and parks until released.
1377        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
1378        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
1379        let in_flight = {
1380            let b = backend.clone();
1381            tokio::spawn(async move {
1382                b.locked(held_stripe, move || {
1383                    let _ = started_tx.send(());
1384                    let _ = release_rx.recv();
1385                    Ok(())
1386                })
1387                .await
1388            })
1389        };
1390        started_rx.await.expect("in-flight op started");
1391
1392        let other_get = tokio::time::timeout(Duration::from_secs(5), backend.get(&other))
1393            .await
1394            .expect("a read of another key must not wait on an unrelated in-flight write");
1395        assert_eq!(other_get.expect("get").as_deref(), Some(b"o".as_slice()));
1396        assert!(
1397            tokio::time::timeout(Duration::from_millis(100), backend.get(held))
1398                .await
1399                .is_err(),
1400            "same-key ops must still serialize behind the in-flight op"
1401        );
1402
1403        release_tx.send(()).expect("release");
1404        in_flight.await.expect("join").expect("in-flight op");
1405        assert_eq!(
1406            backend.get(held).await.expect("get").as_deref(),
1407            Some(b"h".as_slice())
1408        );
1409    }
1410
1411    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1412    async fn interleaved_cross_key_set_get_loses_no_update() {
1413        // Many keys, so tasks land on shared and distinct stripes alike: every
1414        // read sees its own task's latest write, and every final value is the
1415        // last one written.
1416        let (_dir, backend) = private_backend();
1417        let tasks: Vec<_> = (0..16)
1418            .map(|t| {
1419                let b = backend.clone();
1420                tokio::spawn(async move {
1421                    let key = format!("task{t}");
1422                    for round in 0..25u32 {
1423                        let value = format!("{t}:{round}").into_bytes();
1424                        b.set(&key, value.clone(), None).await.expect("set");
1425                        assert_eq!(
1426                            b.get(&key).await.expect("get"),
1427                            Some(value),
1428                            "task {t} round {round}: update lost"
1429                        );
1430                    }
1431                })
1432            })
1433            .collect();
1434        for task in tasks {
1435            task.await.expect("join");
1436        }
1437        for t in 0..16 {
1438            assert_eq!(
1439                backend.get(&format!("task{t}")).await.expect("get"),
1440                Some(format!("{t}:24").into_bytes())
1441            );
1442        }
1443    }
1444
1445    #[tokio::test]
1446    async fn expired_read_racing_fresh_set_never_loses_the_write() {
1447        // The stale-unlink race, end to end through the public API: an expired
1448        // entry is read (deciding to unlink) while a fresh set replaces it.
1449        // Whatever the interleaving, the fresh value must survive.
1450        let dir = tempfile::tempdir().expect("tempdir");
1451        // tempfile dirs follow the umask (e.g. 0775) — the builder's
1452        // ownership/mode gate would rightly reject that; make it 0700.
1453        #[cfg(unix)]
1454        {
1455            use std::os::unix::fs::PermissionsExt;
1456            fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).expect("chmod");
1457        }
1458        let backend = FileBackend::builder()
1459            .cache_dir(dir.path())
1460            .build()
1461            .expect("build");
1462        let key = "raced";
1463        let path = backend.entry_path(key);
1464
1465        for round in 0..50 {
1466            // Plant an already-expired entry directly (py-format bytes).
1467            fs::write(
1468                &path,
1469                hex::decode("434b0100000000000000000000017374616c65").unwrap(),
1470            )
1471            .expect("plant expired entry");
1472
1473            let reader = {
1474                let b = backend.clone();
1475                tokio::spawn(async move { b.get(key).await })
1476            };
1477            let writer = {
1478                let b = backend.clone();
1479                tokio::spawn(async move { b.set(key, b"fresh".to_vec(), None).await })
1480            };
1481            reader.await.expect("join").expect("get");
1482            writer.await.expect("join").expect("set");
1483
1484            assert_eq!(
1485                backend.get(key).await.expect("get").as_deref(),
1486                Some(b"fresh".as_slice()),
1487                "round {round}: fresh write lost to a stale expired-unlink"
1488            );
1489        }
1490    }
1491}