fix(sync): mirror the server's watch position so resume crosses devices (DR-155)
The resume check reads the local user_data row and nothing else, but mirror_user_data -- the only path by which server UserData lands in that table -- mirrored is_favorite alone, and returned early whenever that field was absent, which is exactly the shape of an ordinary watched episode. playback_position_ticks was therefore write-only from this device's perspective: watch 40 minutes in a browser, open JellyTau, and it resumed from whatever this device last saw, or offered no resume at all. Same user-visible symptom as the Android bug fixed earlier on this branch, from an unrelated cause -- which is why resume read as broadly flaky rather than as one defect. The mirror now carries the position alongside the favourite flag under the same pending_sync = 0 conflict rule, so a local position still waiting to be pushed is never pulled backwards by a server that has not yet heard where we got to. COALESCE(excluded.x, user_data.x) keeps the stored value for a field the server omitted rather than nulling it, and a row with neither field is still skipped rather than fabricated as zeroes. Mirroring alone was not sufficient. get_item -- the call the player route makes -- returned the cached copy on a hit and never consulted the server, so for an already-cached item the mirror never ran. It now refreshes in the background on a cache hit via race_with_refresh, the reusable form of what get_items already did inline. That asymmetry is why browsing a season picked up other devices' state while opening the episode directly did not. The refreshed value lands for the next read; the cache-first race still answers immediately. The DR/total counts in extract-traces.test.ts are updated for DR-154 and DR-155 -- that edit is the test's intended signal that the CI gate's denominator is live rather than frozen. Verified red->green in the jellytau-builder image: both new tests failed before the fix. Full Rust suite passes (634), cargo fmt clean, clippy adds no new warnings; frontend suite (933) and svelte-check clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -319,6 +319,49 @@ impl HybridRepository {
|
||||
}
|
||||
}
|
||||
|
||||
/// [`Self::parallel_race`], plus a callback fired on the fast path so the
|
||||
/// caller can refresh the cache in the background.
|
||||
///
|
||||
/// A plain cache hit answers from data that may be arbitrarily old, which
|
||||
/// is right for the *response* and wrong for what it leaves behind: per-user
|
||||
/// state (watch positions, favourites) only reaches the local tables when a
|
||||
/// server result is cached, so a surface that always hits cache never learns
|
||||
/// what another device did. `get_items` had a bespoke version of this; this
|
||||
/// is the same idea, reusable.
|
||||
///
|
||||
/// The callback runs only on a cache hit — on a miss the server result is
|
||||
/// already being fetched and cached by the normal path.
|
||||
///
|
||||
/// TRACES: UR-002, UR-025 | DR-155
|
||||
async fn race_with_refresh<T, F1, F2, R>(
|
||||
&self,
|
||||
cache_future: F1,
|
||||
server_future: F2,
|
||||
on_cache_hit: R,
|
||||
) -> Result<T, RepoError>
|
||||
where
|
||||
T: MeaningfulContent + 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;
|
||||
|
||||
if let Ok(data) = &cache_result {
|
||||
if data.has_content() {
|
||||
debug!("[HybridRepo] Cache hit, returning immediately (refreshing in background)");
|
||||
on_cache_hit();
|
||||
return Ok(data.clone());
|
||||
}
|
||||
}
|
||||
|
||||
debug!("[HybridRepo] Cache miss, querying server");
|
||||
match server_future.await {
|
||||
Ok(data) => Ok(data),
|
||||
Err(e) => cache_result.or(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Simple timeout wrapper for cache queries (100ms timeout)
|
||||
///
|
||||
/// @req: DR-013 - Repository pattern (cache-first with timeout)
|
||||
@@ -489,6 +532,21 @@ impl MediaRepository for HybridRepository {
|
||||
}
|
||||
}
|
||||
|
||||
/// A single item, cache-first — and, on a cache hit, refreshed in the
|
||||
/// background so the stored copy keeps up with the server.
|
||||
///
|
||||
/// The background refresh is what carries per-user state home: caching an
|
||||
/// item runs `mirror_user_data`, which is the only path by which a watch
|
||||
/// position set on another device reaches the local `user_data` row the
|
||||
/// resume check reads. Without it a cache hit returned this device's own
|
||||
/// stale position forever and cross-device resume silently did nothing —
|
||||
/// `get_items` already refreshes this way, so browsing a season worked
|
||||
/// while opening the episode directly did not.
|
||||
///
|
||||
/// The refreshed value lands for the *next* read rather than this one: the
|
||||
/// point of the cache-first race is to answer immediately.
|
||||
///
|
||||
/// TRACES: UR-025, UR-002 | DR-155 | UT-152
|
||||
async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
|
||||
let offline = Arc::clone(&self.offline);
|
||||
let online = Arc::clone(&self.online);
|
||||
@@ -497,9 +555,32 @@ impl MediaRepository for HybridRepository {
|
||||
|
||||
let cache_future = self.cache_with_timeout(async move { offline.get_item(&item_id).await });
|
||||
|
||||
let online_for_refresh = Arc::clone(&self.online);
|
||||
let offline_for_save = Arc::clone(&self.offline);
|
||||
let refresh_id = item_id_clone.clone();
|
||||
let on_cache_hit = move || {
|
||||
tokio::spawn(async move {
|
||||
match online_for_refresh.get_item(&refresh_id).await {
|
||||
Ok(fresh) => {
|
||||
// `save_to_cache` files the row under a parent; the item's
|
||||
// own parent keeps it where a later listing expects it.
|
||||
let parent = fresh
|
||||
.parent_id
|
||||
.clone()
|
||||
.unwrap_or_else(|| "item".to_string());
|
||||
if let Err(e) = offline_for_save.save_to_cache(&parent, &[fresh]).await {
|
||||
debug!("[HybridRepo] Background item refresh failed: {:?}", e);
|
||||
}
|
||||
}
|
||||
Err(e) => debug!("[HybridRepo] Background item refresh unavailable: {:?}", e),
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
let server_future = async move { online.get_item(&item_id_clone).await };
|
||||
|
||||
self.parallel_race(cache_future, server_future).await
|
||||
self.race_with_refresh(cache_future, server_future, on_cache_hit)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_latest_items(
|
||||
|
||||
@@ -665,40 +665,63 @@ impl OfflineRepository {
|
||||
}
|
||||
|
||||
/// Mirror the server's per-user state for an item into the local
|
||||
/// `user_data` table, so favourites marked on any other client are visible
|
||||
/// here — including offline, where the local table is the only source.
|
||||
/// `user_data` table, so favourites marked — and positions watched — on any
|
||||
/// other client are visible here, including offline, where the local table
|
||||
/// is the only source.
|
||||
///
|
||||
/// The `WHERE user_data.pending_sync = 0` on the conflict clause is the
|
||||
/// conflict rule: a toggle made while the server was unreachable is still
|
||||
/// conflict rule: a change made while the server was unreachable is still
|
||||
/// waiting to be pushed, and must not be clobbered by the stale value the
|
||||
/// server is still reporting. Rows carrying no favourite state are skipped
|
||||
/// entirely rather than written as `0`, which would fabricate an
|
||||
/// "unfavourited" record from an endpoint that simply omits `UserData`.
|
||||
/// server is still reporting. For a position that means it is never pulled
|
||||
/// *backwards* by a server that has not yet heard where we got to.
|
||||
///
|
||||
/// TRACES: UR-069 | DR-114 | UT-102
|
||||
/// Each field is mirrored only when the server actually reported it —
|
||||
/// `COALESCE(excluded.x, user_data.x)` keeps the stored value for anything
|
||||
/// absent, and a row with neither field is skipped outright rather than
|
||||
/// written as zeroes, which would fabricate an "unfavourited, unwatched"
|
||||
/// record from an endpoint that simply omits `UserData`.
|
||||
///
|
||||
/// The position half is what makes cross-device resume work: the resume
|
||||
/// check reads this table alone, so before it was mirrored an item watched
|
||||
/// elsewhere resumed from whatever *this* device last saw, or not at all.
|
||||
///
|
||||
/// TRACES: UR-025, UR-069 | DR-114, DR-155 | UT-102, UT-152
|
||||
async fn mirror_user_data(&self, item: &MediaItem, now: &str) -> Result<(), RepoError> {
|
||||
let Some(is_favorite) = item.user_data.as_ref().and_then(|ud| ud.is_favorite) else {
|
||||
let user_data = item.user_data.as_ref();
|
||||
let is_favorite = user_data.and_then(|ud| ud.is_favorite);
|
||||
let position_ticks = user_data.and_then(|ud| ud.playback_position_ticks);
|
||||
|
||||
// Nothing the server actually told us about — do not invent a row.
|
||||
if is_favorite.is_none() && position_ticks.is_none() {
|
||||
return Ok(());
|
||||
};
|
||||
}
|
||||
|
||||
let query = Query::with_params(
|
||||
"INSERT INTO user_data (user_id, item_id, is_favorite, synced_at, pending_sync)
|
||||
VALUES (?1, ?2, ?3, ?4, 0)
|
||||
"INSERT INTO user_data
|
||||
(user_id, item_id, is_favorite, playback_position_ticks, synced_at, pending_sync)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, 0)
|
||||
ON CONFLICT(user_id, item_id) DO UPDATE SET
|
||||
is_favorite = excluded.is_favorite,
|
||||
is_favorite = COALESCE(excluded.is_favorite, user_data.is_favorite),
|
||||
playback_position_ticks = COALESCE(
|
||||
excluded.playback_position_ticks, user_data.playback_position_ticks),
|
||||
synced_at = excluded.synced_at
|
||||
WHERE user_data.pending_sync = 0",
|
||||
vec![
|
||||
QueryParam::String(self.user_id.clone()),
|
||||
QueryParam::String(item.id.clone()),
|
||||
QueryParam::Int(if is_favorite { 1 } else { 0 }),
|
||||
is_favorite
|
||||
.map(|f| QueryParam::Int(if f { 1 } else { 0 }))
|
||||
.unwrap_or(QueryParam::Null),
|
||||
position_ticks
|
||||
.map(QueryParam::Int64)
|
||||
.unwrap_or(QueryParam::Null),
|
||||
QueryParam::String(now.to_string()),
|
||||
],
|
||||
);
|
||||
|
||||
// A missing item row (FK) is not fatal here — the favourite mirror is
|
||||
// best-effort metadata, and failing the whole cache write over it would
|
||||
// break browsing.
|
||||
// A missing item row (FK) is not fatal here — the mirror is best-effort
|
||||
// metadata, and failing the whole cache write over it would break
|
||||
// browsing.
|
||||
if let Err(e) = self.db_service.execute(query).await {
|
||||
debug!(
|
||||
"[OfflineRepo] user_data mirror skipped for {}: {}",
|
||||
@@ -4438,4 +4461,142 @@ mod tests {
|
||||
"an unsynced local toggle must survive a cache write"
|
||||
);
|
||||
}
|
||||
|
||||
/// UT-152 — the server's watch position is mirrored locally, so an item
|
||||
/// watched on another device resumes here.
|
||||
///
|
||||
/// The resume check reads only the local `user_data` row, and the mirror
|
||||
/// previously carried `is_favorite` alone — so a position set on any other
|
||||
/// client never reached this device and cross-device resume silently did
|
||||
/// nothing. The `pending_sync` guard is the same conflict rule favourites
|
||||
/// use: a local position still waiting to be pushed must not be pulled
|
||||
/// backwards by the stale value the server is still reporting.
|
||||
///
|
||||
/// TRACES: UR-025, UR-069 | DR-155 | UT-152
|
||||
#[tokio::test]
|
||||
async fn test_save_to_cache_mirrors_playback_position_without_clobbering_pending() {
|
||||
use crate::storage::db_service::DatabaseService;
|
||||
let db_service = create_test_db();
|
||||
let repo = OfflineRepository::new(
|
||||
db_service.clone(),
|
||||
"test-server".to_string(),
|
||||
"test-user".to_string(),
|
||||
);
|
||||
|
||||
let position = |id: &'static str| {
|
||||
let db = db_service.clone();
|
||||
async move {
|
||||
db.query_optional(
|
||||
Query::with_params(
|
||||
"SELECT playback_position_ticks, pending_sync FROM user_data \
|
||||
WHERE user_id = ? AND item_id = ?",
|
||||
vec![
|
||||
QueryParam::String("test-user".to_string()),
|
||||
QueryParam::String(id.to_string()),
|
||||
],
|
||||
),
|
||||
|row| Ok((row.get::<_, Option<i64>>(0)?, row.get::<_, Option<i32>>(1)?)),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
};
|
||||
|
||||
// Watched 20 minutes into this episode on another device.
|
||||
let mut watched = create_test_item("ep-1", "Watched Elsewhere", None);
|
||||
watched.user_data = Some(UserData {
|
||||
playback_position_ticks: Some(12_000_000_000),
|
||||
..Default::default()
|
||||
});
|
||||
// No user data at all — must not fabricate a position of 0.
|
||||
let untouched = create_test_item("ep-2", "No User Data", None);
|
||||
|
||||
repo.save_to_cache("parent-1", &[watched.clone(), untouched])
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
position("ep-1").await,
|
||||
Some((Some(12_000_000_000), Some(0))),
|
||||
"the server's position should be mirrored as synced"
|
||||
);
|
||||
assert_eq!(
|
||||
position("ep-2").await,
|
||||
None,
|
||||
"an item without UserData should not get an invented position"
|
||||
);
|
||||
|
||||
// Watched further here while the server was unreachable: pending_sync = 1.
|
||||
db_service
|
||||
.execute(Query::with_params(
|
||||
"UPDATE user_data SET playback_position_ticks = ?, pending_sync = 1 \
|
||||
WHERE user_id = ? AND item_id = ?",
|
||||
vec![
|
||||
QueryParam::Int64(30_000_000_000),
|
||||
QueryParam::String("test-user".to_string()),
|
||||
QueryParam::String("ep-1".to_string()),
|
||||
],
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// The server still reports the older position; caching must not win.
|
||||
repo.save_to_cache("parent-1", &[watched]).await.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
position("ep-1").await,
|
||||
Some((Some(30_000_000_000), Some(1))),
|
||||
"an unsynced local position must not be pulled backwards"
|
||||
);
|
||||
}
|
||||
|
||||
/// UT-152 — a server item carrying *only* a position (no favourite flag)
|
||||
/// still gets mirrored.
|
||||
///
|
||||
/// The mirror used to return early whenever `is_favorite` was absent, which
|
||||
/// is exactly the shape of an ordinary watched episode: Jellyfin reports
|
||||
/// `PlaybackPositionTicks` with no favourite state. That early return is why
|
||||
/// the position never landed.
|
||||
///
|
||||
/// TRACES: UR-025 | DR-155 | UT-152
|
||||
#[tokio::test]
|
||||
async fn test_position_is_mirrored_even_when_no_favourite_flag_is_present() {
|
||||
use crate::storage::db_service::DatabaseService;
|
||||
let db_service = create_test_db();
|
||||
let repo = OfflineRepository::new(
|
||||
db_service.clone(),
|
||||
"test-server".to_string(),
|
||||
"test-user".to_string(),
|
||||
);
|
||||
|
||||
let mut watched = create_test_item("ep-3", "Position Only", None);
|
||||
watched.user_data = Some(UserData {
|
||||
is_favorite: None,
|
||||
playback_position_ticks: Some(9_000_000_000),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
repo.save_to_cache("parent-1", &[watched]).await.unwrap();
|
||||
|
||||
let stored = db_service
|
||||
.query_optional(
|
||||
Query::with_params(
|
||||
"SELECT playback_position_ticks FROM user_data \
|
||||
WHERE user_id = ? AND item_id = ?",
|
||||
vec![
|
||||
QueryParam::String("test-user".to_string()),
|
||||
QueryParam::String("ep-3".to_string()),
|
||||
],
|
||||
),
|
||||
|row| row.get::<_, Option<i64>>(0),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
stored,
|
||||
Some(Some(9_000_000_000)),
|
||||
"a position with no favourite flag must still be mirrored"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user