From 86971c574d22f8773faa5c1c8e036fc18282c36e Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 11 Sep 2026 18:05:30 +0200 Subject: [PATCH] fix(api): progress checkpoints past the int4 column are a 422, and idle HLS sessions are reaped on a timer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deep-review round 3 of bug hunt 2026-09. - The YouTube and Plex progress bodies are bounded by POSITION_MAX_S (the int4 position columns): a finite position of 2^31 or more was a DataError, i.e. a 500 — the sibling of round 2's non-finite fix. - plex/stream.py promised "a periodic reaper", but reap_idle ran only inside the NEXT session request: after the last playback an abandoned session (its ffmpeg still working, its directory a whole segmented copy of the film) lived until someone pressed play again. A lifespan task reaps every 60 s. - Tests for what round 2 left untested: the 422's body keeps the list shape the UI localizes, and stream_session refuses non-finite start/aoff. --- backend/app/main.py | 25 ++++++++++++++-- backend/app/models.py | 6 ++++ backend/app/routes/feed.py | 8 ++++-- backend/app/routes/plex/playback.py | 8 ++++-- backend/tests/test_hls_reaper.py | 41 +++++++++++++++++++++++++++ backend/tests/test_progress_bounds.py | 29 +++++++++++++++++++ 6 files changed, 108 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_hls_reaper.py diff --git a/backend/app/main.py b/backend/app/main.py index 3acb4bde..a8add1d1 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -81,6 +81,7 @@ async def lifespan(app: FastAPI): # alone doesn't buy that; not waiting does. The sweep only touches directories no live session # owns, so racing it against early playback is safe. sweep = asyncio.create_task(_sweep_hls_orphans()) + reaper = asyncio.create_task(_reap_hls_sessions()) start_scheduler() try: yield @@ -90,15 +91,33 @@ async def lifespan(app: FastAPI): # pending ("Task was destroyed but it is pending!" — noise that reads like a bug during # triage). Cancelling does NOT interrupt the rmtree thread itself; this only makes our own # bookkeeping honest. - sweep.cancel() - with suppress(asyncio.CancelledError): - await sweep + for task in (sweep, reaper): + task.cancel() + with suppress(asyncio.CancelledError): + await task 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() +# How often idle HLS sessions are reaped (see plex/stream.py `reap_idle`). +_HLS_REAP_EVERY_S = 60.0 + + +async def _reap_hls_sessions() -> None: + """The periodic reaper plex/stream.py's docstring promises. `reap_idle` used to run only inside + the NEXT session request, so after the last playback of the day an abandoned session — its + ffmpeg still working through the film, its directory a whole segmented copy of it — lived until + somebody pressed play again or the process restarted.""" + while True: + await asyncio.sleep(_HLS_REAP_EVERY_S) + try: + await asyncio.to_thread(plex_stream.reap_idle) + except Exception: + log.exception("HLS session reap failed") + + async def _sweep_hls_orphans() -> None: """Background half of the startup HLS sweep (see lifespan). Owns its own errors: a failure here must never take the app down, and a cancelled shutdown is not one.""" diff --git a/backend/app/models.py b/backend/app/models.py index 02a2895a..6ea02966 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -24,6 +24,11 @@ from app.db import Base from app.utils import canonical_email +# The resume-position columns (VideoState / PlexState.position_seconds) are int4. The checkpoint +# routes bound their bodies by this, so an out-of-range position is a 422 rather than a 500. +POSITION_MAX_S = 2**31 - 1 + + class TimestampMixin: """Adds a server-defaulted `created_at`. Models that index created_at or order it differently (QuotaEvent, Notification, Message) keep their own column on purpose.""" @@ -423,6 +428,7 @@ class VideoState(Base, UpdatedAtMixin): # status: a partially-watched video keeps status "new" but a non-zero position, so # it can render a progress bar and match the "in progress" feed filter. 0 = not # started / finished (trivially-early and near-finished positions are not stored). + # int4: a checkpoint body must stay under POSITION_MAX_S, or the flush is a DataError (a 500). position_seconds: Mapped[int] = mapped_column( Integer, default=0, server_default="0", nullable=False ) diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index e969de25..a1a8a8c3 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -1,7 +1,7 @@ from datetime import date, datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import and_, func, or_, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session @@ -22,6 +22,7 @@ from app.feedquery import ( sort_keys, ) from app.models import ( + POSITION_MAX_S, ChannelTag, Tag, User, @@ -310,8 +311,9 @@ class VideoProgressIn(BaseModel): # Accept a float and truncate (below) — the player may post a raw currentTime (a float), which # the old `int(payload.get(...))` silently truncated. A plain `int` field would 422 that on the # 10s progress hot path, losing the resume position; `float` keeps the old tolerance. - position_seconds: float = 0 - duration_seconds: float = 0 + # Bounded above by the int4 column (see POSITION_MAX_S): past it the flush was a 500. + position_seconds: float = Field(0, le=POSITION_MAX_S) + duration_seconds: float = Field(0, le=POSITION_MAX_S) @router.post("/videos/{video_id}/progress") diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index f0e03ef9..248855a3 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -15,13 +15,14 @@ from fastapi import ( Response, ) from fastapi.responses import FileResponse -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import select from sqlalchemy.orm import Session from app.auth import current_user, resolved_user_id from app.db import SessionLocal, get_db from app.models import ( + POSITION_MAX_S, PlexItem, PlexSeason, PlexShow, @@ -293,8 +294,9 @@ class ItemProgressIn(BaseModel): # 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 - duration_seconds: float = 0 + # Bounded above by the int4 column (see POSITION_MAX_S): past it the flush was a 500. + position_seconds: float = Field(0, le=POSITION_MAX_S) + duration_seconds: float = Field(0, le=POSITION_MAX_S) final: bool = False diff --git a/backend/tests/test_hls_reaper.py b/backend/tests/test_hls_reaper.py new file mode 100644 index 00000000..51521ed5 --- /dev/null +++ b/backend/tests/test_hls_reaper.py @@ -0,0 +1,41 @@ +"""Idle HLS sessions are reaped on a timer, not only when the next session is requested. + +plex/stream.py promises "a periodic reaper kills idle ones", but `reap_idle` ran only inside +POST /session: after the last playback, an abandoned session (its ffmpeg still working through the +film, its directory a whole segmented copy of it) lived until somebody pressed play again. +""" + +import asyncio + +import app.main as app_main +from app.main import app + + +class _Configured: + def __enter__(self): + return None + + def __exit__(self, *exc): + return False + + +def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch): + async def _no_sweep(): + return None + + reaped: list[int] = [] + 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: None) + monkeypatch.setattr(app_main, "_HLS_REAP_EVERY_S", 0.01) + monkeypatch.setattr(app_main.plex_stream, "reap_idle", lambda: reaped.append(1) or 0) + + async def _cycle(): + async with app_main.lifespan(app): + await asyncio.sleep(0.2) + + asyncio.run(_cycle()) + assert len(reaped) >= 2, "no periodic reap ran while the app was up" diff --git a/backend/tests/test_progress_bounds.py b/backend/tests/test_progress_bounds.py index aea5dee5..7736cb7c 100644 --- a/backend/tests/test_progress_bounds.py +++ b/backend/tests/test_progress_bounds.py @@ -32,3 +32,32 @@ def test_a_non_finite_position_is_refused(client, path, raw): headers={"content-type": "application/json"}, ) assert r.status_code == 422, r.text + + +@pytest.mark.parametrize("path", ["/api/videos/v1/progress", "/api/plex/item/98614/progress"]) +@pytest.mark.parametrize("field", ["position_seconds", "duration_seconds"]) +def test_a_position_past_the_int4_column_is_refused_not_a_500(client, path, field): + """Finite but past the int4 column: the flush was a DataError, i.e. a 500 with a traceback.""" + body = {"position_seconds": 10, "duration_seconds": 3000, field: 2**31} + assert client.post(path, json=body).status_code == 422 + + +def test_the_refusal_keeps_the_shape_the_ui_reads(client): + """The handler replaces FastAPI's own for every validation error in the app, and the UI turns + exactly this shape — a LIST of errors — into its localized "check your input" sentence. The + refused value is echoed by its name, because JSON cannot carry NaN.""" + r = client.post( + "/api/videos/v1/progress", + content='{"position_seconds": NaN, "duration_seconds": 3000}', + headers={"content-type": "application/json"}, + ) + detail = r.json()["detail"] + assert isinstance(detail, list) and detail[0]["loc"] == ["body", "position_seconds"] + assert detail[0]["input"] == "nan" + + +@pytest.mark.parametrize("q", ["start=inf", "start=nan", "aoff=inf", "aoff=-inf", "aoff=nan"]) +def test_a_session_at_an_impossible_offset_is_refused(client, q): + """`start=inf` reached `int()` in the session builder (a 500); `aoff=inf` spawned an ffmpeg + with a non-finite offset.""" + assert client.post(f"/api/plex/stream/98614/session?{q}").status_code == 422