diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py index f9048565..bf659372 100644 --- a/backend/app/plex/subtitles.py +++ b/backend/app/plex/subtitles.py @@ -73,6 +73,9 @@ def srt_to_vtt(text: str) -> str: def _vtt_ts_to_s(ts: str) -> float: + # Also reads ffmpeg's spelling of a NEGATIVE time (a windowed pass, see `_ffmpeg_extract`): its + # webvtt muxer splits the milliseconds with C's truncating division, so every field carries the + # sign — -5.005 s is `00:-5.-05`, -65.3 s is `-1:-5.-300` — and summing the signed fields is exact. parts = ts.split(":") h = "0" if len(parts) == 2 else parts[0] m, rest = parts[-2], parts[-1] @@ -81,16 +84,12 @@ def _vtt_ts_to_s(ts: str) -> float: def _vtt_s_to_ts(x: float) -> str: - x = max(0.0, x) - h = int(x // 3600) - x -= h * 3600 - m = int(x // 60) - x -= m * 60 - s = int(x) - ms = int(round((x - s) * 1000)) - if ms >= 1000: - s += 1 - ms -= 1000 + # Rounded to whole milliseconds FIRST, then split: rounding the fraction last carried a rounded-up + # millisecond into the seconds and no further, so 59.9996 came out as `00:00:60.000` — a timestamp + # the WebVTT parser rejects, silently dropping that cue. + h, rest = divmod(int(round(max(0.0, x) * 1000)), 3_600_000) + m, rest = divmod(rest, 60_000) + s, ms = divmod(rest, 1000) return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}" @@ -314,6 +313,12 @@ def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) -> with _PUBLISH_LOCK: old = cache_read(key) old_ranges, was_complete = read_meta(key) + if old is None: + # A VTT with no sidecar reads as a COMPLETE track (see read_meta — how older builds + # cached), and a publish is two renames. So before a key's first VTT lands, its sidecar + # already claims nothing: dying between the two then leaves a window that is extracted + # again, never one taken for the whole track for good. + _write_meta(key, old_ranges, False) cache_write(key, merge_vtt(old or "", vtt)) _write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete) @@ -330,19 +335,46 @@ WINDOW_LEN_S = 300.0 # Ceiling on one ffmpeg run so a wedged extraction can't hold a pool thread forever (the old # in-request extraction had no timeout at all). EXTRACT_TIMEOUT_S = 900.0 -# A failed track is remembered briefly so a 1 s status poll doesn't respawn ffmpeg every second, and -# forgotten again soon after so fixing the file (or Plex coming back) doesn't need a restart. +# A failed track (or window) is remembered briefly so a 1 s status poll doesn't respawn ffmpeg every +# second, and forgotten again soon after so fixing the file (or Plex coming back) doesn't need a +# restart. FAIL_TTL_S = 60.0 +# …except a FULL pass that failed. That is a read of the whole container (minutes), a failure that +# reproduces does so every time, and the viewer is typically inside what it did publish — so the +# short memo re-armed a whole-file read every minute for as long as they watched. Windows around the +# playhead still run meanwhile; only the full pass waits this out. +FULL_RETRY_S = 3600.0 +# The furthest position the status poll accepts. No film is this long, and an absurd one used to +# reach ffmpeg's `-ss` (which rejects anything past its int64 microseconds). +MAX_POSITION_S = 100_000.0 # Two items' full passes plus a window for a third: enough that a seek into an un-extracted region is # served while a full pass runs, bounded so a burst can't thrash the array. _POOL = ThreadPoolExecutor(max_workers=3, thread_name_prefix="plex-subs") +# Bounds on what may wait for that pool. Its queue is unbounded and every job is a read of the +# library: a viewer scrubbing through a film queued a window per stop (the one where they stopped ran +# LAST), and a script could queue thousands. Past either bound nothing is queued — the next poll asks +# again, by which time something has finished. +_MAX_INFLIGHT = 12 +_MAX_TRACK_WINDOWS = 2 _LOCK = threading.Lock() -_INFLIGHT: set[tuple] = set() +# Queued or running jobs → the absolute span a window job reads (None for a full pass). +_INFLIGHT: dict[tuple, tuple[float, float] | None] = {} +# 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). +_FULL_FAILED: dict[str, float] = {} +# cache key → the spans windows failed to read, with when. A failed window says nothing about the rest +# of the track — the full pass may be about to deliver it — so it is remembered as a SPAN: it must be +# neither respawned every poll nor taken as a failure of positions it never tried. (It used to set the +# track-level memo, which told every viewer at any uncovered position "failed" — a final answer that +# ends the player's poll — including one planted by a request for an absurd position.) +_WIN_FAILED: dict[str, list[tuple[float, float, float]]] = {} _PUBLISH_LOCK = threading.Lock() -# The ffmpeg children currently running, so a shutdown can kill them (see `shutdown`). +# The ffmpeg children currently running, so a shutdown can kill them (see `shutdown`), and whether it +# has: a worker that reaches its Popen after the kill sweep must not start a read. _PROCS: set[subprocess.Popen] = set() +_STOPPING = False class SubtitleUnavailable(Exception): @@ -368,9 +400,13 @@ UNAVAILABLE_STATUS = { def reset_state() -> None: """Drop the in-flight/failure registries (tests; nothing in the app needs it).""" + global _STOPPING with _LOCK: _INFLIGHT.clear() _FAILURES.clear() + _FULL_FAILED.clear() + _WIN_FAILED.clear() + _STOPPING = False def shutdown() -> None: @@ -378,11 +414,14 @@ def shutdown() -> None: 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) + workers at once — and a worker still BEFORE its ffmpeg (the Plex lookup, the DB) finds the flag + and never starts one. Nothing that matters is lost: an interrupted publish cannot pass for a whole + track (see `publish`), and an interrupted track is simply extracted again on the next poll.""" + global _STOPPING with _LOCK: + _STOPPING = True procs = list(_PROCS) + _POOL.shutdown(wait=False, cancel_futures=True) for p in procs: p.kill() @@ -401,6 +440,36 @@ def _recent_failure(key: str) -> str | None: return None +def _full_failed(rating_key: str) -> bool: + with _LOCK: + when = _FULL_FAILED.get(rating_key) + if when is not None and time.monotonic() - when < FULL_RETRY_S: + return True + _FULL_FAILED.pop(rating_key, None) + return False + + +def _live_window_failures(key: str) -> list[tuple[float, float, float]]: + """(Caller holds _LOCK.) The key's failed-window spans that are still remembered.""" + now = time.monotonic() + live = [f for f in _WIN_FAILED.get(key, []) if now - f[2] < FAIL_TTL_S] + if live: + _WIN_FAILED[key] = live + else: + _WIN_FAILED.pop(key, None) + return live + + +def _window_fail(key: str, span: tuple[float, float]) -> None: + with _LOCK: + _WIN_FAILED[key] = [*_live_window_failures(key), (span[0], span[1], time.monotonic())] + + +def _window_failed(key: str, at: float) -> bool: + with _LOCK: + 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: """State of one subtitle track, scheduling whatever extraction is still missing. NEVER blocks. @@ -416,8 +485,21 @@ def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: code = _recent_failure(key) if code and not covered: return {"state": "failed", "code": code, "gen": generation(key)} + # "Failed" is a FINAL answer — the player stops asking — so for an ffmpeg failure it is given only + # once nothing is left that could still cover the position: the full pass failed, AND so did a + # window here. + full_failed = _full_failed(rating_key) + window_failed = not covered and _window_failed(key, at) + if full_failed and window_failed: + return {"state": "failed", "code": "extract_failed", "gen": generation(key)} if not code: - _schedule(rating_key, sub_ord, at, want_window=not covered) + _schedule( + rating_key, + sub_ord, + at, + want_window=not covered and not window_failed, + want_full=not full_failed, + ) return {"state": "partial" if covered else "preparing", "code": None, "gen": generation(key)} @@ -434,28 +516,42 @@ def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None] 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: +def _schedule( + rating_key: str, sub_ord: int, at: float, want_window: bool, want_full: bool = True +) -> 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. + full pass that makes it complete. A 1 s status poll (or a second viewer) joins a job already + queued or running instead of starting another ffmpeg. - A window's identity is its track plus WHICH STRETCH of the film it is for — never the exact - 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; bucketing the - start by the window's own length makes a playhead creeping along join the job already running. It - must not be the track alone either: then a window cut around the wrong spot (the first poll after - a resume can still report 0) made the right one wait for it. The full pass is per ITEM, because - one read fans out to every embedded text track of the file.""" + A window is joined when one of this track's in-flight windows will COVER the position asked about + — decided by what it reads, not by an identity derived from the position. The poll sends the live + playhead, so a position-keyed window was a new job every second; bucketing the position instead + still queued a second read when the playhead crossed a bucket edge, and joined a backward seek to + a window that could not cover it. Nor may it be one window per track: a window cut around the + wrong spot (the first poll after a resume can still report 0) made the right one wait for it. 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, sub_ord, int(w_start // WINDOW_LEN_S)), w_start)) - jobs.append((("full", rating_key), None)) + with _LOCK: + if _STOPPING or len(_INFLIGHT) >= _MAX_INFLIGHT: + return + if want_window: + spans = [ + span + for ident, span in _INFLIGHT.items() + if span is not None and ident[1:3] == (rating_key, sub_ord) + ] + if not any(a <= at <= b for a, b in spans) and len(spans) < _MAX_TRACK_WINDOWS: + # Rounded to what ffmpeg is told (`-ss %.3f`): the cues come back on ffmpeg's zero, + # and shifting them by an unrounded start put them up to half a millisecond off — + # enough to spell a timestamp differently from the full pass and double the cue. + w_start = float(f"{max(0.0, at - WINDOW_PRE_S):.3f}") + 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 ("full", rating_key) not in _INFLIGHT: + _INFLIGHT[("full", rating_key)] = None + 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, at) @@ -472,23 +568,43 @@ def _already_done(key: str, window_start: float | None, at: float) -> bool: def _run( ident: tuple, rating_key: str, sub_ord: int, window_start: float | None, at: float ) -> None: + key = cache_key(rating_key, sub_ord) try: - if _already_done(cache_key(rating_key, sub_ord), window_start, at): + if _STOPPING or _already_done(key, 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) + if e.code == "extract_failed": + _pass_failed(rating_key, [key], window_start) + else: + _fail(key, e.code) except PlexError as e: log.warning("subtitle extraction failed for %s/%s: %s", rating_key, sub_ord, e) - _fail(cache_key(rating_key, sub_ord), "plex_error") + _fail(key, "plex_error") except ( Exception ): # a worker thread has nobody to raise to — record it, let the next poll see it log.exception("subtitle extraction crashed for %s/%s", rating_key, sub_ord) - _fail(cache_key(rating_key, sub_ord), "extract_failed") + _pass_failed(rating_key, [key], window_start) finally: with _LOCK: - _INFLIGHT.discard(ident) + _INFLIGHT.pop(ident, None) + + +def _pass_failed( + rating_key: str, + keys: list[str], + window_start: float | None, + span: tuple[float, float] | None = None, +) -> None: + """Record an ffmpeg pass that failed: a full pass backs off the whole item (FULL_RETRY_S); a + window marks only the stretch it failed to read, for the tracks it was reading.""" + if window_start is None: + with _LOCK: + _FULL_FAILED[rating_key] = time.monotonic() + return + for key in keys: + _window_fail(key, span or (window_start, window_start + WINDOW_PRE_S + WINDOW_LEN_S)) def _text_streams(db: Session, rating_key: str) -> tuple[PlexItem, list[dict]]: @@ -554,7 +670,7 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No clean and length is None, ) if not clean: - _fail(key, "extract_failed") + _pass_failed(rating_key, [key], window_start, (covered_end, end)) def _sidecar_vtt(db: Session, stream: dict) -> str: @@ -595,13 +711,26 @@ def _ffmpeg_extract( cmd += ["-map", f"0:s:{emb_ord}"] if length is not None: cmd += ["-t", f"{length:.3f}"] + # The input seek lands on the keyframe at or BEFORE `start`, so a cue already running + # at `start` comes out at a negative time — and by default the muxer then shifts the + # WHOLE output so that cue sits at 0: every cue of the window late by the distance to + # that keyframe (1–5 s measured on a real library file, a whole GOP on a synthetic + # one), and doubled for good once the full pass merged in beside it. Kept negative, the + # times are exact on the zero-based clock (`_vtt_ts_to_s` reads the sign). + cmd += ["-avoid_negative_ts", "disabled"] cmd += ["-f", "webvtt", "-y", str(out)] - # Popen rather than run(): the child is registered so `shutdown` can kill it. + # Popen rather than run(): the child is registered so `shutdown` can kill it — and one that + # would start after the shutdown's kill sweep never does, or is killed right here. + with _LOCK: + if _STOPPING: + raise SubtitleUnavailable("extract_failed") proc = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE ) with _LOCK: _PROCS.add(proc) + if _STOPPING: + proc.kill() try: try: _, err = proc.communicate(timeout=EXTRACT_TIMEOUT_S) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 35fab8f3..f0e03ef9 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -15,7 +15,7 @@ from fastapi import ( Response, ) from fastapi.responses import FileResponse -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict from sqlalchemy import select from sqlalchemy.orm import Session @@ -101,8 +101,9 @@ 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), + # Bounded both ways: a non-finite position used to reach the scheduler's arithmetic and 500, and + # an absurd one reached ffmpeg's `-ss`, which refused it. + at: float = Query(0.0, ge=0.0, le=subtitles.MAX_POSITION_S, 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 @@ -115,7 +116,12 @@ def subtitle_status( @router.get("/subtitle/{rating_key}/{sub_ord}") def subtitle( - rating_key: str, sub_ord: int, user: User = Depends(current_user), offset: float = 0.0 + rating_key: str, + sub_ord: int, + user: User = Depends(current_user), + # Finite, like every float this router takes: `-inf` crashed the shift (a 500), `nan` stacked + # every cue at 0. It can legitimately be negative (a subtitle nudged later than the session start). + offset: float = Query(0.0, allow_inf_nan=False), ) -> 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 @@ -281,6 +287,10 @@ def _push_watch( class ItemProgressIn(BaseModel): + # Finite only: JSON parsing here accepts `Infinity`/`NaN`, and the handler's `int()` turned + # one into a 500. + model_config = ConfigDict(allow_inf_nan=False) + # float, truncated in the handler — the player may post a raw currentTime (a float); a plain # `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated). position_seconds: float = 0 @@ -528,9 +538,10 @@ def season_state( @router.post("/stream/{rating_key}/session") def stream_session( rating_key: str, - start: float = 0.0, + # Finite: `start=inf` reached `int()` in the session builder and came back as a 500. + start: float = Query(0.0, allow_inf_nan=False), audio: int | None = None, - aoff: float = 0.0, + aoff: float = Query(0.0, allow_inf_nan=False), multi: bool = False, user: User = Depends(current_user), db: Session = Depends(get_db), diff --git a/backend/tests/test_plex_subtitle_prepare.py b/backend/tests/test_plex_subtitle_prepare.py index 13c572b1..52a3385d 100644 --- a/backend/tests/test_plex_subtitle_prepare.py +++ b/backend/tests/test_plex_subtitle_prepare.py @@ -18,6 +18,8 @@ behind an unrelated pass). Pure — no Postgres, no ffmpeg (the auth half is in test_plex_subtitle_auth.py, which needs the DB). """ +import asyncio +import shutil import subprocess import sys import threading @@ -30,10 +32,13 @@ import pytest from fastapi.testclient import TestClient import app.state as app_state +import app.main as app_main from app.auth import current_user from app.main import app from app.plex import subtitles +needs_ffmpeg = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available") + @pytest.fixture(autouse=True) def _isolated_cache(tmp_path, monkeypatch): @@ -373,14 +378,19 @@ def test_one_window_publishes_every_embedded_text_track_at_absolute_time(monkeyp 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.""" + ffmpeg had died, in every later session too. Now it covers what it provably reached; past that + a window is tried, and only when that fails too is the position answered as failed.""" _stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000")) - monkeypatch.setattr(subtitles, "_POOL", _ManualPool()) + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) 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" + assert subtitles.ensure("98614", 3, at=3000.0)["state"] == "preparing" + assert [a[3] for _, a in pool.jobs] == [2970.0], "a window past the break — not the full pass" + _window_raises(monkeypatch) + pool.run_all() after = subtitles.ensure("98614", 3, at=3000.0) assert (after["state"], after["code"]) == ("failed", "extract_failed") @@ -430,7 +440,7 @@ def test_a_sidecar_does_not_wait_behind_an_embedded_full_pass(monkeypatch): pool = _ManualPool() monkeypatch.setattr(subtitles, "_POOL", pool) _stub_item(monkeypatch) - subtitles._INFLIGHT.add(("full", "98614")) # an embedded track's full pass is running + subtitles._INFLIGHT[("full", "98614")] = None # 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" @@ -528,3 +538,254 @@ def test_an_impossible_position_is_rejected(client, monkeypatch, at): monkeypatch.setattr(subtitles, "_POOL", pool) assert client.get(f"/api/plex/subtitle/98614/0/status?at={at}").status_code == 422 assert pool.jobs == [] + + +@pytest.mark.parametrize("q", ["offset=-inf", "offset=nan", "offset=inf"]) +def test_an_impossible_offset_is_rejected(client, q): + """The sibling of `at`: `-inf` crashed the shift with a 500, `nan` stacked every cue at 0.""" + assert client.get(f"/api/plex/subtitle/98614/0?{q}").status_code == 422 + + +def test_an_absurd_position_is_rejected_before_it_reaches_ffmpeg(client, monkeypatch): + """ffmpeg refuses an `-ss` past its int64 microseconds, and that failure was recorded against the + TRACK — every other viewer at an uncovered position was told "failed" for a minute.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + assert client.get("/api/plex/subtitle/98614/0/status?at=1e13").status_code == 422 + assert pool.jobs == [] + + +# --- the second deep-review round --------------------------------------------------------------- + + +def _window_raises(monkeypatch): + """From here on every window read fails outright; a full pass still works.""" + real = subtitles._ffmpeg_extract + + def _extract(src, emb_ords, start, length): + if length is not None: + raise subtitles.SubtitleUnavailable("extract_failed") + return real(src, emb_ords, start, length) + + monkeypatch.setattr(subtitles, "_ffmpeg_extract", _extract) + + +def _later(monkeypatch, seconds: float) -> None: + """Move the module's clock forward (the failure memos are monotonic-time TTLs).""" + now = time.monotonic() + seconds + monkeypatch.setattr(subtitles.time, "monotonic", lambda: now) + + +def test_a_full_pass_that_failed_is_not_read_again_while_its_output_is_watched(monkeypatch): + """Published as partial, a failed full pass left the viewer inside what it DID publish — so the + status said `partial`, the poll kept going, and each time the one-minute failure memo lapsed it + re-armed a read of the whole container. A failure that reproduces did so every time: a full + read of the file every few minutes for as long as they watched.""" + _stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000")) + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + subtitles._extract_and_publish("98614", 3, None) + _later(monkeypatch, subtitles.FAIL_TTL_S + 1) + for _ in range(3): + assert subtitles.ensure("98614", 3, at=100.0)["state"] == "partial" + assert pool.jobs == [], "a failed full pass was queued again" + + +def test_a_failed_window_does_not_end_the_poll_while_the_full_pass_may_cover_it(monkeypatch): + """ "failed" is final — the player stops asking. A window that failed used to set the TRACK's + failure memo, so a viewer anywhere outside the covered part got "failed" and never heard about + the full pass that landed a minute later.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + _stub_item(monkeypatch) + _window_raises(monkeypatch) + subtitles.ensure("98614", 3, at=3000.0) + pool.run_next() # the window fails; the full pass is still queued behind it + assert subtitles.ensure("98614", 3, at=3000.0)["state"] == "preparing" + assert subtitles.ensure("98614", 3, at=500.0)["state"] == "preparing" + assert [a[3] for _, a in pool.jobs] == [None, 470.0], "the failed stretch was not tried again" + + +def test_a_backward_seek_gets_its_own_window(monkeypatch): + """A window bucketed by position joined a seek back to 6:40 onto a running window at 9:10 that + could not cover it — the right one waited for the wrong one.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + subtitles.ensure("98614", 1, at=580.0) + subtitles.ensure("98614", 1, at=400.0) + assert sorted(a[3] for _, a in pool.jobs if a[3] is not None) == [370.0, 550.0] + + +def test_a_playhead_crossing_a_bucket_edge_joins_the_window_that_covers_it(monkeypatch): + """…and the same bucketing queued a second read when a playhead crossed an edge while the first + window — which already covered it — was still running.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + for at in (628.0, 629.0, 630.0, 631.0, 632.0): + subtitles.ensure("98614", 1, at=at) + assert len(pool.jobs) == 2, "one window + the one full pass" + + +def test_a_scrubbing_viewer_cannot_queue_a_window_per_stop(monkeypatch): + """The pool's queue is unbounded and every job is a read of the library.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + for at in (0.0, 1000.0, 2000.0, 3000.0, 4000.0): + subtitles.ensure("98614", 1, at=at) + assert len([1 for _, a in pool.jobs if a[3] is not None]) == subtitles._MAX_TRACK_WINDOWS + + +def test_nothing_more_is_queued_while_the_extractor_is_saturated(monkeypatch): + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + for rk in range(20): + subtitles.ensure(str(rk), 0, at=0.0) + assert len(pool.jobs) == subtitles._MAX_INFLIGHT + + +def test_a_window_starts_where_ffmpeg_is_told_to_seek(monkeypatch): + """ffmpeg gets `-ss` to the millisecond, and its cues come back on THAT zero. Shifted back by an + unrounded start they landed up to half a millisecond off the full pass's — spelled differently, + and so kept twice by the merge.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + subtitles.ensure("98614", 1, at=1198.4375) + (window_start,) = [a[3] for _, a in pool.jobs if a[3] is not None] + assert window_start == 1168.438 + + +def test_a_millisecond_rounded_up_carries_into_the_minute(): + """`00:00:60.000` is not a WebVTT timestamp — the browser drops the cue.""" + assert subtitles._vtt_s_to_ts(59.9996) == "00:01:00.000" + assert subtitles._vtt_s_to_ts(3599.9996) == "01:00:00.000" + shifted = subtitles.shift_vtt(_vtt(_cue("00:01:00.000", "00:01:02.000", "x")), 0.0004) + assert "00:01:00.000 --> 00:01:02.000" in shifted, shifted + + +def test_ffmpegs_spelling_of_a_negative_time_reads_exactly(): + """With `-avoid_negative_ts disabled` a cue already running at the window's start comes out + before zero, and ffmpeg's muxer puts the sign on every field.""" + assert subtitles._vtt_ts_to_s("00:-5.-05") == pytest.approx(-5.005) + assert subtitles._vtt_ts_to_s("-1:-5.-300") == pytest.approx(-65.3) + + +def test_dying_between_the_two_renames_cannot_pass_a_window_off_as_the_whole_track(monkeypatch): + """A VTT without a sidecar is how an older build cached a COMPLETE track. A publish is two + renames, the VTT first — so dying between them on a key's first publish left a window that + read as complete, "ready" for good.""" + key = subtitles.cache_key("98614", 0) + real_write = subtitles.cache_write + + def _die_after(k, vtt): + real_write(k, vtt) + raise KeyboardInterrupt # the process goes away right after the VTT landed + + monkeypatch.setattr(subtitles, "cache_write", _die_after) + with pytest.raises(KeyboardInterrupt): + subtitles.publish(key, _vtt(_cue("10:00.000", "10:02.000", "x")), (570.0, 900.0), False) + assert subtitles.read_meta(key) == ([], False) + + +def test_shutdown_drops_queued_extractions(monkeypatch): + """Only the kill half used to be pinned. A job still QUEUED at shutdown would otherwise start + after the kill sweep — a fresh read of the file holding the process open.""" + pool = ThreadPoolExecutor(max_workers=1) + monkeypatch.setattr(subtitles, "_POOL", pool) + busy, release, ran = threading.Event(), threading.Event(), threading.Event() + pool.submit(lambda: (busy.set(), release.wait(timeout=5))) + pool.submit(ran.set) + assert busy.wait(timeout=5) + subtitles.shutdown() + release.set() + pool.shutdown(wait=True) + assert not ran.is_set(), "a queued extraction ran after shutdown" + + +def test_a_worker_that_reaches_ffmpeg_after_shutdown_never_starts_it(monkeypatch): + """A worker still in its Plex lookup when the kill sweep ran used to start its ffmpeg anyway.""" + monkeypatch.setattr(subtitles, "_POOL", ThreadPoolExecutor(max_workers=1)) + started: list = [] + monkeypatch.setattr(subtitles.subprocess, "Popen", lambda cmd, **kw: started.append(cmd)) + subtitles.shutdown() + with pytest.raises(subtitles.SubtitleUnavailable): + subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None) + assert started == [] + + +def test_the_app_stops_the_extractor_when_it_shuts_down(monkeypatch): + """No test ran the lifespan, so deleting this one call switched all of the above off with every + lane green.""" + + class _Configured: + def __enter__(self): + return None + + def __exit__(self, *exc): + return False + + async def _no_sweep(): + return None + + calls: list[str] = [] + 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: calls.append("stopped")) + + async def _cycle(): + async with app_main.lifespan(app): + pass + + asyncio.run(_cycle()) + assert calls == ["stopped"] + + +# --- a window against a REAL ffmpeg ------------------------------------------------------------- + + +def _make_mkv(tmp_path: Path) -> Path: + """A minute of video with a keyframe every 10 s and a subtitle every 2 s — so a seek to 21.3 s + lands on the keyframe at 20 s, with the 20.0–21.5 s cue already running at the window's start.""" + srt = tmp_path / "s.srt" + srt.write_text( + "".join( + f"{i + 1}\n00:00:{2 * i:02d},000 --> 00:00:{2 * i + 1:02d},500\nline {i}\n\n" + for i in range(30) + ), + "utf-8", + ) + out = tmp_path / "f.mkv" + subprocess.run( + [ + "ffmpeg", "-nostdin", "-loglevel", "error", + "-f", "lavfi", "-i", "color=c=black:s=32x32:r=10:d=60", + "-i", str(srt), + "-map", "0:v", "-map", "1:s", "-c:v", "mpeg4", "-g", "100", "-c:s", "srt", + "-y", str(out), + ], + check=True, + ) # fmt: skip + return out + + +@needs_ffmpeg +def test_a_window_lines_up_with_the_full_pass_on_a_real_file(tmp_path, monkeypatch): + """THE late-and-doubled window. The input seek lands on the keyframe before the window's start, + a cue already running there comes out at a negative time, and by default ffmpeg's muxer shifted + the whole window so that cue sat at 0 — every cue late by the distance to the keyframe (1–5 s on + a real library file), and kept twice for good once the full pass merged in beside it.""" + src = _make_mkv(tmp_path) + monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path / "cache") + full, clean = subtitles._ffmpeg_extract(str(src), [0], None, None) + assert clean + win, clean = subtitles._ffmpeg_extract(str(src), [0], 21.3, 20.0) + assert clean + key = subtitles.cache_key("1", 0) + subtitles.publish(key, subtitles.shift_vtt(win[0], -21.3), (21.3, 41.3), False) + served = subtitles.cache_read(key) or "" + assert "00:00:22.000 --> 00:00:23.500\nline 11" in served, served # on time, not 1.3 s late + subtitles.publish(key, full[0], (0.0, subtitles._FULL_END_S), True) + merged = subtitles.cache_read(key) or "" + assert merged.count("-->") == 30, "a window cue survived beside its full-pass twin"