1use 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
72const MAGIC: &[u8; 2] = b"CK";
74const FORMAT_VERSION: u8 = 1;
75const HEADER_SIZE: usize = 14;
76const EXPIRY_OFFSET: u64 = 6;
77
78const MAX_TTL_SECS: u64 = 10 * 365 * 24 * 60 * 60;
80
81const TEMP_FILE_MAX_AGE: Duration = Duration::from_secs(60);
84
85static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
88
89fn now_secs() -> u64 {
92 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 h[6..14].copy_from_slice(&expiry_unix_secs.to_be_bytes());
105 h
106}
107
108enum ParsedHeader {
109 Corrupt,
111 Expired,
113 Unsupported,
116 Live { expiry: u64 },
118}
119
120fn 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 if expiry > 0 && now >= expiry {
133 return ParsedHeader::Expired;
134 }
135 ParsedHeader::Live { expiry }
136}
137
138fn 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, 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
153fn classify_io(e: &std::io::Error) -> BackendErrorKind {
157 use std::io::ErrorKind as K;
158 match e.kind() {
159 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
169fn 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
191fn is_missing(e: &std::io::Error) -> bool {
195 e.kind() == std::io::ErrorKind::NotFound || is_symlink_loop(e)
196}
197
198fn 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 opts.custom_flags(libc::O_NOFOLLOW);
208 }
209 opts
210}
211
212fn 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
246fn unlink_if_same_inode(path: &Path, opened: &fs::File) {
254 if still_linked(path, opened) {
255 unlink_quiet(path);
256 }
257}
258
259fn 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
277fn 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
300fn 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 let mut opts = open_opts(false, true);
327 opts.create_new(true); #[cfg(unix)]
329 {
330 use std::os::unix::fs::OpenOptionsExt;
331 opts.mode(0o600); }
333 let mut file = opts
334 .open(&temp)
335 .map_err(|e| file_err(e, "failed to create cache temp file"))?;
336
337 let write_all = |file: &mut fs::File| -> std::io::Result<()> {
339 file.write_all(&header)?;
340 file.write_all(value)?;
341 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 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 unlink_if_same_inode(path, &file);
385 return Ok(false);
386 }
387 ParsedHeader::Unsupported => return Ok(false),
389 ParsedHeader::Live { .. } => {}
390 }
391
392 if !still_linked(path, &file) {
397 return Ok(false);
398 }
399
400 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 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
429fn 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 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#[derive(Debug, Clone)]
464pub struct FileBackend {
465 cache_dir: PathBuf,
466 stripes: Arc<[Arc<tokio::sync::Mutex<()>>; LOCK_STRIPES]>,
470 swept: Arc<AtomicBool>,
473}
474
475const 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
485fn 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 pub fn builder() -> FileBackendBuilder {
498 FileBackendBuilder::default()
499 }
500
501 #[must_use]
503 pub fn cache_dir(&self) -> &Path {
504 &self.cache_dir
505 }
506
507 fn entry_path(&self, key: &str) -> PathBuf {
510 self.cache_dir.join(crate::metrics::key_hash(key))
511 }
512
513 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#[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 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 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 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#[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 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#[derive(Default)]
681#[must_use]
682pub struct FileBackendBuilder {
683 cache_dir: Option<PathBuf>,
684}
685
686impl FileBackendBuilder {
687 pub fn cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
690 self.cache_dir = Some(dir.into());
691 self
692 }
693
694 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); }
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 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 #[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)] mod tests {
780 use super::*;
781
782 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 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); assert_eq!(header[3], 0); assert_eq!(&header[4..6], &[0, 0]); assert_eq!(&header[6..14], &[1, 2, 3, 4, 5, 6, 7, 8]); }
820
821 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 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 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 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 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 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 const FILE_BACKEND_VECTORS: &str = include_str!("../../tests/vectors/file-backend.json");
903
904 const FILE_BACKEND_SHA256: &str =
907 "b7b0c51935a3a00ab340005ae1e091797b14938f92eecae23346bdccf0d12af4"; #[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)] 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 "miss_expired" => assert!(!found, "{name}/{probe}"),
968 other => panic!("{name}: unknown reader_action {other}"),
969 }
970 }
971
972 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 #[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 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 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 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 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 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 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 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 assert!(!rewrite_expiry(&dir.path().join("nope"), new_expiry).expect("no error"));
1137 }
1138
1139 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 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()); 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 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 #[cfg(not(feature = "unsync"))]
1206 #[tokio::test]
1207 async fn first_set_sweeps_without_waiting_for_its_stripe() {
1208 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 #[cfg(unix)]
1238 #[test]
1239 fn builder_rejects_group_or_other_writable_directory() {
1240 use std::os::unix::fs::PermissionsExt;
1241
1242 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 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 let dir = tempfile::tempdir().expect("tempdir");
1279 let path = write_raw(
1280 dir.path(),
1281 "entry",
1282 "434b0100000000000000000000017374616c65", );
1284
1285 let stale_fd = open_opts(true, false).open(&path).expect("open");
1287
1288 write_entry(&path, build_header(0), b"fresh").expect("replace");
1290
1291 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 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 let holder = fs::File::open(&path).expect("open");
1319 flock(&holder, FlockOperation::NonBlockingLockExclusive).expect("lock");
1320
1321 let err = read_entry(&path).expect_err("contended read must error");
1324 assert_eq!(err.kind, BackendErrorKind::Timeout, "{err}");
1325 }
1326
1327 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 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 #[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 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 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 let dir = tempfile::tempdir().expect("tempdir");
1451 #[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 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}