peter/siftlode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(plex): a subtitle failure about one position no longer ends the track, and windows are read ahead and never wait behind a full pass

Browse Source 
Deep-review round 3 of bug hunt 2026-09.

- `failed` is now final only for a track that can never be text (bitmap, no item/track/file). A Plex
  hiccup or a stretch ffmpeg could not read answers `unavailable`: said, polled again every 5 s, and
  the <track> stays mounted when cues exist elsewhere (gen > 0). Re-picking the selected track asks
  the server again at once.
- Within LOOKAHEAD_S of a covered stretch's end the next window is read ahead, instead of only once
  the playhead has fallen off it (subtitles dropped out at every edge while the full pass was backed
  off or slower than a window lasts).
- Windows and full passes run on separate pools with separate in-flight caps, so a window never
  queues behind (or is refused by) minute-long full passes — the pool comment promised that and
  nothing enforced it.
- A pass that times out is TERMinated and keeps what it reached (published as partial) instead of
  being killed and discarded.
- Lapsed failure memos are swept on every write (they were only pruned when the same key was read
  again, so polling fresh keys grew the process without bound).
- `publish` no longer merges into a sidecar with no VTT behind it, or claims a region whose VTT write
  failed, or overwrites an unreadable track with one window.
- A NUL in a rating_key names no row (it was a DataError, a 500, on every Plex route) and no longer
  500s the status poll.

Tests: 16 new, each mutation-checked (18 mutations, all red on the test that names them).
This commit is contained in:
peter
2026-09-11 18:04:30 +02:00
parent f7faa19ee2
commit 4f37c41021
7 changed files with 453 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/plex/subtitles.py
+115 -43
View File
@@ -166,14 +166,17 @@ def cache_read(key: str) -> str | None:
return None
def cache_write(key: str, vtt: str) -> None:
def cache_write(key: str, vtt: str) -> bool:
"""Atomically replace the cached VTT; False when it could not be written (`publish` must then
not claim the region it was carrying)."""
try:
CACHE_DIR.mkdir(parents=True, exist_ok=True)
tmp = CACHE_DIR / f"{key}.vtt.tmp"
tmp.write_text(vtt, encoding="utf-8")
tmp.replace(CACHE_DIR / f"{key}.vtt") # atomic publish
return True
except OSError:
pass
return False
# --- cache identity -----------------------------------------------------------------------------
@@ -222,7 +225,7 @@ def generation(key: str) -> int:
window is superseded by a wider one the browser refetches instead of keeping the short VTT."""
try:
return _vtt_path(key).stat().st_mtime_ns
except OSError:
except (OSError, ValueError): # ValueError: a NUL in the key — `stat` refuses the path outright
return 0
@@ -312,14 +315,22 @@ def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) ->
staging names safe."""
with _PUBLISH_LOCK:
old = cache_read(key)
if old is None and _vtt_path(key).exists():
# There, but unreadable (`cache_read` swallows the OSError). Merging into "" would replace
# a whole track with this one region while its sidecar still claimed all of it.
log.warning("subtitle cache %s unreadable; not publishing into it", key)
return
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))
# A sidecar with no VTT behind it claims cues that are not there, so it is not merged
# into. And a VTT with no sidecar reads as a COMPLETE track (see read_meta — how older
# builds cached), while a publish is two renames: so before a key's first VTT lands, its
# sidecar is made to claim nothing. Dying between the two then leaves a window that is
# extracted again, never one taken for the whole track for good.
old_ranges, was_complete = [], False
_write_meta(key, [], False)
if not cache_write(key, merge_vtt(old or "", vtt)):
return # the region never landed, so the sidecar must not claim it
_write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete)
@@ -332,9 +343,16 @@ def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) ->
# any realistic full pass.
WINDOW_PRE_S = 30.0
WINDOW_LEN_S = 300.0
# …but it does not always land: a full pass can be backed off (FULL_RETRY_S) or simply slower than
# the window lasts (a file past ~40 GB, or several passes sharing the array). A window was only asked
# for once the playhead had LEFT coverage, so every edge unmounted the track for a poll plus an
# extraction. Within this distance of a covered stretch's end, the next window is read ahead.
LOOKAHEAD_S = 60.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
# How long a timed-out ffmpeg gets, after a TERM, to close its outputs before it is killed.
_TERM_GRACE_S = 10.0
# 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.
@@ -347,14 +365,20 @@ 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
# Windows and full passes run on SEPARATE pools, so a window never waits behind a full pass. They
# shared one FIFO pool of three whose comment promised "room for a window" that nothing enforced:
# stepping through three episodes with subtitles on left three full passes (minutes each) holding every
# worker, and a seek window on the third queued behind them. Two of each: bounded, so a burst can't
# thrash the array.
_WIN_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-win")
_FULL_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-full")
# Bounds on what may wait for those pools — separate too, so held full-pass slots (minutes each)
# cannot refuse a window (seconds). The queues are 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 a bound nothing is queued — the next poll asks again, by
# which time something has finished.
_MAX_WINDOWS_INFLIGHT = 8
_MAX_FULL_INFLIGHT = 6
_MAX_TRACK_WINDOWS = 2
_LOCK = threading.Lock()
@@ -396,6 +420,12 @@ UNAVAILABLE_STATUS = {
"extract_failed": 422,
"plex_error": 422,
}
# The codes that are a FINAL answer about the track: it can never be served as text, so the player
# stops asking. The others (`plex_error`, `extract_failed`) are about now or about one stretch — a
# memo that lapses, a window that failed where the rest is fine — and answered as `unavailable`: said,
# but asked again. Given as `failed`, one Plex hiccup ended the subtitles for the rest of the session,
# at cached positions too, and after the full pass had completed behind it.
FINAL_CODES = frozenset({"no_item", "no_track", "no_file", "not_fetchable", "bitmap"})
def reset_state() -> None:
@@ -421,13 +451,28 @@ def shutdown() -> None:
with _LOCK:
_STOPPING = True
procs = list(_PROCS)
_POOL.shutdown(wait=False, cancel_futures=True)
for pool in (_WIN_POOL, _FULL_POOL):
pool.shutdown(wait=False, cancel_futures=True)
for p in procs:
p.kill()
def _sweep_memos() -> None:
"""(Caller holds _LOCK.) Drop every lapsed failure memo. They were only ever pruned when the SAME
key was read again, and the status route takes any rating_key/ordinal — so a script polling fresh
keys grew the API process without bound. Swept on every write, they hold at most a TTL's worth."""
now = time.monotonic()
for k in [k for k, (_, t) in _FAILURES.items() if now - t >= FAIL_TTL_S]:
del _FAILURES[k]
for k in [k for k, t in _FULL_FAILED.items() if now - t >= FULL_RETRY_S]:
del _FULL_FAILED[k]
for k in list(_WIN_FAILED):
_live_window_failures(k)
def _fail(key: str, code: str) -> None:
with _LOCK:
_sweep_memos()
_FAILURES[key] = (code, time.monotonic())
@@ -462,6 +507,7 @@ def _live_window_failures(key: str) -> list[tuple[float, float, float]]:
def _window_fail(key: str, span: tuple[float, float]) -> None:
with _LOCK:
_sweep_memos()
_WIN_FAILED[key] = [*_live_window_failures(key), (span[0], span[1], time.monotonic())]
@@ -474,33 +520,41 @@ 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.
``ready`` = fully extracted · ``partial`` = usable, covers ``at``, still growing · ``preparing``
= nothing to show for ``at`` yet · ``failed`` = ``code`` says why. ``gen`` changes whenever the
cached file grows, so the client can refetch a window that has been superseded."""
= nothing to show for ``at`` yet · ``unavailable`` = nothing for ``at`` and nothing being read
for it right now (``code`` says why), ask again · ``failed`` = the track can never be shown
(``code`` says why) — the only final answer. ``gen`` changes whenever the cached file grows, so
the client can refetch a window that has been superseded; 0 = nothing is cached at all."""
key = cache_key(rating_key, sub_ord)
ranges, complete = read_meta(key)
have = cache_read(key) is not None
covered = have and covers(ranges, at)
gen = generation(key)
if complete and have:
return {"state": "ready", "code": None, "gen": generation(key)}
return {"state": "ready", "code": None, "gen": gen}
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.
return {
"state": "failed" if code in FINAL_CODES else "unavailable",
"code": code,
"gen": gen,
}
# Nothing left that could cover the position for now: the full pass failed, AND so did a window
# here. Still not final — the memos lapse, and the rest of the track may be right there.
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)}
return {"state": "unavailable", "code": "extract_failed", "gen": gen}
if not code:
_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)}
probe, want_window = at, not covered and not window_failed
if covered:
# Read ahead of an edge instead of waiting to fall off it (see LOOKAHEAD_S). The probe is
# just PAST the edge: `covers` is inclusive, so the edge itself reads as already done.
edge = max(b for a, b in ranges if a <= at <= b)
if edge - at <= LOOKAHEAD_S and edge < _FULL_END_S:
probe = edge + 1.0
want_window = not _window_failed(key, probe)
_schedule(rating_key, sub_ord, probe, want_window=want_window, want_full=not full_failed)
return {"state": "partial" if covered else "preparing", "code": None, "gen": gen}
def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None]:
@@ -532,9 +586,11 @@ def _schedule(
full pass is per ITEM, because one read fans out to every embedded text track of the file."""
jobs: list[tuple[tuple, float | None]] = []
with _LOCK:
if _STOPPING or len(_INFLIGHT) >= _MAX_INFLIGHT:
if _STOPPING:
return
if want_window:
n_windows = sum(1 for span in _INFLIGHT.values() if span is not None)
n_full = len(_INFLIGHT) - n_windows
if want_window and n_windows < _MAX_WINDOWS_INFLIGHT:
spans = [
span
for ident, span in _INFLIGHT.items()
@@ -548,11 +604,12 @@ def _schedule(
ident = ("win", rating_key, sub_ord, w_start)
_INFLIGHT[ident] = (w_start, w_start + WINDOW_PRE_S + WINDOW_LEN_S)
jobs.append((ident, w_start))
if want_full and ("full", rating_key) not in _INFLIGHT:
if want_full and n_full < _MAX_FULL_INFLIGHT and ("full", rating_key) not in _INFLIGHT:
_INFLIGHT[("full", rating_key)] = None
jobs.append((("full", rating_key), None))
for ident, window_start in jobs:
_POOL.submit(_run, ident, rating_key, sub_ord, window_start, at)
pool = _FULL_POOL if window_start is None else _WIN_POOL
pool.submit(_run, ident, rating_key, sub_ord, window_start, at)
def _already_done(key: str, window_start: float | None, at: float) -> bool:
@@ -601,6 +658,7 @@ def _pass_failed(
window marks only the stretch it failed to read, for the tracks it was reading."""
if window_start is None:
with _LOCK:
_sweep_memos()
_FULL_FAILED[rating_key] = time.monotonic()
return
for key in keys:
@@ -731,13 +789,23 @@ def _ffmpeg_extract(
_PROCS.add(proc)
if _STOPPING:
proc.kill()
timed_out = False
try:
try:
_, err = proc.communicate(timeout=EXTRACT_TIMEOUT_S)
except subprocess.TimeoutExpired:
proc.kill()
proc.communicate()
raise
# A pass that ran out of time is a pass that died partway: what it reached is kept and
# published as partial (the caller's unclean path), like any other non-zero exit. It
# used to be killed and its outputs deleted unread — and with the full-pass backoff, the
# same doomed read came back every hour. TERM first: ffmpeg closes its outputs on it,
# whereas a KILL leaves only what was already flushed in whole blocks (measured: none).
timed_out = True
proc.terminate()
try:
_, err = proc.communicate(timeout=_TERM_GRACE_S)
except subprocess.TimeoutExpired:
proc.kill()
_, err = proc.communicate()
finally:
with _LOCK:
_PROCS.discard(proc)
@@ -748,7 +816,11 @@ def _ffmpeg_extract(
}
if not got:
raise SubtitleUnavailable("extract_failed")
clean = proc.returncode == 0
clean = proc.returncode == 0 and not timed_out
if not clean:
log.warning("ffmpeg subtitle pass exited %s: %s", proc.returncode, (err or b"")[:200])
log.warning(
"ffmpeg subtitle pass %s: %s",
"timed out" if timed_out else f"exited {proc.returncode}",
(err or b"")[:200],
)
return got, clean