diff --git a/src-tauri/src/repository/hybrid.rs b/src-tauri/src/repository/hybrid.rs index bac0442d4..a5eb76e97 100644 --- a/src-tauri/src/repository/hybrid.rs +++ b/src-tauri/src/repository/hybrid.rs @@ -18,6 +18,38 @@ use tokio::time::{timeout, Duration}; use super::exclusions::ExcludeHidden; use super::{types::*, MediaRepository, OfflineRepository, OnlineRepository}; +/// The cache side of a cache-first query. +/// +/// Either the cache answered inside the fast path, or it is still working and +/// the query can be collected later. Keeping the slow case *addressable* rather +/// than discarding it is what lets an offline query fall back to cached content +/// after the server leg fails. +/// +/// TRACES: UR-002 | DR-013 +enum CacheLeg { + /// The cache answered within [`HybridRepository::CACHE_FAST_PATH`]. + Ready(Result), + /// Still running. Awaiting the handle yields the answer eventually. + Slow(tokio::task::JoinHandle>), +} + +impl CacheLeg { + /// Split into the fast-path answer and the still-running query. Exactly one + /// side is `Some`. + #[allow(clippy::type_complexity)] + fn split( + self, + ) -> ( + Option>, + Option>>, + ) { + match self { + CacheLeg::Ready(result) => (Some(result), None), + CacheLeg::Slow(handle) => (None, Some(handle)), + } + } +} + /// Hybrid repository combining online and offline data sources /// /// Uses cache-first parallel racing strategy: @@ -379,35 +411,12 @@ impl HybridRepository { /// @req: DR-013 - Repository pattern for online/offline data access /// /// TRACES: UR-002, UR-076 | DR-013, DR-209 - async fn parallel_race( - &self, - cache_future: F1, - server_future: F2, - ) -> Result + async fn parallel_race(cache: CacheLeg, server_future: F2) -> Result where T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static, - F1: std::future::Future> + Send, F2: std::future::Future> + Send, { - // Try cache first (100ms timeout already applied by callers) - let cache_result = cache_future.await.map(ExcludeHidden::without_excluded); - - if let Ok(data) = &cache_result { - if data.has_content() { - debug!("[HybridRepo] Cache hit, returning immediately"); - return Ok(data.clone()); - } - } - - // Cache miss — fall back to server - debug!("[HybridRepo] Cache miss, querying server"); - match server_future.await { - Ok(data) => Ok(data.without_excluded()), - Err(e) => { - // Server failed, try to return cache even if empty - cache_result.or(Err(e)) - } - } + Self::race_with_refresh(cache, server_future, || {}).await } /// [`Self::parallel_race`], plus a callback fired on the fast path so the @@ -424,21 +433,20 @@ impl HybridRepository { /// already being fetched and cached by the normal path. /// /// TRACES: UR-002, UR-025, UR-076 | DR-155, DR-209 - async fn race_with_refresh( - &self, - cache_future: F1, + async fn race_with_refresh( + cache: CacheLeg, server_future: F2, on_cache_hit: R, ) -> Result where T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static, - F1: std::future::Future> + Send, F2: std::future::Future> + Send, R: FnOnce(), { - let cache_result = cache_future.await.map(ExcludeHidden::without_excluded); + let (fast, slow) = cache.split(); + let fast = fast.map(|r| r.map(ExcludeHidden::without_excluded)); - if let Ok(data) = &cache_result { + if let Some(Ok(data)) = &fast { if data.has_content() { debug!("[HybridRepo] Cache hit, returning immediately (refreshing in background)"); on_cache_hit(); @@ -446,21 +454,81 @@ impl HybridRepository { } } - debug!("[HybridRepo] Cache miss, querying server"); + debug!("[HybridRepo] Cache miss or slow, querying server"); match server_future.await { Ok(data) => Ok(data.without_excluded()), - Err(e) => cache_result.or(Err(e)), + Err(e) => { + // The server cannot answer. If the cache is still working, it is + // now the only thing that can, so wait it out rather than + // reporting the server's failure over data we are about to hold. + // This is the offline path: a cache read slowed by a concurrent + // write used to surface as a network error. + if let Some(handle) = slow { + debug!("[HybridRepo] Server failed; waiting for the slow cache query"); + return match handle.await { + Ok(Ok(data)) => Ok(data.without_excluded()), + Ok(Err(cache_err)) => { + debug!("[HybridRepo] Slow cache query also failed: {cache_err}"); + Err(e) + } + Err(join) => { + debug!("[HybridRepo] Slow cache query panicked: {join}"); + Err(e) + } + }; + } + // Cache answered in time but had nothing: return that, so an + // empty-but-valid cached listing still beats a network error. + fast.unwrap_or(Err(e)) + } } } - /// Simple timeout wrapper for cache queries (100ms timeout) + /// How long the cache gets to answer before a query falls through to the + /// server. Short on purpose: this bounds how long a *cache hit* may delay + /// the UI, not how long the query is allowed to take. + const CACHE_FAST_PATH: Duration = Duration::from_millis(100); + + /// Start a cache query and give it [`Self::CACHE_FAST_PATH`] to answer. /// - /// @req: DR-013 - Repository pattern (cache-first with timeout) + /// Missing the deadline does **not** cancel the query — it keeps running on + /// its own task and [`CacheLeg::settle`] can still collect it. That + /// distinction is the whole point. The database is one SQLite connection + /// behind one mutex, so a concurrent write (a sync drain, a bulk + /// `save_to_cache`) blocks reads for its duration and this deadline trips + /// routinely on slow storage. Treating that as "the cache is empty" while + /// throwing the answer away meant that offline — where the server leg also + /// fails — browsing surfaced a network error instead of the cached content + /// sitting right there on disk. + /// + /// TRACES: UR-002 | DR-013 + async fn cache_leg( + future: impl std::future::Future> + Send + 'static, + ) -> CacheLeg + where + T: Send + 'static, + { + // `&mut handle` so the timeout borrows the join handle rather than + // consuming it: on expiry the task is still ours to collect. + let mut handle = tokio::spawn(future); + match timeout(Self::CACHE_FAST_PATH, &mut handle).await { + Ok(Ok(result)) => CacheLeg::Ready(result), + Ok(Err(join)) => CacheLeg::Ready(Err(RepoError::Database { + message: format!("Cache query failed: {join}"), + })), + Err(_) => { + debug!("[HybridRepo] Cache missed the fast path; leaving it running"); + CacheLeg::Slow(handle) + } + } + } + + /// Await a cache query that is still running, however long it takes. async fn cache_with_timeout( &self, future: impl std::future::Future> + Send, ) -> Result { - timeout(Duration::from_millis(100), future) + timeout(Self::CACHE_FAST_PATH, future) .await .unwrap_or_else(|_| { Err(RepoError::Database { @@ -657,7 +725,7 @@ impl MediaRepository for HybridRepository { let item_id = item_id.to_string(); let item_id_clone = item_id.clone(); - let cache_future = self.cache_with_timeout(async move { offline.get_item(&item_id).await }); + let cache_future = Self::cache_leg(async move { offline.get_item(&item_id).await }).await; let online_for_refresh = Arc::clone(&self.online); let offline_for_save = Arc::clone(&self.offline); @@ -683,8 +751,7 @@ impl MediaRepository for HybridRepository { let server_future = async move { online.get_item(&item_id_clone).await }; - self.race_with_refresh(cache_future, server_future, on_cache_hit) - .await + Self::race_with_refresh(cache_future, server_future, on_cache_hit).await } async fn get_latest_items( @@ -698,13 +765,13 @@ impl MediaRepository for HybridRepository { let parent_id_clone = parent_id.clone(); let limit_clone = limit; - let cache_future = self - .cache_with_timeout(async move { offline.get_latest_items(&parent_id, limit).await }); + let cache_future = + Self::cache_leg(async move { offline.get_latest_items(&parent_id, limit).await }).await; let server_future = async move { online.get_latest_items(&parent_id_clone, limit_clone).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_resume_items( @@ -718,11 +785,12 @@ impl MediaRepository for HybridRepository { let parent_id_clone = parent_id_str.clone(); let limit_clone = limit; - let cache_future = self.cache_with_timeout(async move { + let cache_future = Self::cache_leg(async move { offline .get_resume_items(parent_id_str.as_deref(), limit) .await - }); + }) + .await; let server_future = async move { online @@ -730,7 +798,7 @@ impl MediaRepository for HybridRepository { .await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_next_up_episodes( @@ -754,11 +822,11 @@ impl MediaRepository for HybridRepository { let limit_clone = limit; let cache_future = - self.cache_with_timeout(async move { offline.get_recently_played_audio(limit).await }); + Self::cache_leg(async move { offline.get_recently_played_audio(limit).await }).await; let server_future = async move { online.get_recently_played_audio(limit_clone).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_resume_movies(&self, limit: Option) -> Result, RepoError> { @@ -767,11 +835,11 @@ impl MediaRepository for HybridRepository { let limit_clone = limit; let cache_future = - self.cache_with_timeout(async move { offline.get_resume_movies(limit).await }); + Self::cache_leg(async move { offline.get_resume_movies(limit).await }).await; let server_future = async move { online.get_resume_movies(limit_clone).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_rediscover_albums( @@ -784,11 +852,12 @@ impl MediaRepository for HybridRepository { let parent_id_owned = parent_id.map(|s| s.to_string()); let parent_id_clone = parent_id_owned.clone(); - let cache_future = self.cache_with_timeout(async move { + let cache_future = Self::cache_leg(async move { offline .get_rediscover_albums(parent_id_owned.as_deref(), limit) .await - }); + }) + .await; let server_future = async move { online @@ -796,7 +865,7 @@ impl MediaRepository for HybridRepository { .await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_genres(&self, parent_id: Option<&str>) -> Result, RepoError> { @@ -869,11 +938,11 @@ impl MediaRepository for HybridRepository { let opts_clone = options.clone(); let cache_future = - self.cache_with_timeout(async move { offline.search(&query, opts_clone).await }); + Self::cache_leg(async move { offline.search(&query, opts_clone).await }).await; let server_future = async move { online.search(&query_clone, options).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_playback_info(&self, item_id: &str) -> Result { @@ -1019,11 +1088,11 @@ impl MediaRepository for HybridRepository { let person_id_clone = person_id.clone(); let cache_future = - self.cache_with_timeout(async move { offline.get_person(&person_id).await }); + Self::cache_leg(async move { offline.get_person(&person_id).await }).await; let server_future = async move { online.get_person(&person_id_clone).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } async fn get_items_by_person( @@ -1037,14 +1106,16 @@ impl MediaRepository for HybridRepository { let person_id_clone = person_id.clone(); let opts_clone = options.clone(); - let cache_future = self.cache_with_timeout(async move { - offline.get_items_by_person(&person_id, opts_clone).await - }); + let cache_future = + Self::cache_leg( + async move { offline.get_items_by_person(&person_id, opts_clone).await }, + ) + .await; let server_future = async move { online.get_items_by_person(&person_id_clone, options).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } /// TRACES: UR-067 | DR-115 @@ -1092,12 +1163,12 @@ impl MediaRepository for HybridRepository { let item_id = item_id.to_string(); let item_id_clone = item_id.clone(); - let cache_future = self - .cache_with_timeout(async move { offline.get_similar_items(&item_id, limit).await }); + let cache_future = + Self::cache_leg(async move { offline.get_similar_items(&item_id, limit).await }).await; let server_future = async move { online.get_similar_items(&item_id_clone, limit).await }; - self.parallel_race(cache_future, server_future).await + Self::parallel_race(cache_future, server_future).await } // ===== Playlist Methods ===== @@ -1228,6 +1299,87 @@ mod tests { use super::*; use std::sync::Mutex; + /// Offline, a cache read slowed past the fast path must still answer. + /// + /// The database is one SQLite connection behind one mutex, so a concurrent + /// write blocks reads for its duration and the 100 ms fast path trips on + /// slow storage. The deadline used to *cancel* the read and report it as a + /// miss; with the server leg also failing (offline), the user got a network + /// error over cached content that was sitting on disk. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn a_slow_cache_still_answers_when_the_server_is_gone() { + let cache = HybridRepository::cache_leg(async { + tokio::time::sleep(Duration::from_millis(250)).await; + Ok(vec![MediaItem { + id: "cached-item".to_string(), + ..Default::default() + }]) + }) + .await; + + let server = async { + Err(RepoError::Network { + message: "offline".to_string(), + }) + }; + + let got = HybridRepository::parallel_race(cache, server) + .await + .expect("a slow cache read must still be delivered when the server is gone"); + assert_eq!(got.len(), 1); + assert_eq!(got[0].id, "cached-item"); + } + + /// A cache that beats the deadline still short-circuits the server. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn a_fast_cache_hit_never_reaches_the_server() { + let cache = HybridRepository::cache_leg(async { + Ok(vec![MediaItem { + id: "fast".to_string(), + ..Default::default() + }]) + }) + .await; + + let server = async { + panic!("the server leg must not run on a cache hit"); + }; + + let got = HybridRepository::parallel_race(cache, server) + .await + .unwrap(); + assert_eq!(got[0].id, "fast"); + } + + /// When both sides fail, the server's error is what the caller sees. + /// + /// TRACES: UR-002 | DR-013 + #[tokio::test] + async fn a_failing_slow_cache_reports_the_server_error() { + let cache: CacheLeg> = HybridRepository::cache_leg(async { + tokio::time::sleep(Duration::from_millis(250)).await; + Err(RepoError::Database { + message: "disk gone".to_string(), + }) + }) + .await; + + let server = async { + Err(RepoError::Network { + message: "offline".to_string(), + }) + }; + + let err = HybridRepository::parallel_race(cache, server) + .await + .unwrap_err(); + assert!(matches!(err, RepoError::Network { .. }), "got {err:?}"); + } + /// Mock offline repository that tracks queries and saves struct MockOfflineRepo { items: Arc>>,