diff --git a/backend/app/config.py b/backend/app/config.py index f01f7216..edb511eb 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -232,7 +232,9 @@ class Settings(BaseSettings): plex_sync_interval_min: int = 360 # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C): # a cheap history + on-deck pull plus a re-push of any states that failed their immediate push. - plex_watch_sync_interval_min: int = 30 + # 15 min (S2): with Plex now the source of truth (prompt write-back on the ~10s checkpoint + + # Plex-authoritative pull), a faster incremental pull keeps Siftlode close to Plex-side changes. + plex_watch_sync_interval_min: int = 15 # How often the FULL watch-state reconcile runs, in minutes (Phase C). A whole-section rescan # that catches what the incremental history feed can't — notably an un-watch done on the Plex # side — so it runs far less often (daily) than the incremental pass. diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 1d0755ed..1ce461f6 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -11,11 +11,14 @@ already pages) and upsert them into the owner's `plex_states`. Plex wins on over exist only in Siftlode are left untouched (union, Plex-authoritative on the intersection). Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→Siftlode via the watch -history + onDeck, last-write-wins) build on this and land in later ships. +history + onDeck) build on this and land in later ships. As of S2 Plex is the source of truth: the +player pushes the resume position back to Plex on the periodic checkpoint (not just the final flush), +and the incremental pull resolves a genuine divergence to Plex's state (see `_pull_apply`) rather than +running a last-write-wins comparison. """ import logging -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from sqlalchemy import func, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert @@ -293,12 +296,6 @@ def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) - # --- Phase C: incremental + full Plex ↔ Siftlode reconcile (scheduler jobs) ------------------------ -# Clock-skew / round-trip slack when comparing a Plex timestamp to a Siftlode one. A state we just -# pushed (Plex stamps viewedAt a beat after our watched_at) must not read as "newer on Plex" and get -# pulled back with a different shape. Genuine Plex-side edits are minutes from our writes — far -# outside this — so the window only absorbs skew, never real changes. -_LWW_TOLERANCE_S = 90 - def _active_sync_links(db: Session) -> list[PlexLink]: """Every link with two-way sync actually running (owner MVP = one row). Iterated uniformly so @@ -338,11 +335,6 @@ def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int: return acct_id -def _sift_ts(st: PlexState) -> datetime | None: - """The freshest Siftlode-side timestamp on a row, for last-write-wins.""" - return max([t for t in (st.watched_at, st.progress_updated_at) if t], default=None) - - def _same_state(st: PlexState | None, status: str, pos: int) -> bool: """Whether a row already matches a target (status, pos) — the core ping-pong guard: when nothing would change we skip the write entirely (no row churn, no timestamp bump, no re-pull).""" @@ -370,10 +362,13 @@ def _pull_apply( if st is not None and not st.synced_to_plex: st.synced_to_plex = True # already matches Plex → just settle the flag, no churn return None - # Conflict: keep Siftlode when it's newer than Plex (plus skew slack). - sts = _sift_ts(st) if st else None - if sts and plex_ts and plex_ts <= sts + timedelta(seconds=_LWW_TOLERANCE_S): - return None + # Plex-authoritative (S2): on a genuine divergence for an item Plex knows about, Plex wins — no + # last-write-wins tug-of-war. This is safe because prompt write-back (playback.py pushes the resume + # position to Plex on the ~10s checkpoint, not just on the final flush) keeps Plex current, so a + # local edit that's "newer" than Plex has almost always already been pushed and matched above by + # `_same_state`; a row that reaches here really is out of step with the source of truth. The + # `_same_state` + `synced_to_plex` short-circuit above still absorbs the freshly-pushed case (Plex + # echoes the same shape a beat later) so a just-pushed row never bounces. if st is None: st = PlexState(user_id=user_id, item_id=item_id) db.add(st) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 8d690ce8..7314087a 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -3,6 +3,7 @@ import hashlib import os import tempfile +import threading from datetime import datetime, timezone import httpx @@ -155,6 +156,29 @@ _PROGRESS_MIN_S = 5 _FINISH_MARGIN_S = 30 +# S2 write-back throttle: the resume position is now pushed to Plex on the periodic (non-final) +# checkpoint too — not only on the final pause/unmount flush — so Plex (the source of truth) stays +# current in near-real-time. To keep a single watch from spraying `/:/timeline` calls, a push is +# rate-limited to at most one per item per user every _PUSH_THROTTLE_S seconds; a `final` checkpoint +# (pause / seek / stop) always pushes immediately and resets the window. This is an in-memory, single- +# process rate-limiter (the deployment is one FastAPI process): after a restart the map is empty, so at +# most one extra push happens on the first checkpoint — harmless. It is NOT durable domain state. +_PUSH_THROTTLE_S = 10 +_last_push_at: dict[tuple[int, int], datetime] = {} +_push_throttle_lock = threading.Lock() + + +def _should_push_resume(user_id: int, item_id: int, now: datetime, final: bool) -> bool: + """Whether to mirror this resume checkpoint to Plex now. `final` always pushes; a periodic tick + pushes only when ≥_PUSH_THROTTLE_S has elapsed since the last push for this (user, item). Records + `now` as the last-push time whenever it returns True so the window is measured from the real push.""" + with _push_throttle_lock: + last = _last_push_at.get((user_id, item_id)) + if final or last is None or (now - last).total_seconds() >= _PUSH_THROTTLE_S: + _last_push_at[(user_id, item_id)] = now + return True + return False + @router.get("/person-image") def person_image(u: str, request: Request) -> Response: @@ -232,9 +256,10 @@ def item_progress( ) -> dict: """Checkpoint the resume position while playing (near the end → mark watched + clear it). - `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what - debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so - a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds.""" + S2 write-back: the resume position is mirrored to Plex (the source of truth) on the periodic + checkpoint too, not only on the final flush — throttled to ≥_PUSH_THROTTLE_S per item (see + `_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and + resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks.""" it = item_or_404(db, rating_key) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) @@ -280,10 +305,13 @@ def item_progress( db.add(st) st.position_seconds = position st.progress_updated_at = now - if final: + push_now = _should_push_resume(user.id, it.id, now, final) + if push_now: + # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so + # `push_state_to_plex` can settle synced_to_plex once Plex confirms. st.synced_to_plex = False db.commit() - if final: + if push_now: _push_watch( background, db, diff --git a/backend/tests/test_plex_watch_sync.py b/backend/tests/test_plex_watch_sync.py new file mode 100644 index 00000000..e4047066 --- /dev/null +++ b/backend/tests/test_plex_watch_sync.py @@ -0,0 +1,148 @@ +"""S2 watch-sync semantics: Plex is the source of truth. + +Two independent pieces: +- `_pull_apply` (DB lane): on a genuine divergence for an item Plex knows about, Plex now WINS — + the old "keep Siftlode when it's newer within a tolerance" tug-of-war is gone. The anti-bounce + short-circuit (`_same_state` + `synced_to_plex`) must still keep a freshly-pushed row from bouncing. +- `_should_push_resume` (pure): the in-memory write-back throttle — a `final` checkpoint always pushes, + a periodic one pushes at most once per _PUSH_THROTTLE_S seconds per (user, item). +""" + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from app.models import PlexItem, PlexLibrary, PlexState, User +from app.plex.watch_sync import _pull_apply +from app.routes.plex.playback import _PUSH_THROTTLE_S, _should_push_resume + + +def _seed_item_and_user(db): + lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + db.add(lib) + db.flush() + item = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="A Movie") + db.add(item) + user = User(email="watch@example.com") + db.add(user) + db.flush() + return item.id, user.id + + +def test_pull_plex_wins_over_newer_local_resume(db): + """The core S2 change: a locally-in-progress item whose local edit is NEWER than Plex's still + resolves to PLEX's position on the incremental pull — Plex is authoritative, no LWW.""" + item_id, user_id = _seed_item_and_user(db) + now = datetime.now(timezone.utc) + db.add( + PlexState( + user_id=user_id, + item_id=item_id, + status="new", + position_seconds=300, + progress_updated_at=now, # local touched just now — "newer" than Plex + synced_to_plex=False, + ) + ) + db.flush() + + # Plex reports a DIFFERENT (older) resume position for the same item. + changed = _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now - timedelta(seconds=60)) + db.flush() + + assert changed == "in_progress" + st = ( + db.execute( + select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + ) + .scalars() + .one() + ) + assert st.position_seconds == 120, "Plex's position must win, even though local was newer" + assert st.synced_to_plex is True + + +def test_pull_plex_watched_wins_over_local_in_progress(db): + """Plex marking the item watched propagates even over a fresh local in-progress row.""" + item_id, user_id = _seed_item_and_user(db) + now = datetime.now(timezone.utc) + db.add( + PlexState( + user_id=user_id, + item_id=item_id, + status="new", + position_seconds=300, + progress_updated_at=now, + synced_to_plex=False, + ) + ) + db.flush() + + changed = _pull_apply(db, user_id, item_id, "watched", 0, plex_ts=now - timedelta(seconds=60)) + db.flush() + + assert changed == "watched" + st = ( + db.execute( + select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + ) + .scalars() + .one() + ) + assert st.status == "watched" + assert st.position_seconds == 0 + + +def test_pull_no_bounce_on_matching_row(db): + """Anti-bounce short-circuit: when the row already matches what Plex reports (a just-pushed + resume), the pull makes NO change and only settles the synced flag — no churn, no re-pull.""" + item_id, user_id = _seed_item_and_user(db) + now = datetime.now(timezone.utc) + db.add( + PlexState( + user_id=user_id, + item_id=item_id, + status="new", + position_seconds=100, + progress_updated_at=now, + synced_to_plex=True, # freshly pushed + ) + ) + db.flush() + + # Plex echoes the same position a beat later (well within the _PROGRESS_MIN_S fuzz). + changed = _pull_apply(db, user_id, item_id, "new", 102, plex_ts=now + timedelta(seconds=2)) + db.flush() + + assert changed is None, "a matching row must not bounce" + st = ( + db.execute( + select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + ) + .scalars() + .one() + ) + assert st.position_seconds == 100, "no churn to the stored position" + + +def test_should_push_resume_final_always_pushes(): + """A final checkpoint (pause / seek / stop) always pushes, bypassing the throttle.""" + now = datetime.now(timezone.utc) + key = (9001, 9001) + assert _should_push_resume(*key, now, final=True) is True + # A second final a fraction of a second later still pushes. + assert _should_push_resume(*key, now + timedelta(seconds=1), final=True) is True + + +def test_should_push_resume_periodic_is_throttled(): + """Periodic (non-final) checkpoints push at most once per _PUSH_THROTTLE_S per (user, item).""" + now = datetime.now(timezone.utc) + key = (9002, 9002) + assert _should_push_resume(*key, now, final=False) is True # first tick pushes + # A tick within the window is suppressed. + assert ( + _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S - 1), final=False) + is False + ) + # Once the window elapses, the next tick pushes again. + assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True diff --git a/frontend/src/lib/plexPlayerHooks.ts b/frontend/src/lib/plexPlayerHooks.ts index 91baf42a..3705b049 100644 --- a/frontend/src/lib/plexPlayerHooks.ts +++ b/frontend/src/lib/plexPlayerHooks.ts @@ -102,6 +102,16 @@ export function usePlexProgress(opts: { const onPauseFlush = () => flush(true); const video = videoRef.current; video?.addEventListener("pause", onPauseFlush); + // A seek is a settled position too (S2: Plex is the source of truth, so it must see the new spot + // right away, not up to 10s later). Debounced so scrubbing — many `seeked` events in a row — pushes + // just ONCE, once the user lands. `final=true` bypasses the backend's ≥10s push-throttle, which is + // exactly why the debounce lives here: it bounds a scrub to one immediate Plex timeline call. + let seekTimer: number | undefined; + const onSeekFlush = () => { + window.clearTimeout(seekTimer); + seekTimer = window.setTimeout(() => flush(true), 800); + }; + video?.addEventListener("seeked", onSeekFlush); // Save on page unload too (F5/close/navigate): React effect cleanup does NOT run on a full reload, // so without this the resume point lags up to 10s — e.g. a seek right before F5 was lost. Uses a // keepalive beacon so the POST survives the unload (a final checkpoint — the user is leaving). @@ -112,8 +122,10 @@ export function usePlexProgress(opts: { window.addEventListener("pagehide", onHide); return () => { window.clearInterval(iv); + window.clearTimeout(seekTimer); window.removeEventListener("pagehide", onHide); video?.removeEventListener("pause", onPauseFlush); + video?.removeEventListener("seeked", onSeekFlush); flush(true); // Back (unmount) persists the position server-side via flush(), but reopening the same item in // the SAME session read react-query's CACHED detail — whose position_seconds predated this