dtourolle/jellytau
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 44 Wiki Activity

fix(repository): stop a slow cache read surfacing as a network error offline

Browse Source 
The cache leg of a cache-first query had a hard 100 ms deadline that
cancelled the read and reported it as a miss. That conflates "the cache
has nothing" with "the cache was slow", and the two want opposite
answers: offline the server leg fails too, so browsing surfaced a network
error over cached content that was sitting on disk.

It is not a rare race. The database is a single SQLite connection behind
a single mutex, so a concurrent write — a sync drain, a bulk
save_to_cache, a thumbnail write — blocks every read for its duration,
and 100 ms is easily exceeded on phone storage. It also compounded:
`spawn_blocking` work is not cancellable, so an abandoned query still ran
and still held the mutex, making the next one slower.

The deadline now bounds only the *fast path*. A cache query that misses it
keeps running on its own task, and when the server leg fails the race
waits that query out instead of discarding it. A cache that answers in
time still short-circuits the server exactly as before, and when both
sides genuinely fail the server's error is still what the caller sees.

`parallel_race`/`race_with_refresh` no longer need `&self`, so they are
associated functions and directly testable without constructing a
repository.

Not addressed here: one connection behind one mutex makes `PRAGMA
journal_mode = WAL` inert, since reads and writes fully serialise
regardless. A read pool is an architecture change and wants a spec.
This commit is contained in:
dtourolle
2026-09-07 22:24:45 +02:00
parent c72ca86865
commit bc92eb4dea
1 changed files with 215 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/repository/hybrid.rs
+215 -63
View File
@@ -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<T> {
/// The cache answered within [`HybridRepository::CACHE_FAST_PATH`].
Ready(Result<T, RepoError>),
/// Still running. Awaiting the handle yields the answer eventually.
Slow(tokio::task::JoinHandle<Result<T, RepoError>>),
}
impl<T> CacheLeg<T> {
/// Split into the fast-path answer and the still-running query. Exactly one
/// side is `Some`.
#[allow(clippy::type_complexity)]
fn split(
self,
) -> (
Option<Result<T, RepoError>>,
Option<tokio::task::JoinHandle<Result<T, RepoError>>>,
) {
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<T, F1, F2>(
&self,
cache_future: F1,
server_future: F2,
) -> Result<T, RepoError>
async fn parallel_race<T, F2>(cache: CacheLeg<T>, server_future: F2) -> Result<T, RepoError>
where
T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
F1: std::future::Future<Output = Result<T, RepoError>> + Send,
F2: std::future::Future<Output = Result<T, RepoError>> + 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<T, F1, F2, R>(
&self,
cache_future: F1,
async fn race_with_refresh<T, F2, R>(
cache: CacheLeg<T>,
server_future: F2,
on_cache_hit: R,
) -> Result<T, RepoError>
where
T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
F1: std::future::Future<Output = Result<T, RepoError>> + Send,
F2: std::future::Future<Output = Result<T, RepoError>> + 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<T>(
future: impl std::future::Future<Output = Result<T, RepoError>> + Send + 'static,
) -> CacheLeg<T>
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<T>(
&self,
future: impl std::future::Future<Output = Result<T, RepoError>> + Send,
) -> Result<T, RepoError> {
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<usize>) -> Result<Vec<MediaItem>, 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<Vec<Genre>, 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<PlaybackInfo, RepoError> {
@@ -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<Vec<MediaItem>> = 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<Mutex<Vec<MediaItem>>>,