diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index d08b81ca..bd23e609 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -17,7 +17,8 @@ from fastapi import ( ) from fastapi.responses import FileResponse from pydantic import BaseModel, ConfigDict, Field -from sqlalchemy import select +from sqlalchemy import select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app.auth import current_user, resolved_user_id @@ -295,6 +296,37 @@ def _push_watch( ) +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 + checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a + read-then-write let both see "not watched" and both scrobble. The transition is decided by the + write itself: an insert that did not conflict, or an update conditional on the old status — + under READ COMMITTED the second writer re-reads the committed row and matches nothing.""" + values = { + "status": "watched", + "watched_at": now, + "position_seconds": 0, + "progress_updated_at": now, + "synced_to_plex": False, + } + inserted = db.execute( + pg_insert(PlexState) + .values(user_id=user_id, item_id=item_id, **values) + .on_conflict_do_nothing(constraint="uq_plex_user_item") + .returning(PlexState.id) + ).first() + if inserted is not None: + return True + mine = (PlexState.user_id == user_id) & (PlexState.item_id == item_id) + if db.execute( + update(PlexState).where(mine, PlexState.status != "watched").values(values) + ).rowcount: + return True + # Already watched: a re-finish still restamps the row, as it always did — just no scrobble. + db.execute(update(PlexState).where(mine).values(values)) + return False + + class ItemProgressIn(BaseModel): # Finite only: JSON parsing here accepts `Infinity`/`NaN`, and the handler's `int()` turned # one into a 500. @@ -335,6 +367,22 @@ def item_progress( duration = int(body.duration_seconds) final = body.final now = datetime.now(timezone.utc) + near_end = duration > 0 and position > duration - _FINISH_MARGIN_S + if near_end: + flipped = _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 + 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. + if flipped: + _push_watch(background, db, user.id, it.id, rating_key, "watched") + return {"status": "watched", "position_seconds": 0} + st = ( db.execute( select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) @@ -342,28 +390,6 @@ def item_progress( .scalars() .first() ) - was_watched = st is not None and st.status == "watched" - - near_end = duration > 0 and position > duration - _FINISH_MARGIN_S - if near_end: - if st is None: - st = PlexState(user_id=user.id, item_id=it.id) - db.add(st) - st.status = "watched" - st.watched_at = now - st.position_seconds = 0 - st.progress_updated_at = now - st.synced_to_plex = False - # Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't - # retain a dead key for every finished item. - with _push_throttle_lock: - _last_push_at.pop((user.id, it.id), None) - db.commit() - # Immediate on the watched transition; skip a repeat scrobble if it was already watched - # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again). - if not was_watched: - _push_watch(background, db, user.id, it.id, rating_key, "watched") - return {"status": "watched", "position_seconds": 0} if position < _PROGRESS_MIN_S: if st is not None and st.status == "new": diff --git a/backend/tests/test_plex_finish.py b/backend/tests/test_plex_finish.py new file mode 100644 index 00000000..3958eb1a --- /dev/null +++ b/backend/tests/test_plex_finish.py @@ -0,0 +1,80 @@ +"""The near-end checkpoint that finishes a Plex item. + +- Exactly one scrobble per finish, even when two near-end checkpoints race (the 10 s tick and the + debounced `seeked` flush arrive together): the transition is decided by a conditional write, not a + read taken before it. +- The finish supersedes every resume push still in flight, so none can land at Plex after the + scrobble and revive a resume point on a watched item. +""" + +import threading +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from app.db import SessionLocal +from app.models import PlexItem, PlexLibrary, PlexState +from app.routes.plex import playback + + +def _item(db) -> PlexItem: + 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) + db.commit() + return item + + +def test_two_racing_finishes_flip_the_row_once(db, make_user): + user = make_user() + item = _item(db) + db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=5000)) + db.commit() + now = datetime.now(timezone.utc) + + first, second = SessionLocal(), SessionLocal() + try: + assert playback._mark_finished(first, user.id, item.id, now) is True # holds the row lock + out: list[bool] = [] + racer = threading.Thread( + target=lambda: out.append(playback._mark_finished(second, user.id, item.id, now)) + ) + racer.start() + racer.join(0.5) + assert racer.is_alive(), "the second writer must wait on the first one's row lock" + first.commit() + racer.join(5) + second.commit() + finally: + first.close() + second.close() + assert out == [False], "the second checkpoint saw the committed finish: no second scrobble" + + +def test_a_first_ever_finish_inserts_and_counts_once(db, make_user): + user = make_user() + item = _item(db) + now = datetime.now(timezone.utc) + assert playback._mark_finished(db, user.id, item.id, now) is True + db.commit() + assert playback._mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False + db.commit() + st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one() + assert (st.status, st.position_seconds) == ("watched", 0) + + +def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_user, monkeypatch): + user = make_user() + item = _item(db) + monkeypatch.setattr(playback.plex_stream, "touch_session", lambda key: None) + earlier = datetime.now(timezone.utc) - timedelta(seconds=5) + # A resume push was scheduled a moment ago and has not reached Plex yet. + assert playback._should_push_resume(user.id, item.id, earlier, final=True) + + r = as_user(user).post( + "/api/plex/item/m1/progress", json={"position_seconds": 2995, "duration_seconds": 3000} + ) + assert r.json() == {"status": "watched", "position_seconds": 0} + assert playback._resume_push_superseded(user.id, item.id, earlier) is True