From 257dc5741d4b2426134eed18139ef65870755f78 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 28 Aug 2026 01:34:02 +0200 Subject: [PATCH] fix(plex): resolve movies stuck "in progress" on the 15-min incremental sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The incremental watch-sync (run_plex_watch_sync) learned a movie's watched transition only from the history feed + onDeck. When a movie is finished on Plex (commonly by scrubbing through the end credits) Plex sets viewCount>0, clears viewOffset and DROPS it from Continue-Watching, so onDeck no longer lists it (and onDeck can only ever ADD an in-progress row, never resolve one), while a scrub-to-end frequently emits no history session row. The stale local status='new', position>0 row then survived until the DAILY full reconcile — which is why movies kept showing "in progress". Add _reconcile_inprogress_movies: for each local in-progress MOVIE that Plex's onDeck no longer lists, read its current per-account viewCount/viewOffset with one metadata() call and apply it via the shared _pull_apply, flipping to watched when Plex says so. Bounded by a probe cap (100), most-recently-touched first; episodes are left to onDeck/history. No-signal rows are preserved (an un-watch stays the daily reconcile's job). Tests: watched-flip, onDeck+episode skip, and no-signal-preserved. --- backend/app/plex/watch_sync.py | 87 +++++++++++++++- backend/tests/test_plex_watch_sync.py | 137 +++++++++++++++++++++++++- 2 files changed, 221 insertions(+), 3 deletions(-) diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 85ef6399..0091b9db 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -439,15 +439,89 @@ def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int: return n +# How many stale in-progress movies we probe per incremental run (one metadata() call each). The set +# is normally tiny (movies started but no longer on Plex's Continue-Watching); the daily full reconcile +# is the backstop for anything beyond the cap. Most-recently-touched are probed first. +_MOVIE_PROBE_CAP = 100 + + +def _reconcile_inprogress_movies( + db: Session, plex: PlexClient, link: PlexLink, on_deck_ids: set[int] +) -> dict[str, int]: + """Close the movie "stuck in progress" gap that the history + onDeck feeds structurally miss. + + When a movie is finished on Plex (often by scrubbing through the end credits) Plex sets + ``viewCount>0``, clears ``viewOffset`` and DROPS the movie from Continue-Watching. So onDeck no + longer lists it (and onDeck can only ever ADD an in-progress row, never resolve one to watched — + see `_pull_apply`), and a scrub-to-end frequently emits no `/status/sessions/history` session row, + so the history feed misses it too. The stale local ``status='new', position>0`` row then survives + until the *daily* full reconcile. Movies are hit hardest: a movie is a single leaf with no + episode-rollup to mask the stuck row, and once watched it leaves onDeck entirely. + + Fix: for each local in-progress MOVIE that Plex's onDeck no longer lists, read the item's current + per-(token-)account ``viewCount``/``viewOffset`` with one cheap `metadata()` call and apply it via + the shared `_pull_apply` — flipping to watched when Plex says so. Returns ``{"watched", "in_progress", + "probed"}`` counts. Episodes are intentionally left to onDeck/history (they promote the next episode + and aren't affected). Note: like the rest of this job, `metadata()` reads the TOKEN account (the + owner) — a HTPC playing under a managed Plex user (accountID>1) is out of scope here as elsewhere.""" + conds = [ + PlexState.user_id == link.user_id, + PlexState.status.notin_(("watched", "hidden")), + PlexState.position_seconds > _PROGRESS_MIN_S, + PlexItem.kind == "movie", + ] + if on_deck_ids: + conds.append(PlexState.item_id.notin_(on_deck_ids)) + rows = db.execute( + select(PlexState, PlexItem.rating_key) + .join(PlexItem, PlexItem.id == PlexState.item_id) + .where(*conds) + .order_by(PlexState.progress_updated_at.desc().nullslast()) + .limit(_MOVIE_PROBE_CAP + 1) + ).all() + if len(rows) > _MOVIE_PROBE_CAP: + rows = rows[:_MOVIE_PROBE_CAP] + log.warning( + "in-progress movie probe hit the %d cap (user %s); reconcile catches the tail", + _MOVIE_PROBE_CAP, + link.user_id, + ) + out = {"watched": 0, "in_progress": 0, "probed": 0} + for st, rk in rows: + try: + meta = plex.metadata(rk) + except PlexError as e: + log.warning( + "movie watch-probe failed (user %s, item %s): %s", link.user_id, st.item_id, e + ) + continue + out["probed"] += 1 + if meta is None: + continue + res = _plex_watch_to_state( + meta.get("viewCount"), meta.get("viewOffset"), meta.get("lastViewedAt") + ) + if res is None: + # Plex holds no watch signal now (viewCount 0, no offset) — don't clear it here; an un-watch + # is the daily reconcile's job, and a Siftlode-only in-progress row must be preserved. + continue + status, pos, watched_at, prog_at = res + ch = _pull_apply(db, link.user_id, st.item_id, status, pos, watched_at or prog_at) + if ch: + out[ch] += 1 + return out + + def run_plex_watch_sync(db: Session) -> dict: """Scheduler job (default 15 min) — the cheap incremental pass. Pulls Plex-side changes since the high-water mark (watch history + on-deck, filtered to the owner account) into Siftlode, resolving a - divergence to Plex (the source of truth — see `_pull_apply`), then re-pushes any still-unsynced + divergence to Plex (the source of truth — see `_pull_apply`), reconciles stale in-progress movies + Plex has since finished (see `_reconcile_inprogress_movies`), then re-pushes any still-unsynced local states. Does NOT propagate un-watches (the history feed only carries positive views) — that's `run_plex_watch_reconcile`'s job.""" if not sysconfig.get_bool(db, "plex_enabled"): return {"skipped": "disabled"} - total = {"links": 0, "watched": 0, "in_progress": 0, "repushed": 0} + total = {"links": 0, "watched": 0, "in_progress": 0, "repushed": 0, "movie_probes": 0} for link in _active_sync_links(db): total["links"] += 1 try: @@ -467,10 +541,14 @@ def run_plex_watch_sync(db: Session) -> dict: if ch: total[ch] += 1 + # onDeck resolves resume-position changes AND tells us which items Plex still considers + # in-progress (needed by the movie reconcile below to know which local rows are stale). + on_deck_ids: set[int] = set() for row in plex.on_deck(): iid = rk_to_id.get(str(row.get("ratingKey") or "")) if iid is None: continue + on_deck_ids.add(iid) off = int(row.get("viewOffset") or 0) // 1000 if off < _PROGRESS_MIN_S: continue @@ -480,6 +558,11 @@ def run_plex_watch_sync(db: Session) -> dict: if ch: total[ch] += 1 + mv = _reconcile_inprogress_movies(db, plex, link, on_deck_ids) + total["watched"] += mv["watched"] + total["in_progress"] += mv["in_progress"] + total["movie_probes"] += mv["probed"] + total["repushed"] += _repush_dirty(db, plex, link) link.last_watch_sync_at = now db.commit() diff --git a/backend/tests/test_plex_watch_sync.py b/backend/tests/test_plex_watch_sync.py index da20bfee..c812d147 100644 --- a/backend/tests/test_plex_watch_sync.py +++ b/backend/tests/test_plex_watch_sync.py @@ -9,11 +9,12 @@ Two independent pieces: """ from datetime import datetime, timedelta, timezone +from types import SimpleNamespace from sqlalchemy import select from app.models import PlexItem, PlexLibrary, PlexState, User -from app.plex.watch_sync import _pull_apply +from app.plex.watch_sync import _pull_apply, _reconcile_inprogress_movies from app.routes.plex.playback import ( _PUSH_THROTTLE_S, _resume_push_superseded, @@ -33,6 +34,16 @@ def _seed_item_and_user(db): return item.id, user.id +class _FakePlex: + """Minimal PlexClient stand-in for the movie-probe tests: metadata(rating_key) → a leaf dict.""" + + def __init__(self, by_rk: dict[str, dict]): + self._by_rk = by_rk + + def metadata(self, rating_key: str, markers: bool = False): + return self._by_rk.get(str(rating_key)) + + def test_pull_plex_wins_over_diverging_synced_local_resume(db): """The core S2 change: a SYNCED local row (its state already reached Plex) that later diverges from Plex = a real external Plex-side change, so the incremental pull resolves to PLEX's position — Plex @@ -198,6 +209,130 @@ def test_pull_settles_flag_on_matching_dirty_row(db): assert st.synced_to_plex is True, "the dirty flag is settled so it isn't re-pushed endlessly" +def test_movie_probe_flips_stale_in_progress_to_watched(db): + """S1 core: a local in-progress MOVIE that Plex has since finished (viewCount>0) and dropped from + Continue-Watching flips to watched on the incremental pass — without waiting for the daily reconcile. + This is the "movie stuck in progress after scrubbing the credits" bug.""" + 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=True, # was created by a prior pull; the user then watched it on Plex + ) + ) + db.flush() + + # Plex now reports the movie watched (viewCount>0, no offset) and it is NOT on the deck. + plex = _FakePlex({"m1": {"viewCount": 1, "lastViewedAt": int(now.timestamp())}}) + out = _reconcile_inprogress_movies( + db, plex, SimpleNamespace(user_id=user_id), on_deck_ids=set() + ) + db.flush() + + assert out == {"watched": 1, "in_progress": 0, "probed": 1} + 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_movie_probe_skips_ondeck_and_episodes(db): + """The probe must NOT touch (a) a movie Plex still lists on the deck — it's legitimately in + progress — nor (b) an in-progress EPISODE (episodes ride onDeck/history + rollup). Only stale + in-progress movies are probed, and metadata() is called for those alone.""" + lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + db.add(lib) + db.flush() + on_deck_movie = PlexItem(rating_key="mDeck", library_id=lib.id, kind="movie", title="On deck") + episode = PlexItem(rating_key="e1", library_id=lib.id, kind="episode", title="An episode") + db.add_all([on_deck_movie, episode]) + user = User(email="probe2@example.com") + db.add(user) + db.flush() + now = datetime.now(timezone.utc) + for iid in (on_deck_movie.id, episode.id): + db.add( + PlexState( + user_id=user.id, + item_id=iid, + status="new", + position_seconds=300, + progress_updated_at=now, + synced_to_plex=True, + ) + ) + db.flush() + + # Plex would report both watched — but neither should be probed at all, so the fake returning + # "watched" must have NO effect. + plex = _FakePlex( + { + "mDeck": {"viewCount": 1, "lastViewedAt": int(now.timestamp())}, + "e1": {"viewCount": 1, "lastViewedAt": int(now.timestamp())}, + } + ) + out = _reconcile_inprogress_movies( + db, plex, SimpleNamespace(user_id=user.id), on_deck_ids={on_deck_movie.id} + ) + db.flush() + + assert out == {"watched": 0, "in_progress": 0, "probed": 0}, "neither item is probed" + for iid in (on_deck_movie.id, episode.id): + st = ( + db.execute( + select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == iid) + ) + .scalars() + .one() + ) + assert st.status == "new" and st.position_seconds == 300, "left untouched" + + +def test_movie_probe_leaves_row_when_plex_has_no_signal(db): + """If Plex reports no watch signal (viewCount 0, no offset) for a probed movie, the incremental pass + must NOT clear the local in-progress row — un-watch is the daily reconcile's job, and a Siftlode-only + in-progress state must survive.""" + 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=True, + ) + ) + db.flush() + + plex = _FakePlex({"m1": {"viewCount": 0}}) + out = _reconcile_inprogress_movies( + db, plex, SimpleNamespace(user_id=user_id), on_deck_ids=set() + ) + db.flush() + + assert out == {"watched": 0, "in_progress": 0, "probed": 1} + st = ( + db.execute( + select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + ) + .scalars() + .one() + ) + assert st.status == "new" and st.position_seconds == 300, "in-progress row preserved" + + def test_should_push_resume_scopes_by_user_and_item(): """The throttle window is keyed by the FULL (user, item) tuple: pushing for one key must not suppress a distinct user or a distinct item within the same window."""