Files
jellytau/src-tauri/src/download/cache.rs
T
dtourolle 8500da1a42 chore(rust): clear the clippy backlog and finish the poison-tolerant lock sweep
`cargo clippy --all-targets` went from 51 warnings (23 in the lib) to zero.
Most were mechanical — needless borrows, `assert_eq!` against a bool literal,
`vec!` where an array does, `or_insert_with(Vec::new)`, a loop index used only
to index — and were applied with `clippy --fix`, then reviewed line by line.
That review caught one auto-fix that was *not* semantically neutral: dropping
the redundant `use hostname;` left its `#[cfg(target_os = "linux")]` orphaned
directly above `SERVICE_NAME`, which would have silently cfg'd the constant out
of every non-Linux build. Removed the stray attribute with the import.

Where a lint asked for a risky change rather than a better one, it is suppressed
with a comment saying why:

- `too_many_arguments` on five `#[tauri::command]` handlers and
  `ThumbnailCache::save_thumbnail` — most of the arity is `State<'_, _>`
  injection, and a parameter struct would change the IPC contract and the
  generated TypeScript for no readability gain.
- `large_enum_variant` on `PlayerStatusEvent` and `AutoplayDecision` — both are
  serde + specta wire types emitted a handful of times a second, never bulk
  allocated; boxing would have to stay invisible to the generated bindings while
  every match arm gained a deref.
- `await_holding_lock` on the `hybrid`/`offline` test modules — the guard is a
  test-only serialisation lock for the process-global `INCLUDE_CATALOG_BROWSE`
  flag, and the await it spans *is* the critical section. Each `#[tokio::test]`
  gets its own single-threaded runtime, so this is not the production deadlock
  class the lint targets; restructuring would reintroduce the flag race.

Real fixes elsewhere: `JellyfinItem::to_media_item` takes `self` by value, so it
is now `into_media_item`; the five-tuple episode row in the download commands
has a named `EpisodeRow` alias; the mpv `PropertyChange` arm matches
`name: "pause"` instead of guarding on it.

Also converted the last 27 raw `.lock().unwrap()` call sites to `lock_safe()`,
completing the `MutexSafe`/`RwLockSafe` convention. All of them turned out to be
in test modules — production code was already clean — so this is consistency
rather than a fix. The two raw locks in `utils/lock.rs` stay raw on purpose:
those tests deliberately poison a mutex to prove the helpers recover from it.

Pure refactoring: all 698 tests still pass.
2026-08-16 23:05:13 +02:00

670 lines
23 KiB
Rust

//! Smart caching engine for predictive downloads
#[cfg(test)]
use crate::utils::lock::MutexSafe;
use log::{debug, info};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
/// Smart caching configuration
#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CacheConfig {
/// Enable queue pre-caching
pub queue_precache_enabled: bool,
/// Number of tracks to pre-cache from queue
pub queue_precache_count: usize,
/// Enable album affinity detection
pub album_affinity_enabled: bool,
/// Threshold for album affinity (tracks played before caching)
pub album_affinity_threshold: usize,
/// Storage limit in bytes (0 = unlimited)
pub storage_limit: u64,
/// Only cache on WiFi
pub wifi_only: bool,
/// How long a temporary (`download_source = 'auto'`) download lives before
/// it is reclaimed, in hours. 0 disables expiry, leaving space pressure as
/// the only reclaim trigger.
///
/// TRACES: UR-071 | DR-127
pub temporary_ttl_hours: u64,
}
impl Default for CacheConfig {
fn default() -> Self {
Self {
queue_precache_enabled: true,
queue_precache_count: 3, // Preload next 3 tracks by default
album_affinity_enabled: true,
album_affinity_threshold: 3,
storage_limit: 10 * 1024 * 1024 * 1024, // 10GB
wifi_only: false, // Allow preloading on any connection by default
// A week: long enough that re-watching over a weekend still hits
// disk, short enough that a one-off play does not hold space
// indefinitely.
temporary_ttl_hours: 24 * 7,
}
}
}
/// Smart caching engine
#[derive(Clone)]
pub struct SmartCache {
config: Arc<Mutex<CacheConfig>>,
/// Track recently played items per album
album_play_history: Arc<Mutex<HashMap<String, Vec<String>>>>,
}
impl SmartCache {
pub fn new(config: CacheConfig) -> Self {
Self {
config: Arc::new(Mutex::new(config)),
album_play_history: Arc::new(Mutex::new(HashMap::new())),
}
}
/// Update configuration
pub fn update_config(&self, config: CacheConfig) {
if let Ok(mut cfg) = self.config.lock() {
*cfg = config;
}
}
/// Check if should pre-cache queue items.
///
/// Note this deliberately does NOT consult `wifi_only`. It used to return
/// `queue_precache_enabled && !wifi_only`, which disabled precaching
/// outright whenever the user enabled WiFi-only — regardless of the network
/// actually in use. The network check now lives in the download queue pump
/// (`downloads_allowed_on_current_network`), which is the single gate for
/// all download traffic, so this only answers "is precaching enabled?".
///
/// TRACES: UR-053 | DR-074
pub fn should_precache_queue(&self) -> bool {
self.config
.lock()
.map(|cfg| cfg.queue_precache_enabled)
.unwrap_or(false)
}
/// Get number of queue items to pre-cache
pub fn queue_precache_count(&self) -> usize {
self.config
.lock()
.map(|cfg| cfg.queue_precache_count)
.unwrap_or(5)
}
/// Track that an item was played
pub fn track_play(&self, item_id: &str, album_id: Option<&str>) {
if let Some(album) = album_id {
if let Ok(mut history) = self.album_play_history.lock() {
let plays = history.entry(album.to_string()).or_insert_with(Vec::new);
if !plays.contains(&item_id.to_string()) {
plays.push(item_id.to_string());
}
}
}
}
/// Check if album affinity threshold reached for caching
pub fn should_cache_album(&self, album_id: &str) -> Option<bool> {
let config = self.config.lock().ok()?;
if !config.album_affinity_enabled {
return Some(false);
}
let history = self.album_play_history.lock().ok()?;
let play_count = history.get(album_id).map(|v| v.len()).unwrap_or(0);
Some(play_count >= config.album_affinity_threshold)
}
/// Get configuration
pub fn get_config(&self) -> Option<CacheConfig> {
self.config.lock().ok().map(|cfg| cfg.clone())
}
/// Get all tracked albums with their play counts
/// Returns Vec<(album_id, unique_tracks_played)>
pub fn get_album_play_history(&self) -> Vec<(String, usize)> {
self.album_play_history
.lock()
.ok()
.map(|history| {
history
.iter()
.map(|(album_id, tracks)| (album_id.clone(), tracks.len()))
.collect()
})
.unwrap_or_default()
}
// ============= Async versions for DatabaseService =============
/// Get total download size for a user (async version)
pub async fn get_total_download_size_async<S: DatabaseService>(
&self,
db_service: &Arc<S>,
user_id: &str,
) -> Result<u64, String> {
let query = Query::with_params(
"SELECT COALESCE(SUM(file_size), 0) FROM downloads
WHERE user_id = ? AND status = 'completed'",
vec![QueryParam::String(user_id.to_string())],
);
let size: i64 = db_service
.query_one(query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
Ok(size as u64)
}
/// Check if storage limit allows download (async version)
pub async fn can_download_async<S: DatabaseService>(
&self,
db_service: &Arc<S>,
user_id: &str,
new_size: u64,
) -> bool {
// Clone config to avoid holding lock across await
let storage_limit = {
match self.config.lock() {
Ok(cfg) => cfg.storage_limit,
Err(_) => return true,
}
};
if storage_limit == 0 {
return true; // Unlimited
}
let current_size = self
.get_total_download_size_async(db_service, user_id)
.await
.unwrap_or(0);
current_size + new_size <= storage_limit
}
/// Reclaim temporary downloads whose life limit has passed.
///
/// The time-based half of the temporary tier (DR-127); [`evict_lru_async`]
/// is the space-pressure half. A row is reclaimed by whichever fires first.
///
/// Scoped to `download_source = 'auto'` for the same reason eviction is: a
/// `'user'` row is someone's own download and has no expiry. `COALESCE`
/// guards rows predating migration 012, whose source is NULL and whose
/// provenance must therefore be treated as the user's.
///
/// Expiry is normally *derived* — `completed_at` plus the configured TTL —
/// rather than stamped at completion. That means a TTL change applies to
/// entries already on disk instead of only to future ones, and entries
/// predating the column expire without a backfill. `expires_at` is honoured
/// as a per-row override when something sets it.
///
/// `now` is passed in rather than read from the clock so the policy is
/// testable without sleeping. Both sides go through SQLite's `datetime()`
/// because `completed_at` is written as `CURRENT_TIMESTAMP`
/// (`YYYY-MM-DD HH:MM:SS`) while callers pass RFC-3339 (`…T…+00:00`) — a raw
/// string comparison between the two formats is wrong, since `' ' < 'T'`.
///
/// Returns the number of entries reclaimed.
///
/// TRACES: UR-071 | DR-127 | UT-120
pub async fn reclaim_expired_async<S: DatabaseService>(
&self,
db_service: &Arc<S>,
user_id: &str,
now: &str,
) -> Result<usize, String> {
let ttl_hours = {
let config = self.config.lock().map_err(|e| e.to_string())?;
config.temporary_ttl_hours
};
// 0 disables time-based reclaim; space pressure remains the only trigger.
if ttl_hours == 0 {
return Ok(0);
}
let expired: Vec<(i64, String)> = db_service
.query_many(
Query::with_params(
"SELECT id, file_path FROM downloads
WHERE user_id = ?
AND COALESCE(download_source, 'user') = 'auto'
AND status = 'completed'
AND datetime(
COALESCE(expires_at, datetime(completed_at, '+' || ? || ' hours'))
) < datetime(?)",
vec![
QueryParam::String(user_id.to_string()),
QueryParam::String(ttl_hours.to_string()),
QueryParam::String(now.to_string()),
],
),
|row| Ok((row.get(0)?, row.get(1)?)),
)
.await
.map_err(|e| e.to_string())?;
let mut reclaimed = 0usize;
for (id, file_path) in expired {
// Best-effort on the file: a missing one still needs its row gone,
// or the sweep retries it forever.
let _ = std::fs::remove_file(&file_path);
db_service
.execute(Query::with_params(
"DELETE FROM downloads WHERE id = ?",
vec![QueryParam::Int64(id)],
))
.await
.map_err(|e| e.to_string())?;
reclaimed += 1;
}
if reclaimed > 0 {
info!("[SmartCache] Reclaimed {} expired entries", reclaimed);
}
Ok(reclaimed)
}
/// Evict least recently used items to make space (async version).
///
/// The space-pressure half of the temporary tier; [`reclaim_expired_async`]
/// is the time-based half.
pub async fn evict_lru_async<S: DatabaseService>(
&self,
db_service: &Arc<S>,
user_id: &str,
space_needed: u64,
) -> Result<u64, String> {
let current_size = self
.get_total_download_size_async(db_service, user_id)
.await?;
// Get limit without holding lock across await
let limit = {
let config = self.config.lock().map_err(|e| e.to_string())?;
config.storage_limit
};
if limit == 0 || current_size + space_needed <= limit {
return Ok(0); // No eviction needed
}
let to_free = (current_size + space_needed) - limit;
let mut freed: u64 = 0;
// Only the *temporary* tier is evictable. `download_source = 'auto'` is
// precache — the cache put it there, the cache may reclaim it. A 'user'
// row is a download someone explicitly asked for; deleting it to make
// room for a predictive fetch is data loss, and because the old query
// ordered purely by `completed_at ASC` it took the oldest — typically
// exactly the film saved for a flight.
//
// COALESCE, not `= 'auto'` alone: migration 012 added the column with a
// 'user' default, but rows predating it can be NULL, and an unknown
// provenance must be treated as the user's, never as disposable.
//
// Freeing less than requested is the correct outcome when only user
// downloads remain — the caller surfaces "unable to free enough space"
// rather than silently deleting them.
//
// TRACES: UR-071 | DR-126 | UT-108
let query = Query::with_params(
"SELECT id, file_size, file_path FROM downloads
WHERE user_id = ? AND status = 'completed'
AND COALESCE(download_source, 'user') = 'auto'
ORDER BY completed_at ASC",
vec![QueryParam::String(user_id.to_string())],
);
let downloads: Vec<(i64, i64, String)> = db_service
.query_many(query, |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
.await
.map_err(|e| e.to_string())?;
for (id, size, file_path) in downloads {
if freed >= to_free {
break;
}
// Delete file
let _ = std::fs::remove_file(&file_path);
debug!("[SmartCache] Evicted: {} ({} bytes)", file_path, size);
// Delete from database
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE id = ?",
vec![QueryParam::Int64(id)],
);
db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
freed += size as u64;
}
info!("[SmartCache] Freed {} bytes ({} needed)", freed, to_free);
Ok(freed)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_default_config() {
let config = CacheConfig::default();
assert_eq!(config.queue_precache_count, 3);
assert_eq!(config.album_affinity_threshold, 3);
assert!(!config.wifi_only); // wifi_only is false by default for easier preloading
}
#[test]
fn test_album_affinity_tracking() {
let cache = SmartCache::new(CacheConfig::default());
// Track plays from same album
cache.track_play("track1", Some("album1"));
cache.track_play("track2", Some("album1"));
// Below threshold
assert!(!cache.should_cache_album("album1").unwrap_or(false));
cache.track_play("track3", Some("album1"));
// At threshold - should cache
assert!(cache.should_cache_album("album1").unwrap_or(false));
}
#[test]
fn test_queue_precache_config() {
let config = CacheConfig {
queue_precache_enabled: false,
..CacheConfig::default()
};
let cache = SmartCache::new(config);
assert!(!cache.should_precache_queue());
let new_config = CacheConfig {
wifi_only: false,
..CacheConfig::default()
};
cache.update_config(new_config);
assert!(cache.should_precache_queue());
}
#[test]
fn test_wifi_only_does_not_disable_precaching() {
// wifi_only must not short-circuit precaching: the network gate lives in
// the download pump, which checks the *actual* transport. Enabling
// WiFi-only while on WiFi should still precache.
let config = CacheConfig {
queue_precache_enabled: true,
wifi_only: true,
..CacheConfig::default()
};
let cache = SmartCache::new(config);
assert!(cache.should_precache_queue());
}
/// Expiry reclaims only temporary entries that are actually past their life
/// limit — never a user's download (which has no expiry), and never a
/// temporary entry still within its life.
///
/// TRACES: UR-071 | DR-127 | UT-120
#[tokio::test]
async fn test_reclaim_expired_only_takes_expired_temporary_entries() {
use crate::storage::db_service::RusqliteService;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
let conn = Connection::open_in_memory().unwrap();
conn.execute(
"CREATE TABLE downloads (
id INTEGER PRIMARY KEY,
user_id TEXT,
status TEXT,
file_size INTEGER,
file_path TEXT,
completed_at TEXT,
download_source TEXT DEFAULT 'user',
expires_at TEXT
)",
[],
)
.unwrap();
// `completed_at` is in SQLite's CURRENT_TIMESTAMP format (space, not
// 'T'), deliberately: the query has to compare it against an RFC-3339
// "now" and must not do so as raw strings.
for (path, source, completed, expires) in [
// Completed long ago, no override => derived expiry has passed.
("/tmp/jt-expired.mp4", "auto", "2026-01-01 00:00:00", None),
// Completed yesterday => still inside the 7-day default TTL.
("/tmp/jt-fresh.mp4", "auto", "2026-05-31 00:00:00", None),
// Old, but an explicit override keeps it alive.
(
"/tmp/jt-override.mp4",
"auto",
"2026-01-01 00:00:00",
Some("2026-12-01T00:00:00+00:00"),
),
// A user download must never carry an expiry, but assert the sweep
// ignores it even if one were somehow set.
(
"/tmp/jt-user.mp4",
"user",
"2026-01-01 00:00:00",
Some("2026-01-01T00:00:00+00:00"),
),
(
"/tmp/jt-user-noexp.mp4",
"user",
"2026-01-01 00:00:00",
None,
),
] {
conn.execute(
"INSERT INTO downloads (user_id, status, file_size, file_path, download_source, completed_at, expires_at)
VALUES ('user1', 'completed', 10, ?1, ?2, ?3, ?4)",
rusqlite::params![path, source, completed, expires],
)
.unwrap();
}
let conn_arc = Arc::new(Mutex::new(conn));
let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
let cache = SmartCache::new(CacheConfig::default());
let reclaimed = cache
.reclaim_expired_async(&db_service, "user1", "2026-06-01T00:00:00+00:00")
.await
.unwrap();
assert_eq!(
reclaimed, 1,
"only the expired temporary entry is reclaimed"
);
let surviving: Vec<String> = {
let guard = conn_arc.lock_safe();
let mut stmt = guard
.prepare("SELECT file_path FROM downloads ORDER BY id")
.unwrap();
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.unwrap()
.map(|r| r.unwrap())
.collect();
rows
};
assert_eq!(
surviving,
vec![
"/tmp/jt-fresh.mp4".to_string(),
"/tmp/jt-override.mp4".to_string(),
"/tmp/jt-user.mp4".to_string(),
"/tmp/jt-user-noexp.mp4".to_string(),
],
"entries within their life, those with a later override, and every user download must survive"
);
// TTL of 0 disables time-based reclaim entirely.
let cache_no_ttl = SmartCache::new(CacheConfig {
temporary_ttl_hours: 0,
..Default::default()
});
assert_eq!(
cache_no_ttl
.reclaim_expired_async(&db_service, "user1", "2027-01-01T00:00:00+00:00")
.await
.unwrap(),
0,
"a zero TTL leaves space pressure as the only reclaim trigger"
);
}
/// Eviction must only reclaim *temporary* (`download_source = 'auto'`)
/// downloads — the precache tier. A download the user explicitly asked for
/// is their file: it may be deleted by them, never by the cache making room
/// for a predictive fetch.
///
/// Before the fix, `evict_lru_async` selected every completed row ordered by
/// `completed_at ASC` with no source filter, so hitting the storage limit
/// deleted the *oldest* download — typically the film someone downloaded for
/// a flight — in favour of a newer auto-precached track.
///
/// TRACES: UR-071 | DR-126 | UT-108
#[tokio::test]
async fn test_evict_lru_never_deletes_user_downloads() {
use crate::storage::db_service::RusqliteService;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
let conn = Connection::open_in_memory().unwrap();
conn.execute(
"CREATE TABLE downloads (
id INTEGER PRIMARY KEY,
user_id TEXT,
status TEXT,
file_size INTEGER,
file_path TEXT,
completed_at TEXT,
download_source TEXT DEFAULT 'user'
)",
[],
)
.unwrap();
// The user's own download is the OLDEST, so a purely time-ordered
// eviction would take it first.
conn.execute(
"INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-user.mp4', '2026-01-01', 'user')",
[],
)
.unwrap();
// A newer, auto-precached item.
conn.execute(
"INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-auto.mp4', '2026-06-01', 'auto')",
[],
)
.unwrap();
let conn_arc = Arc::new(Mutex::new(conn));
let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
let cache = SmartCache::new(CacheConfig {
storage_limit: 1000,
..Default::default()
});
// 1200 bytes held against a 1000 limit: eviction must free something.
let freed = cache
.evict_lru_async(&db_service, "user1", 0)
.await
.unwrap();
assert!(freed > 0, "eviction should have reclaimed the auto entry");
let surviving: Vec<String> = {
let guard = conn_arc.lock_safe();
let mut stmt = guard
.prepare("SELECT download_source FROM downloads ORDER BY id")
.unwrap();
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.unwrap()
.map(|r| r.unwrap())
.collect();
rows
};
assert_eq!(
surviving,
vec!["user".to_string()],
"the user's own download must survive; only the 'auto' entry is evictable"
);
}
#[tokio::test]
async fn test_storage_limit_check() {
use crate::storage::db_service::RusqliteService;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
let conn = Connection::open_in_memory().unwrap();
conn.execute(
"CREATE TABLE downloads (
id INTEGER PRIMARY KEY,
user_id TEXT,
status TEXT,
file_size INTEGER
)",
[],
)
.unwrap();
let conn_arc = Arc::new(Mutex::new(conn));
let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
let config = CacheConfig {
storage_limit: 1000,
..Default::default()
};
let cache = SmartCache::new(config);
// Empty - can download
assert!(cache.can_download_async(&db_service, "user1", 500).await);
// Add some downloads
{
let conn_guard = conn_arc.lock_safe();
conn_guard.execute(
"INSERT INTO downloads (user_id, status, file_size) VALUES ('user1', 'completed', 600)",
[],
)
.unwrap();
}
// Total would be 1100 > 1000
assert!(!cache.can_download_async(&db_service, "user1", 500).await);
// Smaller size fits
assert!(cache.can_download_async(&db_service, "user1", 300).await);
}
}