From e96a562f14cefc9dbf381bb645c83dfa41869413 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 28 Oct 2026 18:48:21 +0100 Subject: [PATCH] fix(plex): a segment name ffmpeg never spells is refused; teardown no longer holds the registry lock The artifact guard parsed the sequence with int(), so a zero-padded or non-ASCII-digit name read as a segment the session had reached and parked the sync route for the full 20 s wait. It now also requires the canonical spelling segment_name() produces. Stopping, replacing, culling or reaping a session waited up to 3 s for ffmpeg and deleted a whole segmented film under _lock, which every segment request and checkpoint takes. Teardown now signals the process and moves its directory aside under the lock, and waits/deletes after it. A new session on the same path still waits for the dying ffmpeg first (it writes by path), and the orphan sweep recognises a moved-aside directory left by a crash. --- backend/app/plex/stream.py | 111 ++++++++++++--- backend/app/routes/plex/playback.py | 7 +- backend/tests/test_hls_reaper.py | 1 + backend/tests/test_hls_teardown.py | 158 ++++++++++++++++++++++ backend/tests/test_plex_hls_file_route.py | 27 +++- backend/tests/test_plex_session_stop.py | 1 + backend/tests/test_plex_stream.py | 8 +- 7 files changed, 289 insertions(+), 24 deletions(-) create mode 100644 backend/tests/test_hls_teardown.py diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index f273710d..0f957a03 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -66,7 +66,9 @@ _SESSION_IDLE_S = 600 # reap a session with no access for this long # restart leaves them behind like any other. `sweep_orphans` deletes ONLY names # matching one of those two shapes: `_HLS_ROOT` is admin-configurable (PLEX_HLS_DIR), so it may # legitimately point at a shared scratch path we do not own, and a blanket "delete every directory -# under the root" would take the neighbours with it. +# under the root" would take the neighbours with it. A torn-down session's directory is renamed to +# its own name + `_TOMB` + 8 hex digits before it is deleted (see `_detach`), so a crash between the +# two leaves a name this sweep still recognises. # # The KEY is anchored to digits (a Plex rating_key is a number) — that is what keeps foreign names # out. A `.+` there let `snapshot_2024_backup_+1.00` through, and pairing `.+` with an optional tail @@ -74,8 +76,10 @@ _SESSION_IDLE_S = 600 # reap a session with no access for this long # non-numeric rating_key would merely leave its directory unswept, which is the safe direction to # fail. Covered by test_plex_stream.py — this regex is the only thing standing between the sweep # and someone else's data. +_TOMB = ".dead-" _SESSION_DIR_RE = re.compile( - r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$|^\d+_u\d+_\d+_[^_]+_[+-]\d+\.\d{2}$" + r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$" + rf"|^\d+_u\d+_\d+_[^_]+_[+-]\d+\.\d{{2}}({re.escape(_TOMB)}[0-9a-f]{{8}})?$" ) # When this process started. `sweep_orphans` only deletes directories older than this: everything # newer belongs to a session THIS run created, and since `start_session` reuses a directory path @@ -93,7 +97,7 @@ _MTIME_GRACE_S = 2.0 _lock = threading.Lock() # Keyed by VIEWER and item, (user_id, rating_key). Keyed by the item alone, a second account opening -# the same film `_kill`ed the first one's session: the displaced viewer got a 404, reclaimed at its +# the same film tore down the first one's session: the displaced viewer got a 404, reclaimed at its # playhead, and so evicted the other in turn — two people could not watch one film at once. (One # account on two devices still shares a session; that case ends in the player's bounded reclaim.) _sessions: dict[tuple[int, str], "HlsSession"] = {} @@ -641,29 +645,82 @@ def _probe_first_pts(seg: Path) -> float | None: return None -def _kill(s: HlsSession) -> None: +# Teardown is split in two, because every segment request and checkpoint takes `_lock`: waiting up to +# 3 s for an ffmpeg to exit and deleting a whole segmented film under it stalled every other +# viewer's playback for one viewer's seek or one reap pass. `_detach` (under the lock) signals the +# process and renames its directory aside — so the deletion can never hit a NEW session that reuses +# the path — and `_finish` (after the lock is released) waits and deletes. +# +# What the rename cannot cover is the dying ffmpeg itself: it writes by path, so until it has exited +# it can still put a segment, or its final ENDLIST playlist, into a new session's directory on the +# same path. `_dying` remembers it by that path, and `_spawn` waits it out — under the lock, on +# purpose — before it creates the directory again. +_dying: dict[Path, subprocess.Popen] = {} + + +def _await_exit(key: str, proc: subprocess.Popen) -> None: try: - s.proc.terminate() try: - s.proc.wait(timeout=3) + proc.wait(timeout=3) # silent-ok: the escalation IS the handling. A transcode that ignores TERM for three # seconds gets KILL, which is exactly what this branch does. except subprocess.TimeoutExpired: - s.proc.kill() + proc.kill() except Exception as exc: # noqa: BLE001 — teardown finishes whatever the process does # Not nothing: what survives this is an ffmpeg nobody owns any more, holding a VAAPI # handle and a core, and the transcode cap that is supposed to bound them counts rows in # `_sessions` rather than processes. One line is the only way that ever surfaces. + log.info( + "could not stop the transcode process", + extra=logs.event("plex.stream.kill_failed", key=key, reason=type(exc).__name__), + ) + + +def _detach(s: HlsSession) -> tuple[HlsSession, Path | None]: + """Caller holds _lock and has taken `s` out of `_sessions`. Hand the result to `_finish` once the + lock is released.""" + try: + s.proc.terminate() + except Exception as exc: # noqa: BLE001 — teardown finishes whatever the process does log.info( "could not stop the transcode process", extra=logs.event("plex.stream.kill_failed", key=s.key, reason=type(exc).__name__), ) - shutil.rmtree(s.dir, ignore_errors=True) + _dying[s.dir] = s.proc + tomb = s.dir.with_name(f"{s.dir.name}{_TOMB}{secrets.token_hex(4)}") + try: + s.dir.rename(tomb) + # silent-ok: nothing on disk (ffmpeg died before writing, or a sweep took it) — nothing to delete. + except FileNotFoundError: + return s, None + except OSError as exc: + # Cannot be moved aside, so it cannot be deleted outside the lock either: the path may be a + # new session's a moment from now. Deleted here instead, the old way. + log.info( + "a session directory could not be moved aside; deleted under the lock", + extra=logs.event("plex.stream.tomb_failed", key=s.key, reason=type(exc).__name__), + ) + shutil.rmtree(s.dir, ignore_errors=True) + return s, None + return s, tomb -def _enforce_caps(user_id: int, max_transcodes: int, max_sessions: int) -> None: - """Caller holds _lock. Three bounds, applied after a session is registered, each dropping the - least recently accessed sessions over it: +def _finish(doomed: "list[tuple[HlsSession, Path | None]]") -> None: + """The slow half of a teardown, called WITHOUT `_lock`.""" + for s, tomb in doomed: + _await_exit(s.key, s.proc) + with _lock: + if _dying.get(s.dir) is s.proc: + del _dying[s.dir] + if tomb is not None: + shutil.rmtree(tomb, ignore_errors=True) + + +def _enforce_caps( + user_id: int, max_transcodes: int, max_sessions: int +) -> "list[tuple[HlsSession, Path | None]]": + """Caller holds _lock, and `_finish`es what this returns once it has released it. Three bounds, + applied after a session is registered, each dropping the least recently accessed sessions over it: * this account's own sessions (`_MAX_SESSIONS_PER_USER`) — first, so a single account filling the box evicts itself before anyone else; @@ -676,6 +733,8 @@ def _enforce_caps(user_id: int, max_transcodes: int, max_sessions: int) -> None: and the player restarts at the playhead, so from the server nothing failed — and "my film keeps rebuffering" has no other explanation anywhere, hence the line.""" + doomed: list[tuple[HlsSession, Path | None]] = [] + def cull(pool: list[tuple[tuple[int, str], HlsSession]], cap: int, bound: str) -> None: if len(pool) <= cap: return @@ -690,12 +749,13 @@ def _enforce_caps(user_id: int, max_transcodes: int, max_sessions: int) -> None: sessions=len(pool), ), ) - _kill(sess) _sessions.pop(reg, None) + doomed.append(_detach(sess)) cull([kv for kv in _sessions.items() if kv[0][0] == user_id], _MAX_SESSIONS_PER_USER, "user") cull([kv for kv in _sessions.items() if kv[1].transcode], max_transcodes, "transcode") cull(list(_sessions.items()), max_sessions, "total") + return doomed def start_session( @@ -759,12 +819,18 @@ def start_session( # can't starve the co-located API/playback. preexec_fn runs in the forked child (POSIX); on a # platform without fork (the API always runs in a Linux container in prod) it's simply omitted. preexec = _sw_child_preexec if (transcode and not hw and os.name == "posix") else None + doomed: list[tuple[HlsSession, Path | None]] = [] with _lock: if replace_only is not None and _sessions.get(reg) is not replace_only: return None # superseded/culled while we waited out of lock — don't clobber the newer one old = _sessions.pop(reg, None) if old is not None: - _kill(old) + doomed.append(_detach(old)) + # An ffmpeg still exiting from THIS path (the old session, or one stopped a moment ago) + # would write into the directory about to be made; see `_dying`. + prev = _dying.pop(directory, None) + if prev is not None: + _await_exit(key, prev) shutil.rmtree(directory, ignore_errors=True) directory.mkdir(parents=True, exist_ok=True) proc = subprocess.Popen( @@ -816,11 +882,12 @@ def start_session( ) # Admin-configurable caps (Configuration → Plex); at least 1 each so playback can't be # capped to zero. - _enforce_caps( + doomed += _enforce_caps( user_id, max(1, sysconfig.get_int(db, "plex_max_transcodes")), max(1, sysconfig.get_int(db, "plex_max_sessions")), ) + _finish(doomed) return s s = _spawn(*_select_encode(db, item, transcode, src)) @@ -974,6 +1041,11 @@ def artifact_may_arrive(s: "HlsSession", filename: str) -> bool: # session, absent for a single-audio one. if variant not in (s.variants or (None,)): return False + # And the name must be the one ffmpeg SPELLS for that sequence: `int()` also reads `seg_007.ts` + # and a non-ASCII digit run as a sequence the session may have reached, and neither file is + # ever written — a yes there parked the request for the whole wait. + if filename != segment_name(seq, variant): + return False if seq == 0: return True try: @@ -1108,16 +1180,16 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: `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 + doomed: list[tuple[HlsSession, Path | None]] = [] with _lock: for reg, s in list(_sessions.items()): done = s.proc.poll() is not None idle = now - s.last_access if idle > _SESSION_IDLE_S or (done and idle > finished_idle_s): - _kill(s) _sessions.pop(reg, None) - dropped += 1 - return dropped + doomed.append(_detach(s)) + _finish(doomed) + return len(doomed) def stop_session(user_id: int, key: str, session_id: str) -> bool: @@ -1132,8 +1204,9 @@ def stop_session(user_id: int, key: str, session_id: str) -> bool: s = _sessions.get((user_id, key)) if s is None or s.id != session_id: return False - _kill(s) _sessions.pop((user_id, key), None) + doomed = [_detach(s)] + _finish(doomed) return True diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 77b7c231..e946d25d 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -632,9 +632,10 @@ def stream_hls_file( 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 + # The session was torn down between the wait and the read — teardown moves the whole + # directory aside and deletes it, 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 diff --git a/backend/tests/test_hls_reaper.py b/backend/tests/test_hls_reaper.py index c7dbac3e..844f5499 100644 --- a/backend/tests/test_hls_reaper.py +++ b/backend/tests/test_hls_reaper.py @@ -83,6 +83,7 @@ class _Proc: class _Session: def __init__(self, tmp_path, name: str, finished: bool, idle_s: float): + self.key = name self.dir = tmp_path / name self.dir.mkdir() self.proc = _Proc(finished) diff --git a/backend/tests/test_hls_teardown.py b/backend/tests/test_hls_teardown.py new file mode 100644 index 00000000..58fe2aa6 --- /dev/null +++ b/backend/tests/test_hls_teardown.py @@ -0,0 +1,158 @@ +"""Tearing a session down does not stall everybody else's playback. + +Every segment request (`current_session`) and every checkpoint (`touch_session`) takes the one +registry `_lock`. Stopping a session waits up to 3 s for its ffmpeg to exit and then deletes a whole +segmented film, ~1200 files — and that used to happen under the same lock, so one viewer's seek, a +cap cull or a reap pass held up every other viewer's HLS requests for seconds, each one a parked +threadpool worker and its pooled DB connection. + +What still has to be atomic is pinned too: a session directory is reused for the same item, offset +and tag, so a new session on that path must not start before the old ffmpeg — which writes by path +— has exited. +""" + +import threading +import time + +import pytest + +from app.plex import stream + + +class _SlowProc: + """An ffmpeg that takes a while to exit after TERM: `wait` blocks until `exited` is set.""" + + def __init__(self, order: list[str], name: str): + self.order = order + self.name = name + self.waiting = threading.Event() + self.exited = threading.Event() + + def poll(self): + return 0 if self.exited.is_set() else None + + def terminate(self): + pass + + def kill(self): + self.exited.set() + + def wait(self, timeout=None): + self.waiting.set() + if not self.exited.wait(timeout): + raise stream.subprocess.TimeoutExpired("ffmpeg", timeout) + self.order.append(f"{self.name} exited") + return 0 + + +class _Session: + def __init__(self, directory, proc, key: str): + self.id = f"sid-{key}" + self.key = key + self.dir = directory + self.proc = proc + self.last_access = time.time() + self.transcode = False + + +@pytest.fixture +def sessions(monkeypatch): + live: dict = {} + monkeypatch.setattr(stream, "_sessions", live) + monkeypatch.setattr(stream, "_dying", {}, raising=False) + return live + + +def test_stopping_a_session_does_not_hold_up_another_viewers_requests(tmp_path, sessions): + order: list[str] = [] + slow = _SlowProc(order, "a") + d = tmp_path / "1_u1_0_0_+0.00" + d.mkdir() + (d / "seg_0.ts").write_bytes(b"x") + sessions[(1, "a")] = _Session(d, slow, "a") + sessions[(2, "b")] = _Session(tmp_path / "2_u2_0_0_+0.00", _SlowProc(order, "b"), "b") + + stopper = threading.Thread(target=stream.stop_session, args=(1, "a", "sid-a")) + stopper.start() + try: + assert slow.waiting.wait(2.0), "the stop never waited for its ffmpeg" + got: list = [] + reader = threading.Thread(target=lambda: got.append(stream.current_session(2, "b"))) + reader.start() + reader.join(1.0) + assert not reader.is_alive(), "a segment request waited on another viewer's teardown" + assert got and got[0] is sessions[(2, "b")] + finally: + slow.exited.set() + stopper.join(5.0) + # The teardown still finishes: the session is gone and nothing of its directory is left behind. + assert (1, "a") not in sessions + assert sorted(p.name for p in tmp_path.iterdir()) == [] + + +def test_a_reap_pass_does_not_hold_up_another_viewers_requests(tmp_path, sessions): + order: list[str] = [] + slow = _SlowProc(order, "a") + d = tmp_path / "1_u1_0_0_+0.00" + d.mkdir() + gone = _Session(d, slow, "a") + gone.last_access = time.time() - stream._SESSION_IDLE_S - 5 + sessions[(1, "a")] = gone + sessions[(2, "b")] = _Session(tmp_path / "2_u2_0_0_+0.00", _SlowProc(order, "b"), "b") + + reaper = threading.Thread(target=stream.reap_idle) + reaper.start() + try: + assert slow.waiting.wait(2.0), "the reap never waited for its ffmpeg" + toucher = threading.Thread(target=stream.touch_session, args=(2, "b")) + toucher.start() + toucher.join(1.0) + assert not toucher.is_alive(), "a checkpoint waited on another viewer's teardown" + finally: + slow.exited.set() + reaper.join(5.0) + assert list(sessions) == [(2, "b")] + assert not d.exists() + + +class _FakeItem: + rating_key = "12345" + file_path = "/media/movie.mkv" + playable = "remux" + codec_video = "h264" + + +def test_a_new_session_on_a_dying_ones_path_starts_only_after_it_exited( + tmp_path, sessions, monkeypatch +): + """The same item at the same second gets the same directory. The old ffmpeg writes by PATH, so + while it is still winding down after TERM it can write a segment — or, on a clean exit, a final + playlist with ENDLIST — into the NEW session's directory.""" + order: list[str] = [] + procs = iter([_SlowProc(order, "old"), _SlowProc(order, "new")]) + + def popen(*a, **k): + order.append("popen") + return next(procs) + + monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path) + src = tmp_path / "movie.mkv" + src.write_bytes(b"x") + monkeypatch.setattr(stream.paths, "local_media_path", lambda db, fp: src) + monkeypatch.setattr(stream.sysconfig, "get_int", lambda db, k: 2) + monkeypatch.setattr(stream.sysconfig, "get_bool", lambda db, k: False) + monkeypatch.setattr(stream.subprocess, "Popen", popen) + monkeypatch.setattr(stream, "wait_for", lambda *a, **k: True) + + first = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0) + assert first is not None + stopper = threading.Thread(target=stream.stop_session, args=(7, "12345", first.id)) + stopper.start() + assert first.proc.waiting.wait(2.0) + threading.Timer(0.3, first.proc.exited.set).start() + second = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0) + stopper.join(5.0) + assert second is not None and second.dir == first.dir + assert order.index("old exited") < order.index("popen", 1) + assert second.dir.is_dir(), "the old teardown deleted the new session's directory" + second.proc.exited.set() diff --git a/backend/tests/test_plex_hls_file_route.py b/backend/tests/test_plex_hls_file_route.py index 56571696..bf58d4b4 100644 --- a/backend/tests/test_plex_hls_file_route.py +++ b/backend/tests/test_plex_hls_file_route.py @@ -198,6 +198,31 @@ def test_a_name_the_session_cannot_produce_fails_at_once(client, tmp_path, sessi assert elapsed < 2.0, f"waited {elapsed:.1f}s for a playlist this session never writes" +@pytest.mark.parametrize( + ("name", "audio_langs"), + [ + ("seg_00.ts", None), # `int()` reads it as segment 0, which needs no predecessor + ("seg_007.ts", None), # …as segment 7, whose predecessor seg_6.ts is on disk + ("seg_٧.ts", None), # an Arabic-Indic 7: `\d` and `int()` both take it + ("seg_a0_07.ts", ["hu", "en"]), + ("seg_0_00.ts", ["hu", "en"]), + ], +) +def test_a_segment_name_ffmpeg_never_spells_is_refused(tmp_path, name, audio_langs): + """ffmpeg writes `seg_%d.ts`: no padding, ASCII digits. A name that only PARSES to a sequence the + session has reached is still a file that will never appear, and saying yes to it parked the + request for the full 20 s wait — the parking this guard exists to end.""" + d = tmp_path / "m1_0_0_+0.00" + d.mkdir() + for v in plex_stream.session_variants(audio_langs) or (None,): + (d / plex_stream.segment_name(6, v)).write_bytes(b"x") + sess = _Session("sid-a", d, running=True, audio_langs=audio_langs) + assert not plex_stream.artifact_may_arrive(sess, name) + # The canonical spelling of the same next segment is still accepted. + v = (plex_stream.session_variants(audio_langs) or (None,))[0] + assert plex_stream.artifact_may_arrive(sess, plex_stream.segment_name(7, v)) + + @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 @@ -213,7 +238,7 @@ def test_an_absurd_segment_index_is_a_404_not_a_500(client, tmp_path, sessions, 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, + """Teardown moves the whole directory aside 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".""" diff --git a/backend/tests/test_plex_session_stop.py b/backend/tests/test_plex_session_stop.py index b375be7d..7f6c2c06 100644 --- a/backend/tests/test_plex_session_stop.py +++ b/backend/tests/test_plex_session_stop.py @@ -39,6 +39,7 @@ class _Proc: class _Session: def __init__(self, session_id: str, directory): self.id = session_id + self.key = "m1" self.dir = directory self.proc = _Proc() self.last_access = time.time() diff --git a/backend/tests/test_plex_stream.py b/backend/tests/test_plex_stream.py index baffdf28..9d56e8ca 100644 --- a/backend/tests/test_plex_stream.py +++ b/backend/tests/test_plex_stream.py @@ -26,6 +26,9 @@ from app.plex.stream import _SESSION_DIR_RE "12345_0_2_+0.00", # current shape, explicit audio_ord "12345_842_multi_-1.25", # multi-audio, negative offset "12345_842_multi_+12.50", + # a torn-down session's directory, moved aside to be deleted outside the lock — left behind + # only if the process died between the rename and the delete + "12345_u7_842_multi_+12.50.dead-0123abcd", ], ) def test_our_session_directories_are_swept(name): @@ -47,6 +50,9 @@ def test_our_session_directories_are_swept(name): "snapshot_2024_backup_+1.00", # session-ish tail, but the key isn't a rating_key "backup_2024_multi_+0.00", ".config", + "12345_u7_0_0_+0.00.dead-", # a tombstone suffix with no id + "12345_u7_0_0_+0.00.dead-notours!", + "backup_u7_0_0_+0.00.dead-0123abcd", ], ) def test_foreign_directories_are_left_alone(name): @@ -159,7 +165,7 @@ class TestSweepOrphans: class _FakeProc: - """Stand-in for the ffmpeg Popen — start_session only stores it (and _kill/reap poll it later).""" + """Stand-in for the ffmpeg Popen — start_session only stores it (and teardown/reap poll it later).""" def poll(self): return None