peter/siftlode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(plex): an HLS session's file names are one vocabulary, and a name it cannot produce fails at once

Browse Source 
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.
This commit is contained in:
peter
2026-09-12 05:49:38 +02:00
parent 1f81684b6e
commit 7f17b17c13
3 changed files with 222 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_plex_hls_file_route.py
+131 -45
View File
@@ -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_<seq>_<variant>.ts`, but ffmpeg substitutes `%v` FIRST — so the
real names are `seg_<variant>_<seq>.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)