diff --git a/backend/app/main.py b/backend/app/main.py index 1386116c..7c2e6038 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -27,6 +27,7 @@ from app import state from app.config import settings from app.db import SessionLocal from app.plex import stream as plex_stream +from app.plex import subtitles as plex_subtitles from app.routes import ( admin, audit as audit_routes, @@ -90,6 +91,9 @@ async def lifespan(app: FastAPI): with suppress(asyncio.CancelledError): await sweep shutdown_scheduler() + # Kills a running subtitle extraction: its worker would otherwise hold interpreter exit for + # the rest of the read (see plex_subtitles.shutdown). + plex_subtitles.shutdown() async def _sweep_hls_orphans() -> None: diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py index ea685cfd..8c749c1c 100644 --- a/backend/app/plex/subtitles.py +++ b/backend/app/plex/subtitles.py @@ -245,35 +245,77 @@ def covers(ranges: list[list[float]], at: float) -> bool: def merge_vtt(*sources: str) -> str: """Union of the cue blocks of several absolute-time VTTs, ordered by start time. - Two extraction passes over the same file emit identical cues for the region they share, so - de-duplication keys off the TIMING line — the one part of a cue that identifies it uniquely and - that neither pass can renumber (ffmpeg's webvtt muxer writes no cue identifiers).""" + Two extraction passes over the same file emit identical cues for the region they share, so they + are de-duplicated — but NOT on the raw timing text. The two passes do not spell a timestamp the + same way: a window comes back through `shift_vtt`, which always writes `HH:MM:SS.mmm`, while a + full pass is ffmpeg's own output, and its webvtt muxer drops the hour below one hour + (`10:00.123`). Keyed on the text, every cue of a window in the first hour survived twice and the + viewer saw each line stacked on itself. So each timing line is parsed and re-emitted in one + canonical form, and the key is that canonical timing PLUS the cue text: two different cues that + happen to share a time (two speakers, a sign over dialogue) are both kept.""" seen: set[str] = set() cues: list[tuple[float, str]] = [] for src in sources: for block in re.split(r"\n[ \t]*\n", src or ""): - b = block.strip() - if not b or "-->" not in b: + canon = _canonical_cue(block.strip()) + if canon is None: continue # header / NOTE / STYLE — the merged file gets its own single header - timing = next(ln for ln in b.split("\n") if "-->" in ln) - if timing in seen: + start, key, text = canon + if key in seen: continue - seen.add(timing) - try: - start = _vtt_ts_to_s(timing.partition("-->")[0].strip()) - except (ValueError, IndexError): - continue - cues.append((start, b)) + seen.add(key) + cues.append((start, text)) cues.sort(key=lambda c: c[0]) return "WEBVTT\n\n" + "\n\n".join(b for _, b in cues) + "\n" +def _canonical_cue(block: str) -> tuple[float, str, str] | None: + """(start seconds, identity, the block with its timing line canonicalised) — or None for a block + that is not a cue. The identity leaves out anything before the timing line (an SRT-style cue + number), which is a label, not content.""" + lines = block.split("\n") + for i, ln in enumerate(lines): + if "-->" not in ln: + continue + left, _, right = ln.partition("-->") + rparts = right.strip().split(None, 1) + settings = (" " + rparts[1]) if len(rparts) > 1 else "" + try: + st = _vtt_ts_to_s(left.strip()) + en = _vtt_ts_to_s(rparts[0]) + except (ValueError, IndexError): + return None + lines[i] = f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}" + return st, "\n".join(lines[i:]), "\n".join(lines) + return None + + +def _last_cue_end(vtt: str) -> float: + """Where the last cue of a VTT ends — how far a pass that died early provably got.""" + end = 0.0 + for ln in vtt.split("\n"): + if "-->" not in ln: + continue + try: + end = max(end, _vtt_ts_to_s(ln.partition("-->")[2].strip().split(None, 1)[0])) + except (ValueError, IndexError): + continue + return end + + def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) -> None: - """Merge a freshly extracted region into the cached track and widen its covered ranges.""" - old = cache_read(key) - old_ranges, was_complete = read_meta(key) - cache_write(key, merge_vtt(old or "", vtt)) - _write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete) + """Merge a freshly extracted region into the cached track and widen its covered ranges. + + A read-merge-write, so it runs under `_PUBLISH_LOCK`: two jobs publish into the same keys (a + window and the full pass, and every embedded track of the file rides each of them), and without + the lock the later writer drops whatever the earlier one merged — permanently, once `complete` is + set. It is also the only writer of the cache files, which is what keeps their fixed `.tmp` + staging names safe.""" + with _PUBLISH_LOCK: + old = cache_read(key) + old_ranges, was_complete = read_meta(key) + cache_write(key, merge_vtt(old or "", vtt)) + _write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete) # --- background extraction ---------------------------------------------------------------------- @@ -298,6 +340,9 @@ _POOL = ThreadPoolExecutor(max_workers=3, thread_name_prefix="plex-subs") _LOCK = threading.Lock() _INFLIGHT: set[tuple] = set() _FAILURES: dict[str, tuple[str, float]] = {} +_PUBLISH_LOCK = threading.Lock() +# The ffmpeg children currently running, so a shutdown can kill them (see `shutdown`). +_PROCS: set[subprocess.Popen] = set() class SubtitleUnavailable(Exception): @@ -328,6 +373,20 @@ def reset_state() -> None: _FAILURES.clear() +def shutdown() -> None: + """Stop extracting, for the app's shutdown. The pool's workers are non-daemon threads and Python + joins them at interpreter exit — so a worker still inside a full pass (minutes on a big file, up + to EXTRACT_TIMEOUT_S) held the process open after uvicorn had stopped, until the container runtime + SIGKILLed it. Queued jobs are dropped and running ffmpeg children are killed, which returns their + workers at once. Nothing that matters is lost: a publish is an atomic rename, and an interrupted + track is simply extracted again on the next poll.""" + _POOL.shutdown(wait=False, cancel_futures=True) + with _LOCK: + procs = list(_PROCS) + for p in procs: + p.kill() + + def _fail(key: str, code: str) -> None: with _LOCK: _FAILURES[key] = (code, time.monotonic()) @@ -362,25 +421,57 @@ def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: return {"state": "partial" if covered else "preparing", "code": None, "gen": generation(key)} +def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None]: + """What the fetch can serve RIGHT NOW, scheduling nothing: the cached VTT (whatever it + covers), else the recent failure that explains why there is none. + + Scheduling belongs to `ensure` — the status poll, which knows where the viewer is. The fetch does + not: when it scheduled too, it judged readiness at position 0, so a viewer who resumed an hour in + got an unrequested window over the head of the file (plus a remount when it landed), and with a + failure memo alive it was refused a window that WAS on disk while the poll called it showable.""" + key = cache_key(rating_key, sub_ord) + hit = cache_read(key) + return (hit, None) if hit is not None else (None, _recent_failure(key)) + + def _schedule(rating_key: str, sub_ord: int, at: float, want_window: bool) -> 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. Each is deduplicated by identity, so a 1 s status poll (or a - second viewer) joins the running job instead of starting another ffmpeg.""" + second viewer) joins the running job instead of starting another ffmpeg. + + A window's identity is its TRACK, never its position. The poll sends the live playhead, so a + position-keyed window minted a new identity every second and queued one more read of the file per + poll until the first window landed. One window per track is in flight at a time; a seek past it + is picked up by the next poll after it publishes. The full pass is per ITEM, because one read + fans out to every embedded text track of the file.""" jobs: list[tuple[tuple, float | None]] = [] if want_window: - w_start = max(0.0, at - WINDOW_PRE_S) - jobs.append((("win", rating_key, round(w_start)), w_start)) + jobs.append((("win", rating_key, sub_ord), max(0.0, at - WINDOW_PRE_S))) jobs.append((("full", rating_key), None)) for ident, window_start in jobs: with _LOCK: if ident in _INFLIGHT: continue _INFLIGHT.add(ident) - _POOL.submit(_run, ident, rating_key, sub_ord, window_start) + _POOL.submit(_run, ident, rating_key, sub_ord, window_start, at) -def _run(ident: tuple, rating_key: str, sub_ord: int, window_start: float | None) -> None: +def _already_done(key: str, window_start: float | None, at: float) -> bool: + """A queued job looks again before it reads anything. While it waited, another job may have + covered it: every embedded track rides each pass over its file, so a window for one language also + publishes the others.""" + if not _vtt_path(key).exists(): + return False + ranges, complete = read_meta(key) + return complete or (window_start is not None and covers(ranges, at)) + + +def _run( + ident: tuple, rating_key: str, sub_ord: int, window_start: float | None, at: float +) -> None: try: + if _already_done(cache_key(rating_key, sub_ord), window_start, at): + return _extract_and_publish(rating_key, sub_ord, window_start) except SubtitleUnavailable as e: _fail(cache_key(rating_key, sub_ord), e.code) @@ -422,9 +513,11 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No want = subs[sub_ord] if _is_bitmap(want): raise SubtitleUnavailable("bitmap") - if want.get("key"): # external sidecar → one fetch, always the complete track - if window_start is not None: - return # nothing to window — the full job serves it in a single fetch + if want.get("key"): + # An external sidecar is one small fetch from Plex, never a read of the media file, so a + # window job fetches the whole thing too. It must not be left to the full job: that + # identity is per ITEM (the embedded fan-out shares it), so a sidecar picked while an + # embedded track's full pass was running sat behind that pass for minutes. vtt = _sidecar_vtt(db, want) publish(cache_key(rating_key, sub_ord), vtt, (0.0, _FULL_END_S), True) return @@ -443,13 +536,22 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No start = window_start or 0.0 length = (WINDOW_LEN_S + WINDOW_PRE_S) if window_start is not None else None end = _FULL_END_S if length is None else start + length - for emb_ord, vtt in _ffmpeg_extract(str(src), sorted(targets), window_start, length).items(): + texts, clean = _ffmpeg_extract(str(src), sorted(targets), window_start, length) + for emb_ord, vtt in texts.items(): + key = cache_key(rating_key, targets[emb_ord]) + # A pass that exited non-zero holds only what ffmpeg reached before it died. Claim no more + # than its last cue, never call it complete, and remember the failure: published as complete, + # the track read "ready" for good while its subtitles stopped wherever ffmpeg did (the cache + # has no eviction, so nothing would ever have extracted it again). + covered_end = end if clean else start + _last_cue_end(vtt) publish( - cache_key(rating_key, targets[emb_ord]), + key, shift_vtt(vtt, -start), # zero-based window clock → absolute cue times - (start, end), - length is None, + (start, covered_end), + clean and length is None, ) + if not clean: + _fail(key, "extract_failed") def _sidecar_vtt(db: Session, stream: dict) -> str: @@ -470,14 +572,16 @@ def _sidecar_vtt(db: Session, stream: dict) -> str: def _ffmpeg_extract( src: str, emb_ords: list[int], start: float | None, length: float | None -) -> dict[int, str]: +) -> tuple[dict[int, str], bool]: """Extract several embedded subtitle streams to WebVTT in ONE ffmpeg run (= one read of the file). + Returns the texts and whether ffmpeg exited cleanly — a non-zero exit can still leave partial + output, which is worth publishing but must never be taken for the whole track. `start`/`length` bound the read to a window. Without `-copyts` the cues come out on a zero-based clock (the caller shifts them back) — which is exactly why `-copyts` is absent: with it ffmpeg ignores `-t` and reads the whole container anyway (measured).""" if not emb_ords: - return {} + return {}, True with tempfile.TemporaryDirectory(prefix="plexsub-") as tmpdir: cmd = ["ffmpeg", "-nostdin", "-loglevel", "error"] if start: @@ -489,7 +593,22 @@ def _ffmpeg_extract( if length is not None: cmd += ["-t", f"{length:.3f}"] cmd += ["-f", "webvtt", "-y", str(out)] - r = subprocess.run(cmd, capture_output=True, timeout=EXTRACT_TIMEOUT_S) + # Popen rather than run(): the child is registered so `shutdown` can kill it. + proc = subprocess.Popen( + cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE + ) + with _LOCK: + _PROCS.add(proc) + try: + try: + _, err = proc.communicate(timeout=EXTRACT_TIMEOUT_S) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + raise + finally: + with _LOCK: + _PROCS.discard(proc) got = { o: p.read_text(encoding="utf-8", errors="replace") for o, p in outs.items() @@ -497,7 +616,7 @@ def _ffmpeg_extract( } if not got: raise SubtitleUnavailable("extract_failed") - if r.returncode != 0: - # Partial output with a non-zero exit still beats nothing (a later pass widens it) — say so. - log.warning("ffmpeg subtitle pass exited %s: %s", r.returncode, (r.stderr or b"")[:200]) - return got + clean = proc.returncode == 0 + if not clean: + log.warning("ffmpeg subtitle pass exited %s: %s", proc.returncode, (err or b"")[:200]) + return got, clean diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index c6e6d1dd..35fab8f3 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -10,6 +10,7 @@ from fastapi import ( BackgroundTasks, Depends, HTTPException, + Query, Request, Response, ) @@ -87,40 +88,52 @@ def image(rating_key: str, request: Request, variant: str = "thumb") -> Response return imgcache.response(data, ct) +# Both subtitle routes authenticate with the full `current_user`, not the cookie-only +# `_require_session` the image proxies use. That lighter gate exists for bursts of low-sensitivity +# poster fetches and skips the is_active / is_suspended / session-epoch checks — acceptable for a +# poster, not for a route that starts reads of the media library: a suspended or "log out +# everywhere"-revoked cookie kept scheduling them. They are not bursty (one poll a second, one fetch +# per track), so the DB user-load costs nothing that matters. + + @router.get("/subtitle/{rating_key}/{sub_ord}/status") -def subtitle_status(rating_key: str, sub_ord: int, request: Request, at: float = 0.0) -> dict: +def subtitle_status( + rating_key: str, + sub_ord: int, + user: User = Depends(current_user), + # Bounded: a non-finite position used to reach the scheduler's arithmetic and 500. + at: float = Query(0.0, ge=0.0, allow_inf_nan=False), +) -> dict: """Is this subtitle track showable yet — and if not, get it moving. Extraction runs in the background (see plex/subtitles.py for the measurements that forced it there), so the player polls this while it waits instead of hanging on a that will not load for a minute or more. `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.""" - _require_session(request) return subtitles.ensure(rating_key, sub_ord, at) @router.get("/subtitle/{rating_key}/{sub_ord}") def subtitle( - rating_key: str, sub_ord: int, request: Request, offset: float = 0.0, at: float = 0.0 + rating_key: str, sub_ord: int, user: User = Depends(current_user), offset: float = 0.0 ) -> Response: """Serve the ord-th subtitle track as WebVTT for the browser's — so choosing a subtitle overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file) AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached. - NEVER extracts inline. A cold track answers 202 and the extraction proceeds in the background; the - player mounts the only once /status says there is something to show, so in practice this - is only ever hit warm (or on a cache eviction race, which the 202 sends back to polling).""" - _require_session(request) + A pure READ: it serves whatever is cached and schedules nothing — the status poll owns that, + because only it knows where the viewer is (see `subtitles.cached_track`). The player mounts the + only once /status says there is something to show, so in practice this is only ever hit + warm; on a cache eviction race the 202 sends the player back to polling.""" # The cached VTT holds ABSOLUTE cue times (per rk+ord); the per-session cue-shift is applied on # the way out, so a resume/seek to any offset reuses the same track (no cache-per-offset explosion). - st = subtitles.ensure(rating_key, sub_ord, at) - if st["state"] == "failed": + hit, code = subtitles.cached_track(rating_key, sub_ord) + if hit is None and code: raise HTTPException( - status_code=subtitles.UNAVAILABLE_STATUS.get(st["code"], 422), - detail={"code": "subtitle", "reason": st["code"]}, + status_code=subtitles.UNAVAILABLE_STATUS.get(code, 422), + detail={"code": "subtitle", "reason": code}, ) - hit = subtitles.cache_read(subtitles.cache_key(rating_key, sub_ord)) if hit is None: return Response( status_code=202, media_type="application/json", content='{"state":"preparing"}' diff --git a/backend/tests/test_plex_subtitle_auth.py b/backend/tests/test_plex_subtitle_auth.py new file mode 100644 index 00000000..0ddb5ee8 --- /dev/null +++ b/backend/tests/test_plex_subtitle_auth.py @@ -0,0 +1,44 @@ +"""The subtitle routes must refuse a session the rest of the app refuses. + +They used to sit behind `_require_session`, the cookie-only gate the poster proxies use — a signed +cookie was enough, with none of `current_user`'s checks (is_active, is_suspended, the session epoch +that "log out everywhere" bumps). Fine for a poster; not for a route that starts reads of the media +library. Found by the 2026-09 bug-hunt deep-review; the gate itself pre-dates that round. +""" + +from fastapi.testclient import TestClient + +import app.auth as auth +import app.state as app_state +from app.db import get_db +from app.main import app +from app.plex import subtitles +from app.routes.plex import playback + + +class _NoPool: + def __init__(self) -> None: + self.jobs: list = [] + + def submit(self, fn, *args) -> None: + self.jobs.append((fn, args)) + + +def test_a_suspended_account_cannot_drive_the_subtitle_extractor( + db, make_user, override_dep, monkeypatch, tmp_path +): + user = make_user(is_suspended=True) + override_dep(get_db, lambda: db) + monkeypatch.setattr(app_state, "_configured_cache", True) + # A validly SIGNED cookie for that account — which is all the old gate looked at. + for mod in (auth, playback): + monkeypatch.setattr(mod, "resolved_user_id", lambda request: (user.id, True)) + monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path) + pool = _NoPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + subtitles.reset_state() + + client = TestClient(app) + assert client.get("/api/plex/subtitle/98614/0/status?at=600").status_code == 401 + assert client.get("/api/plex/subtitle/98614/0").status_code == 401 + assert pool.jobs == [], "a refused session still scheduled an extraction" diff --git a/backend/tests/test_plex_subtitle_prepare.py b/backend/tests/test_plex_subtitle_prepare.py index 128a84eb..633c2587 100644 --- a/backend/tests/test_plex_subtitle_prepare.py +++ b/backend/tests/test_plex_subtitle_prepare.py @@ -12,21 +12,27 @@ thread sat with it), and picking the second language cost the whole read again. These tests pin the three things that fixes it: the request never waits for ffmpeg, the extraction publishes a window around the playhead before the whole track, and one ffmpeg run covers every -embedded text track. Pure — no Postgres, no ffmpeg. +embedded text track — plus what the deep-review of that fix found (cues doubled at the window/full +seam, a truncated pass cached as complete, a poll that respawned ffmpeg every second, a sidecar stuck +behind an unrelated pass). Pure — no Postgres, no ffmpeg (the auth half is in +test_plex_subtitle_auth.py, which needs the DB). """ import subprocess +import sys import threading import time +from concurrent.futures import ThreadPoolExecutor from pathlib import Path +from types import SimpleNamespace import pytest from fastapi.testclient import TestClient import app.state as app_state +from app.auth import current_user from app.main import app from app.plex import subtitles -from app.routes.plex import playback @pytest.fixture(autouse=True) @@ -39,10 +45,10 @@ def _isolated_cache(tmp_path, monkeypatch): @pytest.fixture -def client(monkeypatch) -> TestClient: - # The subtitle endpoints authenticate off the signed session cookie directly (not a FastAPI - # dependency), so there is no dependency_overrides hook to use. - monkeypatch.setattr(playback, "resolved_user_id", lambda request: (1, None)) +def client(monkeypatch, override_dep) -> TestClient: + # Neither route reads the user; the real `current_user` (and what it rejects) is pinned in + # test_plex_subtitle_auth.py against the DB. + override_dep(current_user, lambda: SimpleNamespace(id=1)) monkeypatch.setattr(app_state, "_configured_cache", True) # pass the instance-setup gate, no DB return TestClient(app) @@ -92,15 +98,15 @@ def test_a_cold_track_answers_at_once_instead_of_waiting_for_ffmpeg(client, monk monkeypatch.setattr(subtitles, "_extract_and_publish", _slow) try: t0 = time.monotonic() - resp = client.get("/api/plex/subtitle/98614/0") + st = client.get("/api/plex/subtitle/98614/0/status") elapsed = time.monotonic() - t0 - assert resp.status_code == 202, resp.text + assert st.status_code == 200 and st.json()["state"] == "preparing", st.text assert elapsed < 2.0, f"the request waited {elapsed:.1f}s on the extraction" assert started.wait(timeout=5), "the extraction never started in the background" - # …and the status endpoint answers just as fast while that extraction is still running. + # …and the fetch answers just as fast while that extraction is still running. t0 = time.monotonic() - st = client.get("/api/plex/subtitle/98614/0/status") - assert st.status_code == 200 and st.json()["state"] == "preparing" + resp = client.get("/api/plex/subtitle/98614/0") + assert resp.status_code == 202, resp.text assert time.monotonic() - t0 < 2.0 finally: release.set() @@ -114,7 +120,7 @@ def test_a_burst_of_polls_starts_one_window_and_one_full_pass(monkeypatch): for _ in range(5): subtitles.ensure("98614", 0, at=600.0) assert len(pool.jobs) == 2 # one window (at 600 s) + the one full pass - # submit(_run, ident, rating_key, sub_ord, window_start) — the last arg is what distinguishes them + # submit(_run, ident, rating_key, sub_ord, window_start, at) — window_start tells them apart assert [a[3] for _, a in pool.jobs] == [570.0, None] # the window first, then the full pass @@ -191,18 +197,33 @@ def test_an_unshowable_track_says_why_and_maps_to_a_status(client, monkeypatch): # --- the ffmpeg pass itself --------------------------------------------------------------------- -def _capture_ffmpeg(monkeypatch) -> list[list[str]]: +class _FakeFfmpeg: + """Stands in for the ffmpeg child: writes the outputs its argv asked for, exits with `returncode`.""" + + def __init__(self, cmd: list[str], returncode: int) -> None: + self.cmd, self.returncode = cmd, returncode + + def communicate(self, timeout=None): + for i, arg in enumerate(self.cmd): + if arg == "-y": + Path(self.cmd[i + 1]).write_text( + _vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8" + ) + return b"", b"" + + def kill(self) -> None: + pass + + +def _capture_ffmpeg(monkeypatch, returncode: int = 0) -> list[list[str]]: """Record the argv of each ffmpeg run and satisfy it by writing the outputs it asked for.""" runs: list[list[str]] = [] - def _fake_run(cmd, **kwargs): + def _fake_popen(cmd, **kwargs): runs.append(cmd) - for i, arg in enumerate(cmd): - if arg == "-y": - Path(cmd[i + 1]).write_text(_vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8") - return subprocess.CompletedProcess(cmd, 0, b"", b"") + return _FakeFfmpeg(cmd, returncode) - monkeypatch.setattr(subtitles.subprocess, "run", _fake_run) + monkeypatch.setattr(subtitles.subprocess, "Popen", _fake_popen) return runs @@ -210,12 +231,20 @@ def test_one_ffmpeg_run_extracts_every_embedded_text_track(monkeypatch): """The read is the cost and it is SHARED. Extracting one track at a time is what made picking Hungarian cost a full pass and then switching to English cost another one.""" runs = _capture_ffmpeg(monkeypatch) - out = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None) + out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None) assert len(runs) == 1, "each extra ffmpeg run is another full read of the file" - assert set(out) == {0, 1} + assert set(out) == {0, 1} and clean assert runs[0].count("-map") == 2 +def test_a_non_zero_exit_is_reported_even_when_it_left_output(monkeypatch): + """ffmpeg can die partway and still leave a VTT behind. That text is worth keeping — but the + caller has to be TOLD, or it publishes a truncated track as the whole thing.""" + _capture_ffmpeg(monkeypatch, returncode=1) + out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None) + assert out and not clean + + def test_a_windowed_pass_bounds_the_read_and_never_uses_copyts(monkeypatch): """`-copyts` is the trap: with it ffmpeg IGNORES -t and reads the whole container anyway (measured — a [0, 600 s] window still took the full 145 s). So the window is extracted on a @@ -261,3 +290,230 @@ def test_a_cached_vtt_from_an_older_build_counts_as_complete(tmp_path): Reading one as "covers nothing" would re-extract every already-cached track after the upgrade.""" (tmp_path / "sub_98614_0.vtt").write_text(_vtt(_cue("00:01.000", "00:02.000", "x")), "utf-8") assert subtitles.read_meta("sub_98614_0") == ([[0.0, 1_000_000_000.0]], True) + + +# --- the deep-review round: the seams between the window and the full pass ---------------------- + + +def test_a_window_and_the_full_pass_do_not_double_a_cue(): + """THE doubled-subtitle bug. The two passes spell the same timestamp differently — a window comes + back through shift_vtt (always HH:MM:SS.mmm), a full pass is ffmpeg's own output, whose webvtt + muxer drops the hour below one hour — so de-duplicating on the text kept both, and every cue of a + window in the first hour was on screen twice.""" + window = subtitles.shift_vtt(_vtt(_cue("00:30.123", "00:32.000", "hello")), -570.0) + full = _vtt(_cue("10:00.123", "10:02.000", "hello")) # ffmpeg's spelling of the same cue + merged = subtitles.merge_vtt(window, full) + assert merged.count("-->") == 1, merged + assert "00:10:00.123 --> 00:10:02.000" in merged # one canonical spelling in the served file + + +def test_two_different_cues_at_the_same_time_are_both_kept(): + """De-duplication is by timing AND text: a sign and a line of dialogue can share a timestamp.""" + both = _vtt(_cue("00:10.000", "00:12.000", "sign"), _cue("00:10.000", "00:12.000", "dialogue")) + merged = subtitles.merge_vtt(both, both) + assert merged.count("-->") == 2 and "sign" in merged and "dialogue" in merged + + +class _NullSession: + def __enter__(self): + return None + + def __exit__(self, *exc): + return False + + +# A file's subtitle streams as Plex lists them: a downloaded sidecar first (no `index` — it is not in +# the file), then the embedded ones, one of them a bitmap track that can never become text. +_STREAMS = [ + {"key": "/library/streams/1", "codec": "srt"}, # sub_ord 0 — sidecar + {"index": 3, "codec": "subrip"}, # sub_ord 1 — embedded ordinal 0 (Hungarian) + {"index": 4, "codec": "pgs"}, # sub_ord 2 — embedded ordinal 1, bitmap + {"index": 5, "codec": "subrip"}, # sub_ord 3 — embedded ordinal 2 (English) +] + + +def _stub_item(monkeypatch, *, clean: bool = True, cues=("00:05.000", "00:07.000")): + """Everything around `_extract_and_publish` stubbed, so the function itself is what runs.""" + monkeypatch.setattr(subtitles, "SessionLocal", _NullSession) + monkeypatch.setattr( + subtitles, "_text_streams", lambda db, rk: (SimpleNamespace(file_path="/f.mkv"), _STREAMS) + ) + monkeypatch.setattr( + subtitles.plex_paths, "local_media_path", lambda db, fp: "/plex-media/f.mkv" + ) + monkeypatch.setattr( + subtitles, "_sidecar_vtt", lambda db, s: _vtt(_cue("00:01.000", "00:03.000", "side")) + ) + calls: list[tuple] = [] + + def _extract(src, emb_ords, start, length): + calls.append((list(emb_ords), start, length)) + return {o: _vtt(_cue(*cues, f"track {o}")) for o in emb_ords}, clean + + monkeypatch.setattr(subtitles, "_ffmpeg_extract", _extract) + return calls + + +def test_one_window_publishes_every_embedded_text_track_at_absolute_time(monkeypatch): + """What `_extract_and_publish` itself is for — every other test stubs it out. One read over the + file's EMBEDDED text streams (the sidecar is not in the file and must not shift the `-map` + ordinals; the bitmap track is skipped), each published under its own sub_ord, and a window's + zero-based cues shifted back to where they sit in the film.""" + calls = _stub_item(monkeypatch) + subtitles._extract_and_publish("98614", 3, 3570.0) + assert calls == [([0, 2], 3570.0, 330.0)], "one ffmpeg run over embedded ordinals 0 and 2" + english = subtitles.cache_read(subtitles.cache_key("98614", 3)) or "" + assert "00:59:35.000 --> 00:59:37.000" in english, english # 3570 + 5 s + assert "track 2" in english + assert "track 0" in (subtitles.cache_read(subtitles.cache_key("98614", 1)) or "") + for never in (0, 2): # the sidecar and the bitmap track come from nowhere in this pass + assert subtitles.cache_read(subtitles.cache_key("98614", never)) is None + + +def test_a_full_pass_that_died_is_not_cached_as_the_whole_track(monkeypatch): + """A truncated full pass used to be published as complete: `ensure` answered "ready" forever, the + player stopped polling, and — the cache has no eviction — the subtitles simply ended wherever + ffmpeg had died, in every later session too. Now it covers what it provably reached, and the + rest of the film honestly reads as failed.""" + _stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000")) + monkeypatch.setattr(subtitles, "_POOL", _ManualPool()) + subtitles._extract_and_publish("98614", 3, None) + key = subtitles.cache_key("98614", 3) + assert subtitles.read_meta(key) == ([[0.0, 2400.0]], False) + assert subtitles.ensure("98614", 3, at=100.0)["state"] == "partial" + after = subtitles.ensure("98614", 3, at=3000.0) + assert (after["state"], after["code"]) == ("failed", "extract_failed") + + +def test_polls_that_follow_the_playhead_join_one_window(monkeypatch): + """The poll sends the LIVE position, so it moves a second per poll. A window keyed on that + position was a new job every second — another read of the file per poll until the first landed.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + for at in (600.0, 601.0, 602.0, 603.0, 604.0): + subtitles.ensure("98614", 1, at=at) + assert len(pool.jobs) == 2, "one window + the one full pass" + + +def test_a_queued_window_that_was_covered_meanwhile_reads_nothing(monkeypatch): + """Every embedded track rides each pass over its file, so by the time a queued window for one + language runs, another language's window may already have published it.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + calls = _stub_item(monkeypatch) + subtitles.ensure("98614", 3, at=600.0) + subtitles.publish( + subtitles.cache_key("98614", 3), + _vtt(_cue("10:00.000", "10:02.000", "x")), + (570.0, 900.0), + False, + ) + pool.run_next() # the window job + assert calls == [] + + +def test_a_sidecar_does_not_wait_behind_an_embedded_full_pass(monkeypatch): + """The full pass is one job per ITEM (the embedded fan-out shares it). A sidecar picked while an + embedded track's pass was running used to find that identity taken and wait out the whole read — + minutes — for a track that is one small fetch.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + _stub_item(monkeypatch) + subtitles._INFLIGHT.add(("full", "98614")) # an embedded track's full pass is running + subtitles.ensure("98614", 0, at=600.0) + pool.run_all() + assert subtitles.ensure("98614", 0, at=600.0)["state"] == "ready" + + +def test_publishes_into_one_track_do_not_lose_each_other(monkeypatch): + """`publish` is a read-merge-write. Two jobs publishing into the same track at once — the window + and the full pass both fan out to every embedded track — used to let the later writer drop the + earlier one's cues. Forced interleaving: the first publisher pauses right after its read.""" + key = subtitles.cache_key("98614", 0) + real_read = subtitles.cache_read + paused, resume = threading.Event(), threading.Event() + reads = {"n": 0} + + def _read(k): + reads["n"] += 1 + out = real_read(k) + if reads["n"] == 1: + paused.set() + resume.wait(timeout=5) + return out + + monkeypatch.setattr(subtitles, "cache_read", _read) + first = threading.Thread( + target=subtitles.publish, + args=(key, _vtt(_cue("00:10.000", "00:12.000", "alpha")), (0.0, 330.0), False), + ) + first.start() + assert paused.wait(timeout=5) + second = threading.Thread( + target=subtitles.publish, + args=(key, _vtt(_cue("10:00.000", "10:02.000", "bravo")), (570.0, 900.0), False), + ) + second.start() + time.sleep(0.3) # without the lock, the second publisher finishes in here + resume.set() + first.join(timeout=5) + second.join(timeout=5) + merged = real_read(key) or "" + assert "alpha" in merged and "bravo" in merged, merged + assert subtitles.read_meta(key)[0] == [[0.0, 330.0], [570.0, 900.0]] + + +def test_shutdown_kills_a_running_extraction(monkeypatch): + """The pool's workers are non-daemon threads that Python joins at exit, so one still inside a + full pass held a restarting process open for the rest of the read. Stand-in child: a real + process that would sleep for a minute.""" + real_popen = subprocess.Popen + monkeypatch.setattr( + subtitles.subprocess, + "Popen", + lambda cmd, **kw: real_popen([sys.executable, "-c", "import time; time.sleep(60)"], **kw), + ) + monkeypatch.setattr(subtitles, "_POOL", ThreadPoolExecutor(max_workers=1)) + finished = threading.Event() + + def _extract(): + try: + subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None) + except subtitles.SubtitleUnavailable: + pass # killed before it wrote anything — exactly what a shutdown does + finally: + finished.set() + + threading.Thread(target=_extract, daemon=True).start() + deadline = time.monotonic() + 10 + while not subtitles._PROCS and time.monotonic() < deadline: + time.sleep(0.02) + assert subtitles._PROCS, "the extraction never registered its child" + subtitles.shutdown() + assert finished.wait(timeout=10), "the extraction was still running after shutdown" + + +def test_the_track_fetch_serves_what_is_cached_and_schedules_nothing(client, monkeypatch): + """The URL carries no position, and it used to run the scheduler anyway — judging + readiness at position 0. A viewer resumed an hour in got an unrequested extraction over the head + of the file, and with a failure memo alive the fetch refused a window that WAS on disk while the + status poll (which knows the position) called the track showable: silently blank subtitles.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + assert client.get("/api/plex/subtitle/98614/0").status_code == 202 + assert pool.jobs == [], "the fetch scheduled an extraction" + key = subtitles.cache_key("98614", 0) + subtitles.publish(key, _vtt(_cue("59:54.187", "59:58.650", "window")), (3570.0, 3900.0), False) + subtitles._fail(key, "extract_failed") # the full pass behind that window failed + resp = client.get("/api/plex/subtitle/98614/0") + assert resp.status_code == 200 and "window" in resp.text + assert pool.jobs == [] + + +@pytest.mark.parametrize("at", ["inf", "-1", "nan"]) +def test_an_impossible_position_is_rejected(client, monkeypatch, at): + """`?at=inf` reached the scheduler's arithmetic and came back as a 500 with a traceback.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + assert client.get(f"/api/plex/subtitle/98614/0/status?at={at}").status_code == 422 + assert pool.jobs == [] diff --git a/frontend/src/lib/api/plex.ts b/frontend/src/lib/api/plex.ts index 89aae403..5cbebde4 100644 --- a/frontend/src/lib/api/plex.ts +++ b/frontend/src/lib/api/plex.ts @@ -650,7 +650,8 @@ export const plexApi = { // cue times onto the HLS session's zero-based clock so subtitles line up with speech after a seek. // `gen` is the cached track's generation: extraction publishes a window around the playhead first // and widens it in the background, so the URL has to change when it grows or the browser would keep - // serving the first, short VTT from its own cache. + // serving the first, short VTT from its own cache. It deliberately carries NO position: the fetch + // is a pure read of what is cached — scheduling is the status poll's job, since only it knows `at`. plexSubtitleUrl: (id: string, ord: number, offset = 0, gen = 0): string => { const p = new URLSearchParams(); if (offset) p.set("offset", String(offset));