From 52b93d001404b6f8bf9a6efbd46eecb9bd5c48aa Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 28 Oct 2026 21:53:59 +0100 Subject: [PATCH] fix(plex): a restart that fails to start still deletes the session it moved aside --- backend/app/plex/stream.py | 132 +++++++++++++++--------------- backend/tests/test_plex_stream.py | 60 ++++++++++++++ 2 files changed, 128 insertions(+), 64 deletions(-) diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index 0f957a03..7c4ec00a 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -820,74 +820,78 @@ def start_session( # 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: - 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( - _ffmpeg_cmd( - src, - start_s, + # Finished even when the new session fails to start (no space, no ffmpeg, a DB error): what + # was moved aside under the lock is otherwise left for a restart's sweep. + try: + 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: + 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( + _ffmpeg_cmd( + src, + start_s, + directory, + audio_ord, + aoff, + audio_langs if is_multi else None, + transcode=transcode, + hw=hw, + tonemap=tonemap, + preset=preset, + crf=crf, + max_height=max_height, + loudnorm=loudnorm, + ), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + preexec_fn=preexec, + ) + s = HlsSession( + key, directory, - audio_ord, - aoff, - audio_langs if is_multi else None, + proc, + start_s, + variants, transcode=transcode, hw=hw, tonemap=tonemap, - preset=preset, - crf=crf, - max_height=max_height, - loudnorm=loudnorm, - ), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - preexec_fn=preexec, - ) - s = HlsSession( - key, - directory, - proc, - start_s, - variants, - transcode=transcode, - hw=hw, - tonemap=tonemap, - user_id=user_id, - ) - _sessions[reg] = s - # The one line that says which pipeline a session actually got. Everything downstream - # — the cap, the HW→SW fallback, the subtitle offset — is about this choice, and until - # now it was only inferable from the ffmpeg command nobody keeps. - log.info( - "an HLS session started", - extra=logs.event( - "plex.stream.session_started", - key=key, - transcode=transcode, - hw=hw, - tonemap=tonemap, - variants=len(variants or ()), - start_s=round(start_s, 1), - ), - ) - # Admin-configurable caps (Configuration → Plex); at least 1 each so playback can't be - # capped to zero. - 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) + user_id=user_id, + ) + _sessions[reg] = s + # The one line that says which pipeline a session actually got. Everything downstream + # — the cap, the HW→SW fallback, the subtitle offset — is about this choice, and until + # now it was only inferable from the ffmpeg command nobody keeps. + log.info( + "an HLS session started", + extra=logs.event( + "plex.stream.session_started", + key=key, + transcode=transcode, + hw=hw, + tonemap=tonemap, + variants=len(variants or ()), + start_s=round(start_s, 1), + ), + ) + # Admin-configurable caps (Configuration → Plex); at least 1 each so playback can't be + # capped to zero. + doomed += _enforce_caps( + user_id, + max(1, sysconfig.get_int(db, "plex_max_transcodes")), + max(1, sysconfig.get_int(db, "plex_max_sessions")), + ) + finally: + _finish(doomed) return s s = _spawn(*_select_encode(db, item, transcode, src)) diff --git a/backend/tests/test_plex_stream.py b/backend/tests/test_plex_stream.py index 9d56e8ca..379d02e1 100644 --- a/backend/tests/test_plex_stream.py +++ b/backend/tests/test_plex_stream.py @@ -1058,3 +1058,63 @@ class TestSessionsPerViewer: self._start(monkeypatch, user_id=uid) assert len(stream._sessions) == 3 and (1, "12345") not in stream._sessions assert oldest.proc.killed and not oldest.transcode + + +class TestTeardownFinishes: + """A session taken down by a cap or replaced by its own viewer is only renamed aside under the + lock; it is deleted after. The rename alone already moves the directory and `terminate()` alone + already marks the proc killed — so what is pinned here is the tomb: a whole segmented copy of the + film that only a restart's sweep would otherwise reclaim.""" + + capturing_env = TestTranscodeStartSession.capturing_env + + @staticmethod + def _left_behind(root): + return sorted(p.name for p in root.iterdir() if stream._TOMB in p.name) + sorted( + str(p) for p in stream._dying + ) + + def test_a_culled_session_is_deleted_not_only_moved_aside(self, capturing_env, monkeypatch): + monkeypatch.setattr(stream, "_dying", {}) + first = TestSessionsPerViewer._start(monkeypatch, user_id=1, rating_key="1") + first.last_access -= 60 + TestSessionsPerViewer._start(monkeypatch, user_id=1, rating_key="2") + TestSessionsPerViewer._start(monkeypatch, user_id=1, rating_key="3") + assert first.proc.killed and not first.dir.exists() + assert self._left_behind(capturing_env) == [] + + def test_a_viewers_restart_deletes_the_session_it_replaces(self, capturing_env, monkeypatch): + monkeypatch.setattr(stream, "_dying", {}) + monkeypatch.setattr( + stream, "_select_encode", lambda db, item, transcode, src: (False, False) + ) + monkeypatch.setattr( + stream, "_probe_first_pts", lambda seg: None + ) # no real segment to probe + old = stream.start_session(db=None, user_id=1, item=_FakeItem(), start_s=0.0) + new = stream.start_session(db=None, user_id=1, item=_FakeItem(), start_s=600.0) + assert old.proc.killed and new.dir.exists() and not old.dir.exists() + assert self._left_behind(capturing_env) == [] + + def test_a_restart_that_fails_to_start_still_deletes_what_it_replaced( + self, capturing_env, monkeypatch + ): + # The new ffmpeg cannot start (a full disk, no ffmpeg, no memory): the old session is already + # moved aside, and nothing else in this process would ever delete it. + monkeypatch.setattr(stream, "_dying", {}) + monkeypatch.setattr( + stream, "_select_encode", lambda db, item, transcode, src: (False, False) + ) + monkeypatch.setattr( + stream, "_probe_first_pts", lambda seg: None + ) # no real segment to probe + old = stream.start_session(db=None, user_id=1, item=_FakeItem(), start_s=0.0) + + def no_ffmpeg(*args, **kwargs): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(stream.subprocess, "Popen", no_ffmpeg) + with pytest.raises(OSError): + stream.start_session(db=None, user_id=1, item=_FakeItem(), start_s=600.0) + assert old.proc.killed and not old.dir.exists() + assert self._left_behind(capturing_env) == []