diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 1710c32b..f763eb52 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -74,12 +74,14 @@ def _scan_plex_states( db: Session, plex: PlexClient ) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, bool, set[str]]: """Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at, - prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, - whether EVERY section's walk reached the server's total (or the server said it holds none), and - the keys of the sections walked — the only ones whose absences mean anything. + prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, the + keys of the sections SETTLED — walked to the server's total (or said empty by a server we mirror + nothing from) — and the keys of every section walked. Shared by the one-time import and the full reconcile (both need Plex's whole current picture). - The last one matters to the reconcile alone: it reads "absent from the scan" as "Plex un-watched it", which a walk cut short by a - Plex hiccup does not mean (see `_paginate`).""" + The settled keys matter to the reconcile alone: it reads "absent from the scan" as "Plex + un-watched it", which a walk cut short by a Plex hiccup does not mean (see `_paginate`). Settled + is decided PER SECTION: one section that never settles (a library emptied on Plex while its + mirrored items are kept) must not stop un-watches syncing in all the others (E32 review).""" item_id_by_rk = _rk_to_id(db) wanted = _enabled_section_keys(db) # The sections we mirror something from: a total of 0 from one of those contradicts the mirror. @@ -89,7 +91,7 @@ def _scan_plex_states( ).scalars() ) scanned = 0 - complete = True + settled: set[str] = set() walked: set[str] = set() out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {} for s in plex.sections(): @@ -113,13 +115,15 @@ def _scan_plex_states( ) if res is not None: out[item_id] = res - # A section the server SAYS is empty (total 0, nothing listed) is complete too: without this - # one genuinely empty library kept every run "incomplete" and un-watches never synced again. - # Believed only for a section we mirror nothing from: one whose items we hold answering 0 is - # a Plex that lost the library for a moment, not one that un-watched all of it (E32 review). - said_empty = progress.get("total") == 0 and scanned == before and key not in mirrored - complete = complete and (bool(progress.get("complete")) or said_empty) - return out, scanned, complete, walked + # A section the server SAYS is empty (total 0, nothing listed) settles too. Believed only for + # a section we mirror nothing from: one whose items we hold answering 0 is a Plex that lost + # the library for a moment, not one that un-watched all of it (E32 review) — so it is left + # unsettled, and its rows are not cleared. + listed = scanned > before + said_empty = progress.get("total") == 0 and not listed and key not in mirrored + if (progress.get("complete") and listed) or said_empty: + settled.add(key) + return out, scanned, settled, walked def import_owner_watch_state(db: Session, link: PlexLink) -> dict: @@ -133,7 +137,7 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: try: with PlexClient(db) as plex: # A partial scan is still safe here: the import only ever ADDS what Plex reports. - scanned_states, stats["scanned"], _complete, _walked = _scan_plex_states(db, plex) + scanned_states, stats["scanned"], _settled, _walked = _scan_plex_states(db, plex) # One row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert is # idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice # while the multi-second import runs) can't raise a duplicate-key error — it re-updates. @@ -612,19 +616,19 @@ def run_plex_watch_sync(db: Session) -> dict: return total -def _clear_unwatched(db: Session, user_id: int, plex_states: dict, walked: set[str]) -> int: +def _clear_unwatched(db: Session, user_id: int, plex_states: dict, settled: set[str]) -> int: """Clear the user's rows Plex no longer has a record for — the mirrored ones (Plex un-watched - them); hidden and dirty rows stay (a dirty one is re-pushed by `_repush_dirty`). Called only - after a COMPLETE scan: a walk a Plex hiccup cut short leaves out rows it merely failed to reach, - and clearing those un-watched them here until the next complete run (E32 S2). Only rows of the - sections the scan WALKED: an item of a library the allowlist leaves out is absent from every - scan, and was cleared on every run (E32 review).""" + them); hidden and dirty rows stay (a dirty one is re-pushed by `_repush_dirty`). Only rows of the + sections the scan SETTLED: a walk a Plex hiccup cut short leaves out rows it merely failed to + reach, and clearing those un-watched them here until the next complete run (E32 S2); an item of + a library the allowlist leaves out is absent from every scan, and was cleared on every run (E32 + review).""" n = 0 rows = db.execute( select(PlexState) .join(PlexItem, PlexItem.id == PlexState.item_id) .join(PlexLibrary, PlexLibrary.id == PlexItem.library_id) - .where(PlexState.user_id == user_id, PlexLibrary.plex_key.in_(walked)) + .where(PlexState.user_id == user_id, PlexLibrary.plex_key.in_(settled)) ) for st in rows.scalars().all(): if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex: @@ -652,7 +656,7 @@ def run_plex_watch_reconcile(db: Session) -> dict: try: with PlexClient(db) as plex: _resolve_account_id(db, plex, link) - plex_states, scanned, complete, walked = _scan_plex_states(db, plex) + plex_states, scanned, settled, walked = _scan_plex_states(db, plex) total["scanned"] += scanned for item_id, (status, pos, watched_at, prog_at) in plex_states.items(): @@ -660,15 +664,18 @@ def run_plex_watch_reconcile(db: Session) -> dict: if ch: total["applied"] += 1 - # A whole scan that listed nothing at all is not "Plex un-watched everything" either: - # every section can say 0 only when the server lost its libraries, never as news. - if complete and scanned: - total["cleared"] += _clear_unwatched(db, link.user_id, plex_states, walked) - else: + total["cleared"] += _clear_unwatched(db, link.user_id, plex_states, settled) + if unsettled := sorted(walked - settled): total["incomplete"] = total.get("incomplete", 0) + 1 log.warning( - "the Plex scan was cut short or listed nothing; nothing is cleared this run", - extra=logs.event("plex.watch.reconcile_incomplete", user_id=link.user_id), + "Plex sections %s were cut short or answered empty; their rows are not cleared" + " this run", + ", ".join(unsettled), + extra=logs.event( + "plex.watch.reconcile_incomplete", + user_id=link.user_id, + sections=unsettled, + ), ) total["repushed"] += _repush_dirty(db, plex, link) diff --git a/backend/tests/test_plex_watch_reconcile.py b/backend/tests/test_plex_watch_reconcile.py index 32993ddb..ce50368c 100644 --- a/backend/tests/test_plex_watch_reconcile.py +++ b/backend/tests/test_plex_watch_reconcile.py @@ -79,6 +79,7 @@ def world(db): return { by_id[s.item_id]: (s.status, s.position_seconds, s.synced_to_plex) for s in db.execute(select(PlexState).where(PlexState.user_id == user.id)).scalars() + if s.item_id in by_id # the five movies; a test's own extra items it reads itself } return state, rows @@ -157,20 +158,40 @@ def test_a_genuinely_empty_second_section_still_lets_the_reconcile_clear(db, mon assert out["cleared"] == 1 and "incomplete" not in out -def test_a_section_we_mirror_that_says_0_is_not_believed(db, monkeypatch, world): - # A total of 0 from the very section our items come from is a Plex that lost the library for a - # moment, not one that un-watched all of it (E32 review round 2): the shows section below says - # 0 and lists nothing — while we hold an episode from it. - state, rows = world - _local_states(state) +def _watched_episode_in_a_shows_section(db) -> int: + """A second mirrored section (key "2") holding one episode the owner has watched, synced.""" shows = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True) db.add(shows) db.flush() - db.add(PlexItem(rating_key="e1", library_id=shows.id, kind="episode", title="e1")) + ep = PlexItem(rating_key="e1", library_id=shows.id, kind="episode", title="e1") + db.add(ep) + db.flush() + user_id = db.execute(select(PlexLink.user_id)).scalar_one() + db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True)) + return ep.id + + +def _state_of(db, item_id: int) -> str | None: + db.expire_all() + st = db.execute(select(PlexState).where(PlexState.item_id == item_id)).scalar_one_or_none() + return st.status if st else None + + +def test_a_section_we_mirror_that_says_0_keeps_its_rows_and_does_not_hold_up_the_rest( + db, monkeypatch, world +): + # A total of 0 from a section our items come from is a Plex that lost the library for a moment, + # not one that un-watched all of it (E32 review round 2): its rows stay. And it is decided per + # section: before round 3 one such section (a library emptied on Plex, its mirrored items kept) + # made EVERY run incomplete, and un-watches in every other section never synced again. + state, rows = world + _local_states(state) + ep_id = _watched_episode_in_a_shows_section(db) db.commit() out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)])) - assert rows()["m1"] == ("watched", 0, True) - assert out["cleared"] == 0 and out["incomplete"] == 1 + assert _state_of(db, ep_id) == "watched" + assert "m1" not in rows() # the movies section was walked whole: Plex un-watched m1 + assert out["cleared"] == 1 and out["incomplete"] == 1 def test_a_library_the_allowlist_leaves_out_keeps_its_watch_states(db, monkeypatch, world): @@ -203,8 +224,8 @@ def test_a_library_the_allowlist_leaves_out_keeps_its_watch_states(db, monkeypat def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monkeypatch, world): - # Completeness is every section's, not the last one's: movies cut short (5 said, 3 handed), - # then a complete shows listing — m1 may simply not have been reached. + # Each section settles on its own walk: movies cut short (5 said, 3 handed), then a complete + # shows listing — m1 may simply not have been reached. state, rows = world _local_states(state) db.commit() @@ -217,10 +238,12 @@ def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monke def test_a_listing_without_a_total_is_never_complete(db, monkeypatch, world): state, rows = world _local_states(state) + ep_id = _watched_episode_in_a_shows_section(db) db.commit() out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], None)])) - assert rows()["m1"] == ("watched", 0, True) - assert out["cleared"] == 0 and out["incomplete"] == 1 + assert _state_of(db, ep_id) == "watched" # its section never said how much it holds + assert "m1" not in rows() + assert out["cleared"] == 1 and out["incomplete"] == 1 def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world):