From 7f17b17c13d1b497333a6bac55e97b7acc28d4ba Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 12 Sep 2026 05:49:38 +0200 Subject: [PATCH] fix(plex): an HLS session's file names are one vocabulary, and a name it cannot produce fails at once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 5 added a guard so a request for a segment beyond the produced region answers instead of polling out the full timeout — the route is a SYNC one, so every such wait parks a threadpool worker and, through `current_user`, a pooled DB connection. Measured against real ffmpeg, the guard could not fire where it mattered. `-hls_segment_filename seg_%v_%d.ts` substitutes the VARIANT first: a dual-audio session writes seg_0_0.ts (video) and seg_a0_0.ts / seg_a1_0.ts (the `name:aN` renditions). The regex read those as sequence-first, so every video segment parsed as segment 0 and every audio segment did not parse at all — the guard was inert on every multi-audio item, which is most of this library. `_seg()` carried the same inversion: the warm-up waited 8 s for `seg_1_0.ts`, a name ffmpeg never writes, on every multi-audio session start AND every seek-restart. The layout is now one constant plus one parser (`segment_name` / `_SEG_NAME_RE`), with a test pinning them to each other, instead of three comments that could drift — and they had drifted, all three ways. The guard is also a positive rule now. It was a blocklist, so anything that was not a single-audio segment name was waved through: `GET …/anything.m3u8` still parked a worker for 30 s, while the route's comment claimed a flood of hopeless names was what it prevented. A session produces its playlists and its segments and nothing else. `.vtt` leaves `_HLS_CT` with it — ffmpeg runs with `-sn` and subtitles are a separate route, so that extension could only ever name a file that will not arrive. Also here: - A 5000-digit segment index reached the client as a 500: `int()` on an unbounded digit run raises past 4300 digits and only OSError was caught. The name is bounded on the route and the index in the regex — the same bug class the subtitle routes were bounded for. - A session torn down between the wait and the read is a 404, not a 500. `_kill` rmtrees the directory and no lock spans those two steps; the player's recovery is keyed on the STATUS, so a 500 skipped the reclaim and ended playback with "couldn't play" instead of restarting at the playhead. The trigger is ordinary — `loadSession` awaits the new session before destroying the old hls instance. - Two comments claimed an open player checkpoints every 10 s "whether or not it is playing", which is the stated reason the 30 s finished-idle reap is safe. `flush` sends nothing once the finish latch is up, or before a position is known. The states it does cover — playing or paused mid-item — are the ones the rule needs; the comments now say that instead of more. - The route's tests leave the DB lane: the route reads nothing from the database since the auto-start was removed, and there those four regressions SKIP on a machine with no Postgres while pytest prints green. 14 tests, mutation-checked: the old parse, the old warm-up name, the blocklist, both bounds off the absurd name, the unguarded read, and the route dropping `proc` — each red on the test that names it. --- backend/app/plex/stream.py | 74 ++++++--- backend/app/routes/plex/playback.py | 51 +++++-- backend/tests/test_plex_hls_file_route.py | 176 ++++++++++++++++------ 3 files changed, 222 insertions(+), 79 deletions(-) diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index 6ab7a95d..b6b51161 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -56,6 +56,10 @@ _SEG_SECONDS = 6 # worse than the old zero-buffer behaviour). _WARMUP_SEGMENTS = 3 _WARMUP_SEG_TIMEOUT_S = 8.0 +# The HLS variant ffmpeg gives the VIDEO rendition of a multi-audio session. `-var_stream_map` names +# each audio rendition (`name:a0`, `a1`, …) but leaves the video one unnamed, so ffmpeg falls back to +# its index for that one. See `segment_name` for why this is a constant and not a comment. +_VIDEO_VARIANT = "0" _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)}`. `sweep_orphans` deletes ONLY names @@ -503,7 +507,9 @@ def _ffmpeg_cmd( if audio_langs: # MULTI-RENDITION: map EVERY audio track as an alternate HLS rendition in one group, so hls.js # switches audio CLIENT-SIDE on the same timeline (no session restart, no drift) — the premium - # audio-switch path. Produces master.m3u8 + stream_%v.m3u8 (v0=video, v1..=audio) + seg_%v_%d.ts. + # audio-switch path. Produces master.m3u8 + one stream_%v.m3u8 and seg_%v_*.ts per variant, + # where %v is the video rendition's index ("0") and each audio rendition's `name:` (a0, a1, + # …) — see `_SEG_NAME_RE`, which is the one place that layout is written down. args += ["-map", "0:v:0"] for i in range(len(audio_langs)): args += ["-map", f"{ai}:a:{i}?"] @@ -630,8 +636,7 @@ def start_session( directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" def _seg(i: int) -> Path: - # multi-audio: the video rendition is variant v0 → seg_{i}_0.ts; single-audio → seg_{i}.ts - return directory / (f"seg_{i}_0.ts" if is_multi else f"seg_{i}.ts") + return directory / segment_name(i, is_multi) def _spawn( hw: bool, tonemap: bool = False, *, replace_only: "HlsSession | None" = None @@ -758,32 +763,51 @@ def current_session(key: str) -> HlsSession | None: return s -# A segment file, single-audio (`seg_12.ts`) or one rendition of a multi-audio session -# (`seg_12_0.ts`). ffmpeg writes them strictly in order, which is what makes `segment_may_arrive` -# exact rather than a heuristic. -_SEG_NAME_RE = re.compile(r"^seg_(\d+)(_\d+)?\.ts$") +# The names ffmpeg writes into a session directory, in ONE place. Getting this wrong is invisible — +# every path that reads a name falls back to "wait and see" — so it is a constant plus a matching +# parser, not three comments that can drift (they did: deep-review round 6 found the regex, `_seg` +# and the comment above `_ffmpeg_cmd` all disagreeing). +# +# MEASURED (ffmpeg 8.1.1, this repo's own argv): `-hls_segment_filename seg_%v_%d.ts` substitutes +# `%v` with the VARIANT and `%d` with the SEQUENCE — variant FIRST. A dual-audio session writes +# seg_0_0.ts… (video, the unnamed variant) plus seg_a0_0.ts… / seg_a1_0.ts… (the `name:aN` audio +# renditions), and the playlists beside them are master.m3u8 + stream_0/a0/a1.m3u8. Single-audio +# uses `seg_%d.ts` + index.m3u8, with no variant field at all. +_SEG_NAME_RE = re.compile(r"^seg_(?:([0-9a-z]{1,4})_)?(\d{1,9})\.ts$") +_PLAYLIST_NAME_RE = re.compile(r"^(?:master|index|stream_[0-9a-z]{1,4})\.m3u8$") -def segment_may_arrive(directory: Path, filename: str) -> bool: +def segment_name(seq: int, is_multi: bool) -> str: + """The file ffmpeg writes for the `seq`-th VIDEO segment of a session.""" + return f"seg_{_VIDEO_VARIANT}_{seq}.ts" if is_multi else f"seg_{seq}.ts" + + +def artifact_may_arrive(directory: Path, filename: str) -> bool: """Whether waiting for `filename` can possibly end in a file. - ffmpeg produces segments sequentially, so a segment whose PREDECESSOR does not exist yet is not - the next one — it is somewhere beyond the generated region, and no amount of waiting will - conjure it. Saying so at once matters because the route that waits is a SYNC one: it holds a - thread off the ~40-slot pool, and a pooled DB connection with it, for the whole timeout. An - ordinary seek past the produced region was enough to park one for 20 s, and forty concurrent - requests for a name that will never exist parked the whole API (deep-review round 5). + Positive rule, not a blocklist: a session produces segments and playlists and NOTHING else, so + any other name is refused at once. ffmpeg also produces segments strictly in order, so a segment + whose PREDECESSOR does not exist yet is not the next one — it is beyond the generated region, + and no amount of waiting will conjure it. Playlists are the one thing that must still be waited + for: they are rewritten continuously and have no ordering to read. - Anything that is not a segment (a playlist, a WebVTT track) is left to the timeout: those are - rewritten continuously and have no ordering to read.""" + Saying no at once matters because the route that waits is a SYNC one: it holds a thread off the + ~40-slot pool, and a pooled DB connection with it, for the whole timeout. An ordinary seek past + the produced region was enough to park one for 20 s, and forty concurrent requests for a name + that will never exist parked the whole API (deep-review round 5). Round 5's guard read only + single-audio segment names, so both halves of that were still fully reachable: every multi-audio + name fell through, and so did any `.m3u8`/`.vtt` name at all.""" + if _PLAYLIST_NAME_RE.match(filename): + return True m = _SEG_NAME_RE.match(filename) if m is None: + return False + variant, seq = m.group(1), int(m.group(2)) + if seq == 0: return True - idx = int(m.group(1)) - if idx == 0: - return True + prev = f"seg_{variant}_{seq - 1}.ts" if variant else f"seg_{seq - 1}.ts" try: - return (directory / f"seg_{idx - 1}{m.group(2) or ''}.ts").exists() + return (directory / prev).exists() except OSError: return False @@ -873,9 +897,13 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: 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 + What makes the short rule safe again is that fetching is no longer the only signal: a player + parked MID-ITEM 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). The heartbeat + stops once the item is finished or before a position is known (`usePlexProgress.flush`) — both + states where reaping is correct, and where the player's 404 → reclaim recovers a viewer who + scrubs back in anyway. Do not restate this as "every 10 s whether or not it is playing": round 6 + found that claim in two comments and not in the code. 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() diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 5cd12387..42105e09 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -323,9 +323,13 @@ 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). + # A player parked mid-item — playing or PAUSED — checkpoints every 10 s, so this is the signal + # that such a 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). It is NOT every open player: `usePlexProgress.flush` sends nothing once the + # finish latch is up, or before the first position is known. Those are the states in which + # reaping is the right answer anyway, and a viewer who scrubs back into a reaped session gets + # the same 404 → reclaim the player already handles. plex_stream.touch_session(str(rating_key)) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) @@ -622,11 +626,18 @@ def stream_file( return FileResponse(str(src)) # Starlette honours the Range header (206) +# What a session directory can contain, and nothing else. No `.vtt`: ffmpeg runs with `-sn` and +# subtitles are independent tracks served by GET /subtitle, so a WebVTT name here could only ever be +# a name that will not arrive — and accepting it cost a 20 s parked worker per request. _HLS_CT = { ".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t", - ".vtt": "text/vtt", } +# A real artifact name is ~20 characters. The bound is what keeps an absurd one out of the parser: +# `int()` on an unbounded digit run raises ValueError past 4300 digits (CPython's +# `sys.get_int_max_str_digits`), which reached the client as a 500 — the same shape of bug the +# subtitle routes were bounded for (`_RatingKey`, above). +_HLS_NAME_MAX = 128 @router.get("/stream/{rating_key}/hls/{session_id}/{filename}") @@ -650,17 +661,24 @@ def stream_hls_file( 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: + if ( + len(filename) > _HLS_NAME_MAX + or 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 or s.id != session_id: raise HTTPException(status_code=404, detail="No active playback session") path = s.dir / filename - # A segment beyond the generated region can never arrive: ffmpeg writes them in order, so if the - # predecessor is not there this one is not next. Answering now instead of waiting it out is what - # keeps a seek past the produced region — or a deliberate flood of hopeless names — from parking - # threadpool workers and their DB connections (see `segment_may_arrive`). - if not plex_stream.segment_may_arrive(s.dir, filename): + # Only a name this session can actually produce is worth waiting for: the playlists, and a + # segment whose predecessor is already on disk (ffmpeg writes them in order, so if it is not + # there this one is not next). Answering now instead of waiting it out is what keeps a seek past + # the produced region — or a deliberate flood of hopeless names — from parking threadpool + # workers and their DB connections (see `artifact_may_arrive`). + if not plex_stream.artifact_may_arrive(s.dir, filename): raise HTTPException(status_code=404, detail="Not available yet") timeout = 30 if ext == ".m3u8" else 20 # `proc` so a session whose ffmpeg has already exited answers at once instead of polling out the @@ -669,8 +687,19 @@ def stream_hls_file( # 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") + try: + content = path.read_bytes() + except OSError: + # The session was torn down between the wait and the read — `_kill` rmtrees the whole + # directory, and nothing holds `_lock` across those two steps. It has to present as the same + # 404 a gone session does: the player's recovery is keyed on the STATUS (`code === 404`), so + # a 500 here skipped the reclaim and ended playback with "couldn't play" instead of + # restarting at the playhead. Ordinary trigger: `loadSession` awaits the new session — which + # kills and rmtrees the old one — before destroying the old hls instance, so the outgoing + # instance's in-flight segment request races the rmtree on every seek-restart. + raise HTTPException(status_code=404, detail="No active playback session") from None return Response( - content=path.read_bytes(), + content=content, media_type=_HLS_CT[ext], headers={"Cache-Control": "no-store"}, ) diff --git a/backend/tests/test_plex_hls_file_route.py b/backend/tests/test_plex_hls_file_route.py index 17c3ab72..448b2bfc 100644 --- a/backend/tests/test_plex_hls_file_route.py +++ b/backend/tests/test_plex_hls_file_route.py @@ -7,7 +7,9 @@ session's twelfth segment: bytes produced from a different `-ss` offset under `- 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. +No `db` fixture: the route reads nothing from the database (its own `Depends(get_db)` went with the +removed auto-start, and `current_user` is overridden wholesale), so these must not sit in the DB +lane — there they SKIP on a machine with no Postgres while pytest prints green (round 6). """ import time @@ -17,9 +19,8 @@ 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.models import User from app.plex import stream as plex_stream @@ -52,46 +53,30 @@ def sessions(monkeypatch): return live -def _client(db, monkeypatch): +@pytest.fixture +def client(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) + try: + yield TestClient(app) + finally: + app.dependency_overrides.pop(current_user, None) -def test_a_segment_is_served_from_the_session_the_url_names(db, monkeypatch, tmp_path, sessions): +def test_a_segment_is_served_from_the_session_the_url_names(client, tmp_path, sessions): d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_11.ts").write_bytes(b"") # its predecessor, so seg_12 is the next one (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() + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") 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 + client, 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.""" @@ -100,17 +85,13 @@ def test_a_segment_of_a_replaced_session_is_a_404_not_the_new_sessions_bytes( (theirs / "seg_11.ts").write_bytes(b"") (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() + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") 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 + client, 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, @@ -121,16 +102,12 @@ def test_a_playlist_with_no_live_session_is_a_404_not_a_silent_restart_at_zero( "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() + r = client.get("/api/plex/stream/m1/hls/sid-gone/index.m3u8") 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): +def test_a_segment_that_will_never_arrive_fails_at_once(client, 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. It used to poll out the whole 20 s for a name that could never appear — an ordinary seek past the produced region was enough to park a worker and a connection, and @@ -141,13 +118,122 @@ def test_a_segment_that_will_never_arrive_fails_at_once(db, monkeypatch, tmp_pat d.mkdir() (d / "seg_0.ts").write_bytes(b"x") # the session is producing; seg_99999 is simply not next sessions["m1"] = _Session("sid-a", d, running=True) - 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() + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_99999.ts") 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" + + +@pytest.mark.parametrize( + "name", + [ + "seg_0_9000.ts", # the VIDEO rendition of a multi-audio session: variant 0, segment 9000 + "seg_a0_9000.ts", # the first audio rendition: `name:a0` from -var_stream_map + ], +) +def test_a_multi_audio_segment_beyond_the_region_fails_at_once_too( + client, tmp_path, sessions, name +): + """The round-5 guard read `seg__.ts`, but ffmpeg substitutes `%v` FIRST — so the + real names are `seg__.ts` and every multi-audio request fell through to the full + 20 s wait: the video rendition because its variant "0" parsed as segment 0, the audio renditions + (`a0`, `a1`) because they did not parse at all. Dual-audio is the common shape in this library, + so the guard was inert exactly where it was needed (round 6).""" + d = tmp_path / "m1_0_multi_+0.00" + d.mkdir() + (d / "seg_0_0.ts").write_bytes(b"x") + (d / "seg_a0_0.ts").write_bytes(b"x") + sessions["m1"] = _Session("sid-a", d, running=True) + started = time.monotonic() + r = client.get(f"/api/plex/stream/m1/hls/sid-a/{name}") + elapsed = time.monotonic() - started + assert r.status_code == 404, r.text + assert elapsed < 2.0, f"waited {elapsed:.1f}s for {name}" + + +def test_a_multi_audio_segment_whose_predecessor_exists_is_served(client, tmp_path, sessions): + """The other half of the same rule: reading the names correctly must not start refusing real + ones. `seg_a0_1.ts` follows `seg_a0_0.ts` — the previous SEGMENT of that rendition, not the + previous variant.""" + d = tmp_path / "m1_0_multi_+0.00" + d.mkdir() + (d / "seg_a0_0.ts").write_bytes(b"x") + (d / "seg_a0_1.ts").write_bytes(b"audio") + sessions["m1"] = _Session("sid-a", d, running=True) + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_a0_1.ts") + assert r.status_code == 200, r.text + assert r.content == b"audio" + + +def test_a_name_the_session_cannot_produce_fails_at_once(client, tmp_path, sessions): + """The guard was a blocklist: anything that was not a single-audio segment name was waved + through to the timeout, so `GET …/anything.m3u8` still parked a worker for the full 30 s — and + the route comment claimed a flood of hopeless names was what it prevented. A session produces + its playlists and its segments and nothing else, so any other name is answerable now.""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + sessions["m1"] = _Session("sid-a", d, running=True) + started = time.monotonic() + r = client.get("/api/plex/stream/m1/hls/sid-a/nope.m3u8") + elapsed = time.monotonic() - started + assert r.status_code == 404, r.text + assert elapsed < 2.0, f"waited {elapsed:.1f}s for a playlist this session never writes" + + +@pytest.mark.parametrize("digits", [100, 5000]) +def test_an_absurd_segment_index_is_a_404_not_a_500(client, tmp_path, sessions, digits): + """`int()` on an unbounded digit run raises ValueError past 4300 digits, and only OSError was + caught — so a long enough name reached the client as a 500 with a traceback. Same bug class the + subtitle routes were bounded for; the answer is the same 404 any other unproducible name gets.""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + sessions["m1"] = _Session("sid-a", d, running=True) + r = client.get(f"/api/plex/stream/m1/hls/sid-a/seg_{'9' * digits}.ts") + assert r.status_code == 404, r.text + + +def test_a_session_torn_down_during_the_read_is_a_404_not_a_500( + client, monkeypatch, tmp_path, sessions +): + """`_kill` rmtrees the whole directory and nothing holds `_lock` across the wait and the read, + so the file can vanish in between — on every seek-restart, since `loadSession` awaits the new + session before destroying the old hls instance. The player's recovery is keyed on the 404 + STATUS, so a 500 here skipped the reclaim and ended playback with "couldn't play".""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + (d / "seg_11.ts").write_bytes(b"") + sessions["m1"] = _Session("sid-a", d) + # The file existed when the wait checked, and is gone by the time the route reads it. + monkeypatch.setattr(plex_stream, "wait_for", lambda *a, **k: True) + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") + assert r.status_code == 404, r.text + + +def test_a_dead_producer_is_not_waited_out(client, tmp_path, sessions): + """The route passes `proc` to `wait_for`. Without it a session whose ffmpeg has already exited + — a finished remux, a VAAPI init refusal — polls the whole 20 s for a segment that will never + be written, holding a threadpool worker and its pooled DB connection. The predecessor IS on + disk here, so `artifact_may_arrive` says yes and only `proc` can end the wait.""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + (d / "seg_5.ts").write_bytes(b"x") + sessions["m1"] = _Session("sid-a", d, running=False) + started = time.monotonic() + r = client.get("/api/plex/stream/m1/hls/sid-a/seg_6.ts") + elapsed = time.monotonic() - started + assert r.status_code == 404, r.text + assert elapsed < 2.0, f"waited {elapsed:.1f}s on a producer that had already exited" + + +@pytest.mark.parametrize("is_multi", [False, True]) +def test_the_name_ffmpeg_is_asked_for_is_a_name_the_guard_understands(tmp_path, is_multi): + """One vocabulary, enforced. `segment_name` is what the warm-up and the PTS probe wait for, and + `artifact_may_arrive` is what the route parses — they were three separate statements of the + layout and two of them were wrong. Pin them to each other instead.""" + name = plex_stream.segment_name(3, is_multi) + prev = plex_stream.segment_name(2, is_multi) + assert not plex_stream.artifact_may_arrive(tmp_path, name) + (tmp_path / prev).write_bytes(b"x") + assert plex_stream.artifact_may_arrive(tmp_path, name)