From 47af4d5c054fed62e2c53613a3be7c10ddc65a2f Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 12 Sep 2026 03:57:52 +0200 Subject: [PATCH] fix(plex): an HLS artifact names the session it belongs to, and a paused viewer says so MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deep-review round 5, over round 4's own fixes. - Sessions are keyed by rating_key alone while a session's DIRECTORY is per (key, offset, tag), and the artifact URL carried no session identity — so once a seek-restart or a second viewer of the same item took the registry slot, the old viewer's request for `seg_12.ts` was answered 200 with the new session's twelfth segment: bytes from a different `-ss` offset under `-copyts`, on a timeline the client is not on. Wrong content with a success status is the one failure a player cannot notice, and round 4's reclaim-on-404 could never fire on it. The id is in the path now (so the playlist's relative names resolve back into the same session), a mismatch is a 404, and the player reclaims at the playhead. The entry playlist's auto-start is gone with it: it could only ever begin at 0, silently discarding the viewer's position. - Round 4 exempted paused viewers from the reaper by keeping EVERY finished session for the full ten-minute idle window — one whole segmented copy of a film per item, on the volume the download center shares. The signal it needed already existed: an open player checkpoints its position every 10 s whether or not it is playing, so that route now touches the session and the periodic pass is back on the short rule. - POST /session's opportunistic reap was never scoped to the key being started: it walked the whole registry with the 30 s finished rule, so pressing play on ANY item — even a direct-play one, the call ran before that early return — killed another viewer's paused session. It is gone; the periodic pass owns reaping. - A segment that can never arrive no longer polls out the full 20 s: the producer's handle is passed to `wait_for`, so a finished ffmpeg answers at once. This is a sync route, so each such wait parked a threadpool worker AND a pooled DB connection, and an ordinary seek past the produced region was enough to do it. - The DataError handler's re-raise branch — the half that keeps a genuine DataError a 500 — is pinned, so simplifying it to an unconditional 422 cannot pass in silence. Six regression tests (two route-level, in the DB lane), each mutation-checked red on the assertion it is named after. pytest 1124, vitest 999. --- backend/app/main.py | 7 +- backend/app/plex/stream.py | 40 ++++-- backend/app/routes/plex/playback.py | 47 ++++--- backend/tests/test_hls_reaper.py | 132 +++++++++++++++--- backend/tests/test_not_found_detail.py | 21 +++ backend/tests/test_plex_hls_file_route.py | 145 ++++++++++++++++++++ backend/tests/test_plex_playback_session.py | 7 +- 7 files changed, 348 insertions(+), 51 deletions(-) create mode 100644 backend/tests/test_plex_hls_file_route.py diff --git a/backend/app/main.py b/backend/app/main.py index ca70b442..792d9335 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -112,11 +112,14 @@ async def _reap_hls_sessions() -> None: ffmpeg still working through the film, its directory a whole segmented copy of it — lived until somebody pressed play again or the process restarted. - `reap_abandoned`, not `reap_idle`: a finished session is also a paused viewer's.""" + A FINISHED session is reaped on the short rule here, not kept for the full idle window: an open + player checkpoints every 10 s and that route touches the session, so a paused viewer is no longer + invisible to this pass (see `reap_idle`). Keeping every finished session for ten minutes instead + held one whole segmented copy of a film per item on the download volume (round 5).""" while True: await asyncio.sleep(_HLS_REAP_EVERY_S) try: - await asyncio.to_thread(plex_stream.reap_abandoned) + await asyncio.to_thread(plex_stream.reap_idle) except Exception: log.exception("HLS session reap failed") diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index fdb012a7..224ff5a9 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -21,6 +21,7 @@ CPU-only prod host; only P3's full transcode is CPU-heavy. import logging import os import re +import secrets import shutil import subprocess import threading @@ -99,6 +100,14 @@ class HlsSession: hw: bool = False, tonemap: bool = False, ): + # Unguessable per-session id, carried in the URLs the client is handed. Sessions are keyed + # by rating_key alone, so a seek-restart or a SECOND viewer of the same item replaces the + # registered one — and the old viewer's next request for `seg_12.ts`, resolved by rating_key, + # was answered 200 with the NEW session's twelfth segment, produced from a different offset + # under `-copyts`. Wrong bytes, not an error, so the client's reclaim never fired on it + # (deep-review round 5). A request naming a session that is no longer registered is a 404, + # which the player recovers from by restarting at the playhead. + self.id = secrets.token_urlsafe(9) self.key = key self.dir = directory self.proc = proc @@ -826,16 +835,19 @@ _FINISHED_IDLE_S = 30 def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: - """Kill sessions nobody has fetched from for `_SESSION_IDLE_S` — or, once their ffmpeg has - finished, for `finished_idle_s`. + """Kill sessions nobody has said they still want for `_SESSION_IDLE_S` — or, once their ffmpeg + has finished, for `finished_idle_s`. - The short rule is only safe where a NEW session is being requested (the default, POST /session): - a finished session is also what a PAUSED viewer has. A remux runs far faster than realtime, so + A finished session is also what a PAUSED viewer has: a remux runs far faster than realtime, so ffmpeg is done a minute or two in and writes ENDLIST; hls.js then stops asking for the playlist, - and with its forward buffer full a paused player fetches nothing at all. The periodic reaper - therefore uses `reap_abandoned` — with the 30 s rule, a one-minute pause killed the session and - playback died ~30 s after resuming (deep-review round 4). A viewer paused for longer than - `_SESSION_IDLE_S` still loses it; the player restarts one at the playhead when a fetch 404s.""" + and with its forward buffer full a paused player fetches nothing at all. With the 30 s rule that + killed a one-minute pause, and playback died ~30 s after resuming (deep-review round 4). + + What makes the short rule safe again is that fetching is no longer the only signal: an open + player checkpoints its position every 10 s, and that route calls `touch_session`, so a viewer who + is merely paused says so far more often than this rule asks (round 5). Round 4's blanket + `reap_abandoned` bought the same safety by keeping every finished session for ten minutes — each + one a whole segmented copy of the film on a volume shared with the download center.""" now = time.time() dropped = 0 with _lock: @@ -849,7 +861,11 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: return dropped -def reap_abandoned() -> int: - """The periodic reaper's pass: only what nobody has fetched from for `_SESSION_IDLE_S`, finished - or not (see reap_idle for why a finished session is not yet an abandoned one).""" - return reap_idle(finished_idle_s=_SESSION_IDLE_S) +def touch_session(key: str) -> None: + """Say a viewer is still there for `key`, without fetching anything. `last_access` used to move + only on an HLS file fetch, which a paused player does not do — so the only way to keep a paused + viewer's session alive was to keep EVERY abandoned one just as long.""" + with _lock: + s = _sessions.get(key) + if s is not None: + s.last_access = time.time() diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 3420b1d0..769bf8bd 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -323,6 +323,10 @@ def item_progress( `_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks.""" it = item_or_404(db, rating_key) + # An open player checkpoints every 10 s whether or not it is playing, so this is the signal that a + # PAUSED viewer still wants their HLS session — the one thing `last_access` could not see, since + # it moved only on a file fetch and a paused player fetches nothing (see stream.reap_idle). + plex_stream.touch_session(str(rating_key)) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) final = body.final @@ -561,7 +565,6 @@ def stream_session( HEVC/VP9/AV1/…) take the same HLS path but with the video re-encoded to H.264 (VAAPI or libx264). Selecting a non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio can be muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart.""" - plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback it = item_or_404(db, rating_key) transcode = it.playable == "transcode" # `multi` (item has ≥2 audio tracks) forces the HLS path so ALL audio tracks ship as renditions and @@ -585,7 +588,9 @@ def stream_session( return { "mode": "hls", "start_s": s.media_start_s, - "url": f"/api/plex/stream/{it.rating_key}/hls/{s.entry}", + # The session id is part of the path (not a query param) so the playlist's RELATIVE segment + # names resolve back into the same session — see `stream_hls_file` for why it has to. + "url": f"/api/plex/stream/{it.rating_key}/hls/{s.id}/{s.entry}", "transcode": s.transcode, "hw": s.hw, "tonemap": s.tonemap, @@ -624,35 +629,39 @@ _HLS_CT = { } -@router.get("/stream/{rating_key}/hls/{filename}") +@router.get("/stream/{rating_key}/hls/{session_id}/{filename}") def stream_hls_file( rating_key: str, + session_id: str, filename: str, user: User = Depends(current_user), - db: Session = Depends(get_db), ) -> Response: - """Serve a file (playlist / segment / subtitle) from the current HLS session directory. The - playlist references its segments/subtitles by relative name, all resolving back here.""" + """Serve a file (playlist / segment / subtitle) from ONE HLS session's directory. The playlist + references its segments/subtitles by relative name, all resolving back here. + + The session id in the path is what makes "back here" mean the same session. Sessions are keyed by + rating_key alone and a seek-restart or a second viewer of the same item replaces the registered + one, while the DIRECTORY is per (key, offset, tag) — so resolving by rating_key handed the old + viewer the new session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong + bytes with a 200, which the player has no way to notice (deep-review round 5). A request for a + session that is no longer the registered one is a 404, and the player reclaims at the playhead. + + That reclaim is also why a request with no live session is no longer auto-started here: the + auto-start could only ever begin at 0, silently throwing away the viewer's position.""" # Filename must be a plain hls artifact — no path separators, only the expected extensions. ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else "" if ext not in _HLS_CT or "/" in filename or "\\" in filename or ".." in filename: raise HTTPException(status_code=404, detail="Not found") s = plex_stream.current_session(str(rating_key)) - if s is None: - # No session yet (e.g. reaped): the ENTRY playlist may auto-start one at 0 — master.m3u8 for a - # multi-audio item, index.m3u8 for a single-audio remux item. - it = item_or_404(db, rating_key) - if filename == "master.m3u8": - s = plex_stream.start_session(db, it, 0.0, multi=True) - elif filename == "index.m3u8" and it.playable == "remux": - s = plex_stream.start_session(db, it, 0.0) - else: - raise HTTPException(status_code=404, detail="No active playback session") - if s is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") + if s is None or s.id != session_id: + raise HTTPException(status_code=404, detail="No active playback session") path = s.dir / filename timeout = 30 if ext == ".m3u8" else 20 - if not plex_stream.wait_for(path, timeout): + # `proc` so a session whose ffmpeg has already exited answers at once instead of polling out the + # whole timeout: this is a SYNC route on the ~40-slot threadpool, holding a pooled DB connection + # with it, so every such wait is a worker (and a connection) parked for 20 s — an ordinary seek + # past the produced region was enough to do it (round 5). + if not plex_stream.wait_for(path, timeout, proc=s.proc): raise HTTPException(status_code=404, detail="Not available yet") return Response( content=path.read_bytes(), diff --git a/backend/tests/test_hls_reaper.py b/backend/tests/test_hls_reaper.py index 0c8d22ea..667474cc 100644 --- a/backend/tests/test_hls_reaper.py +++ b/backend/tests/test_hls_reaper.py @@ -1,21 +1,35 @@ -"""Idle HLS sessions are reaped on a timer, not only when the next session is requested. +"""Idle HLS sessions are reaped on a timer, and a viewer who is merely PAUSED keeps theirs. 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. +film, its directory a whole segmented copy of it) lived until somebody pressed play again +(deep-review round 4). 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". +paused player with a full buffer fetches nothing, so "finished and unfetched for 30 s" does not mean +"abandoned". Round 4 bought that safety by keeping EVERY finished session for the full ten-minute +idle window — one segmented copy of a film per item, on the volume the download center shares. Round +5 replaced the blanket retention with the signal that already existed: an open player checkpoints its +position every 10 s whether or not it is playing, and that route says "still here". + +Both call sites are pinned here, because the bug each time was that only one of them moved: the +periodic pass, and the fact that starting a session no longer sweeps OTHER viewers' sessions at all. """ import asyncio import time +from types import SimpleNamespace import pytest +from fastapi.testclient import TestClient import app.main as app_main +import app.state as state +from app.auth import current_user +from app.db import get_db from app.main import app +from app.models import PlexItem, PlexLibrary, User +from app.plex import paths as plex_paths from app.plex import stream @@ -27,7 +41,7 @@ class _Configured: return False -def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch): +def test_the_periodic_pass_is_the_one_that_runs_and_it_uses_the_short_rule(monkeypatch): async def _no_sweep(): return None @@ -39,7 +53,9 @@ 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_abandoned", lambda: reaped.append(1) or 0) + # `reap_idle`, i.e. the SHORT finished rule: a finished session nobody has said they still want + # is a whole copy of a film sitting on the download volume. + monkeypatch.setattr(app_main.plex_stream, "reap_idle", lambda: reaped.append(1) or 0) async def _cycle(): async with app_main.lifespan(app): @@ -79,12 +95,10 @@ def sessions(monkeypatch): 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_reaps_a_finished_session_nobody_has_claimed(tmp_path, sessions): + sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45) + assert stream.reap_idle() == 1 + assert sessions == {} def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limit( @@ -96,11 +110,95 @@ def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limi sessions["2"] = _Session( tmp_path, "gone_live", finished=False, idle_s=stream._SESSION_IDLE_S + 5 ) - assert stream.reap_abandoned() == 2 + assert stream.reap_idle() == 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 == {} +def test_touching_a_session_keeps_it_through_the_short_rule(tmp_path, sessions): + s = _Session(tmp_path, "paused", finished=True, idle_s=90) + sessions["12345"] = s + stream.touch_session("12345") + assert stream.reap_idle() == 0 + assert "12345" in sessions and s.dir.exists() and not s.proc.killed + + +# --- the two call sites, driven through the real routes (DB lane) ------------------------------- + + +def _library_item(db, rating_key: str, *, playable: str = "remux"): + lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + db.add(lib) + db.flush() + it = PlexItem( + rating_key=rating_key, + library_id=lib.id, + kind="movie", + title="A Movie", + file_path="/media/a.mkv", + playable=playable, + ) + db.add(it) + db.commit() + return it + + +def _client(db, user): + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[current_user] = lambda: user + return TestClient(app) + + +def test_a_paused_viewers_checkpoint_keeps_their_session_alive(db, monkeypatch, tmp_path, sessions): + """The whole reason the short rule is safe again: a paused player fetches nothing, so + `last_access` never moved on its own — but it does keep checkpointing its position.""" + monkeypatch.setattr(state, "_configured_cache", True) + it = _library_item(db, "paused1") + user = User(email="paused@example.com") + db.add(user) + db.commit() + s = _Session(tmp_path, "paused", finished=True, idle_s=90) + sessions["paused1"] = s + try: + resp = _client(db, user).post( + f"/api/plex/item/{it.rating_key}/progress", + json={"position_seconds": 600, "duration_seconds": 3000, "final": False}, + ) + finally: + app.dependency_overrides.clear() + assert resp.status_code == 200, resp.text + assert stream.reap_idle() == 0, "the checkpoint did not say the viewer is still there" + assert "paused1" in sessions and not s.proc.killed + + +def test_starting_a_session_leaves_another_viewers_paused_one_alone( + db, monkeypatch, tmp_path, sessions +): + """The opportunistic sweep this route used to run was not scoped to the key being started: it + walked the WHOLE registry with the 30 s finished rule, so pressing play on any item killed a + paused viewer's session — the very case round 4 set out to protect (found in round 5).""" + monkeypatch.setattr(state, "_configured_cache", True) + monkeypatch.setattr(plex_paths, "local_media_path", lambda db_, fp: "/local/a.mkv") + started = _library_item(db, "starting") + user = User(email="starter@example.com") + db.add(user) + db.commit() + paused = _Session(tmp_path, "paused", finished=True, idle_s=90) + sessions["someone-else"] = paused + + def fake_start(db_, item, start, audio=None, aoff=0.0, multi=False, transcode=False): + return SimpleNamespace( + id="sid", + media_start_s=start, + entry="index.m3u8", + transcode=transcode, + hw=False, + tonemap=False, + ) + + monkeypatch.setattr(stream, "start_session", fake_start) + try: + resp = _client(db, user).post(f"/api/plex/stream/{started.rating_key}/session") + finally: + app.dependency_overrides.clear() + assert resp.status_code == 200, resp.text + assert "someone-else" in sessions and not paused.proc.killed diff --git a/backend/tests/test_not_found_detail.py b/backend/tests/test_not_found_detail.py index b3616189..1510fff9 100644 --- a/backend/tests/test_not_found_detail.py +++ b/backend/tests/test_not_found_detail.py @@ -51,3 +51,24 @@ def test_a_nul_in_a_request_string_is_a_422_not_a_500(db, make_user, as_user): 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 + + +def test_any_other_data_error_is_still_a_500(): + """The other half of the app-wide handler, and the half a refactor can delete in silence: it is + registered for EVERY DataError on every route and decides on a substring of the driver's message, + so only a NUL may become a 422. Without this, simplifying it to an unconditional 422 — it reads + like dead defensive code — would answer a numeric overflow or an over-long ingested title with + "your input was bad", losing the traceback that names the real fault (deep-review round 5).""" + import asyncio + + import pytest + from sqlalchemy.exc import DataError + + from app.main import _nul_in_text + + overflow = DataError("SELECT 1", {}, Exception("numeric field overflow")) + with pytest.raises(DataError): + asyncio.run(_nul_in_text(None, overflow)) # re-raised: it stays the 500 it always was + + nul = DataError("SELECT 1", {}, Exception("invalid byte sequence: NUL (0x00)")) + assert asyncio.run(_nul_in_text(None, nul)).status_code == 422 diff --git a/backend/tests/test_plex_hls_file_route.py b/backend/tests/test_plex_hls_file_route.py new file mode 100644 index 00000000..8be286c7 --- /dev/null +++ b/backend/tests/test_plex_hls_file_route.py @@ -0,0 +1,145 @@ +"""An HLS artifact request names the session it belongs to, and gets nothing else. + +Sessions are keyed by rating_key ALONE, while a session's directory is per (key, offset, tag). So a +seek-restart — or a second viewer of the same item — replaces the registered session, and the old +viewer's next request for `seg_12.ts`, resolved by rating_key, was answered 200 with the NEW +session's twelfth segment: bytes produced from a different `-ss` offset under `-copyts`, i.e. a +timeline the client is not on. Wrong content with a success status is the one failure a player cannot +notice, and the reclaim-on-404 added in round 4 could never fire on it (deep-review round 5). + +DB lane (`db` fixture) — skipped when no Postgres is reachable. +""" + +import time + +import pytest +from fastapi.testclient import TestClient + +import app.state as state +from app.auth import current_user +from app.db import get_db +from app.main import app +from app.models import PlexItem, PlexLibrary, User +from app.plex import stream as plex_stream + + +class _Proc: + def poll(self): + return 0 # finished: `wait_for` may not poll out its whole timeout on a missing file + + def terminate(self): + pass + + def wait(self, timeout=None): + return 0 + + +class _Session: + def __init__(self, session_id: str, directory): + self.id = session_id + self.dir = directory + self.proc = _Proc() + self.last_access = time.time() + + +@pytest.fixture +def sessions(monkeypatch): + live: dict = {} + monkeypatch.setattr(plex_stream, "_sessions", live) + return live + + +def _client(db, monkeypatch): + monkeypatch.setattr(state, "_configured_cache", True) + lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + db.add(lib) + db.flush() + db.add( + PlexItem( + rating_key="m1", + library_id=lib.id, + kind="movie", + title="A Movie", + file_path="/media/a.mkv", + playable="remux", + ) + ) + user = User(email="hls@example.com") + db.add(user) + db.commit() + app.dependency_overrides[get_db] = lambda: db + app.dependency_overrides[current_user] = lambda: user + return TestClient(app) + + +def test_a_segment_is_served_from_the_session_the_url_names(db, monkeypatch, tmp_path, sessions): + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + (d / "seg_12.ts").write_bytes(b"mine") + sessions["m1"] = _Session("sid-a", d) + client = _client(db, monkeypatch) + try: + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") + finally: + app.dependency_overrides.clear() + assert r.status_code == 200, r.text + assert r.content == b"mine" + + +def test_a_segment_of_a_replaced_session_is_a_404_not_the_new_sessions_bytes( + db, monkeypatch, tmp_path, sessions +): + """Viewer A is at 0:00; viewer B seeks to 25:00 and their session takes the registry slot. A's + next segment must fail honestly — the player reclaims at the playhead — not arrive as B's.""" + theirs = tmp_path / "m1_1500_0_+0.00" + theirs.mkdir() + (theirs / "seg_12.ts").write_bytes(b"theirs") + sessions["m1"] = _Session("sid-b", theirs) + client = _client(db, monkeypatch) + try: + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") + finally: + app.dependency_overrides.clear() + assert r.status_code == 404, r.text + assert b"theirs" not in r.content + + +def test_a_playlist_with_no_live_session_is_a_404_not_a_silent_restart_at_zero( + db, monkeypatch, sessions +): + """The entry playlist used to auto-start a session when none was registered — always at 0.0, + silently throwing away the viewer's position. A 404 sends the player through its own reclaim, + which restarts at the playhead.""" + started: list[float] = [] + monkeypatch.setattr( + plex_stream, + "start_session", + lambda *a, **k: started.append(a[2] if len(a) > 2 else 0.0), + ) + client = _client(db, monkeypatch) + try: + r = client.get("/api/plex/stream/m1/hls/sid-gone/index.m3u8") + finally: + app.dependency_overrides.clear() + assert r.status_code == 404, r.text + assert started == [] + + +def test_a_segment_that_will_never_arrive_fails_at_once(db, monkeypatch, tmp_path, sessions): + """`stream_hls_file` is a SYNC route: it runs on the ~40-slot threadpool and holds a pooled DB + connection while it waits. Waiting without the producer's handle polled out the whole 20 s even + when that ffmpeg had already exited and no segment could ever appear — one ordinary seek past the + produced region parked a worker and a connection for 20 s (deep-review round 5).""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + sessions["m1"] = _Session("sid-a", d) # its _Proc.poll() reports a finished ffmpeg + client = _client(db, monkeypatch) + started = time.monotonic() + try: + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_99999.ts") + finally: + app.dependency_overrides.clear() + elapsed = time.monotonic() - started + assert r.status_code == 404, r.text + # The wait is 20 s; anything in that neighbourhood means the producer's exit was not consulted. + assert elapsed < 2.0, f"waited {elapsed:.1f}s for a segment that can never arrive" diff --git a/backend/tests/test_plex_playback_session.py b/backend/tests/test_plex_playback_session.py index 51a952ac..ec5d7331 100644 --- a/backend/tests/test_plex_playback_session.py +++ b/backend/tests/test_plex_playback_session.py @@ -58,7 +58,12 @@ def _post(db, user, monkeypatch, rating_key, *, captured): def fake_start(db_, it, start, audio, aoff, multi, transcode=False): captured["transcode"] = transcode return SimpleNamespace( - media_start_s=start, entry="index.m3u8", transcode=transcode, hw=False, tonemap=False + id="sid", + media_start_s=start, + entry="index.m3u8", + transcode=transcode, + hw=False, + tonemap=False, ) monkeypatch.setattr(plex_stream, "start_session", fake_start)