From bc79720e81a1d4e2f4155f7824223bcaa0e7ed00 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 18 Aug 2026 22:47:25 +0200 Subject: [PATCH] fix(e3): address /code-review high findings (5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-1 review fixes on the E3 branch: - prune partial-fetch (correctness): `_paginate` now reports whether it reached the server's reported `total`; `_prune_stale` prunes only on a COMPLETE non-empty listing. A truncated fetch (a mid-sequence empty/short page before total — a transient Plex hiccup) no longer deletes the rows it merely failed to list. New test: a listing that claims 5 but returns 1 prunes nothing. - `_prune_stale` (robustness): `stats.get("pruned", 0) + 1` — no KeyError when a caller's stats dict omits the key. - service transitions (conventions): pause/resume/redownload use the JobStatus StrEnum instead of loose "queued"/"paused"/"error" literals, matching cancel. - ffmpeg progress writer (correctness): a stuck-percent heartbeat now runs on its OWN 1/interval budget, so it can't consume the progress-write slot and delay a real percent change by up to an interval. - backend/app/README.md (conventions): reword the transaction-ownership rule to quote and extend CLAUDE.md's canonical ("handler/job-wrapper commits; helpers flush; only quota owns its session") instead of stating a contradicting "helpers MAY commit" carve-out. ruff clean, full DB-lane 364 pass. --- backend/app/README.md | 18 ++++---- backend/app/downloads/pipeline.py | 21 +++++---- backend/app/downloads/service.py | 12 ++--- backend/app/plex/sync.py | 49 ++++++++++++++------- backend/tests/test_plex_query_conversion.py | 4 +- backend/tests/test_plex_sync_prune.py | 31 +++++++++++++ 6 files changed, 96 insertions(+), 39 deletions(-) diff --git a/backend/app/README.md b/backend/app/README.md index e8aa3c23..db130a7e 100644 --- a/backend/app/README.md +++ b/backend/app/README.md @@ -31,15 +31,17 @@ pre-abstract for a single caller; don't leave shared logic duplicated once there ## Transaction ownership -**A helper that mutates rows as part of a larger operation only `db.add()`s / stages its change; the -CALLER commits, so the whole operation is one transaction.** `audit.record()` is the reference: it adds -the audit row and the caller commits it together with the mutation it records, so an audit entry can -never exist without its change (or vice-versa). +This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush +(only `quota` owns its session)"** — so read it that way, not as an exception to it. The commit belongs +to the **operation owner**; anything called *inside* another transaction only stages its change +(`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction. +`audit.record()` is the reference: it adds the audit row and the caller commits it together with the +mutation it records, so an audit entry can never exist without its change (or vice-versa). -- A self-contained transition that owns its whole unit of work MAY commit (e.g. `service.cancel/…` - each commit their one transition; the worker's short-lived `jobdb` helpers each commit their own - write). But a helper called *in the middle* of a caller's transaction must not commit — that splits - the caller's atomic operation in two. +- The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a + whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't + a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes, + the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes. - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM attribute — see `service.decr_ref_count`. diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py index 2b0e0e40..72eafc7b 100644 --- a/backend/app/downloads/pipeline.py +++ b/backend/app/downloads/pipeline.py @@ -193,21 +193,26 @@ def _cancel_poll(job_id: int, interval_s: float = _CANCEL_POLL_INTERVAL_S, *, no def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic): - """`on_progress` for run_ffmpeg: persist at most one row write per interval, and only when the - whole-percent value actually changed (the bar shows integers).""" - allow = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now) + """`on_progress` for run_ffmpeg: persist a progress row write at most once per interval and only + when the whole-percent value changed (the bar shows integers), plus a separate once-per-interval + heartbeat while the percent is stuck (so a slow re-encode isn't mistaken for a stall).""" + # Two independent 1/interval budgets so a heartbeat never consumes the progress-write slot — a + # percent change right after a heartbeat must still write immediately, not wait out the interval. + allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now) + allow_beat = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now) state = {"pct": -1} def write(pct: float) -> None: - if not allow(): - return whole = int(pct) if whole == state["pct"]: # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264 # or a crop on the CPU-only host) can sit on one percent for minutes. Heartbeat the row's - # updated_at so the stall watchdog doesn't kill a live job as "stalled" (B-sub-4), without - # rewriting the unchanged progress bar. - _touch_job(job_id) + # updated_at (on its OWN budget) so the stall watchdog doesn't kill a live job as "stalled" + # (B-sub-4), without rewriting the unchanged bar or delaying the next real write. + if allow_beat(): + _touch_job(job_id) + return + if not allow_write(): return state["pct"] = whole _set_job(job_id, progress=whole, phase=phase) diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index e496c6f2..46432ea7 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -318,8 +318,8 @@ def release_asset(db: Session, job: DownloadJob) -> None: def pause(db: Session, job: DownloadJob) -> None: """Pause a queued/running job (a running worker aborts cooperatively via the cancel poll); a no-op in any other state.""" - if job.status in ("queued", "running"): - job.status = "paused" + if job.status in (JobStatus.QUEUED, JobStatus.RUNNING): + job.status = JobStatus.PAUSED db.commit() @@ -327,16 +327,16 @@ def resume(db: Session, job: DownloadJob) -> None: """Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would instantly re-inherit the asset's stale error (the worker short-circuits an errored asset).""" - if job.status in ("paused", "error"): + if job.status in (JobStatus.PAUSED, JobStatus.ERROR): job.progress = 0 job.error = None job.error_code = None if job.asset_id: asset = db.get(MediaAsset, job.asset_id) - if asset and asset.status == "error": + if asset and asset.status == "error": # MediaAsset.status (no enum), not a JobStatus asset.status = "pending" asset.error = None - job.status = "queued" + job.status = JobStatus.QUEUED db.commit() @@ -370,7 +370,7 @@ def redownload(db: Session, job: DownloadJob) -> None: release_asset(db, job) incr_ref_count(db, asset.id) job.asset_id = asset.id - job.status = "queued" + job.status = JobStatus.QUEUED job.progress = 0 job.phase = None job.error = None diff --git a/backend/app/plex/sync.py b/backend/app/plex/sync.py index a914e30b..303a8366 100644 --- a/backend/app/plex/sync.py +++ b/backend/app/plex/sync.py @@ -284,8 +284,14 @@ def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary return lib -def _paginate(plex: PlexClient, key: str, item_type: int): +def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None = None): + """Yield a section's items page by page. If `progress` is given, set `progress["complete"]` True + ONLY when the walk reached the server's reported `total` — a truncated fetch (an empty/short page + mid-sequence, e.g. a transient Plex hiccup, before `total` is reached) leaves it False so the + caller can skip pruning. Pruning on a partial listing would delete rows the listing merely failed + to mention (the metadata is re-syncable, but that still cascades watch-state).""" start = 0 + total = None while True: items, total = plex.section_items(key, item_type, start, _PAGE) if not items: @@ -297,6 +303,10 @@ def _paginate(plex: PlexClient, key: str, item_type: int): break if len(items) < _PAGE: break + if progress is not None: + # A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows + # (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative. + progress["complete"] = bool(total) and start >= total # The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this @@ -356,18 +366,21 @@ def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None: row.playable = classify_playable(mf.get("codec_video"), mf.get("codec_audio"), mf.get("container")) -def _prune_stale(db: Session, existing: dict, seen: set[str], stats: dict) -> None: +def _prune_stale( + db: Session, existing: dict, seen: set[str], stats: dict, *, complete: bool +) -> None: """Delete stored rows whose rating_key is no longer in the freshly-listed `seen` set — the titles - that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED: an - empty `seen` (a section that listed zero items — a real empty library, or a transient Plex glitch - we can't tell apart) prunes NOTHING, so a hiccup can never wipe a whole library. Callers pass the - rating_key -> row map they already built for the upsert.""" - if not seen: + that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED on a + COMPLETE, non-empty listing: a truncated fetch (`complete` False — a mid-sequence empty/short page + before the server's total) or a zero-item listing prunes NOTHING, so a transient Plex hiccup can + never wipe a library over rows it merely failed to list. Callers pass the rating_key -> row map + they already built for the upsert.""" + if not complete or not seen: return for rk, row in existing.items(): if rk not in seen: db.delete(row) - stats["pruned"] += 1 + stats["pruned"] = stats.get("pruned", 0) + 1 db.flush() @@ -379,7 +392,8 @@ def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) - ).scalars() } seen: set[str] = set() - for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE): + progress: dict = {} + for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE, progress): rk = str(meta.get("ratingKey") or "") if not rk: continue @@ -390,7 +404,7 @@ def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) - seen.add(rk) stats["movies"] += 1 db.flush() - _prune_stale(db, existing, seen, stats) + _prune_stale(db, existing, seen, stats, complete=progress.get("complete", False)) def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None: @@ -402,7 +416,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> ).scalars() } shows_seen: set[str] = set() - for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW): + shows_progress: dict = {} + for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW, shows_progress): rk = str(meta.get("ratingKey") or "") if not rk: continue @@ -433,7 +448,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> ).scalars() } seasons_seen: set[str] = set() - for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON): + seasons_progress: dict = {} + for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON, seasons_progress): rk = str(meta.get("ratingKey") or "") sid = show_id.get(str(meta.get("parentRatingKey") or "")) if not rk or sid is None: @@ -459,7 +475,8 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> ).scalars() } episodes_seen: set[str] = set() - for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE): + episodes_progress: dict = {} + for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE, episodes_progress): rk = str(meta.get("ratingKey") or "") if not rk: continue @@ -478,9 +495,9 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> # Prune leaf-first (episodes -> seasons -> shows) so a stale row is deleted directly, before its # parent's cascade would reach it — avoids deleting the same ORM object twice. Each level is # independently guarded on a non-empty listing by _prune_stale. - _prune_stale(db, episodes, episodes_seen, stats) - _prune_stale(db, seasons, seasons_seen, stats) - _prune_stale(db, shows, shows_seen, stats) + _prune_stale(db, episodes, episodes_seen, stats, complete=episodes_progress.get("complete", False)) + _prune_stale(db, seasons, seasons_seen, stats, complete=seasons_progress.get("complete", False)) + _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False)) def _sync_overlay( diff --git a/backend/tests/test_plex_query_conversion.py b/backend/tests/test_plex_query_conversion.py index d11392b9..2eea5bf0 100644 --- a/backend/tests/test_plex_query_conversion.py +++ b/backend/tests/test_plex_query_conversion.py @@ -149,7 +149,9 @@ class TestRollup: def _fake_paginate(by_type): - def _p(_plex, _key, typ): + def _p(_plex, _key, typ, progress=None): + if progress is not None: + progress["complete"] = True # the fake returns the whole list — a complete listing return list(by_type.get(typ, [])) return _p diff --git a/backend/tests/test_plex_sync_prune.py b/backend/tests/test_plex_sync_prune.py index d80f37eb..d6a71aac 100644 --- a/backend/tests/test_plex_sync_prune.py +++ b/backend/tests/test_plex_sync_prune.py @@ -37,6 +37,21 @@ class _StubPlex: return items[start : start + size], len(items) +class _TruncatedPlex: + """Simulates a truncated listing (a transient Plex hiccup): it reports a large `total` but returns + only one short page, so pagination stops BEFORE reaching total. Prune must treat this as an + incomplete fetch and delete nothing — the un-listed rows are present, just not mentioned.""" + + def __init__(self, items: list[dict], claimed_total: int): + self._items = items + self._total = claimed_total + + def section_items(self, key, item_type, start, size): + if start == 0: + return list(self._items), self._total # one short page, total says there's more + return [], self._total + + def _seed_movies(db): lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) db.add(lib) @@ -88,6 +103,22 @@ def test_empty_listing_prunes_nothing(db): assert stats["pruned"] == 0 +def test_truncated_listing_prunes_nothing(db): + """A partial fetch (Plex reports 5 items but the listing returns only 1 then stops) must NOT prune + the 4 it failed to mention — the completeness guard, not just the empty-listing guard.""" + lib, m1_id, _m2_id, _user_id = _seed_movies(db) + # Only m1 is listed, but total claims 5 → the walk never reaches total → incomplete → no prune. + stub = _TruncatedPlex([{"ratingKey": "m1", "title": "Kept Movie"}], claimed_total=5) + stats = {"movies": 0, "pruned": 0} + + _sync_movies(db, stub, lib, stats) + db.commit() + + live = {r.rating_key for r in db.execute(select(PlexItem)).scalars()} + assert live == {"m1", "m2"}, "a truncated listing must not prune the un-listed rows" + assert stats["pruned"] == 0 + + def test_show_prune_removes_stale_show_season_and_episode(db): lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True) db.add(lib)