fix(plex): subtitle windows land where the full pass does, and the extractor stops re-reading, respawning and outliving the process
Deep-review round 2 of the 2026-09 bug hunt, over the round-1 fix diff. - A window was LATE by the distance to the keyframe its input seek landed on, and then doubled for good once the full pass merged in: a cue already running at the window's start came out at a negative time, and the webvtt muxer's default shifted the whole output so it sat at 0. Measured on the real Westworld S01E06 at 25:00: every window cue +5.005 s, 682 cues merged instead of 608. `-avoid_negative_ts disabled` keeps the times exact (the parser reads ffmpeg's signed-field spelling of a negative time); now 608/608 on that file, and a real-ffmpeg test pins it. - A full pass that failed was re-armed every minute while the viewer watched what it did publish — a whole-container read per minute. It now backs off for an hour; windows keep running meanwhile. - A failed WINDOW set the track's failure memo, so any uncovered position read "failed" — a final answer that ends the player's poll, even with the full pass about to deliver, and one any account could plant with an absurd `at`. It now marks only the span it failed to read; "failed" is given once the full pass AND a window there have both failed. - Windows join by what they COVER, not a bucketed identity: a backward seek no longer waits behind a window that cannot cover it, and a playhead crossing a bucket edge no longer queues a second read. The queue is bounded per track and overall. - shutdown() sets a flag a worker checks before it spawns ffmpeg (one still in its Plex lookup used to start a read after the kill sweep); queued-job cancellation and the lifespan call are now tested. - A window starts where ffmpeg is told to seek (rounded to the ms it gets), a rounded-up millisecond carries into the minute (no more `00:00:60.000`), a key's first publish claims nothing before its VTT lands, and `at`/`offset`/`start`/`aoff` refuse non-finite values (`at` also an absurd one).
This commit is contained in:
+172
-43
@@ -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)
|
||||
|
||||
@@ -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 <track> — 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),
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user