fix(api): progress checkpoints past the int4 column are a 422, and idle HLS sessions are reaped on a timer
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.
This commit is contained in:
+21
-2
@@ -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()
|
||||
for task in (sweep, reaper):
|
||||
task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await sweep
|
||||
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."""
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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"
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user