From 24190e4c16376e8d0bb81da486b5fd231603a762 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 28 Oct 2026 18:56:12 +0100 Subject: [PATCH] fix(plex): a checkpoint sent just before the finish no longer lands on the watched row The 10 s tick in flight when the player finishes (or a refusal rollback whose state write in fact committed) wrote its mid-item position onto the freshly watched row and scheduled a resume push after the scrobble, so Plex showed the item in progress again. The checkpoint is now a conditional write that skips a row marked watched in the last 30 s and answers like a finish; a rewatch past that is checkpointed as before. Marking an item (or a season/show) watched also supersedes resume pushes still in flight, as the near-end finish already did. --- backend/app/plex/service.py | 39 +++++++++++++++----- backend/app/routes/plex/playback.py | 28 ++++++++++---- backend/tests/test_plex_finish.py | 57 +++++++++++++++++++++++++++++ backend/tests/test_plex_service.py | 6 ++- 4 files changed, 112 insertions(+), 18 deletions(-) diff --git a/backend/app/plex/service.py b/backend/app/plex/service.py index 215303e7..b090166d 100644 --- a/backend/app/plex/service.py +++ b/backend/app/plex/service.py @@ -6,9 +6,9 @@ The rest of the mirror's writes stay with the sync (`plex.sync`): a collection's after a write-back, its local cleanup after a delete. """ -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone -from sqlalchemy import CursorResult, delete, func, select, update +from sqlalchemy import CursorResult, delete, func, or_, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session @@ -271,6 +271,10 @@ def reorder_playlist(db: Session, pl: PlexPlaylist, rating_keys: list[str]) -> N # --- a user's watch state -------------------------------------------------------------------- +# How long after a watched mark a mid-item position is read as the finishing player's late tick +# (`checkpoint`): the tick is every 10 s, plus the request's own latency and a retry. +_LATE_CHECKPOINT_S = 30 + def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool: """Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end @@ -352,16 +356,33 @@ def checkpoint( now: datetime, *, dirty: bool, -) -> PlexState: - """Record a resume position. `dirty` marks it not yet mirrored to Plex — set before a push, so - the reconcile re-pushes it if the push fails and the push can settle it once Plex confirms.""" +) -> bool: + """Record a resume position, and say whether it was recorded. `dirty` marks it not yet mirrored + to Plex — set before a push, so the reconcile re-pushes it if the push fails and the push can + settle it once Plex confirms. + + Not onto a row marked watched in the last `_LATE_CHECKPOINT_S`: that position is the player's + tick sent just before its own finish (or a mark from another tab), and writing it made a watched + item in progress again — on Plex too, through the resume push it carried. Decided by the write + itself, since the mark can commit between any read here and the write. A rewatch is checkpointed + as before once the mark is that far behind it.""" st = state_row(db, user, item_id, st) - st.position_seconds = position - st.progress_updated_at = now + values: dict[str, object] = {"position_seconds": position, "progress_updated_at": now} if dirty: - st.synced_to_plex = False + values["synced_to_plex"] = False + not_just_finished = or_( + PlexState.status != "watched", + PlexState.watched_at.is_(None), + PlexState.watched_at <= now - timedelta(seconds=_LATE_CHECKPOINT_S), + ) + result = db.execute( + update(PlexState) + .where(PlexState.id == st.id, not_just_finished) + .values(values) + .execution_options(synchronize_session="fetch") + ) db.flush() - return st + return isinstance(result, CursorResult) and result.rowcount > 0 def forget_state(db: Session, st: PlexState) -> None: diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index e946d25d..043dc26a 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -234,6 +234,15 @@ def person_image(u: str, request: Request) -> Response: return imgcache.response(data, ct) +def _supersede_resume_pushes(user_id: int, item_id: int, now: datetime) -> None: + """A watched mark supersedes every resume push still in flight for the item: stamping it as the + latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep the map + small) removed the very guard, and an older resume push could then land at Plex AFTER the + scrobble and revive a resume point on a watched item. Eviction bounds the map.""" + with _push_throttle_lock: + _last_push_at[(user_id, item_id)] = now + + def _push_resume_guarded( user_id: int, item_id: int, @@ -329,12 +338,7 @@ def item_progress( near_end = duration > 0 and position > duration - _FINISH_MARGIN_S if near_end: flipped = service.mark_finished(db, user.id, it.id, now) - # The finish supersedes every resume push still in flight for this item: stamping it as the - # latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep - # the map small) removed the very guard, and an older resume push could then land at Plex - # AFTER the scrobble and revive a resume point on a watched item. Eviction bounds the map. - with _push_throttle_lock: - _last_push_at[(user.id, it.id)] = now + _supersede_resume_pushes(user.id, it.id, now) db.commit() # Only the checkpoint that flipped the row scrobbles: a later checkpoint of the same finished # item — or a second one racing this one — would otherwise bump Plex's viewCount again. @@ -352,7 +356,11 @@ def item_progress( push_now = _should_push_resume(user.id, it.id, now, final) # Marked 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. - service.checkpoint(db, user, it.id, st, position, now, dirty=push_now) + if not service.checkpoint(db, user, it.id, st, position, now, dirty=push_now): + # Marked watched a moment ago: this is the finishing player's late tick (see + # `service.checkpoint`), and the answer is the same one a near-end checkpoint gets. + db.commit() + return {"ok": True, "status": "watched", "position_seconds": 0} db.commit() if push_now: _push_watch( @@ -408,6 +416,8 @@ def item_state( _push_watch(background, db, user.id, it.id, rating_key, "unwatched") return {"ok": True, "status": "new"} service.set_status(db, user, it.id, st, status) + if status == "watched": + _supersede_resume_pushes(user.id, it.id, datetime.now(timezone.utc)) db.commit() if status == "watched" and prev_status != "watched": _push_watch(background, db, user.id, it.id, rating_key, "watched") @@ -427,6 +437,10 @@ def _bulk_state( always reads the committed marks.""" pushable = plex_watch.link_for_push(db, user_id) is not None to_push = service.bulk_state(db, user_id, eps, watched) + if watched: + now = datetime.now(timezone.utc) + for item_id, _rating_key, _action in to_push: + _supersede_resume_pushes(user_id, item_id, now) if pushable and to_push: # One coalesced background task for the whole batch (reuses a single session + Plex client), # not one task per episode — a 200-ep show mark is 1 task, not 200. diff --git a/backend/tests/test_plex_finish.py b/backend/tests/test_plex_finish.py index fea93d65..f264ba57 100644 --- a/backend/tests/test_plex_finish.py +++ b/backend/tests/test_plex_finish.py @@ -81,3 +81,60 @@ def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_u ) assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0} assert playback._resume_push_superseded(user.id, item.id, earlier) is True + + +def test_a_checkpoint_sent_before_the_finish_does_not_land_on_the_watched_row( + db, as_user, make_user, monkeypatch +): + """The 10 s tick posted at 2699 of 3000 s (not near the end) races the credits auto-skip's + "watched". When the mark commits first, the late checkpoint used to write its position onto the + watched row and schedule a resume push after the scrobble: Plex showed the item in progress, and + a Plex-authoritative pull imported it back.""" + user = make_user() + item = _item(db) + monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None) + client = as_user(user) + assert client.post("/api/plex/item/m1/state", json={"status": "watched"}).status_code == 200 + + r = client.post( + "/api/plex/item/m1/progress", json={"position_seconds": 2699, "duration_seconds": 3000} + ) + assert r.status_code == 200, r.text + assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0} + db.expire_all() + st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one() + assert (st.status, st.position_seconds) == ("watched", 0) + + +def test_a_rewatch_records_its_position_once_the_finish_is_behind_it( + db, as_user, make_user, monkeypatch +): + """The other half: a watched item played again is still checkpointed — only a position arriving + right on the heels of the mark is read as that mark's late tick.""" + user = make_user() + item = _item(db) + long_ago = datetime.now(timezone.utc) - timedelta(days=3) + db.add(PlexState(user_id=user.id, item_id=item.id, status="watched", watched_at=long_ago)) + db.commit() + monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None) + r = as_user(user).post( + "/api/plex/item/m1/progress", json={"position_seconds": 600, "duration_seconds": 3000} + ) + assert r.json() == {"ok": True, "position_seconds": 600} + db.expire_all() + st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one() + assert (st.status, st.position_seconds) == ("watched", 600) + + +def test_marking_watched_supersedes_a_resume_push_still_in_flight( + db, as_user, make_user, monkeypatch +): + """The same guard the near-end finish has: a resume push scheduled before the mark must not reach + Plex after its scrobble.""" + user = make_user() + item = _item(db) + earlier = datetime.now(timezone.utc) - timedelta(seconds=5) + assert playback._should_push_resume(user.id, item.id, earlier, final=True) + r = as_user(user).post("/api/plex/item/m1/state", json={"status": "watched"}) + assert r.status_code == 200, r.text + assert playback._resume_push_superseded(user.id, item.id, earlier) is True diff --git a/backend/tests/test_plex_service.py b/backend/tests/test_plex_service.py index 3858db86..181e0cc3 100644 --- a/backend/tests/test_plex_service.py +++ b/backend/tests/test_plex_service.py @@ -348,7 +348,8 @@ def test_an_early_checkpoint_drops_a_bare_state_and_zeroes_a_marked_one(db, make user = make_user() lib = _lib(db) bare, hid = _movie(db, lib, "bare"), _movie(db, lib, "hid") - st = service.checkpoint(db, user, bare.id, None, 120, NOW, dirty=True) + assert service.checkpoint(db, user, bare.id, None, 120, NOW, dirty=True) + st = queries.state(db, user, bare.id) assert (st.position_seconds, st.synced_to_plex) == (120, False) _state(db, user, hid, status="hidden", position_seconds=40) service.clear_progress(db, queries.state(db, user, bare.id), NOW) @@ -429,7 +430,8 @@ def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user) user = make_user() it = _movie(db, _lib(db), "m-race") _state(db, user, it, position_seconds=10) # the request that won, committed - st = service.checkpoint(db, user, it.id, None, 42, NOW, dirty=True) # read "none" before it + assert service.checkpoint(db, user, it.id, None, 42, NOW, dirty=True) # read "none" before it + st = queries.state(db, user, it.id) assert (st.position_seconds, st.synced_to_plex) == (42, False) st = service.set_status(db, user, it.id, None, "hidden") assert st.status == "hidden"