fix(r5): address the third /code-review high round — 7 findings

- the sweep guard anchors the rating_key to digits: the round-2 widening had
  degenerated to "anything ending in _<digits>" (backup_2024, release_10)
- RSS counts apply-side failures too, so a read-only database no longer reports
  a wholly failed poll as "ok — new=0, failed=0"
- the player reads/writes its volume directly (readAccount/writeAccount): the
  persisted-object hook rendered nothing and cost a second write per settle
- new tests: the sweep guard's accept/reject table, and run_ffmpeg's threading
  (drain, cancel while silent, stall watchdog, stderr tail on a nonzero exit)
- the breaker sentinel is checked one way (isinstance) at all four sites
- run_rss_poll is typed dict[str, int]

All three new test groups were mutation-checked: reverting each fix turns them
red (or, for the cancel/stall pair, hangs — which is the bug itself).

backend/app/plex/stream.py

@@ -35,12 +35,18 @@ _HLS_ROOT = Path(settings.plex_hls_dir) if settings.plex_hls_dir else Path(setti
 _SEG_SECONDS = 6
 _SESSION_IDLE_S = 600  # reap a session with no access for this long
 # A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session);
-# before multi-audio it was just `{rating_key}_{int(start_s)}`, so the tail is OPTIONAL here — the
-# oldest leftovers on disk carry the old shape and are exactly what the sweep exists to remove.
-# `sweep_orphans` deletes ONLY names matching this: `_HLS_ROOT` is admin-configurable
-# (PLEX_HLS_DIR), so it may legitimately point at a shared scratch path we do not own — a blanket
-# "delete every directory under the root" would take the neighbours with it.
-_SESSION_DIR_RE = re.compile(r"^.+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
+# before multi-audio it was just `{rating_key}_{int(start_s)}`. `sweep_orphans` deletes ONLY names
+# matching one of those two shapes: `_HLS_ROOT` is admin-configurable (PLEX_HLS_DIR), so it may
+# legitimately point at a shared scratch path we do not own, and a blanket "delete every directory
+# under the root" would take the neighbours with it.
+#
+# The KEY is anchored to digits (a Plex rating_key is a number) — that is what keeps foreign names
+# out. A `.+` there let `snapshot_2024_backup_+1.00` through, and pairing `.+` with an optional tail
+# degenerated all the way to "anything ending in _<digits>" (`backup_2024`, `release_10`). A
+# non-numeric rating_key would merely leave its directory unswept, which is the safe direction to
+# fail. Covered by test_plex_stream.py — this regex is the only thing standing between the sweep
+# and someone else's data.
+_SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
 
 _lock = threading.Lock()
 _sessions: dict[str, "HlsSession"] = {}

backend/app/sync/playlists.py

@@ -346,10 +346,11 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             created = 1
             # Brand-new playlist: just append every video in order.
             for vid in desired:
-                if breaker.run(
+                added = breaker.run(
                     lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
                     lambda v=vid: failures.append(v),
-                ) is not FAILED:
+                )
+                if not isinstance(added, _Failed):
                     inserted += 1
             if not failures:
                 pl.synced_fingerprint = fingerprint(pl.name, desired)
@@ -382,10 +383,11 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
         # Delete extras (local is authoritative).
         for it in current:
             if it["video_id"] not in desired_set:
-                if breaker.run(
+                dropped = breaker.run(
                     lambda i=it: yt.delete_playlist_item(i["item_id"]),
                     lambda i=it: failures.append(i["video_id"]),
-                ) is not FAILED:
+                )
+                if not isinstance(dropped, _Failed):
                     deleted += 1
         # Insert missing (append; the reorder pass below fixes positions).
         cur_set = set(cur_vids)
@@ -407,12 +409,13 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
                 continue  # failed insert; skip
             if i < len(model) and model[i] == want:
                 continue
-            if breaker.run(
+            moved = breaker.run(
                 lambda w=want, pos=i: yt.move_playlist_item(
                     item_id_by_vid[w], pl.yt_playlist_id, w, pos
                 ),
                 lambda w=want: failures.append(w),
-            ) is FAILED:
+            )
+            if isinstance(moved, _Failed):
                 continue
             reordered += 1
             j = model.index(want)

backend/app/sync/runner.py

@@ -48,11 +48,15 @@ def get_service_user(db: Session) -> User | None:
     return users[0] if users else None
 
 
-def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict:
-    """Poll every channel's free RSS feed. Returns `{"new": inserted, "failed": unreadable_feeds}`
-    — the failure count is part of the RESULT, not just a log line, so the scheduler card and the
-    audit summary can tell "no new uploads" apart from "every feed was unreachable" (an egress
-    outage otherwise renders identically to a quiet hour)."""
+def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str, int]:
+    """Poll every channel's free RSS feed. Returns `{"new": inserted, "failed": channels_that_did
+    _not_complete}` — the failure count is part of the RESULT, not just a log line, so the scheduler
+    card and the audit summary can tell "no new uploads" apart from "nothing worked" (an egress
+    outage, or a database that can't accept writes, otherwise renders identically to a quiet hour).
+
+    `failed` counts a channel whose feed could not be READ *or* whose rows could not be APPLIED —
+    both leave the channel un-polled, and counting only the first would move the blind spot rather
+    than remove it."""
     if channels is None:
         channels = db.execute(select(Channel)).scalars().all()
     total = len(channels)
@@ -87,12 +91,15 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict:
             try:
                 new += apply_rss_feed(db, channel, entries)
             except Exception:
+                # A write failure (read-only DB, failover) leaves the channel just as un-polled as
+                # an unreachable feed — count it, or a wholly broken run still reports "failed=0".
                 db.rollback()
+                failed += 1
                 log.exception("RSS apply failed for channel %s", channel.id)
             done += 1
             progress.report(done, total, "rss_poll")
     if failed:
-        log.warning("RSS poll: %d/%d channel feeds could not be read", failed, total)
+        log.warning("RSS poll: %d/%d channels could not be polled", failed, total)
     return {"new": new, "failed": failed}
 
 

backend/tests/test_plex_stream.py

@@ -0,0 +1,44 @@
+"""The guard in front of a recursive delete.
+
+`sweep_orphans` rmtree's directories under `_HLS_ROOT`, and that root is admin-configurable
+(PLEX_HLS_DIR) — it can legitimately be a shared scratch path. `_SESSION_DIR_RE` is the ONLY thing
+that keeps the sweep off a neighbour's data, so the names it accepts are pinned here."""
+import pytest
+
+from app.plex.stream import _SESSION_DIR_RE
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        "12345_0",  # legacy shape (pre multi-audio): {rating_key}_{start}
+        "12345_842",
+        "12345_0_None_+0.00",  # current shape, single audio track (audio_ord None)
+        "12345_0_2_+0.00",  # current shape, explicit audio_ord
+        "12345_842_multi_-1.25",  # multi-audio, negative offset
+        "12345_842_multi_+12.50",
+    ],
+)
+def test_our_session_directories_are_swept(name):
+    assert _SESSION_DIR_RE.match(name)
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        "backup_2024",  # the class an "optional tail" pattern wrongly matched
+        "release_10",
+        "important_data_2024",
+        "node_modules",
+        "logs",
+        "plex-hls",
+        "12345",  # no start offset — not a session dir
+        "12345_0_multi",  # truncated: missing the offset
+        "12345_0_multi_1.25",  # offset without its sign
+        "snapshot_2024_backup_+1.00",  # session-ish tail, but the key isn't a rating_key
+        "backup_2024_multi_+0.00",
+        ".config",
+    ],
+)
+def test_foreign_directories_are_left_alone(name):
+    assert not _SESSION_DIR_RE.match(name)

backend/tests/test_run_ffmpeg.py

@@ -0,0 +1,133 @@
+"""run_ffmpeg's process plumbing — the part that can hang a worker thread forever.
+
+No ffmpeg here: a fake Popen stands in, so the reader threads, the 1s poll loop, the stall watchdog
+and the bounded exit wait are all exercised on any machine. These are regression tests for a real
+deadlock: stderr was opened and never read, so a chatty ffmpeg filled the ~64KB pipe, blocked on its
+next write and never exited, and `proc.wait()` never returned."""
+import time
+
+import pytest
+
+from app.downloads import edit
+
+
+class _FakeStream:
+    """A pipe. `block=True` = the process is alive but silent (a wedged ffmpeg): iteration parks
+    until the stream is closed, which is what terminate()/kill() does to a real pipe."""
+
+    def __init__(self, lines=(), block=False):
+        self._lines = list(lines)
+        self._block = block
+        self.closed = False
+        self.consumed = 0
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._lines:
+            self.consumed += 1
+            return self._lines.pop(0)
+        while self._block and not self.closed:
+            time.sleep(0.02)
+        raise StopIteration
+
+    def close(self):
+        self.closed = True
+
+
+class _FakeProc:
+    pid = 4242
+
+    def __init__(self, stdout, stderr, returncode=0):
+        self.stdout = stdout
+        self.stderr = stderr
+        self._rc = returncode
+        self.terminated = False
+        self.killed = False
+
+    def terminate(self):
+        self.terminated = True
+        # A terminated process's pipes hit EOF — that is how the reader threads finish.
+        self.stdout.close()
+        self.stderr.close()
+
+    def kill(self):
+        self.killed = True
+        self.terminate()
+
+    def wait(self, timeout=None):
+        return self._rc
+
+
+@pytest.fixture
+def fake_popen(monkeypatch):
+    """Install a fake Popen and hand the test the process it will produce."""
+
+    def install(proc):
+        monkeypatch.setattr(edit.subprocess, "Popen", lambda *a, **k: proc)
+        return proc
+
+    return install
+
+
+def test_progress_is_reported_and_a_clean_exit_returns(fake_popen):
+    proc = fake_popen(
+        _FakeProc(_FakeStream(["out_time_us=5000000\n", "progress=continue\n"]), _FakeStream())
+    )
+    seen = []
+    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=seen.append, should_cancel=lambda: False)
+    assert seen == [50.0]  # 5s of 10s
+    assert not proc.terminated
+
+
+def test_progress_is_capped_below_100(fake_popen):
+    fake_popen(_FakeProc(_FakeStream(["out_time_us=99000000\n"]), _FakeStream()))
+    seen = []
+    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=seen.append, should_cancel=lambda: False)
+    assert seen == [99.0]  # never reports done before the process actually exits
+
+
+def test_a_nonzero_exit_raises_with_the_stderr_tail(fake_popen):
+    # The tail must come from the drained stderr — reading the pipe only AFTER the exit was the
+    # original deadlock.
+    proc = fake_popen(
+        _FakeProc(_FakeStream([]), _FakeStream(["Invalid data found\n"]), returncode=2)
+    )
+    with pytest.raises(RuntimeError, match="Invalid data found"):
+        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
+    assert proc._rc == 2
+
+
+def test_cancel_is_honoured_while_the_process_is_silent(fake_popen):
+    # The pre-fix loop only checked cancel when a progress line arrived, so a quiet ffmpeg was
+    # uncancellable. The 1s poll tick is what makes this terminate.
+    proc = fake_popen(_FakeProc(_FakeStream(block=True), _FakeStream(block=True)))
+    started = time.monotonic()
+    with pytest.raises(edit.EditAborted):
+        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: True)
+    assert proc.terminated
+    assert time.monotonic() - started < 5  # not "whenever ffmpeg next speaks"
+
+
+def test_a_wedged_process_is_killed_by_the_stall_watchdog(fake_popen):
+    proc = fake_popen(_FakeProc(_FakeStream(block=True), _FakeStream(block=True)))
+    with pytest.raises(RuntimeError, match="stalled"):
+        edit.run_ffmpeg(
+            ["ffmpeg"],
+            total_s=10,
+            on_progress=lambda _p: None,
+            should_cancel=lambda: False,
+            stall_timeout_s=0.5,
+        )
+    assert proc.terminated  # the worker thread is freed, not parked forever
+
+
+def test_stderr_is_drained_while_the_process_runs(fake_popen):
+    # The deadlock case: a chatty ffmpeg that exits 0. EVERY stderr line must be consumed — with a
+    # real pipe, an unread one fills at ~64KB and blocks ffmpeg's next write forever.
+    err = _FakeStream([f"warning {i}\n" for i in range(500)])
+    fake_popen(_FakeProc(_FakeStream(["out_time_us=1000000\n"]), err))
+    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
+    assert err.consumed == 500
+    assert err.closed

backend/tests/test_sync_guards.py

@@ -3,6 +3,7 @@ playlist-push circuit breaker, and RSS telling a failure apart from an empty fee
 import httpx
 import pytest
 
+from app.sync import runner
 from app.sync.playlists import FAILED, _Breaker
 from app.sync.subscriptions import should_prune
 from app.youtube import rss
@@ -85,6 +86,55 @@ class TestBreakerRun:
             _Breaker().run(boom, lambda: None)
 
 
+class _StubDb:
+    """Just enough Session for run_rss_poll when the caller supplies the channel list."""
+
+    def __init__(self):
+        self.rollbacks = 0
+
+    def rollback(self):
+        self.rollbacks += 1
+
+
+class _StubChannel:
+    def __init__(self, cid):
+        self.id = cid
+
+
+class TestRssPollResult:
+    """`failed` must count every channel that did not get polled — feed-side AND apply-side.
+    Counting only the fetch failures would move the "a broken run looks like a quiet hour" blind
+    spot rather than remove it."""
+
+    def _run(self, monkeypatch, *, fetch, apply_):
+        monkeypatch.setattr(runner, "fetch_channel_feed", fetch)
+        monkeypatch.setattr(runner, "apply_rss_feed", apply_)
+        return runner.run_rss_poll(_StubDb(), [_StubChannel("UC1"), _StubChannel("UC2")])
+
+    def test_a_healthy_run_reports_no_failures(self, monkeypatch):
+        out = self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=lambda db, ch, e: 1)
+        assert out == {"new": 2, "failed": 0}
+
+    def test_unreadable_feeds_are_counted(self, monkeypatch):
+        def boom(cid):
+            raise rss.RssError("404")
+
+        out = self._run(monkeypatch, fetch=boom, apply_=lambda db, ch, e: 1)
+        assert out == {"new": 0, "failed": 2}
+
+    def test_apply_failures_are_counted_too(self, monkeypatch):
+        # A read-only database: every feed reads fine, nothing can be stored.
+        def boom(db, ch, entries):
+            raise RuntimeError("read-only transaction")
+
+        out = self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=boom)
+        assert out == {"new": 0, "failed": 2}, "a run where nothing was stored must not read as ok"
+
+    def test_a_quiet_hour_is_distinguishable_from_an_outage(self, monkeypatch):
+        quiet = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0)
+        assert quiet == {"new": 0, "failed": 0}  # same "new", different "failed" — that's the point
+
+
 class TestRssErrorVsEmpty:
     """A failed poll must RAISE (so the caller leaves `last_rss_at` alone); only a genuinely
     empty feed may return []."""