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 "8d9d8c4709baf9ef3a8f2d71d21fb5a56207bc7a05c9bc7a967ac567f2604615"; #[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.1.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(), 5);
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]
995 fn unsupported_header_is_a_miss_on_every_path_and_never_touched() {
996 for v in file_backend_vectors() {
997 if v["reader_action"] != "miss_preserve" {
998 continue;
999 }
1000 let name = v["name"].as_str().unwrap();
1001 let file_hex = v["file_hex"].as_str().unwrap();
1002 let original = hex::decode(file_hex).unwrap();
1003 let dir = tempfile::tempdir().expect("tempdir");
1004 let path = write_raw(dir.path(), "entry", file_hex);
1005
1006 assert_eq!(read_entry(&path).expect("get"), None, "{name}: get");
1007 assert_eq!(probe_header(&path).expect("exists"), None, "{name}: exists");
1008 assert_eq!(entry_ttl(&path).expect("ttl"), None, "{name}: ttl");
1009 assert!(
1010 !rewrite_expiry(&path, now_secs() + 3600).expect("refresh_ttl"),
1011 "{name}: refresh_ttl"
1012 );
1013 assert_eq!(
1014 fs::read(&path).expect("file must remain"),
1015 original,
1016 "{name}"
1017 );
1018 }
1019 }
1020
1021 #[test]
1022 fn expiry_is_reached_at_the_deadline_second() {
1023 let header = build_header(1_000);
1024 assert!(matches!(
1025 parse_header(&header, 999),
1026 ParsedHeader::Live { expiry: 1_000 }
1027 ));
1028 assert!(matches!(
1029 parse_header(&header, 1_000),
1030 ParsedHeader::Expired
1031 ));
1032 assert!(matches!(
1033 parse_header(&header, 1_001),
1034 ParsedHeader::Expired
1035 ));
1036 assert!(matches!(
1038 parse_header(&build_header(0), u64::MAX),
1039 ParsedHeader::Live { expiry: 0 }
1040 ));
1041 }
1042
1043 #[test]
1044 fn corrupt_header_wins_over_flags() {
1045 let mut header = build_header(0);
1048 header[2] = 2;
1049 header[5] = 1;
1050 assert!(matches!(parse_header(&header, 0), ParsedHeader::Corrupt));
1051 }
1052
1053 #[test]
1054 fn ttl_semantics_match_py() {
1055 let dir = tempfile::tempdir().expect("tempdir");
1056
1057 let never = write_raw(
1059 dir.path(),
1060 "never",
1061 "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
1062 );
1063 assert_eq!(entry_ttl(&never).expect("no error"), None);
1064 assert!(never.exists());
1065
1066 let expired = write_raw(
1068 dir.path(),
1069 "expired",
1070 "434b0100000000000000000000017374616c65",
1071 );
1072 assert_eq!(entry_ttl(&expired).expect("no error"), None);
1073 assert!(!expired.exists());
1074
1075 assert_eq!(entry_ttl(&dir.path().join("nope")).expect("no error"), None);
1077 }
1078
1079 #[test]
1080 fn rewrite_expiry_semantics_match_py() {
1081 let dir = tempfile::tempdir().expect("tempdir");
1082
1083 let live = write_raw(dir.path(), "live", "434b0100000000000000f48657006672657368");
1085 let new_expiry = now_secs() + 42;
1086 assert!(rewrite_expiry(&live, new_expiry).expect("no error"));
1087 let remaining = entry_ttl(&live).expect("no error").expect("has TTL");
1088 assert!(remaining <= Duration::from_secs(42));
1089 assert_eq!(
1090 read_entry(&live).expect("no error").as_deref(),
1091 Some(b"fresh".as_slice()),
1092 "payload must survive an in-place expiry rewrite"
1093 );
1094
1095 let expired = write_raw(
1097 dir.path(),
1098 "expired",
1099 "434b0100000000000000000000017374616c65",
1100 );
1101 assert!(!rewrite_expiry(&expired, new_expiry).expect("no error"));
1102 assert!(!expired.exists());
1103
1104 assert!(!rewrite_expiry(&dir.path().join("nope"), new_expiry).expect("no error"));
1106 }
1107
1108 fn plant_stale_temp(dir: &Path) -> PathBuf {
1110 let stale = dir.join("abc.tmp.123.456");
1111 fs::write(&stale, b"orphan").expect("write");
1112 fs::File::options()
1113 .write(true)
1114 .open(&stale)
1115 .expect("open")
1116 .set_modified(SystemTime::now() - Duration::from_secs(120))
1117 .expect("set mtime");
1118 stale
1119 }
1120
1121 #[test]
1122 fn temp_file_sweep_removes_stale_keeps_fresh() {
1123 let dir = tempfile::tempdir().expect("tempdir");
1124
1125 let stale = plant_stale_temp(dir.path());
1126
1127 let fresh = dir.path().join("def.tmp.123.789");
1128 fs::write(&fresh, b"in flight").expect("write");
1129
1130 let entry = dir.path().join("0123abcd");
1131 fs::write(&entry, b"not a temp file").expect("write");
1132
1133 cleanup_temp_files(dir.path());
1134
1135 assert!(!stale.exists(), "stale temp file must be swept");
1136 assert!(fresh.exists(), "fresh temp file must be kept");
1137 assert!(entry.exists(), "cache entries must never be swept");
1138 }
1139
1140 #[tokio::test]
1141 async fn orphan_sweep_runs_on_first_set_not_on_build() {
1142 let dir = tempfile::tempdir().expect("tempdir");
1145 let cache_dir = dir.path().join("cache");
1146 let build = || {
1147 FileBackend::builder()
1148 .cache_dir(&cache_dir)
1149 .build()
1150 .expect("build")
1151 };
1152 drop(build()); let stale = plant_stale_temp(&cache_dir);
1155
1156 let backend = build();
1157 assert!(stale.exists(), "build must not scan the directory");
1158
1159 assert!(backend.get("k").await.expect("get").is_none());
1160 assert!(backend.exists("k").await.is_ok());
1161 assert!(stale.exists(), "reads must not sweep");
1162
1163 let clone = backend.clone();
1165 clone.set("k", b"v".to_vec(), None).await.expect("set");
1166 assert!(!stale.exists(), "first set must sweep stale orphans");
1167 assert!(
1168 backend.swept.load(Ordering::Relaxed),
1169 "clones share the flag"
1170 );
1171 }
1172
1173 #[cfg(not(feature = "unsync"))]
1175 #[tokio::test]
1176 async fn first_set_sweeps_without_waiting_for_its_stripe() {
1177 let (_dir, backend) = private_backend();
1180 let stale = plant_stale_temp(backend.cache_dir());
1181 let stripe = stripe_of(&backend.entry_path("k"));
1182 let held = Arc::clone(&backend.stripes[stripe]).lock_owned().await;
1183
1184 let set = {
1185 let b = backend.clone();
1186 tokio::spawn(async move { b.set("k", b"v".to_vec(), None).await })
1187 };
1188 tokio::time::timeout(Duration::from_secs(5), async {
1189 while stale.exists() {
1190 tokio::time::sleep(Duration::from_millis(5)).await;
1191 }
1192 })
1193 .await
1194 .expect("the sweep must not wait for the set's stripe");
1195 assert!(
1196 !set.is_finished(),
1197 "the write itself still waits for its stripe"
1198 );
1199
1200 drop(held);
1201 set.await.expect("join").expect("set");
1202 }
1203
1204 #[cfg(unix)]
1207 #[test]
1208 fn builder_rejects_group_or_other_writable_directory() {
1209 use std::os::unix::fs::PermissionsExt;
1210
1211 let dir = tempfile::tempdir().expect("tempdir");
1215 let target = dir.path().join("shared");
1216 fs::create_dir(&target).expect("mkdir");
1217 fs::set_permissions(&target, fs::Permissions::from_mode(0o770)).expect("chmod");
1218
1219 let err = FileBackend::builder()
1220 .cache_dir(&target)
1221 .build()
1222 .expect_err("group-writable dir must be rejected");
1223 assert!(err.to_string().contains("writable"), "{err}");
1224
1225 fs::set_permissions(&target, fs::Permissions::from_mode(0o700)).expect("chmod");
1227 FileBackend::builder()
1228 .cache_dir(&target)
1229 .build()
1230 .expect("0700 dir must be accepted");
1231 }
1232
1233 #[test]
1234 fn temp_paths_never_collide() {
1235 let target = Path::new("/cache/abc");
1236 let a = temp_path(target);
1237 let b = temp_path(target);
1238 assert_ne!(a, b, "same-key temp paths must be unique");
1239 assert!(a.to_string_lossy().contains(".tmp."), "py sweep pattern");
1240 }
1241
1242 #[cfg(unix)]
1243 #[test]
1244 fn stale_unlink_decision_spares_replaced_entry() {
1245 let dir = tempfile::tempdir().expect("tempdir");
1248 let path = write_raw(
1249 dir.path(),
1250 "entry",
1251 "434b0100000000000000000000017374616c65", );
1253
1254 let stale_fd = open_opts(true, false).open(&path).expect("open");
1256
1257 write_entry(&path, build_header(0), b"fresh").expect("replace");
1259
1260 unlink_if_same_inode(&path, &stale_fd);
1262 assert_eq!(
1263 read_entry(&path).expect("no error").as_deref(),
1264 Some(b"fresh".as_slice()),
1265 "inode-validated unlink must spare the replaced entry"
1266 );
1267
1268 assert!(!still_linked(&path, &stale_fd));
1270 let live_fd = open_opts(true, false).open(&path).expect("open");
1271 assert!(still_linked(&path, &live_fd));
1272 }
1273
1274 #[cfg(unix)]
1275 #[test]
1276 fn flock_contention_surfaces_as_timeout() {
1277 use rustix::fs::{flock, FlockOperation};
1278
1279 let dir = tempfile::tempdir().expect("tempdir");
1280 let path = write_raw(
1281 dir.path(),
1282 "entry",
1283 "434b01000000000000000000000068656c6c6f2066726f6d20707974686f6e",
1284 );
1285
1286 let holder = fs::File::open(&path).expect("open");
1288 flock(&holder, FlockOperation::NonBlockingLockExclusive).expect("lock");
1289
1290 let err = read_entry(&path).expect_err("contended read must error");
1293 assert_eq!(err.kind, BackendErrorKind::Timeout, "{err}");
1294 }
1295
1296 fn private_backend() -> (tempfile::TempDir, FileBackend) {
1298 let dir = tempfile::tempdir().expect("tempdir");
1299 #[cfg(unix)]
1300 {
1301 use std::os::unix::fs::PermissionsExt;
1302 fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).expect("chmod");
1303 }
1304 let backend = FileBackend::builder()
1305 .cache_dir(dir.path())
1306 .build()
1307 .expect("build");
1308 (dir, backend)
1309 }
1310
1311 #[test]
1312 fn stripe_is_the_filename_first_byte() {
1313 let backend = backend_at(Path::new("/cache"));
1315 assert_eq!(
1316 stripe_of(&backend.entry_path("ns:app:func:m.f:args:abc:v1")),
1317 0xbc % LOCK_STRIPES
1318 );
1319 for i in 0..1000 {
1320 let key = format!("k{i}");
1321 let name = crate::metrics::key_hash(&key);
1322 let first = u8::from_str_radix(&name[..2], 16).expect("hex");
1323 assert_eq!(
1324 stripe_of(&backend.entry_path(&key)),
1325 usize::from(first) % LOCK_STRIPES
1326 );
1327 }
1328 }
1329
1330 #[cfg(not(feature = "unsync"))]
1332 #[tokio::test]
1333 async fn an_in_flight_op_blocks_its_key_and_no_other() {
1334 let (_dir, backend) = private_backend();
1335 let held = "held";
1336 let held_stripe = stripe_of(&backend.entry_path(held));
1337 let other = (0..)
1338 .map(|i| format!("other{i}"))
1339 .find(|k| stripe_of(&backend.entry_path(k)) != held_stripe)
1340 .expect("a key on another stripe");
1341 backend.set(held, b"h".to_vec(), None).await.expect("set");
1342 backend.set(&other, b"o".to_vec(), None).await.expect("set");
1343
1344 let (started_tx, started_rx) = tokio::sync::oneshot::channel();
1347 let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
1348 let in_flight = {
1349 let b = backend.clone();
1350 tokio::spawn(async move {
1351 b.locked(held_stripe, move || {
1352 let _ = started_tx.send(());
1353 let _ = release_rx.recv();
1354 Ok(())
1355 })
1356 .await
1357 })
1358 };
1359 started_rx.await.expect("in-flight op started");
1360
1361 let other_get = tokio::time::timeout(Duration::from_secs(5), backend.get(&other))
1362 .await
1363 .expect("a read of another key must not wait on an unrelated in-flight write");
1364 assert_eq!(other_get.expect("get").as_deref(), Some(b"o".as_slice()));
1365 assert!(
1366 tokio::time::timeout(Duration::from_millis(100), backend.get(held))
1367 .await
1368 .is_err(),
1369 "same-key ops must still serialize behind the in-flight op"
1370 );
1371
1372 release_tx.send(()).expect("release");
1373 in_flight.await.expect("join").expect("in-flight op");
1374 assert_eq!(
1375 backend.get(held).await.expect("get").as_deref(),
1376 Some(b"h".as_slice())
1377 );
1378 }
1379
1380 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1381 async fn interleaved_cross_key_set_get_loses_no_update() {
1382 let (_dir, backend) = private_backend();
1386 let tasks: Vec<_> = (0..16)
1387 .map(|t| {
1388 let b = backend.clone();
1389 tokio::spawn(async move {
1390 let key = format!("task{t}");
1391 for round in 0..25u32 {
1392 let value = format!("{t}:{round}").into_bytes();
1393 b.set(&key, value.clone(), None).await.expect("set");
1394 assert_eq!(
1395 b.get(&key).await.expect("get"),
1396 Some(value),
1397 "task {t} round {round}: update lost"
1398 );
1399 }
1400 })
1401 })
1402 .collect();
1403 for task in tasks {
1404 task.await.expect("join");
1405 }
1406 for t in 0..16 {
1407 assert_eq!(
1408 backend.get(&format!("task{t}")).await.expect("get"),
1409 Some(format!("{t}:24").into_bytes())
1410 );
1411 }
1412 }
1413
1414 #[tokio::test]
1415 async fn expired_read_racing_fresh_set_never_loses_the_write() {
1416 let dir = tempfile::tempdir().expect("tempdir");
1420 #[cfg(unix)]
1423 {
1424 use std::os::unix::fs::PermissionsExt;
1425 fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).expect("chmod");
1426 }
1427 let backend = FileBackend::builder()
1428 .cache_dir(dir.path())
1429 .build()
1430 .expect("build");
1431 let key = "raced";
1432 let path = backend.entry_path(key);
1433
1434 for round in 0..50 {
1435 fs::write(
1437 &path,
1438 hex::decode("434b0100000000000000000000017374616c65").unwrap(),
1439 )
1440 .expect("plant expired entry");
1441
1442 let reader = {
1443 let b = backend.clone();
1444 tokio::spawn(async move { b.get(key).await })
1445 };
1446 let writer = {
1447 let b = backend.clone();
1448 tokio::spawn(async move { b.set(key, b"fresh".to_vec(), None).await })
1449 };
1450 reader.await.expect("join").expect("get");
1451 writer.await.expect("join").expect("set");
1452
1453 assert_eq!(
1454 backend.get(key).await.expect("get").as_deref(),
1455 Some(b"fresh".as_slice()),
1456 "round {round}: fresh write lost to a stale expired-unlink"
1457 );
1458 }
1459 }
1460}