HLS sessions were keyed by the item, so a second account opening the same film killed the first one's session; the displaced player reclaimed at its playhead and evicted the other in turn. The registry is now keyed by (viewer, rating_key), segment requests resolve by both, and the session directory carries the viewer (the orphan sweep still recognises the older shapes). Only transcodes counted toward a cap, so remuxes — each a whole segmented copy of the film on the scratch volume — were bounded by the library size. Three bounds now apply after every start, each dropping the least recently used: two sessions per account, the transcode cap, and a new admin knob, plex_max_sessions (default 8), over every session. A superseded session request returns at once, since it waits on the killed ffmpeg, so an account cannot park more request workers than it may hold sessions. The subtitle extractor's full-pass slots were global and minutes long each; one account opening subtitles on film after film could hold them all. An account may now hold three of the six, and still gets windows past that.
205 lines
7.4 KiB
Python
205 lines
7.4 KiB
Python
"""Idle HLS sessions are reaped on a timer, and a viewer who is merely PAUSED keeps theirs.
|
|
|
|
plex/stream.py promises "a periodic reaper kills idle ones", but `reap_idle` ran only inside
|
|
POST /session: after the last playback, an abandoned session (its ffmpeg still working through the
|
|
film, its directory a whole segmented copy of it) lived until somebody pressed play again
|
|
(deep-review round 4).
|
|
|
|
And it must not take a PAUSED viewer's session: a remux finishes long before the film does, and a
|
|
paused player with a full buffer fetches nothing, so "finished and unfetched for 30 s" does not mean
|
|
"abandoned". Round 4 bought that safety by keeping EVERY finished session for the full ten-minute
|
|
idle window — one segmented copy of a film per item, on the volume the download center shares. Round
|
|
5 replaced the blanket retention with the signal that already existed: an open player checkpoints its
|
|
position every 10 s whether or not it is playing, and that route says "still here".
|
|
|
|
Both call sites are pinned here, because the bug each time was that only one of them moved: the
|
|
periodic pass, and the fact that starting a session no longer sweeps OTHER viewers' sessions at all.
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as app_main
|
|
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.plex import paths as plex_paths
|
|
from app.plex import stream
|
|
|
|
|
|
class _Configured:
|
|
def __enter__(self):
|
|
return None
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
def test_the_periodic_pass_is_the_one_that_runs_and_it_uses_the_short_rule(monkeypatch):
|
|
async def _no_sweep():
|
|
return None
|
|
|
|
reaped: list[int] = []
|
|
monkeypatch.setattr(app_main, "SessionLocal", _Configured)
|
|
monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
|
|
monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
|
|
monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
|
|
monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
|
|
monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
|
|
monkeypatch.setattr(app_main, "_HLS_REAP_EVERY_S", 0.01)
|
|
# `reap_idle`, i.e. the SHORT finished rule: a finished session nobody has said they still want
|
|
# is a whole copy of a film sitting on the download volume.
|
|
monkeypatch.setattr(app_main.plex_stream, "reap_idle", lambda: reaped.append(1) or 0)
|
|
|
|
async def _cycle():
|
|
async with app_main.lifespan(app):
|
|
await asyncio.sleep(0.2)
|
|
|
|
asyncio.run(_cycle())
|
|
assert len(reaped) >= 2, "no periodic reap ran while the app was up"
|
|
|
|
|
|
class _Proc:
|
|
def __init__(self, finished: bool):
|
|
self.finished = finished
|
|
self.killed = False
|
|
|
|
def poll(self):
|
|
return 0 if self.finished else None
|
|
|
|
def terminate(self):
|
|
self.killed = True
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
|
|
class _Session:
|
|
def __init__(self, tmp_path, name: str, finished: bool, idle_s: float):
|
|
self.dir = tmp_path / name
|
|
self.dir.mkdir()
|
|
self.proc = _Proc(finished)
|
|
self.last_access = time.time() - idle_s
|
|
|
|
|
|
@pytest.fixture
|
|
def sessions(monkeypatch):
|
|
live: dict = {}
|
|
monkeypatch.setattr(stream, "_sessions", live)
|
|
return live
|
|
|
|
|
|
def test_the_periodic_pass_reaps_a_finished_session_nobody_has_claimed(tmp_path, sessions):
|
|
sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45)
|
|
assert stream.reap_idle() == 1
|
|
assert sessions == {}
|
|
|
|
|
|
def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limit(
|
|
tmp_path, sessions
|
|
):
|
|
sessions["1"] = _Session(
|
|
tmp_path, "gone_done", finished=True, idle_s=stream._SESSION_IDLE_S + 5
|
|
)
|
|
sessions["2"] = _Session(
|
|
tmp_path, "gone_live", finished=False, idle_s=stream._SESSION_IDLE_S + 5
|
|
)
|
|
assert stream.reap_idle() == 2
|
|
assert sessions == {}
|
|
|
|
|
|
def test_touching_a_session_keeps_it_through_the_short_rule(tmp_path, sessions):
|
|
s = _Session(tmp_path, "paused", finished=True, idle_s=90)
|
|
sessions[(1, "12345")] = s
|
|
stream.touch_session(1, "12345")
|
|
assert stream.reap_idle() == 0
|
|
assert (1, "12345") in sessions and s.dir.exists() and not s.proc.killed
|
|
|
|
|
|
# --- the two call sites, driven through the real routes (DB lane) -------------------------------
|
|
|
|
|
|
def _library_item(db, rating_key: str, *, playable: str = "remux"):
|
|
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
it = PlexItem(
|
|
rating_key=rating_key,
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title="A Movie",
|
|
file_path="/media/a.mkv",
|
|
playable=playable,
|
|
)
|
|
db.add(it)
|
|
db.commit()
|
|
return it
|
|
|
|
|
|
def _client(db, user):
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
app.dependency_overrides[current_user] = lambda: user
|
|
return TestClient(app)
|
|
|
|
|
|
def test_a_paused_viewers_checkpoint_keeps_their_session_alive(db, monkeypatch, tmp_path, sessions):
|
|
"""The whole reason the short rule is safe again: a paused player fetches nothing, so
|
|
`last_access` never moved on its own — but it does keep checkpointing its position."""
|
|
monkeypatch.setattr(state, "_configured_cache", True)
|
|
it = _library_item(db, "paused1")
|
|
user = User(email="paused@example.com")
|
|
db.add(user)
|
|
db.commit()
|
|
s = _Session(tmp_path, "paused", finished=True, idle_s=90)
|
|
sessions[(user.id, "paused1")] = s
|
|
try:
|
|
resp = _client(db, user).post(
|
|
f"/api/plex/item/{it.rating_key}/progress",
|
|
json={"position_seconds": 600, "duration_seconds": 3000, "final": False},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
assert resp.status_code == 200, resp.text
|
|
assert stream.reap_idle() == 0, "the checkpoint did not say the viewer is still there"
|
|
assert (user.id, "paused1") in sessions and not s.proc.killed
|
|
|
|
|
|
def test_starting_a_session_leaves_another_viewers_paused_one_alone(
|
|
db, monkeypatch, tmp_path, sessions
|
|
):
|
|
"""The opportunistic sweep this route used to run was not scoped to the key being started: it
|
|
walked the WHOLE registry with the 30 s finished rule, so pressing play on any item killed a
|
|
paused viewer's session — the very case round 4 set out to protect (found in round 5)."""
|
|
monkeypatch.setattr(state, "_configured_cache", True)
|
|
monkeypatch.setattr(plex_paths, "local_media_path", lambda db_, fp: "/local/a.mkv")
|
|
started = _library_item(db, "starting")
|
|
user = User(email="starter@example.com")
|
|
db.add(user)
|
|
db.commit()
|
|
paused = _Session(tmp_path, "paused", finished=True, idle_s=90)
|
|
sessions[(999, "someone-else")] = paused
|
|
|
|
def fake_start(db_, item, user_id, start, audio=None, aoff=0.0, multi=False, transcode=False):
|
|
return SimpleNamespace(
|
|
id="sid",
|
|
media_start_s=start,
|
|
entry="index.m3u8",
|
|
transcode=transcode,
|
|
hw=False,
|
|
tonemap=False,
|
|
)
|
|
|
|
monkeypatch.setattr(stream, "start_session", fake_start)
|
|
try:
|
|
resp = _client(db, user).post(f"/api/plex/stream/{started.rating_key}/session")
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
assert resp.status_code == 200, resp.text
|
|
assert (999, "someone-else") in sessions and not paused.proc.killed
|