fix(plex): a name this session cannot write fails at once, and the layout has one spelling

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_<variant>_<n>.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/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_<variant>_<n>.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")