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

fix(sync): queue a watch position the server could not be told about (DR-154)

Browse Source 
sync_queue and its drain (DR-131) were built, tested and running, but the
stop-report path never fed them, so closing a video while the server was
unreachable lost the resume point outright.

HybridRepository::report_playback_stopped is a bare pass-through to the
online repository ("Playback reporting goes directly to server"), and on
failure the error surfaced to a frontend catch whose own comment read
"Server error - could queue, but for now just log". Both producers that
would have queued it -- PlaybackReporter::queue_for_sync in Rust and
syncService.queuePlaybackProgress on the frontend -- have no callers on
the playback path. user_data.pending_sync was dutifully set to 1, but
nothing drains that flag for positions the way favourites do (DR-120).

The command layer now enqueues a report_playback_stopped row whenever the
push fails; the existing drain already parses and replays that operation.

The pending row for an item is superseded in place rather than appended
to: progress is reported every 10s, so a server that stays down would
otherwise add a row per tick, all obsoleted by the newest -- the
unbounded queue DR-131 exists to prevent. Only pending/failed rows are
superseded, since reviving an abandoned row restores that same growing
counter. Queueing is best-effort and never fails the command: the local
position is already saved, so a failed queue write must not be reported
as a lost position.

Verified red->green in the jellytau-builder image: the four new tests
failed to compile (enqueue_playback_stopped not found) before the fix.
Full Rust suite passes (627 tests), cargo fmt clean, clippy adds no new
warnings; frontend suite (933) and svelte-check also clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dtourolle
2026-08-12 19:18:11 +02:00
co-authored by Claude Opus 5
parent 2d50744320
commit e4632bb2b2
4 changed files with 243 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/sync_drain.rs
+196
View File
@@ -355,6 +355,65 @@ async fn mark_failed(
Ok(())
}
/// Queue a watch position that could not be reported to the server.
///
/// The stop-report path pushed straight to the server and, on failure, logged
/// and dropped the position — so closing a video while the server was
/// unreachable lost the resume point outright, even though the queue and its
/// drain (DR-131) were built and running. This is the missing producer.
///
/// The pending row for an item is *replaced* rather than appended to. Progress
/// is reported every 10s, so a server that stays down would otherwise grow one
/// row per tick, all of them superseded by the newest — the unbounded queue
/// DR-131 exists to prevent. Only `pending`/`failed` rows are superseded:
/// an `abandoned` row has been given up on and must not be revived, and a
/// `completed` one is history.
///
/// TRACES: UR-025 | DR-154 | UT-151
pub async fn enqueue_playback_stopped(
db: &Arc<RusqliteService>,
user_id: &str,
item_id: &str,
position_ticks: i64,
) -> Result<(), String> {
let payload = format!(r#"{{"position_ticks": {}}}"#, position_ticks);
// Supersede an already-queued position for this item, keeping its place in
// the queue order (created_at) so a later item cannot overtake it.
let updated = db
.execute(Query::with_params(
"UPDATE sync_queue \
SET payload = ?, status = 'pending', error_message = NULL \
WHERE user_id = ? AND item_id = ? AND operation = 'report_playback_stopped' \
AND status IN ('pending', 'failed')",
vec![
QueryParam::String(payload.clone()),
QueryParam::String(user_id.to_string()),
QueryParam::String(item_id.to_string()),
],
))
.await?;
if updated == 0 {
db.execute(Query::with_params(
"INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at) \
VALUES (?, 'report_playback_stopped', ?, ?, 'pending', CURRENT_TIMESTAMP)",
vec![
QueryParam::String(user_id.to_string()),
QueryParam::String(item_id.to_string()),
QueryParam::String(payload),
],
))
.await?;
}
debug!(
"[SyncQueue] Queued unreported stop for {} at {} ticks",
item_id, position_ticks
);
Ok(())
}
/// Drain on every offline→online transition.
///
/// TRACES: UR-025 | DR-131
@@ -781,6 +840,143 @@ mod tests {
assert_eq!(row_state(&db, "theirs").await.0, "pending");
}
/// The bug DR-154 fixes: a stop-report that could not reach the server was
/// logged and dropped, so the watch position was lost outright. It must
/// land in the queue the drain already knows how to push.
///
/// TRACES: UR-025 | DR-154 | UT-151
#[tokio::test]
async fn test_failed_stop_report_is_queued_rather_than_dropped() {
let db = test_db();
enqueue_playback_stopped(&db, "u1", "ep1", 5_000_000_000)
.await
.unwrap();
// The very drain that already exists must be able to push it.
let sink = RecordingSink::new();
let report = drain_sync_queue(&db, &sink, "u1").await.unwrap();
assert_eq!(
sink.calls(),
vec![QueuedOp::PlaybackStopped {
item_id: "ep1".to_string(),
position_ticks: 5_000_000_000,
}]
);
assert_eq!(report.pushed, 1);
assert_eq!(report.remaining, 0);
}
/// Progress is reported every 10s, and a server that stays unreachable
/// would otherwise add a row per tick — an unbounded queue of positions
/// that are all superseded by the newest one. The pending row for an item
/// is replaced in place, so the queue holds the latest position only.
///
/// TRACES: UR-025 | DR-154 | UT-151
#[tokio::test]
async fn test_requeueing_the_same_item_supersedes_the_earlier_position() {
let db = test_db();
enqueue_playback_stopped(&db, "u1", "ep1", 1_000)
.await
.unwrap();
enqueue_playback_stopped(&db, "u1", "ep1", 2_000)
.await
.unwrap();
enqueue_playback_stopped(&db, "u1", "ep1", 3_000)
.await
.unwrap();
let sink = RecordingSink::new();
drain_sync_queue(&db, &sink, "u1").await.unwrap();
// One row, carrying the newest position — not three.
assert_eq!(
sink.calls(),
vec![QueuedOp::PlaybackStopped {
item_id: "ep1".to_string(),
position_ticks: 3_000,
}]
);
}
/// Distinct items must not collide — superseding is per item, not global.
///
/// TRACES: UR-025 | DR-154 | UT-151
#[tokio::test]
async fn test_requeueing_keeps_positions_for_different_items_apart() {
let db = test_db();
enqueue_playback_stopped(&db, "u1", "ep1", 1_000)
.await
.unwrap();
enqueue_playback_stopped(&db, "u1", "ep2", 2_000)
.await
.unwrap();
let sink = RecordingSink::new();
drain_sync_queue(&db, &sink, "u1").await.unwrap();
let mut calls = sink.calls();
calls.sort_by_key(|op| match op {
QueuedOp::PlaybackStopped { item_id, .. } => item_id.clone(),
_ => String::new(),
});
assert_eq!(
calls,
vec![
QueuedOp::PlaybackStopped {
item_id: "ep1".to_string(),
position_ticks: 1_000,
},
QueuedOp::PlaybackStopped {
item_id: "ep2".to_string(),
position_ticks: 2_000,
},
]
);
}
/// A row already abandoned (DR-131 gave up on it) must not be resurrected
/// by a later report — that would restore the queue-that-only-grows this
/// whole area exists to prevent. The new report is queued as its own row.
///
/// TRACES: UR-025 | DR-154 | UT-151
#[tokio::test]
async fn test_requeueing_does_not_revive_an_abandoned_row() {
let db = test_db();
seed(
&db,
&[(
"u1",
"report_playback_stopped",
"ep1",
Some(r#"{"position_ticks": 111}"#),
"abandoned",
MAX_SYNC_ATTEMPTS,
"2026-08-01T10:00:00Z",
)],
)
.await;
enqueue_playback_stopped(&db, "u1", "ep1", 999)
.await
.unwrap();
let sink = RecordingSink::new();
drain_sync_queue(&db, &sink, "u1").await.unwrap();
// Only the fresh row is pushed; the abandoned one stays abandoned.
assert_eq!(
sink.calls(),
vec![QueuedOp::PlaybackStopped {
item_id: "ep1".to_string(),
position_ticks: 999,
}]
);
}
/// Nothing queued means no server calls at all — a reconnect must not
/// generate traffic just because it happened.
///