From 71d7646bf15366f170598bf940e3e3b1bb3e2002 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 12 Sep 2026 16:21:37 +0200 Subject: [PATCH] fix(plex): a name this session cannot write fails at once, and the layout has one spelling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 6 turned `artifact_may_arrive` into a positive rule, but keyed it to the HLS vocabulary in general rather than to what THIS session produces: the playlist regex accepted master/index/stream_* for every session and the `seq == 0` branch accepted any variant token unconditionally. The two argv branches are disjoint — a single-audio session is given `seg_%d.ts` + `index.m3u8` and no `%v` at all, so `master.m3u8`, `stream_*.m3u8` and every `seg__.ts` are names it can never write, and vice versa — so each of those was read as "an HLS name, so wait and see" and held a threadpool worker, plus its pooled DB connection, for the full timeout. That is verbatim the round-5 outage the route's own comment claims it prevents. Measured through the real route against a live single-audio session, before: master.m3u8 404 after 30.0s, stream_q7z.m3u8 404 after 30.0s, seg_a0_0.ts / seg_zzzz_0.ts / seg_0_0.ts 404 after 20.1s each. After, on the running localdev API: 404 in 44-316ms for all of them, while index.m3u8, seg_0.ts and seg_1.ts still answer 200 — nothing is over-refused. The session now carries the rendition names its ffmpeg was told to write, and `entry` is derived from them instead of passed. And "one place" now means the argv BUILDS its patterns from the same functions the parser reads back (`segment_name("%d", "%v")`, `variant_playlist_name("%v")`): the layout was still spelled out independently in six places, with `_VIDEO_VARIANT` read by no consumer, which is the drift that caused round 6's HIGH. A test asserts the argv's own patterns are what those functions produce, and that every name they expand to is one the guard accepts. The load-bearing MEASURED comment cited ffmpeg 8.1.1 — the Windows host's. The binary that writes these files is the API container's 7.1.5; re-measured there, the conclusion holds. --- backend/app/plex/stream.py | 115 ++++++++++++++++------ backend/app/routes/plex/playback.py | 12 ++- backend/tests/test_plex_hls_file_route.py | 107 +++++++++++++++++--- 3 files changed, 189 insertions(+), 45 deletions(-) diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index b6b51161..e65d65e8 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -26,6 +26,7 @@ import shutil import subprocess import threading import time +from collections.abc import Sequence from pathlib import Path from sqlalchemy.orm import Session as DbSession @@ -99,7 +100,7 @@ class HlsSession: directory: Path, proc: subprocess.Popen, start_s: float, - entry: str, + variants: "Sequence[str]", transcode: bool = False, hw: bool = False, tonemap: bool = False, @@ -126,7 +127,13 @@ class HlsSession: # subtitle cue-shift both key off the true content position — no drift. Provisional = X until # the first segment is probed. self.media_start_s = start_s - self.entry = entry # the playlist filename hls.js should load (master.m3u8 with subs, else index.m3u8) + # The rendition names this session's ffmpeg was told to write — () for a single-audio + # session. Held on the session because it is what `artifact_may_arrive` needs to tell a name + # this session CAN produce from one belonging to the other argv branch (see it). + self.variants = tuple(variants) + # The playlist filename hls.js should load: the master of a multi-audio session, else the + # single index. Derived, not passed, so it cannot disagree with the argv. + self.entry = _MASTER_PL if self.variants else _INDEX_PL self.last_access = time.time() @@ -509,7 +516,8 @@ def _ffmpeg_cmd( # switches audio CLIENT-SIDE on the same timeline (no session restart, no drift) — the premium # 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. + # …). Those names are spelled by `segment_name` / `variant_playlist_name` below, which the + # guard and the warm-up read back — see them for the measurement behind the ordering. args += ["-map", "0:v:0"] for i in range(len(audio_langs)): args += ["-map", f"{ai}:a:{i}?"] @@ -523,14 +531,16 @@ def _ffmpeg_cmd( if i == 0: entry += ",default:yes" var.append(entry) + # The name patterns come from the vocabulary functions, with ffmpeg's own placeholders passed + # through them — so the producer and the parser cannot spell the layout differently (round 7). args += _HLS_TAIL + [ "-master_pl_name", - "master.m3u8", + _MASTER_PL, "-var_stream_map", " ".join(var), "-hls_segment_filename", - str(out_dir / "seg_%v_%d.ts"), - str(out_dir / "stream_%v.m3u8"), + str(out_dir / segment_name("%d", "%v")), + str(out_dir / variant_playlist_name("%v")), ] else: ao = audio_ord if audio_ord is not None else 0 @@ -538,8 +548,8 @@ def _ffmpeg_cmd( args += [*venc, *_aenc_opts(loudnorm)] args += _HLS_TAIL + [ "-hls_segment_filename", - str(out_dir / "seg_%d.ts"), - str(out_dir / "index.m3u8"), + str(out_dir / segment_name("%d")), + str(out_dir / _INDEX_PL), ] return args @@ -627,6 +637,9 @@ def start_session( start_s = max(0.0, float(start_s)) audio_langs = _probe_audio_langs(src) if multi else [] is_multi = len(audio_langs) > 1 # only worth a master playlist when there really are ≥2 tracks + # The rendition names the argv below will produce — one source for the argv, the warm-up's + # segment names, the entry playlist and the artifact guard. + variants = session_variants(audio_langs if is_multi else None) # Loudness normalisation changes the AUDIO THE SEGMENTS CONTAIN, so it belongs in the directory # name beside the other parameters that do (offset, audio track, A/V offset): a session left over # from before the admin flipped the knob must not be mistaken for one that honours it. Appended to @@ -636,7 +649,7 @@ def start_session( directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" def _seg(i: int) -> Path: - return directory / segment_name(i, is_multi) + return directory / segment_name(i, _VIDEO_VARIANT if variants else None) def _spawn( hw: bool, tonemap: bool = False, *, replace_only: "HlsSession | None" = None @@ -691,7 +704,7 @@ def start_session( directory, proc, start_s, - "master.m3u8" if is_multi else "index.m3u8", + variants, transcode=transcode, hw=hw, tonemap=tonemap, @@ -763,27 +776,64 @@ def current_session(key: str) -> HlsSession | None: return s -# 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). +# The names ffmpeg writes into a session directory, in ONE place — and "one place" means the argv +# BUILDS its patterns from these same functions (`segment_name("%d", "%v")`), not that a comment +# says so. Getting this wrong is invisible in both directions: a name we failed to anticipate used +# to fall back to "wait and see" (round 6's HIGH), and now that the rule is positive it falls back +# to a 404 instead — so the producer and the parser have to be literally the same spelling. # -# 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. +# MEASURED (ffmpeg 7.1.5 in the API container — the binary that actually writes these files — and +# re-checked on 8.1.1): `-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. +# +# The two argv branches are DISJOINT vocabularies, which is why the guard below is keyed to the +# SESSION and not to HLS in general: a single-audio session can never write `master.m3u8`, +# `stream_*.m3u8` or any `seg__.ts`, and a multi-audio one can never write `index.m3u8` +# or `seg_.ts`. Accepting the union meant a hopeless name of the other shape was waited out for +# the full playlist/segment timeout — the threadpool parking this guard exists to end (round 7). +_MASTER_PL = "master.m3u8" +_INDEX_PL = "index.m3u8" _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_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 session_variants(audio_langs: "Sequence[str | None] | None") -> tuple[str, ...]: + """The `-var_stream_map` rendition names a session will write, or () for a single-audio one. + + The video rendition is left unnamed in the map, so ffmpeg falls back to its index + (`_VIDEO_VARIANT`); each audio rendition is named `aN` in track order.""" + if not audio_langs: + return () + return (_VIDEO_VARIANT, *(f"a{i}" for i in range(len(audio_langs)))) -def artifact_may_arrive(directory: Path, filename: str) -> bool: - """Whether waiting for `filename` can possibly end in a file. +def segment_name(seq: "int | str", variant: "str | None" = None) -> str: + """The file ffmpeg writes for the `seq`-th segment of `variant`. + + `variant=None` is a single-audio session, whose argv carries no `%v` field at all. `seq` and + `variant` are typed loosely because `_ffmpeg_cmd` builds its own `-hls_segment_filename` pattern + through this function, passing ffmpeg's `"%d"` / `"%v"` placeholders — that is what keeps the + producer and the parser from drifting, rather than a comment asking the next editor not to.""" + return f"seg_{variant}_{seq}.ts" if variant else f"seg_{seq}.ts" + + +def variant_playlist_name(variant: str) -> str: + """The sub-playlist ffmpeg writes for one rendition of a multi-audio session.""" + return f"stream_{variant}.m3u8" + + +def playlist_names(variants: "Sequence[str]") -> frozenset[str]: + """Every playlist name this session's argv can write — the master plus one per rendition, or + just `index.m3u8` for a single-audio session.""" + if not variants: + return frozenset({_INDEX_PL}) + return frozenset({_MASTER_PL, *(variant_playlist_name(v) for v in variants)}) + + +def artifact_may_arrive(s: "HlsSession", filename: str) -> bool: + """Whether waiting for `filename` can possibly end in a file OF THIS SESSION. 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 @@ -791,23 +841,32 @@ def artifact_may_arrive(directory: Path, filename: str) -> bool: 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. + Keyed to the SESSION's own vocabulary, not to HLS's: the two argv branches are disjoint, so a + name belonging to the other shape is as hopeless here as a random string. Reading those as "an + HLS name, so wait and see" left the whole class reachable — measured through the real route + against a live single-audio session, `master.m3u8` and `stream_q7z.m3u8` were 404s after 30.0 s + and `seg_a0_0.ts` / `seg_zzzz_0.ts` after 20.1 s each (deep-review round 7). + 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): + if filename in playlist_names(s.variants): return True m = _SEG_NAME_RE.match(filename) if m is None: return False variant, seq = m.group(1), int(m.group(2)) + # The variant field must be one THIS session writes: present and known for a multi-audio + # session, absent for a single-audio one. + if variant not in (s.variants or (None,)): + return False if seq == 0: return True - prev = f"seg_{variant}_{seq - 1}.ts" if variant else f"seg_{seq - 1}.ts" try: - return (directory / prev).exists() + return (s.dir / segment_name(seq - 1, variant)).exists() except OSError: return False diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 42105e09..d08b81ca 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -673,12 +673,14 @@ def stream_hls_file( if s is None or s.id != session_id: raise HTTPException(status_code=404, detail="No active playback session") path = 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 + # Only a name THIS session can actually produce is worth waiting for: its own playlists, and a + # segment of one of its own renditions whose predecessor is already on disk (ffmpeg writes them + # in order, so if it is not there this one is not next). The session is passed, not just its + # directory, because the single- and multi-audio argv branches spell their names differently and + # neither can produce the other's. 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): + if not plex_stream.artifact_may_arrive(s, 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 diff --git a/backend/tests/test_plex_hls_file_route.py b/backend/tests/test_plex_hls_file_route.py index 448b2bfc..81fb645c 100644 --- a/backend/tests/test_plex_hls_file_route.py +++ b/backend/tests/test_plex_hls_file_route.py @@ -39,11 +39,16 @@ class _Proc: class _Session: - def __init__(self, session_id: str, directory, running: bool = False): + def __init__(self, session_id: str, directory, running: bool = False, audio_langs=None): self.id = session_id self.dir = directory self.proc = _Proc(running) self.last_access = time.time() + # The real session derives both of these from the argv it spawned; the guard reads + # `variants` to tell a name this session can produce from one only the OTHER argv branch + # writes, so a stand-in without it would test a shape that cannot exist. + self.variants = plex_stream.session_variants(audio_langs) + self.entry = "master.m3u8" if self.variants else "index.m3u8" @pytest.fixture @@ -145,7 +150,7 @@ def test_a_multi_audio_segment_beyond_the_region_fails_at_once_too( 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) + sessions["m1"] = _Session("sid-a", d, running=True, audio_langs=["hu", "en"]) started = time.monotonic() r = client.get(f"/api/plex/stream/m1/hls/sid-a/{name}") elapsed = time.monotonic() - started @@ -161,7 +166,7 @@ def test_a_multi_audio_segment_whose_predecessor_exists_is_served(client, tmp_pa 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) + sessions["m1"] = _Session("sid-a", d, running=True, audio_langs=["hu", "en"]) 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" @@ -227,13 +232,91 @@ def test_a_dead_producer_is_not_waited_out(client, tmp_path, sessions): 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) +@pytest.mark.parametrize("audio_langs", [None, ["hu", "en"]]) +def test_the_name_ffmpeg_is_asked_for_is_a_name_the_guard_understands(tmp_path, audio_langs): + """One vocabulary, enforced. `segment_name` is what the warm-up and the PTS probe wait for, what + the argv's own `-hls_segment_filename` pattern is built from, and what `artifact_may_arrive` + parses back — they were three separate statements of the layout and two of them were wrong. Pin + them to each other instead.""" + sess = _Session("sid", tmp_path, audio_langs=audio_langs) + variant = sess.variants[0] if sess.variants else None + name = plex_stream.segment_name(3, variant) + prev = plex_stream.segment_name(2, variant) + assert not plex_stream.artifact_may_arrive(sess, name) (tmp_path / prev).write_bytes(b"x") - assert plex_stream.artifact_may_arrive(tmp_path, name) + assert plex_stream.artifact_may_arrive(sess, name) + + +def test_the_argv_asks_for_exactly_the_names_the_guard_accepts(tmp_path): + """The producer and the parser, pinned to each other through the argv itself. + + Round 6's HIGH was a guard that read `seg_%v_%d.ts` sequence-first, so it was inert on every + multi-audio item, and a warm-up that waited 8 s for a name ffmpeg never writes. The fix was to + give the layout one spelling — but a comment saying so is what had just failed, so assert that + the `-hls_segment_filename` pattern in the real argv is literally what `segment_name` produces, + and that every playlist the argv names is one `playlist_names` accepts.""" + for audio_langs in (None, ["hu", "en", "de"]): + args = plex_stream._ffmpeg_cmd( + tmp_path / "in.mkv", 0.0, tmp_path, 0, 0.0, audio_langs, loudnorm=False + ) + variants = plex_stream.session_variants(audio_langs) + sess = _Session("sid", tmp_path, audio_langs=audio_langs) + # The two patterns in the argv are the vocabulary's own spelling, `%d`/`%v` and all. + seg_pattern = args[args.index("-hls_segment_filename") + 1] + pl_pattern = args[-1].rsplit("/", 1)[-1].rsplit("\\", 1)[-1] + if variants: + assert seg_pattern.endswith(plex_stream.segment_name("%d", "%v")), seg_pattern + assert pl_pattern == plex_stream.variant_playlist_name("%v"), pl_pattern + assert args[args.index("-master_pl_name") + 1] == plex_stream._MASTER_PL + else: + assert seg_pattern.endswith(plex_stream.segment_name("%d")), seg_pattern + assert pl_pattern == plex_stream._INDEX_PL, pl_pattern + assert "-master_pl_name" not in args + + # And every name those patterns can actually expand to is one the guard lets through — the + # entry playlist hls.js is handed included, since a session whose own manifest is refused + # cannot start at all. + for v in variants or (None,): + assert plex_stream.artifact_may_arrive(sess, plex_stream.segment_name(0, v)) + if v is not None: + assert plex_stream.artifact_may_arrive(sess, plex_stream.variant_playlist_name(v)) + assert plex_stream.artifact_may_arrive(sess, sess.entry) + + +def test_a_name_only_the_other_argv_branch_writes_is_refused_at_once(tmp_path, client, sessions): + """The guard is keyed to THIS session's vocabulary, not to HLS in general. + + The two argv branches are disjoint: a single-audio session is given `seg_%d.ts` + `index.m3u8` + and no `%v` at all, so `master.m3u8`, `stream_*.m3u8` and every `seg__.ts` are names + it can never write — and vice versa. Accepting the union read them as "an HLS name, so wait and + see", and each one then held a threadpool worker and its pooled DB connection for the full + timeout: measured through this route against a live single-audio session, `master.m3u8` took + 30.0 s and `seg_a0_0.ts` 20.1 s to answer 404 (deep-review round 7).""" + single = tmp_path / "m1_0_0_+0.00" + single.mkdir() + sessions["m1"] = _Session("sid-a", single, running=True) + multi = tmp_path / "m2_0_multi_+0.00" + multi.mkdir() + sessions["m2"] = _Session("sid-b", multi, running=True, audio_langs=["hu", "en"]) + + # Each of these belongs to the other shape (or to no shape at all) and must be refused without + # a wait — the assertion is the ELAPSED time as much as the status. + for key, sid, names in ( + ("m1", "sid-a", ["master.m3u8", "stream_0.m3u8", "seg_0_0.ts", "seg_zzzz_0.ts"]), + ("m2", "sid-b", ["index.m3u8", "seg_0.ts", "seg_a9_0.ts", "stream_zz.m3u8"]), + ): + for name in names: + started = time.monotonic() + r = client.get(f"/api/plex/stream/{key}/hls/{sid}/{name}") + elapsed = time.monotonic() - started + assert r.status_code == 404, f"{name}: {r.text}" + assert elapsed < 2.0, ( + f"{name}: waited {elapsed:.1f}s for a name this session cannot write" + ) + + # And the control: the names each session CAN write are still waited for, not refused. + assert plex_stream.artifact_may_arrive(sessions["m1"], "index.m3u8") + assert plex_stream.artifact_may_arrive(sessions["m1"], "seg_0.ts") + assert plex_stream.artifact_may_arrive(sessions["m2"], "master.m3u8") + assert plex_stream.artifact_may_arrive(sessions["m2"], "stream_a1.m3u8") + assert plex_stream.artifact_may_arrive(sessions["m2"], "seg_a1_0.ts")