Merge #158: bound the TMDB response cache

Closes #158

crates/arr-meta/Cargo.toml

@@ -15,6 +15,7 @@ thiserror.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
+tempfile.workspace = true
 tokio.workspace = true
 wiremock.workspace = true
 

crates/arr-meta/src/cache.rs

@@ -1,69 +1,289 @@
-//! A small time-to-live cache over raw response bodies.
+//! A time-to-live cache over raw response bodies, backed by files on disk.
 //!
 //! Keyed on a logical request key rather than the URL, so the API key never
-//! becomes part of a cache key. Bodies are stored unparsed: parsing again on a
-//! hit costs microseconds and keeps one cache serving every endpoint.
+//! becomes part of a cache key (see `client::request_url`). The key is
+//! hashed into the filename rather than embedded verbatim, so a search
+//! query or a title id never becomes readable in a directory listing.
+//!
+//! Bounded two ways: an entry older than `ttl` is dropped whether or not
+//! anyone reads it, and the directory as a whole is capped at `max_bytes` —
+//! age alone does not stop a busy day filling the disk. Both are enforced by
+//! [`Cache::sweep`], which runs after every write and is also safe to call
+//! on a schedule (the daily metadata tick, DESIGN.md §8).
+//!
+//! File IO here is synchronous. Bodies are TMDB JSON responses — at most a
+//! few hundred KB — and this runs once per miss, not in a hot loop, so
+//! blocking the calling task briefly is a fair trade for not pulling an
+//! async-fs dependency into a crate that otherwise doesn't need one.
 
-use std::collections::HashMap;
+use std::fs;
+use std::io;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
-use std::sync::{Mutex, PoisonError};
-use std::time::{Duration, Instant};
+use std::time::{Duration, SystemTime};
 
-#[derive(Debug)]
-struct Entry {
-    stored_at: Instant,
-    body: Arc<str>,
-}
+const FILE_EXT: &str = "tmdb-cache";
 
 #[derive(Debug)]
 pub(crate) struct Cache {
+    dir: PathBuf,
     ttl: Duration,
-    entries: Mutex<HashMap<String, Entry>>,
+    max_bytes: u64,
+    /// Whether this cache created `dir` itself (the default, process-unique
+    /// temp directory) rather than being pointed at one the caller owns. Only
+    /// a directory we created is ours to delete.
+    owns_dir: bool,
 }
 
 impl Cache {
-    pub(crate) fn new(ttl: Duration) -> Self {
-        Self {
+    pub(crate) fn new(
+        dir: PathBuf,
+        ttl: Duration,
+        max_bytes: u64,
+        owns_dir: bool,
+    ) -> io::Result<Self> {
+        fs::create_dir_all(&dir)?;
+        Ok(Self {
+            dir,
             ttl,
-            entries: Mutex::new(HashMap::new()),
-        }
+            max_bytes,
+            owns_dir,
+        })
     }
 
     /// The cached body for `key`, if one is present and still fresh.
     pub(crate) fn get(&self, key: &str) -> Option<Arc<str>> {
-        let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
-        let entry = entries.get(key)?;
-        if entry.stored_at.elapsed() < self.ttl {
-            return Some(Arc::clone(&entry.body));
+        let path = self.path_for(key);
+        let metadata = fs::metadata(&path).ok()?;
+        let age = metadata.modified().ok()?.elapsed().unwrap_or_default();
+        if age >= self.ttl {
+            let _ = fs::remove_file(&path);
+            return None;
         }
-        entries.remove(key);
-        None
+        fs::read_to_string(&path).ok().map(Arc::from)
     }
 
-    pub(crate) fn insert(&self, key: String, body: Arc<str>) {
-        let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
-        entries.insert(
-            key,
-            Entry {
-                stored_at: Instant::now(),
-                body,
-            },
-        );
+    /// Write `body` for `key`, atomically: a temp file plus a rename, so a
+    /// killed process can never leave a half-written body that later reads
+    /// as valid JSON.
+    pub(crate) fn insert(&self, key: &str, body: &str) {
+        let path = self.path_for(key);
+        let tmp = self.tmp_path();
+        if fs::write(&tmp, body).is_err() {
+            return;
+        }
+        if fs::rename(&tmp, &path).is_err() {
+            let _ = fs::remove_file(&tmp);
+            return;
+        }
+        self.sweep();
     }
 
     pub(crate) fn remove(&self, key: &str) {
-        self.entries
-            .lock()
-            .unwrap_or_else(PoisonError::into_inner)
-            .remove(key);
+        let _ = fs::remove_file(self.path_for(key));
     }
 
-    /// Drop everything. The daily metadata refresh does not need this — entries
-    /// expire on their own — but a forced refresh from the UI does.
+    /// Drop everything. The daily metadata refresh does not need this —
+    /// entries expire on their own — but a forced refresh from the UI does.
     pub(crate) fn clear(&self) {
-        self.entries
-            .lock()
-            .unwrap_or_else(PoisonError::into_inner)
-            .clear();
+        self.for_each_entry(|path, _modified, _len| {
+            let _ = fs::remove_file(path);
+        });
+    }
+
+    /// Evict every expired entry, then, if the directory is still over
+    /// `max_bytes`, the oldest survivors until it is not.
+    pub(crate) fn sweep(&self) {
+        let mut survivors = Vec::new();
+        self.for_each_entry(|path, modified, len| {
+            if modified.elapsed().unwrap_or_default() >= self.ttl {
+                let _ = fs::remove_file(&path);
+            } else {
+                survivors.push((path, modified, len));
+            }
+        });
+
+        let mut total: u64 = survivors.iter().map(|(_, _, len)| len).sum();
+        if total <= self.max_bytes {
+            return;
+        }
+
+        // Oldest first, so the size sweep behaves like the age one: the most
+        // recently fetched title is the one that survives.
+        survivors.sort_by_key(|(_, modified, _)| *modified);
+        for (path, _, len) in survivors {
+            if total <= self.max_bytes {
+                break;
+            }
+            if fs::remove_file(&path).is_ok() {
+                total = total.saturating_sub(len);
+            }
+        }
+    }
+
+    fn for_each_entry(&self, mut visit: impl FnMut(PathBuf, SystemTime, u64)) {
+        let Ok(read_dir) = fs::read_dir(&self.dir) else {
+            return;
+        };
+        for entry in read_dir.flatten() {
+            let path = entry.path();
+            if path.extension().is_none_or(|ext| ext != FILE_EXT) {
+                continue;
+            }
+            let Ok(metadata) = entry.metadata() else {
+                continue;
+            };
+            let Ok(modified) = metadata.modified() else {
+                continue;
+            };
+            visit(path, modified, metadata.len());
+        }
+    }
+
+    fn path_for(&self, key: &str) -> PathBuf {
+        self.dir
+            .join(format!("{:016x}.{FILE_EXT}", fnv1a64(key.as_bytes())))
+    }
+
+    fn tmp_path(&self) -> PathBuf {
+        static COUNTER: AtomicU64 = AtomicU64::new(0);
+        let unique = COUNTER.fetch_add(1, Ordering::Relaxed);
+        self.dir
+            .join(format!("{}-{unique}.{FILE_EXT}.tmp", std::process::id()))
+    }
+}
+
+impl Drop for Cache {
+    fn drop(&mut self) {
+        if self.owns_dir {
+            let _ = fs::remove_dir_all(&self.dir);
+        }
+    }
+}
+
+/// FNV-1a, 64-bit. Not cryptographic — it only needs to pick a filename —
+/// but unlike `std`'s `DefaultHasher` it is stable across runs, so a
+/// restarted process still hits its own cache from before.
+fn fnv1a64(bytes: &[u8]) -> u64 {
+    const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
+    const PRIME: u64 = 0x0000_0100_0000_01b3;
+    bytes.iter().fold(OFFSET_BASIS, |hash, &byte| {
+        (hash ^ u64::from(byte)).wrapping_mul(PRIME)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Cache;
+    use std::time::Duration;
+
+    fn cache(ttl: Duration, max_bytes: u64) -> (tempfile::TempDir, Cache) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let cache = Cache::new(dir.path().to_owned(), ttl, max_bytes, false).expect("cache");
+        (dir, cache)
+    }
+
+    #[test]
+    fn hit_returns_what_was_inserted() {
+        let (_dir, cache) = cache(Duration::from_secs(60), 1024);
+        cache.insert("movie/1", "hello");
+        assert_eq!(cache.get("movie/1").as_deref(), Some("hello"));
+    }
+
+    #[test]
+    fn miss_is_none() {
+        let (_dir, cache) = cache(Duration::from_secs(60), 1024);
+        assert_eq!(cache.get("movie/absent"), None);
+    }
+
+    #[test]
+    fn an_expired_entry_is_deleted_from_disk_not_just_hidden() {
+        let (dir, cache) = cache(Duration::from_millis(1), 1024);
+        cache.insert("movie/1", "hello");
+        std::thread::sleep(Duration::from_millis(50));
+
+        assert_eq!(cache.get("movie/1"), None, "a stale entry must not hit");
+
+        let files: Vec<_> = std::fs::read_dir(dir.path())
+            .expect("read dir")
+            .filter_map(Result::ok)
+            .collect();
+        assert!(
+            files.is_empty(),
+            "expired entry should have been removed from disk, found {files:?}"
+        );
+    }
+
+    #[test]
+    fn sweep_evicts_expired_entries_without_being_read() {
+        let (dir, cache) = cache(Duration::from_millis(1), 1024);
+        cache.insert("movie/1", "hello");
+        std::thread::sleep(Duration::from_millis(50));
+
+        cache.sweep();
+
+        let count = std::fs::read_dir(dir.path())
+            .expect("read dir")
+            .filter_map(Result::ok)
+            .count();
+        assert_eq!(count, 0, "sweep should have deleted the expired entry");
+    }
+
+    #[test]
+    fn over_the_size_budget_evicts_the_oldest_entry_first() {
+        // A generous TTL, so only the size budget can be doing the evicting.
+        let (dir, cache) = cache(Duration::from_secs(60), 12);
+        cache.insert("movie/old", "0123456789"); // 10 bytes, written first
+        std::thread::sleep(Duration::from_millis(10));
+        cache.insert("movie/new", "0123456789"); // pushes total to 20 > 12
+
+        assert_eq!(
+            cache.get("movie/old"),
+            None,
+            "the older entry should have been evicted to stay under budget"
+        );
+        assert_eq!(
+            cache.get("movie/new").as_deref(),
+            Some("0123456789"),
+            "the newer entry should survive"
+        );
+
+        let total: u64 = std::fs::read_dir(dir.path())
+            .expect("read dir")
+            .filter_map(Result::ok)
+            .map(|entry| entry.metadata().expect("metadata").len())
+            .sum();
+        assert!(
+            total <= 12,
+            "directory should be back under budget, was {total}"
+        );
+    }
+
+    #[test]
+    fn dropping_an_owned_cache_removes_its_directory() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let owned_subdir = dir.path().join("owned");
+        let cache =
+            Cache::new(owned_subdir.clone(), Duration::from_secs(60), 1024, true).expect("cache");
+        cache.insert("movie/1", "hello");
+        drop(cache);
+
+        assert!(
+            !owned_subdir.exists(),
+            "an owned cache directory should be cleaned up on drop"
+        );
+    }
+
+    #[test]
+    fn dropping_a_borrowed_cache_leaves_its_directory() {
+        let (dir, cache) = cache(Duration::from_secs(60), 1024);
+        cache.insert("movie/1", "hello");
+        drop(cache);
+
+        assert!(
+            dir.path().exists(),
+            "a caller-supplied cache directory must survive the client that used it"
+        );
     }
 }

crates/arr-meta/src/client.rs

@@ -1,6 +1,7 @@
 //! The TMDB HTTP client.
 
-use std::sync::Arc;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
 use reqwest::{StatusCode, Url};
@@ -21,12 +22,21 @@ pub const DEFAULT_BASE_URL: &str = "https://api.themoviedb.org/3/";
 /// One day. Metadata refresh is a daily tick (§8), not a per-request cost.
 pub const DEFAULT_CACHE_TTL: Duration = Duration::from_hours(24);
 
+/// 64 MiB. Dokploy's samples for `arr.n62.casa` showed resident memory going
+/// from 16.1-16.5 MiB to 48.77 MiB across the TV-tracking and rich-metadata
+/// milestones (issue #158) — roughly 32 MiB of growth attributable to the
+/// same cache this now bounds, on one household's ordinary usage. Doubling
+/// that for a busier day, and rounding up, gives a ceiling that is real —
+/// unlike the memory it replaces — without being tight; `DEFAULT_CACHE_TTL`
+/// clears the whole directory every day regardless.
+pub const DEFAULT_CACHE_MAX_BYTES: u64 = 64 * 1024 * 1024;
+
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 
 /// How much of an unexpected response body is worth keeping in an error.
 const MAX_ERROR_BODY: usize = 512;
 
-/// A TMDB client with an in-process response cache.
+/// A TMDB client with an on-disk response cache.
 ///
 /// Deliberately not `Clone`: the cache lives inside it, so build one and share
 /// it behind an `Arc` rather than handing out copies that each miss.
@@ -66,6 +76,8 @@ impl TmdbClient {
             api_key: api_key.into(),
             base_url: DEFAULT_BASE_URL.to_owned(),
             cache_ttl: DEFAULT_CACHE_TTL,
+            cache_dir: None,
+            cache_max_bytes: DEFAULT_CACHE_MAX_BYTES,
             timeout: DEFAULT_TIMEOUT,
         }
     }
@@ -254,6 +266,15 @@ impl TmdbClient {
         self.cache.clear();
     }
 
+    /// Evict expired cache entries, and the oldest survivors if the
+    /// directory is still over its size budget. `insert` already does this
+    /// after every write, so calling this is only needed to reclaim space an
+    /// idle cache is sitting on — the daily metadata tick (§8) is the
+    /// intended caller.
+    pub fn sweep_cache(&self) {
+        self.cache.sweep();
+    }
+
     /// Fetch and decode, caching only what decoded.
     ///
     /// Decoding before the insert matters: a malformed response that got into
@@ -279,7 +300,7 @@ impl TmdbClient {
 
         let body = self.fetch(url, path).await?;
         let value = serde_json::from_str(&body)?;
-        self.cache.insert(cache_key, Arc::from(body));
+        self.cache.insert(&cache_key, &body);
         Ok(value)
     }
 
@@ -364,6 +385,8 @@ pub struct TmdbClientBuilder {
     api_key: String,
     base_url: String,
     cache_ttl: Duration,
+    cache_dir: Option<PathBuf>,
+    cache_max_bytes: u64,
     timeout: Duration,
 }
 
@@ -383,6 +406,30 @@ impl TmdbClientBuilder {
         self
     }
 
+    /// Where cache files live.
+    ///
+    /// Left unset, `build` picks a process-unique directory under the OS
+    /// temp dir and removes it when the client is dropped — a warm cache
+    /// never outlives one process, same as before this cache moved to disk.
+    /// Production wants this pointed explicitly at a directory alongside the
+    /// database (DESIGN.md §10), from bootstrap config, so a restart keeps a
+    /// warm cache; a directory set this way is the caller's and is left
+    /// alone on drop.
+    #[must_use]
+    pub fn cache_dir(mut self, cache_dir: impl Into<PathBuf>) -> Self {
+        self.cache_dir = Some(cache_dir.into());
+        self
+    }
+
+    /// Upper bound on total cache size in bytes, enforced by evicting the
+    /// oldest entries first. See [`DEFAULT_CACHE_MAX_BYTES`] for how the
+    /// default was chosen.
+    #[must_use]
+    pub fn cache_max_bytes(mut self, cache_max_bytes: u64) -> Self {
+        self.cache_max_bytes = cache_max_bytes;
+        self
+    }
+
     /// Per-request timeout.
     #[must_use]
     pub fn timeout(mut self, timeout: Duration) -> Self {
@@ -395,7 +442,8 @@ impl TmdbClientBuilder {
     /// # Errors
     ///
     /// [`Error::BaseUrl`] if the base URL will not parse, [`Error::Transport`]
-    /// if the HTTP client cannot be built.
+    /// if the HTTP client cannot be built, [`Error::Cache`] if the cache
+    /// directory cannot be created.
     pub fn build(self) -> Result<TmdbClient> {
         // Without a trailing slash `Url::join` replaces the last path segment
         // instead of appending, which silently drops the `/3`.
@@ -410,11 +458,33 @@ impl TmdbClientBuilder {
             .user_agent(concat!("arr/", env!("CARGO_PKG_VERSION")))
             .build()?;
 
+        let (cache_dir, owns_cache_dir) = match self.cache_dir {
+            Some(dir) => (dir, false),
+            None => (default_cache_dir(), true),
+        };
+        let cache = Cache::new(
+            cache_dir,
+            self.cache_ttl,
+            self.cache_max_bytes,
+            owns_cache_dir,
+        )?;
+
         Ok(TmdbClient {
             http,
             base_url,
             api_key: self.api_key,
-            cache: Cache::new(self.cache_ttl),
+            cache,
         })
     }
 }
+
+/// A process-unique directory under the OS temp dir, used when the caller
+/// does not configure `cache_dir` explicitly. Unique per built client (not
+/// just per process) so that tests building several clients in the same
+/// process, against the same TMDB paths but different mock responses, do not
+/// share a cache and see each other's bodies.
+fn default_cache_dir() -> PathBuf {
+    static COUNTER: AtomicU64 = AtomicU64::new(0);
+    let unique = COUNTER.fetch_add(1, Ordering::Relaxed);
+    std::env::temp_dir().join(format!("arr-meta-cache-{}-{unique}", std::process::id()))
+}

crates/arr-meta/src/error.rs

@@ -53,4 +53,8 @@ pub enum Error {
     /// The configured base URL is not a URL.
     #[error("invalid TMDB base URL: {0}")]
     BaseUrl(String),
+
+    /// The on-disk response cache directory could not be created.
+    #[error("TMDB cache directory: {0}")]
+    Cache(#[from] std::io::Error),
 }

crates/arr-meta/src/lib.rs

@@ -7,8 +7,11 @@
 //! - The digital release date gates targeted search (§6.2). A movie with no
 //!   digital release date must get zero searches.
 //!
-//! Responses are cached in process with a time-to-live, defaulting to a day,
+//! Responses are cached on disk with a time-to-live, defaulting to a day,
 //! because metadata refresh is a daily tick (§8) and not a per-request cost.
+//! The cache directory is bounded by both age and total size (see
+//! [`DEFAULT_CACHE_MAX_BYTES`]) so a service meant to run for months does
+//! not grow without limit.
 
 // `unused_crate_dependencies` is a per-target lint and the library's own test
 // target links the dev-dependencies without using them. The real uses are in
@@ -21,7 +24,9 @@ mod client;
 mod error;
 mod model;
 
-pub use client::{TmdbClient, TmdbClientBuilder, DEFAULT_BASE_URL, DEFAULT_CACHE_TTL};
+pub use client::{
+    TmdbClient, TmdbClientBuilder, DEFAULT_BASE_URL, DEFAULT_CACHE_MAX_BYTES, DEFAULT_CACHE_TTL,
+};
 pub use error::{Error, Result};
 pub use model::{
     CastMember, Episode, ExternalIds, FindResults, Genre, Movie, MovieDetail, MovieSearchResult,

crates/arr-meta/tests/tmdb.rs

@@ -3,7 +3,7 @@
 
 // Same per-target quirk as in `lib.rs`: an integration test links the library's
 // dependencies without using them directly.
-use {reqwest as _, serde as _, serde_json as _, thiserror as _, tracing as _};
+use {reqwest as _, serde as _, serde_json as _, tempfile as _, thiserror as _, tracing as _};
 
 use std::time::Duration;