fix(api): a paused viewer keeps its HLS session, a refused subtitle publish backs off, and a NUL is a 422 everywhere

Deep-review round 4, backend half.

- The periodic HLS reaper (round 3) applied the "finished and unfetched for 30 s" rule that only a
  NEW session request may apply. A remux finishes long before the film, and a paused player with a
  full buffer fetches nothing, so a one-minute pause killed the session and playback died ~30 s after
  resuming. The periodic pass is `reap_abandoned` now (the long idle limit, finished or not).
- A subtitle `publish` that could not keep its output (full / read-only cache volume) returned
  quietly, so every poll queued the window and the whole-container pass again. It reports failure
  and the pass is backed off like a failed read; a cached VTT that cannot be READ is answered
  `unavailable` (`cache_unreadable`) instead of scheduling reads that can never land.
- The subtitle routes bound `rating_key` / `sub_ord` (they become a cache file name: a 250-digit
  one was ENAMETOOLONG, which pathlib's exists() re-raises -> a 500), and the cache helpers no longer
  throw on such a name.
- One app-wide handler turns psycopg's "text fields cannot contain NUL" DataError into a 422: round 3
  guarded only the by-rating_key lookup, and `contains=`, `contains_group=`, reorder/bulk key lists
  and the collection search still 500'd.
- The hand-written `db.get(...) is None -> 404 "Unknown ..."` siblings use `get_or_404` with a
  `what` (channel / video / user), so the Hungarian UI no longer shows English for a gone row.

backend/tests/test_hls_reaper.py

@@ -3,12 +3,20 @@
 plex/stream.py promises "a periodic reaper kills idle ones", but `reap_idle` ran only inside
 POST /session: after the last playback, an abandoned session (its ffmpeg still working through the
 film, its directory a whole segmented copy of it) lived until somebody pressed play again.
+
+And it must not take a PAUSED viewer's session: a remux finishes long before the film does, and a
+paused player with a full buffer fetches nothing — so "finished and unfetched for 30 s" (the rule a
+new session request applies) is not "abandoned".
 """
 
 import asyncio
+import time
+
+import pytest
 
 import app.main as app_main
 from app.main import app
+from app.plex import stream
 
 
 class _Configured:
@@ -31,7 +39,7 @@ def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch):
     monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
     monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
     monkeypatch.setattr(app_main, "_HLS_REAP_EVERY_S", 0.01)
-    monkeypatch.setattr(app_main.plex_stream, "reap_idle", lambda: reaped.append(1) or 0)
+    monkeypatch.setattr(app_main.plex_stream, "reap_abandoned", lambda: reaped.append(1) or 0)
 
     async def _cycle():
         async with app_main.lifespan(app):
@@ -39,3 +47,60 @@ def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch):
 
     asyncio.run(_cycle())
     assert len(reaped) >= 2, "no periodic reap ran while the app was up"
+
+
+class _Proc:
+    def __init__(self, finished: bool):
+        self.finished = finished
+        self.killed = False
+
+    def poll(self):
+        return 0 if self.finished else None
+
+    def terminate(self):
+        self.killed = True
+
+    def wait(self, timeout=None):
+        return 0
+
+
+class _Session:
+    def __init__(self, tmp_path, name: str, finished: bool, idle_s: float):
+        self.dir = tmp_path / name
+        self.dir.mkdir()
+        self.proc = _Proc(finished)
+        self.last_access = time.time() - idle_s
+
+
+@pytest.fixture
+def sessions(monkeypatch):
+    live: dict = {}
+    monkeypatch.setattr(stream, "_sessions", live)
+    return live
+
+
+def test_the_periodic_pass_keeps_a_paused_viewers_finished_session(tmp_path, sessions):
+    # A remux that finished a minute ago, its viewer paused with a full buffer since.
+    s = _Session(tmp_path, "paused", finished=True, idle_s=90)
+    sessions["12345"] = s
+    assert stream.reap_abandoned() == 0
+    assert "12345" in sessions and s.dir.exists() and not s.proc.killed
+
+
+def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limit(
+    tmp_path, sessions
+):
+    sessions["1"] = _Session(
+        tmp_path, "gone_done", finished=True, idle_s=stream._SESSION_IDLE_S + 5
+    )
+    sessions["2"] = _Session(
+        tmp_path, "gone_live", finished=False, idle_s=stream._SESSION_IDLE_S + 5
+    )
+    assert stream.reap_abandoned() == 2
+    assert sessions == {}
+
+
+def test_a_new_session_request_still_reaps_finished_sessions_quickly(tmp_path, sessions):
+    sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45)
+    assert stream.reap_idle() == 1
+    assert sessions == {}

backend/tests/test_not_found_detail.py

@@ -18,3 +18,36 @@ def test_a_gone_view_tag_or_pin_says_what_is_gone(db, make_user, as_user):
         r = getattr(client, method)(url, **({"json": {"name": "x"}} if method == "patch" else {}))
         assert r.status_code == 404, (url, r.text)
         assert r.json()["detail"] == {"code": "not_found", "reason": what}, url
+
+
+def test_a_gone_channel_video_or_user_says_what_is_gone(db, make_user, as_user):
+    """The siblings that still hand-wrote `db.get(...) is None → 404 "Unknown …"` (deep-review
+    round 4): the same gone row, but the Hungarian UI showed the English."""
+    admin = make_user(role="admin")
+    client = as_user(admin)
+    for method, url, body, what in [
+        ("post", "/api/channels/UCgone000000000000000000/block", None, "channel"),
+        ("post", "/api/videos/gone0000000/state", {"status": "watched"}, "video"),
+        (
+            "post",
+            "/api/videos/gone0000000/progress",
+            {"position_seconds": 60, "duration_seconds": 600},
+            "video",
+        ),
+        ("put", "/api/admin/downloads/quota/999999", {"max_bytes": 1}, "user"),
+    ]:
+        r = getattr(client, method)(url, **({"json": body} if body is not None else {}))
+        assert r.status_code == 404, (url, r.text)
+        assert r.json()["detail"] == {"code": "not_found", "reason": what}, url
+
+
+def test_a_nul_in_a_request_string_is_a_422_not_a_500(db, make_user, as_user):
+    """Postgres cannot hold NUL in text, so the driver refuses the query outright: a DataError, i.e.
+    a 500 with a traceback, wherever a `%00` or an escaped NUL reached a comparison (deep-review
+    round 4 — round 3 had guarded only the by-rating_key lookup). One app-wide handler now answers
+    all of them."""
+    client = as_user(make_user())
+    r = client.get("/api/plex/playlists?contains=%00")
+    assert r.status_code == 422, r.text
+    assert r.json()["detail"] == {"code": "invalid_input", "reason": "nul"}
+    assert client.get("/api/plex/playlists?contains_group=1,%00").status_code == 422

backend/tests/test_plex_subtitle_prepare.py

@@ -1002,3 +1002,75 @@ def test_a_nul_rating_key_names_no_row():
     from app.plex.lookups import by_rating_key
 
     assert by_rating_key(None, PlexItem, "12\x0034") is None  # type: ignore[arg-type]
+
+
+# --- the fourth deep-review round ---------------------------------------------------------------
+
+
+def test_a_pass_whose_output_could_not_be_kept_is_backed_off_like_a_failed_one(monkeypatch):
+    """`publish` refusing (a full or read-only cache volume) returned quietly, so `_run` recorded
+    nothing and every 1 s poll queued the window AND the whole-container pass again — the file read
+    end to end over and over, the player on "preparing" for good."""
+    _stub_item(monkeypatch)
+    pool = _manual_pool(monkeypatch)
+    monkeypatch.setattr(subtitles, "cache_write", lambda key, vtt: False)
+    assert subtitles.ensure("98614", 3, at=600.0)["state"] == "preparing"
+    assert [a[3] for _, a in pool.jobs] == [570.0, None]
+    pool.run_all()
+    after = subtitles.ensure("98614", 3, at=600.0)
+    assert (after["state"], after["code"]) == ("unavailable", "extract_failed")
+    for _ in range(3):
+        subtitles.ensure("98614", 3, at=600.0)
+    assert pool.jobs == [], "the same reads were queued again"
+
+
+def test_a_sidecar_that_could_not_be_kept_is_recorded_as_failed(monkeypatch):
+    _stub_item(monkeypatch)
+    pool = _manual_pool(monkeypatch)
+    monkeypatch.setattr(subtitles, "cache_write", lambda key, vtt: False)
+    subtitles.ensure("98614", 0, at=600.0)
+    pool.run_all()
+    subtitles.ensure("98614", 0, at=600.0)
+    assert pool.jobs == [], "a refused sidecar publish was queued again"
+
+
+def test_an_unreadable_cached_track_is_said_once_and_not_rescheduled(monkeypatch):
+    """A VTT on disk that cannot be read: `publish` will not merge into it and a job whose sidecar
+    covers the position skips as done, so scheduling could only ever leave the player on
+    "preparing"."""
+    pool = _manual_pool(monkeypatch)
+    key = subtitles.cache_key("98614", 0)
+    subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
+    monkeypatch.setattr(subtitles, "cache_read", lambda k: None)  # an I/O error on the read
+    st = subtitles.ensure("98614", 0, at=1000.0)
+    assert (st["state"], st["code"], st["gen"]) == ("unavailable", "cache_unreadable", 0)
+    assert pool.jobs == []
+
+
+def test_an_absurd_rating_key_or_ordinal_is_refused_not_a_500(client, monkeypatch):
+    """They become a cache FILE name: a 250-digit key outgrew the OS name limit, and pathlib's
+    `exists()` re-raises that (ENAMETOOLONG) instead of answering False — a 500 on every poll."""
+    pool = _manual_pool(monkeypatch)
+    long_key = "9" * 250
+    for url in (
+        f"/api/plex/subtitle/{long_key}/0/status",
+        f"/api/plex/subtitle/98614/{long_key}/status",
+        f"/api/plex/subtitle/{long_key}/0",
+    ):
+        assert client.get(url).status_code == 422, url
+    assert pool.jobs == []
+    # …and the cache helpers themselves no longer throw on such a name.
+    assert subtitles.read_meta(subtitles.cache_key(long_key, 0)) == ([], False)
+
+
+def test_a_read_ahead_window_that_failed_is_not_queued_again(monkeypatch):
+    """Near a covered edge the probe moves just past it, and `_schedule` has no failed-span check of
+    its own: without the guard, every poll re-queued the same failing window read."""
+    pool = _manual_pool(monkeypatch)
+    subtitles._FULL_FAILED["98614"] = time.monotonic()
+    key = subtitles.cache_key("98614", 0)
+    subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
+    subtitles._window_fail(key, (1271.0, 1601.0))
+    for at in (1250.0, 1252.0, 1254.0):
+        assert subtitles.ensure("98614", 0, at=at)["state"] == "partial"
+    assert pool.jobs == [], "the failed read-ahead window was queued again"