diff --git a/Cargo.lock b/Cargo.lock index ec522c3..1042d70 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -172,6 +172,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "tempfile", "thiserror", "tokio", "tracing", diff --git a/DESIGN.md b/DESIGN.md index c2b07f4..457278a 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -578,8 +578,11 @@ thing that can be down. fragments, never URLs; the browser composes the URL and chooses the size. No image proxy and no image cache in the service. -**Rich detail is not persisted.** It is served through arr-meta's existing -24-hour response cache. The single exception is `poster_path`, `backdrop_path` +**Rich detail is not persisted.** It is served through arr-meta's 24-hour +response cache, which lives on disk in a directory alongside the database +(§10), bounded by both age and total size (#158) — not the database itself, +and not an image cache; images stay hotlinked as above. The single exception +is `poster_path`, `backdrop_path` and `vote_average`, stored on `movies` and `series` and written by the daily metadata refresh (§8), so library views render without a TMDB call. @@ -624,7 +627,7 @@ service. Policy lives in the database, not a config file — size targets and DV rules get tuned by hand during testing and a restart-to-reload loop gets old immediately. Only bootstrap settings (bind address, Prowlarr URL, Transmission URL, TMDB key, -media root) come from config/env. +media root, TMDB response cache directory) come from config/env. Backup is `sqlite3 .backup` on a timer. diff --git a/crates/arr-meta/Cargo.toml b/crates/arr-meta/Cargo.toml index d6e9525..e9ee44c 100644 --- a/crates/arr-meta/Cargo.toml +++ b/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 diff --git a/crates/arr-meta/src/cache.rs b/crates/arr-meta/src/cache.rs index 7519272..6566b83 100644 --- a/crates/arr-meta/src/cache.rs +++ b/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, -} +const FILE_EXT: &str = "tmdb-cache"; #[derive(Debug)] pub(crate) struct Cache { + dir: PathBuf, ttl: Duration, - entries: Mutex>, + 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 { + 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> { - 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) { - 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" + ); } } diff --git a/crates/arr-meta/src/client.rs b/crates/arr-meta/src/client.rs index f0edbe7..782ca27 100644 --- a/crates/arr-meta/src/client.rs +++ b/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, + 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) -> 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 { // 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())) +} diff --git a/crates/arr-meta/src/error.rs b/crates/arr-meta/src/error.rs index 38716da..e6c617f 100644 --- a/crates/arr-meta/src/error.rs +++ b/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), } diff --git a/crates/arr-meta/src/lib.rs b/crates/arr-meta/src/lib.rs index ca6d0f6..8eba96d 100644 --- a/crates/arr-meta/src/lib.rs +++ b/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, diff --git a/crates/arr-meta/tests/tmdb.rs b/crates/arr-meta/tests/tmdb.rs index 95f1c6f..88885b7 100644 --- a/crates/arr-meta/tests/tmdb.rs +++ b/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;