diff --git a/backend/app/config.py b/backend/app/config.py index 2490b51a..8933fd62 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -315,6 +315,10 @@ class Settings(BaseSettings): # re-encode is CPU-heavy (the VAAPI hardware path is far cheaper and could sustain more); direct-serve # and remux (stream-copy) are not capped by this. plex_max_transcodes: int = 4 + # Max concurrent playback sessions of ANY kind — remux (stream-copy) included. Cheap on the CPU but + # each one is a whole segmented copy of the film on the scratch volume, written faster than + # realtime. One account holds at most two regardless (plex/stream.py `_MAX_SESSIONS_PER_USER`). + plex_max_sessions: int = 8 # --- Transcode (playable="transcode") re-encode tuning (E6 S1) --- # Hardware-encoder policy: "auto" = VAAPI (h264_vaapi) when the iGPU render node is present AND the # source codec is hardware-decodable, else libx264 software; "on" = force VAAPI wherever the render diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index 9975e1be..656dabd9 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -60,8 +60,10 @@ _WARMUP_SEG_TIMEOUT_S = 8.0 # its index for that one. See `segment_name` for why this is a constant and not a comment. _VIDEO_VARIANT = "0" _SESSION_IDLE_S = 600 # reap a session with no access for this long -# A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session); -# before multi-audio it was just `{rating_key}_{int(start_s)}`. `sweep_orphans` deletes ONLY names +# A session directory is named `{rating_key}_u{user_id}_{int(start_s)}_{tag}_{aoff:+.2f}` (see +# start_session); before per-viewer sessions it was `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}`, +# and before multi-audio just `{rating_key}_{int(start_s)}` — the older shapes stay sweepable, since a +# 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. @@ -72,7 +74,9 @@ _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. -_SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$") +_SESSION_DIR_RE = re.compile( + r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$|^\d+_u\d+_\d+_[^_]+_[+-]\d+\.\d{2}$" +) # 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 # for the same (rating_key, offset, tag), "stale name" is not the same as "stale directory". @@ -88,7 +92,16 @@ _PROCESS_START = time.time() _MTIME_GRACE_S = 2.0 _lock = threading.Lock() -_sessions: dict[str, "HlsSession"] = {} +# 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 +# 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"] = {} +# How many live sessions one account may hold at once (two devices, or a next item starting while the +# last one winds down). A new one past it replaces that account's least recently used — whose waiting +# session request then returns at once, since `wait_for` watches the killed ffmpeg — so one account +# can neither pile up film-sized scratch copies nor park more than this many request workers. +_MAX_SESSIONS_PER_USER = 2 class HlsSession: @@ -102,16 +115,17 @@ class HlsSession: transcode: bool = False, hw: bool = False, tonemap: bool = False, + user_id: int = 0, ): - # Unguessable per-session id, carried in the URLs the client is handed. Sessions are keyed - # by rating_key alone, so a seek-restart or a SECOND viewer of the same item replaces the - # registered one — and the old viewer's next request for `seg_12.ts`, resolved by rating_key, - # was answered 200 with the NEW session's twelfth segment, produced from a different offset + # Unguessable per-session id, carried in the URLs the client is handed. A seek-restart + # replaces the viewer's registered session for the item — and the old player's next request + # for `seg_12.ts`, resolved by (viewer, rating_key) alone, was answered 200 with the NEW session's twelfth segment, produced from a different offset # under `-copyts`. Wrong bytes, not an error, so the client's reclaim never fired on it # (deep-review round 5). A request naming a session that is no longer registered is a 404, # which the player recovers from by restarting at the playhead. self.id = secrets.token_urlsafe(9) - self.key = key + self.key = key # the item's rating_key (what the logs name) + self.user_id = user_id # whose session it is: the registry key is (user_id, key) self.dir = directory self.proc = proc self.transcode = transcode # video re-encoded (non-h264 source), not stream-copied @@ -637,29 +651,47 @@ def _kill(s: HlsSession) -> None: shutil.rmtree(s.dir, ignore_errors=True) -def _enforce_cap(cap: int) -> None: - # Caller holds _lock. Only TRANSCODE sessions count toward plex_max_transcodes — remux - # (stream-copy) and direct-serve are cheap and deliberately uncapped (see config.py: - # plex_max_transcodes, and the user-facing config hint). Drop the least-recently-accessed - # transcode sessions over the cap, leaving any concurrent remux streams untouched. - tx = [(key, s) for key, s in _sessions.items() if s.transcode] - if len(tx) <= cap: - return - for key, s in sorted(tx, key=lambda kv: kv[1].last_access)[: len(tx) - cap]: - # Somebody's playback stops here. The route turns the missing session into a retryable - # 404 and the player restarts at the playhead, so from the server nothing failed — and - # "my film keeps rebuffering" has no other explanation anywhere. - log.info( - "a transcode session was dropped to stay under the cap", - extra=logs.event("plex.stream.session_culled", key=key, cap=cap, sessions=len(tx)), - ) - _kill(s) - _sessions.pop(key, None) +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: + + * this account's own sessions (`_MAX_SESSIONS_PER_USER`) — first, so a single account filling the + box evicts itself before anyone else; + * TRANSCODE sessions (plex_max_transcodes): the CPU/GPU-heavy ones; + * EVERY session (plex_max_sessions). A remux is cheap on the CPU but not on the disk: each one is a + whole segmented copy of the film under `_HLS_ROOT`, written faster than realtime. Counted only + by the transcode cap, N distinct items were N ffmpegs and N film-sized scratch directories. + + Somebody's playback stops at each drop. The route turns the missing session into a retryable 404 + 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.""" + + def cull(pool: list[tuple[tuple[int, str], HlsSession]], cap: int, bound: str) -> None: + if len(pool) <= cap: + return + for reg, sess in sorted(pool, key=lambda kv: kv[1].last_access)[: len(pool) - cap]: + log.info( + "a session was dropped to stay under a cap", + extra=logs.event( + "plex.stream.session_culled", + key=sess.key, + bound=bound, + cap=cap, + sessions=len(pool), + ), + ) + _kill(sess) + _sessions.pop(reg, None) + + 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") def start_session( db: DbSession, item: PlexItem, + user_id: int, start_s: float, audio_ord: int | None = None, aoff: float = 0.0, @@ -679,6 +711,7 @@ def start_session( # session respawn (the HW->SW fallback) must not pick up a mid-playback config change. loudnorm = sysconfig.get_bool(db, "plex_audio_loudnorm") key = item.rating_key + reg = (user_id, key) # the registry key: one session per viewer per item start_s = max(0.0, float(start_s)) audio_langs = _probe_audio_langs(src) if multi else [] is_multi = len(audio_langs) > 1 # only worth a master playlist when there really are ≥2 tracks @@ -691,7 +724,7 @@ def start_session( # the tag with a DASH — `_SESSION_DIR_RE` treats the tag as one underscore-free field, and that # regex is the only thing keeping the orphan sweep off a shared scratch path (see its comment). tag = ("multi" if is_multi else str(audio_ord)) + ("-ln" if loudnorm else "") - directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" + directory = _HLS_ROOT / f"{key}_u{user_id}_{int(start_s)}_{tag}_{aoff:+.2f}" def _seg(i: int) -> Path: return directory / segment_name(i, _VIDEO_VARIANT if variants else None) @@ -717,9 +750,9 @@ 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 with _lock: - if replace_only is not None and _sessions.get(key) is not replace_only: + 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(key, None) + old = _sessions.pop(reg, None) if old is not None: _kill(old) shutil.rmtree(directory, ignore_errors=True) @@ -753,8 +786,9 @@ def start_session( transcode=transcode, hw=hw, tonemap=tonemap, + user_id=user_id, ) - _sessions[key] = s + _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. @@ -770,9 +804,13 @@ def start_session( start_s=round(start_s, 1), ), ) - # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes); - # at least 1 so playback can't be capped to zero. - _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes"))) + # Admin-configurable caps (Configuration → Plex); at least 1 each so playback can't be + # capped to zero. + _enforce_caps( + user_id, + max(1, sysconfig.get_int(db, "plex_max_transcodes")), + max(1, sysconfig.get_int(db, "plex_max_sessions")), + ) return s s = _spawn(*_select_encode(db, item, transcode, src)) @@ -798,7 +836,7 @@ def start_session( # turns a None return into a "Media file not found on disk" 404 — dishonest for a file that # exists. A (dead) session instead degrades to a retryable 404 ("No active playback session" / # "Not available yet") the client re-requests. - return current_session(key) or s + return current_session(user_id, key) or s s = respawned ready = wait_for(_seg(0), timeout=25.0, proc=s.proc) if ready: @@ -829,9 +867,9 @@ def start_session( return s -def current_session(key: str) -> HlsSession | None: +def current_session(user_id: int, key: str) -> HlsSession | None: with _lock: - s = _sessions.get(key) + s = _sessions.get((user_id, key)) if s is not None: s.last_access = time.time() return s @@ -1060,21 +1098,21 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: now = time.time() dropped = 0 with _lock: - for key, s in list(_sessions.items()): + 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(key, None) + _sessions.pop(reg, None) dropped += 1 return dropped -def touch_session(key: str) -> None: - """Say a viewer is still there for `key`, without fetching anything. `last_access` used to move +def touch_session(user_id: int, key: str) -> None: + """Say this viewer is still there for `key`, without fetching anything. `last_access` used to move only on an HLS file fetch, which a paused player does not do — so the only way to keep a paused viewer's session alive was to keep EVERY abandoned one just as long.""" with _lock: - s = _sessions.get(key) + s = _sessions.get((user_id, key)) if s is not None: s.last_access = time.time() diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py index 67f62cd2..374a731e 100644 --- a/backend/app/plex/subtitles.py +++ b/backend/app/plex/subtitles.py @@ -461,11 +461,18 @@ _FULL_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-ful # which time something has finished. _MAX_WINDOWS_INFLIGHT = 8 _MAX_FULL_INFLIGHT = 6 +# …of which one account may hold this many. The full-pass slots are minutes each, and a hostile (or +# merely curious) account opening subtitles on film after film could otherwise hold all of them while +# everyone else's tracks never completed. A viewer over their share still gets windows — the part +# that makes a track watchable now — just not another whole-file read until one of theirs ends. +_MAX_FULL_PER_USER = 3 _MAX_TRACK_WINDOWS = 2 _LOCK = threading.Lock() # Queued or running jobs → the absolute span a window job reads (None for a full pass). _INFLIGHT: dict[tuple, tuple[float, float] | None] = {} +# rating_key → the account whose poll queued that item's full pass (see _MAX_FULL_PER_USER). +_FULL_OWNERS: dict[str, int] = {} # Track-level failures: a track that cannot be served at all (no item, a bitmap track, Plex down…). _FAILURES: dict[str, tuple[str, float]] = {} # rating_key → when its full pass last failed (see FULL_RETRY_S). @@ -515,6 +522,7 @@ def reset_state() -> None: global _STOPPING with _LOCK: _INFLIGHT.clear() + _FULL_OWNERS.clear() _FAILURES.clear() _FULL_FAILED.clear() _WIN_FAILED.clear() @@ -598,7 +606,7 @@ def _window_failed(key: str, at: float) -> bool: return any(a <= at <= b for a, b, _ in _live_window_failures(key)) -def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: +def ensure(rating_key: str, sub_ord: int, at: float = 0.0, viewer: int = 0) -> dict: """State of one subtitle track, scheduling whatever extraction is still missing. NEVER blocks. ``ready`` = fully extracted · ``partial`` = usable, covers ``at``, still growing · ``preparing`` @@ -641,7 +649,14 @@ def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: if edge - at <= LOOKAHEAD_S and edge < _FULL_END_S: probe = edge + 1.0 want_window = not _window_failed(key, probe) - _schedule(rating_key, sub_ord, probe, want_window=want_window, want_full=not full_failed) + _schedule( + rating_key, + sub_ord, + probe, + want_window=want_window, + want_full=not full_failed, + viewer=viewer, + ) # `growing`: whether the track will fill in on its own. False once the full pass has failed and # is backed off — a partial track then stays partial until a window is asked for, so the player # may poll slowly (only to read ahead of an edge) instead of every two seconds for the whole film. @@ -667,7 +682,12 @@ def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None] def _schedule( - rating_key: str, sub_ord: int, at: float, want_window: bool, want_full: bool = True + rating_key: str, + sub_ord: int, + at: float, + want_window: bool, + want_full: bool = True, + viewer: int = 0, ) -> None: """Kick off the jobs this track still needs: the window that makes it watchable now, and the one full pass that makes it complete. A 1 s status poll (or a second viewer) joins a job already @@ -700,8 +720,15 @@ def _schedule( ident = ("win", rating_key, sub_ord, w_start) _INFLIGHT[ident] = (w_start, w_start + WINDOW_PRE_S + WINDOW_LEN_S) jobs.append((ident, w_start)) - if want_full and n_full < _MAX_FULL_INFLIGHT and ("full", rating_key) not in _INFLIGHT: + mine = sum(1 for owner in _FULL_OWNERS.values() if owner == viewer) + if ( + want_full + and n_full < _MAX_FULL_INFLIGHT + and ("full", rating_key) not in _INFLIGHT + and mine < _MAX_FULL_PER_USER + ): _INFLIGHT[("full", rating_key)] = None + _FULL_OWNERS[rating_key] = viewer jobs.append((("full", rating_key), None)) for ident, window_start in jobs: pool = _FULL_POOL if window_start is None else _WIN_POOL @@ -765,6 +792,8 @@ def _run( finally: with _LOCK: _INFLIGHT.pop(ident, None) + if ident[0] == "full": + _FULL_OWNERS.pop(rating_key, None) def _pass_failed( diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index bd23e609..fbda4a1d 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -121,7 +121,7 @@ def subtitle_status( `at` is the viewer's absolute position: the extractor windows around it, so what gets read first is the part actually about to be watched.""" - return subtitles.ensure(rating_key, sub_ord, at) + return subtitles.ensure(rating_key, sub_ord, at, viewer=user.id) @router.get("/subtitle/{rating_key}/{sub_ord}") @@ -362,7 +362,7 @@ def item_progress( # finish latch is up, or before the first position is known. Those are the states in which # reaping is the right answer anyway, and a viewer who scrubs back into a reaped session gets # the same 404 → reclaim the player already handles. - plex_stream.touch_session(str(rating_key)) + plex_stream.touch_session(user.id, str(rating_key)) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) final = body.final @@ -607,7 +607,7 @@ def stream_session( } if plex_paths.local_media_path(db, it.file_path) is None: raise HTTPException(status_code=404, detail="Media file not found on disk") - s = plex_stream.start_session(db, it, start, audio, aoff, multi, transcode=transcode) + s = plex_stream.start_session(db, it, user.id, start, audio, aoff, multi, transcode=transcode) if s is None: raise HTTPException(status_code=404, detail="Media file not found on disk") # Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the @@ -677,9 +677,9 @@ def stream_hls_file( references its segments/subtitles by relative name, all resolving back here. The session id in the path is what makes "back here" mean the same session. Sessions are keyed by - rating_key alone and a seek-restart or a second viewer of the same item replaces the registered - one, while the DIRECTORY is per (key, offset, tag) — so resolving by rating_key handed the old - viewer the new session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong + (viewer, rating_key) and a seek-restart replaces the viewer's registered one, while the DIRECTORY + is per (key, viewer, offset, tag) — so resolving by the key alone handed the old player the new + session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong bytes with a 200, which the player has no way to notice (deep-review round 5). A request for a session that is no longer the registered one is a 404, and the player reclaims at the playhead. @@ -695,7 +695,7 @@ def stream_hls_file( or ".." in filename ): raise HTTPException(status_code=404, detail="Not found") - s = plex_stream.current_session(str(rating_key)) + s = plex_stream.current_session(user.id, str(rating_key)) if s is None or s.id != session_id: raise HTTPException(status_code=404, detail="No active playback session") path = s.dir / filename diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index 41f1b7bd..4eeb0585 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -176,6 +176,7 @@ SPECS: tuple[ConfigSpec, ...] = ( ConfigSpec("plex_overlay_map", "str", "plex", "plex_overlay_map"), # (plex_sync_interval_min is tuned from the Scheduler dashboard, not here — it's a scheduler job.) ConfigSpec("plex_max_transcodes", "int", "plex", "plex_max_transcodes", min=0, max=16), + ConfigSpec("plex_max_sessions", "int", "plex", "plex_max_sessions", min=1, max=64), # Transcode re-encode tuning (E6 S1). The libx264 knobs apply only to the software fallback. # crf 0–51 (x264 range); max_height 0 = source. ConfigSpec("plex_transcode_hw", "choice", "plex", "plex_transcode_hw"), diff --git a/backend/tests/test_hls_reaper.py b/backend/tests/test_hls_reaper.py index 667474cc..b42b2ee4 100644 --- a/backend/tests/test_hls_reaper.py +++ b/backend/tests/test_hls_reaper.py @@ -116,10 +116,10 @@ def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limi 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["12345"] = s - stream.touch_session("12345") + sessions[(1, "12345")] = s + stream.touch_session(1, "12345") assert stream.reap_idle() == 0 - assert "12345" in sessions and s.dir.exists() and not s.proc.killed + 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) ------------------------------- @@ -157,7 +157,7 @@ def test_a_paused_viewers_checkpoint_keeps_their_session_alive(db, monkeypatch, db.add(user) db.commit() s = _Session(tmp_path, "paused", finished=True, idle_s=90) - sessions["paused1"] = s + sessions[(user.id, "paused1")] = s try: resp = _client(db, user).post( f"/api/plex/item/{it.rating_key}/progress", @@ -167,7 +167,7 @@ def test_a_paused_viewers_checkpoint_keeps_their_session_alive(db, monkeypatch, 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 "paused1" in sessions and not s.proc.killed + assert (user.id, "paused1") in sessions and not s.proc.killed def test_starting_a_session_leaves_another_viewers_paused_one_alone( @@ -183,9 +183,9 @@ def test_starting_a_session_leaves_another_viewers_paused_one_alone( db.add(user) db.commit() paused = _Session(tmp_path, "paused", finished=True, idle_s=90) - sessions["someone-else"] = paused + sessions[(999, "someone-else")] = paused - def fake_start(db_, item, start, audio=None, aoff=0.0, multi=False, transcode=False): + 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, @@ -201,4 +201,4 @@ def test_starting_a_session_leaves_another_viewers_paused_one_alone( finally: app.dependency_overrides.clear() assert resp.status_code == 200, resp.text - assert "someone-else" in sessions and not paused.proc.killed + assert (999, "someone-else") in sessions and not paused.proc.killed diff --git a/backend/tests/test_plex_finish.py b/backend/tests/test_plex_finish.py index 3958eb1a..464e695a 100644 --- a/backend/tests/test_plex_finish.py +++ b/backend/tests/test_plex_finish.py @@ -68,7 +68,7 @@ def test_a_first_ever_finish_inserts_and_counts_once(db, make_user): def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_user, monkeypatch): user = make_user() item = _item(db) - monkeypatch.setattr(playback.plex_stream, "touch_session", lambda key: None) + monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None) earlier = datetime.now(timezone.utc) - timedelta(seconds=5) # A resume push was scheduled a moment ago and has not reached Plex yet. assert playback._should_push_resume(user.id, item.id, earlier, final=True) diff --git a/backend/tests/test_plex_hls_file_route.py b/backend/tests/test_plex_hls_file_route.py index 81fb645c..a7a680c7 100644 --- a/backend/tests/test_plex_hls_file_route.py +++ b/backend/tests/test_plex_hls_file_route.py @@ -38,6 +38,9 @@ class _Proc: return 0 +VIEWER = 41 # the id of the account the test client is signed in as + + class _Session: def __init__(self, session_id: str, directory, running: bool = False, audio_langs=None): self.id = session_id @@ -61,7 +64,7 @@ def sessions(monkeypatch): @pytest.fixture def client(monkeypatch): monkeypatch.setattr(state, "_configured_cache", True) - user = User(email="hls@example.com") + user = User(id=VIEWER, email="hls@example.com") app.dependency_overrides[current_user] = lambda: user try: yield TestClient(app) @@ -74,12 +77,23 @@ def test_a_segment_is_served_from_the_session_the_url_names(client, tmp_path, se 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) + sessions[(VIEWER, "m1")] = _Session("sid-a", d) 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_another_viewers_session_of_the_same_item_is_not_this_viewers(client, tmp_path, sessions): + """Sessions are per viewer: a second account playing the same film has its own, and knowing its + id does not make it yours — resolution is by (viewer, item) before the id is compared.""" + d = tmp_path / "m1_u7_0_0_+0.00" + d.mkdir() + (d / "index.m3u8").write_bytes(b"#EXTM3U") + sessions[(7, "m1")] = _Session("sid-a", d) + r = client.get("/api/plex/stream/m1/hls/sid-a/index.m3u8") + assert r.status_code == 404, r.text + + def test_a_segment_of_a_replaced_session_is_a_404_not_the_new_sessions_bytes( client, tmp_path, sessions ): @@ -89,7 +103,7 @@ def test_a_segment_of_a_replaced_session_is_a_404_not_the_new_sessions_bytes( theirs.mkdir() (theirs / "seg_11.ts").write_bytes(b"") (theirs / "seg_12.ts").write_bytes(b"theirs") - sessions["m1"] = _Session("sid-b", theirs) + sessions[(VIEWER, "m1")] = _Session("sid-b", theirs) 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 @@ -105,7 +119,7 @@ def test_a_playlist_with_no_live_session_is_a_404_not_a_silent_restart_at_zero( monkeypatch.setattr( plex_stream, "start_session", - lambda *a, **k: started.append(a[2] if len(a) > 2 else 0.0), + lambda *a, **k: started.append(a[3] if len(a) > 3 else 0.0), ) r = client.get("/api/plex/stream/m1/hls/sid-gone/index.m3u8") assert r.status_code == 404, r.text @@ -122,7 +136,7 @@ def test_a_segment_that_will_never_arrive_fails_at_once(client, tmp_path, sessio d = tmp_path / "m1_0_0_+0.00" 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) + sessions[(VIEWER, "m1")] = _Session("sid-a", d, running=True) started = time.monotonic() r = client.get("/api/plex/stream/m1/hls/sid-a/seg_99999.ts") elapsed = time.monotonic() - started @@ -150,7 +164,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, audio_langs=["hu", "en"]) + sessions[(VIEWER, "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 @@ -166,7 +180,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, audio_langs=["hu", "en"]) + sessions[(VIEWER, "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" @@ -179,7 +193,7 @@ def test_a_name_the_session_cannot_produce_fails_at_once(client, tmp_path, sessi 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) + sessions[(VIEWER, "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 @@ -194,7 +208,7 @@ def test_an_absurd_segment_index_is_a_404_not_a_500(client, tmp_path, sessions, 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) + sessions[(VIEWER, "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 @@ -209,7 +223,7 @@ def test_a_session_torn_down_during_the_read_is_a_404_not_a_500( d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_11.ts").write_bytes(b"") - sessions["m1"] = _Session("sid-a", d) + sessions[(VIEWER, "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") @@ -224,7 +238,7 @@ def test_a_dead_producer_is_not_waited_out(client, tmp_path, sessions): 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) + sessions[(VIEWER, "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 @@ -294,10 +308,10 @@ def test_a_name_only_the_other_argv_branch_writes_is_refused_at_once(tmp_path, c 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) + sessions[(VIEWER, "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"]) + sessions[(VIEWER, "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. @@ -315,8 +329,8 @@ def test_a_name_only_the_other_argv_branch_writes_is_refused_at_once(tmp_path, c ) # 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") + assert plex_stream.artifact_may_arrive(sessions[(VIEWER, "m1")], "index.m3u8") + assert plex_stream.artifact_may_arrive(sessions[(VIEWER, "m1")], "seg_0.ts") + assert plex_stream.artifact_may_arrive(sessions[(VIEWER, "m2")], "master.m3u8") + assert plex_stream.artifact_may_arrive(sessions[(VIEWER, "m2")], "stream_a1.m3u8") + assert plex_stream.artifact_may_arrive(sessions[(VIEWER, "m2")], "seg_a1_0.ts") diff --git a/backend/tests/test_plex_playback_session.py b/backend/tests/test_plex_playback_session.py index ec5d7331..c868e329 100644 --- a/backend/tests/test_plex_playback_session.py +++ b/backend/tests/test_plex_playback_session.py @@ -55,7 +55,7 @@ def _post(db, user, monkeypatch, rating_key, *, captured): monkeypatch.setattr(plex_stream, "reap_idle", lambda: None) monkeypatch.setattr(plex_paths, "local_media_path", lambda db_, fp: "/local/a.mkv") - def fake_start(db_, it, start, audio, aoff, multi, transcode=False): + def fake_start(db_, it, user_id, start, audio, aoff, multi, transcode=False): captured["transcode"] = transcode return SimpleNamespace( id="sid", diff --git a/backend/tests/test_plex_stream.py b/backend/tests/test_plex_stream.py index 5c354d9b..143f3eb5 100644 --- a/backend/tests/test_plex_stream.py +++ b/backend/tests/test_plex_stream.py @@ -108,7 +108,7 @@ class TestSweepOrphans: def test_a_live_sessions_directory_is_never_deleted(self, hls_root): # Registered session, but with an OLD directory — only the live check can save it. d = _session_dir(hls_root, "12345_0_multi_+0.00", age_s=3600) - stream._sessions["k"] = _FakeSession(d) + stream._sessions[(1, "k")] = _FakeSession(d) assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0 assert d.exists() @@ -179,10 +179,20 @@ class _CapturingProc: def __init__(self, *args, **kwargs): _CapturingProc.calls.append((args, kwargs)) + self.killed = False def poll(self): return None + def terminate(self): + self.killed = True + + def wait(self, timeout=None): + return 0 + + def kill(self): + self.killed = True + def _venc(cmd): """The video codec ffmpeg was told to use (the value after the first `-c:v`).""" @@ -495,7 +505,7 @@ class TestColdStartWarmup: return True monkeypatch.setattr(stream, "wait_for", spy) - s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0) + s = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0) assert s is not None # seg_0 (unconditional now) + the warm-up run up to _WARMUP_SEGMENTS. assert waited == [f"seg_{i}.ts" for i in range(stream._WARMUP_SEGMENTS)] @@ -508,7 +518,7 @@ class TestColdStartWarmup: return path.name != "seg_1.ts" # seg_1 never appears → warm-up gives up there monkeypatch.setattr(stream, "wait_for", spy) - s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0) + s = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0) assert s is not None assert waited == ["seg_0.ts", "seg_1.ts"] # never asked for seg_2 @@ -522,7 +532,7 @@ class TestColdStartWarmup: return False monkeypatch.setattr(stream, "wait_for", spy) - s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0) + s = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0) assert s is not None assert waited == ["seg_0.ts"] @@ -548,6 +558,7 @@ class TestTranscodeStartSession: "plex_transcode_crf": 18, "plex_transcode_max_height": 480, "plex_max_transcodes": 4, + "plex_max_sessions": 8, }[k], ) monkeypatch.setattr( @@ -573,7 +584,7 @@ class TestTranscodeStartSession: stream, "_select_encode", lambda db, item, transcode, src: (False, False) ) s = stream.start_session( - db=None, item=_FakeItem(codec_video="mpeg4"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="mpeg4"), start_s=0.0, transcode=True ) assert s is not None and s.transcode is True and s.hw is False (cmd, *_), kwargs = _CapturingProc.calls[0] @@ -590,7 +601,7 @@ class TestTranscodeStartSession: stream, "_select_encode", lambda db, item, transcode, src: (True, False) ) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert s is not None and s.transcode is True and s.hw is True and s.tonemap is False assert len(_CapturingProc.calls) == 1 # first segment arrived — no HW→SW fallback respawn @@ -609,7 +620,7 @@ class TestTranscodeStartSession: # tonemap=True on the session, still on the GPU (hw=True) and never niced. monkeypatch.setattr(stream, "_select_encode", lambda db, item, transcode, src: (True, True)) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert s is not None and s.transcode is True and s.hw is True and s.tonemap is True (cmd, *_), kwargs = _CapturingProc.calls[0] @@ -635,7 +646,7 @@ class TestTranscodeStartSession: monkeypatch.setattr(stream, "wait_for", flaky_wait) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert ( s is not None and s.transcode is True and s.hw is False @@ -666,7 +677,7 @@ class TestTranscodeStartSession: monkeypatch.setattr(stream, "wait_for", flaky_wait) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert s is not None and s.transcode is True and s.hw is False and s.tonemap is False assert len(_CapturingProc.calls) == 2 # the HW tonemap attempt, then the SW respawn @@ -699,17 +710,17 @@ class TestTranscodeStartSession: def wait_then_supersede(path, timeout=20.0, proc=None): # Simulate the concurrent replacement happening during our out-of-lock wait, then report the # (now-killed) HW proc produced nothing. - stream._sessions["12345"] = newer + stream._sessions[(7, "12345")] = newer return False monkeypatch.setattr(stream, "wait_for", wait_then_supersede) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) # No SW respawn — only the initial HW _spawn ran; the newer session was left intact. assert len(_CapturingProc.calls) == 1 assert s is newer # yielded the currently-registered session, did not clobber it - assert stream._sessions["12345"] is newer + assert stream._sessions[(7, "12345")] is newer def test_a_cap_culled_hw_session_does_not_respawn_and_is_not_reported_missing( self, capturing_env, monkeypatch @@ -727,16 +738,16 @@ class TestTranscodeStartSession: def wait_then_cull(path, timeout=20.0, proc=None): captured["s"] = stream._sessions[ - "12345" + (7, "12345") ] # the HW session the initial _spawn just registered del stream._sessions[ - "12345" + (7, "12345") ] # simulate _enforce_cap dropping the LRU under cap pressure return False monkeypatch.setattr(stream, "wait_for", wait_then_cull) s = stream.start_session( - db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True + db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True ) assert ( len(_CapturingProc.calls) == 1 @@ -752,7 +763,9 @@ class TestTranscodeStartSession: monkeypatch.setattr( stream, "_select_encode", lambda db, item, transcode, src: (False, False) ) - s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0) # transcode defaults False + s = stream.start_session( + db=None, user_id=7, item=_FakeItem(), start_s=0.0 + ) # transcode defaults False assert s is not None and s.transcode is False (cmd, *_), kwargs = _CapturingProc.calls[0] assert _venc(cmd) == "copy" @@ -896,7 +909,7 @@ class TestAudioLoudnorm: ) if multi: monkeypatch.setattr(stream, "_probe_audio_langs", lambda src: ["hun", "eng"]) - s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0, multi=multi) + s = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0, multi=multi) assert s is not None (cmd, *_), _ = _CapturingProc.calls[0] return s, cmd @@ -945,3 +958,51 @@ class TestAudioLoudnorm: # sweep no longer recognises is one that never gets cleaned up. s, _ = self._spawn(monkeypatch, on=on) assert _SESSION_DIR_RE.match(s.dir.name) + + +class TestSessionsPerViewer: + """HLS sessions were keyed by the ITEM, so a second account opening the same film killed the + first one's session; each then reclaimed and evicted the other. And only transcodes counted + toward a cap, so remuxes — each a whole segmented copy of the film on disk — were unbounded.""" + + capturing_env = TestTranscodeStartSession.capturing_env + + @staticmethod + def _start(monkeypatch, user_id, rating_key="12345"): + monkeypatch.setattr( + stream, "_select_encode", lambda db, item, transcode, src: (False, False) + ) + return stream.start_session( + db=None, user_id=user_id, item=_FakeItem(rating_key=rating_key), start_s=0.0 + ) + + def test_two_viewers_of_one_film_each_keep_their_own_session(self, capturing_env, monkeypatch): + monkeypatch.setattr(stream, "_sessions", {}) + a = self._start(monkeypatch, user_id=1) + b = self._start(monkeypatch, user_id=2) + assert stream._sessions == {(1, "12345"): a, (2, "12345"): b} + assert not a.proc.killed + assert a.dir != b.dir and _SESSION_DIR_RE.match(a.dir.name) + + def test_one_account_holds_at_most_two_sessions(self, capturing_env, monkeypatch): + monkeypatch.setattr(stream, "_sessions", {}) + first = self._start(monkeypatch, user_id=1, rating_key="1") + first.last_access -= 60 # the least recently used of the three + self._start(monkeypatch, user_id=1, rating_key="2") + self._start(monkeypatch, user_id=1, rating_key="3") + assert sorted(stream._sessions) == [(1, "2"), (1, "3")] + assert first.proc.killed + + def test_remux_sessions_count_toward_the_overall_ceiling(self, capturing_env, monkeypatch): + monkeypatch.setattr(stream, "_sessions", {}) + monkeypatch.setattr( + stream.sysconfig, + "get_int", + lambda db, k: {"plex_max_transcodes": 4, "plex_max_sessions": 3}[k], + ) + oldest = self._start(monkeypatch, user_id=1) + oldest.last_access -= 60 + for uid in (2, 3, 4): + 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 diff --git a/backend/tests/test_plex_subtitle_prepare.py b/backend/tests/test_plex_subtitle_prepare.py index 9a0dbcc8..c6ccc90f 100644 --- a/backend/tests/test_plex_subtitle_prepare.py +++ b/backend/tests/test_plex_subtitle_prepare.py @@ -690,8 +690,8 @@ def test_a_scrubbing_viewer_cannot_queue_a_window_per_stop(monkeypatch): def test_nothing_more_is_queued_while_the_extractor_is_saturated(monkeypatch): pool = _manual_pool(monkeypatch) - for rk in range(20): - subtitles.ensure(str(rk), 0, at=0.0) + for rk in range(20): # twenty viewers, so no one account's share is what stops the passes + subtitles.ensure(str(rk), 0, at=0.0, viewer=rk) windows = [a for _, a in pool.jobs if a[3] is not None] assert len(windows) == subtitles._MAX_WINDOWS_INFLIGHT assert len(pool.jobs) - len(windows) == subtitles._MAX_FULL_INFLIGHT @@ -971,6 +971,27 @@ def test_a_window_is_not_refused_while_full_passes_hold_their_slots(monkeypatch) assert full.jobs == [], "past its bound no full pass is queued" +def test_one_account_cannot_hold_every_full_pass_slot(monkeypatch): + """The full-pass slots are global and minutes long each: one account opening subtitles on film + after film held them all, and nobody else's track ever completed. Past its share an account + still gets windows — what makes a track watchable now — and the next account still gets a pass.""" + win, full = _ManualPool(), _ManualPool() + monkeypatch.setattr(subtitles, "_WIN_POOL", win) + monkeypatch.setattr(subtitles, "_FULL_POOL", full) + monkeypatch.setattr(subtitles, "_extract_and_publish", lambda *a: None) + for rk in range(subtitles._MAX_FULL_PER_USER + 1): + subtitles.ensure(str(1000 + rk), 1, at=600.0, viewer=7) + assert len(full.jobs) == subtitles._MAX_FULL_PER_USER + assert len(win.jobs) == subtitles._MAX_FULL_PER_USER + 1, ( + "the account over its share still gets windows" + ) + subtitles.ensure("2000", 1, at=600.0, viewer=8) + assert len(full.jobs) == subtitles._MAX_FULL_PER_USER + 1, "another account is not shut out" + full.run_all() # the account's passes end: its share is free again + subtitles.ensure("3000", 1, at=600.0, viewer=7) + assert any(a[1] == "3000" and a[3] is None for _, a in full.jobs) + + def test_a_poll_after_shutdown_queues_nothing(monkeypatch): pool = _manual_pool(monkeypatch) subtitles.shutdown() diff --git a/frontend/src/i18n/locales/en/config.json b/frontend/src/i18n/locales/en/config.json index 586f0393..a342b443 100644 --- a/frontend/src/i18n/locales/en/config.json +++ b/frontend/src/i18n/locales/en/config.json @@ -193,7 +193,11 @@ }, "plex_max_transcodes": { "label": "Max concurrent transcodes", - "hint": "Cap on simultaneous transcode sessions for non-H.264 files. Direct-play and remux (stream-copy) files aren't limited." + "hint": "Cap on simultaneous transcode sessions for non-H.264 files. Remux (stream-copy) sessions count only toward the overall session limit; direct-play files aren't limited." + }, + "plex_max_sessions": { + "label": "Max concurrent playback sessions", + "hint": "Cap on all simultaneous playback sessions, remux (stream-copy) included — each one writes a full segmented copy of the film to the scratch volume. One account can hold at most two. Direct-play files aren't sessions and aren't limited." }, "plex_transcode_hw": { "label": "Hardware transcode", diff --git a/frontend/src/i18n/locales/hu/config.json b/frontend/src/i18n/locales/hu/config.json index 320d9f3c..6318540e 100644 --- a/frontend/src/i18n/locales/hu/config.json +++ b/frontend/src/i18n/locales/hu/config.json @@ -193,7 +193,11 @@ }, "plex_max_transcodes": { "label": "Max párhuzamos transcode", - "hint": "A nem-H.264 fájlok egyidejű transcode-munkameneteinek felső korlátja. A direktben játszható és a remux (stream-copy) fájlokra nincs korlát." + "hint": "A nem-H.264 fájlok egyidejű transcode-munkameneteinek felső korlátja. A remux (stream-copy) munkamenetek csak az összesített munkamenet-korlátba számítanak bele; a direktben játszható fájlokra nincs korlát." + }, + "plex_max_sessions": { + "label": "Max párhuzamos lejátszási munkamenet", + "hint": "Az összes egyidejű lejátszási munkamenet felső korlátja, a remuxot (stream-copy) is beleértve — mindegyik a film teljes, szegmentált másolatát írja a munkakötetre. Egy fiók legfeljebb kettőt tarthat. A direktben játszható fájlok nem munkamenetek, rájuk nincs korlát." }, "plex_transcode_hw": { "label": "Hardveres transcode",