fix(r5): address the eighth /code-review high round — 7 findings
The worst one was, again, the previous round's own fix. - Round 7 stopped awaiting the HLS sweep so startup could serve immediately — which also made it run CONCURRENTLY with playback, and `sweep_orphans` snapshotted the live-session set BEFORE listing the root and rmtree'd outside the lock. Since `start_session` reuses a directory path for the same (key, offset, tag), a stale name can become a live session at any moment. Two guards now: nothing whose mtime is newer than this process's start is ever deleted, and the live check + rmtree happen together under the lock, one directory at a time. - The sweep task is awaited after cancel, so shutdown can't leave it pending. - `_rate_limiter` used a truthiness test on its timestamp, so a `now()` of exactly 0.0 read as "never ran" — unreachable in production, but `now` exists to be injected and a test clock starting at 0 is the obvious choice. - The RSS total-outage error now carries the first underlying error: "all N channels failed" is also what ONE dead feed id looks like on a small instance, and a red card the user can't act on is its own kind of dishonest. - The WebKit fullscreen handling moved into `lib/fullscreen` and PlexPlayer uses it too — it had the same unprefixed-only reads in three places, including an Escape branch that would have closed the whole player instead of leaving fullscreen. - The `--max` CSS comment claimed `inset: 0` did the sizing; Tailwind's `w-full` is still on the element and wins. Comment now states the real mechanism. Verified in real Chrome (the pane can't composite or do native fullscreen): - F fills the 1920x889 viewport; the exit pill renders top-right on the letterbox band and overlaps NO YouTube control — the collision this round suspected does not reproduce in this embed (its own buttons sit bottom-left / bottom-right). - Four real ArrowDown presses while maximised: persisted volume 100 -> 80, with the level flash visible on screen. - A real click on the exit pill un-maximises and returns focus to the card. - lib/fullscreen driven from a real click: enter (innerHeight 889 -> 1024), the change listener fires exactly once per transition, exit returns to 889. Backend suite 150 -> 158; the new sweep tests are mutation-checked, and they read their cutoff from the filesystem rather than a clock (the container VM and the bind-mounted host disagree by enough to make a clock-based assertion flaky).
This commit is contained in:
+7
-1
@@ -1,7 +1,7 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from contextlib import asynccontextmanager
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
@@ -82,7 +82,13 @@ async def lifespan(app: FastAPI):
|
||||
yield
|
||||
finally:
|
||||
log.info("Siftlode shutting down")
|
||||
# Awaited, not just cancelled: a bare `.cancel()` lets the loop close with the task still
|
||||
# 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
|
||||
shutdown_scheduler()
|
||||
|
||||
|
||||
|
||||
@@ -47,6 +47,10 @@ _SESSION_IDLE_S = 600 # reap a session with no access for this long
|
||||
# fail. Covered by test_plex_stream.py — this regex is the only thing standing between the sweep
|
||||
# and someone else's data.
|
||||
_SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
|
||||
# When this process started. `sweep_orphans` only deletes directories older than this: everything
|
||||
# newer belongs to a session THIS run created, and since `start_session` reuses a directory path
|
||||
# for the same (rating_key, offset, tag), "stale name" is not the same as "stale directory".
|
||||
_PROCESS_START = time.time()
|
||||
|
||||
_lock = threading.Lock()
|
||||
_sessions: dict[str, "HlsSession"] = {}
|
||||
@@ -268,7 +272,7 @@ def wait_for(path: Path, timeout: float = 20.0) -> bool:
|
||||
return path.exists()
|
||||
|
||||
|
||||
def sweep_orphans() -> int:
|
||||
def sweep_orphans(older_than: float | None = None) -> int:
|
||||
"""Delete OUR segment directories that no live session owns.
|
||||
|
||||
Sessions live only in THIS process, so at startup every session directory under `_HLS_ROOT` is
|
||||
@@ -278,18 +282,39 @@ def sweep_orphans() -> int:
|
||||
|
||||
Only entries whose name matches `_SESSION_DIR_RE` are touched: the root is admin-configurable,
|
||||
so anything else under it belongs to someone else and is left alone. Blocking (rmtree) — call
|
||||
it off the event loop."""
|
||||
with _lock:
|
||||
live = {s.dir for s in _sessions.values()}
|
||||
it off the event loop.
|
||||
|
||||
TWO guards against deleting a LIVE session, because this now runs CONCURRENTLY with request
|
||||
handling (the caller stopped awaiting it so startup isn't held up):
|
||||
* `older_than` (default: this process's start) — a directory touched since we booted was made
|
||||
by THIS run, so it is by definition not a leftover. This is the guard that matters, because
|
||||
`start_session` REUSES the same path for the same (key, offset, tag): a stale directory can
|
||||
become a live one at any moment.
|
||||
* the live-session check and the rmtree happen together under `_lock`, so a session can't be
|
||||
registered in the gap between them. The lock is held for one directory at a time — long
|
||||
enough to be atomic, short enough that a boot-time sweep of a big backlog doesn't stall
|
||||
playback."""
|
||||
cutoff = _PROCESS_START if older_than is None else older_than
|
||||
dropped = 0
|
||||
try:
|
||||
entries = list(_HLS_ROOT.iterdir())
|
||||
except OSError: # the root doesn't exist yet — nothing to sweep
|
||||
return 0
|
||||
for p in entries:
|
||||
if p.is_dir() and p not in live and _SESSION_DIR_RE.match(p.name):
|
||||
if not _SESSION_DIR_RE.match(p.name):
|
||||
continue
|
||||
if not p.is_dir():
|
||||
continue
|
||||
try:
|
||||
if p.stat().st_mtime >= cutoff:
|
||||
continue # this run made (or reused) it — never ours to delete
|
||||
except OSError: # vanished under us (a concurrent restart of that session)
|
||||
continue
|
||||
with _lock:
|
||||
if any(s.dir == p for s in _sessions.values()):
|
||||
continue
|
||||
shutil.rmtree(p, ignore_errors=True)
|
||||
dropped += 1
|
||||
dropped += 1
|
||||
return dropped
|
||||
|
||||
|
||||
|
||||
@@ -60,13 +60,19 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str
|
||||
|
||||
Raises RuntimeError when NOTHING could be polled (`failed == total`, total > 0): a run that
|
||||
achieved literally nothing is a failed run, and only an exception turns the scheduler card red
|
||||
— a returned count leaves it green with the outage buried in the summary line."""
|
||||
— a returned count leaves it green with the outage buried in the summary line. The message
|
||||
carries the FIRST underlying error, because on a small instance "all channels failed" is also
|
||||
what a single dead feed id looks like."""
|
||||
if channels is None:
|
||||
channels = db.execute(select(Channel)).scalars().all()
|
||||
total = len(channels)
|
||||
new = 0
|
||||
done = 0
|
||||
failed = 0
|
||||
# Kept for the total-outage message: "failed for all N channels" alone can't tell an egress
|
||||
# outage from one permanently dead feed id — which is the whole difference on an instance with
|
||||
# a single subscription, where BOTH render as 100% failed.
|
||||
first_error: str | None = None
|
||||
# Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
|
||||
# each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
|
||||
# plain str) to the workers; the ORM instance is touched only here on the main thread.
|
||||
@@ -86,17 +92,20 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str
|
||||
except RssError as exc:
|
||||
log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
|
||||
failed += 1
|
||||
except Exception:
|
||||
first_error = first_error or f"{channel.id}: {exc}"
|
||||
except Exception as exc:
|
||||
log.exception("RSS fetch failed for channel %s", channel.id)
|
||||
failed += 1
|
||||
first_error = first_error or f"{channel.id}: {exc}"
|
||||
else:
|
||||
try:
|
||||
new += apply_rss_feed(db, channel, entries)
|
||||
except Exception:
|
||||
except Exception as exc:
|
||||
# A write failure (read-only DB, failover) leaves the channel just as un-polled
|
||||
# as an unreachable feed — count it, or a wholly broken run reports "failed=0".
|
||||
db.rollback()
|
||||
failed += 1
|
||||
first_error = first_error or f"{channel.id} (apply): {exc}"
|
||||
log.exception("RSS apply failed for channel %s", channel.id)
|
||||
# One place that advances the counter, whatever happened above.
|
||||
done += 1
|
||||
@@ -106,9 +115,12 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str
|
||||
if total and failed == total:
|
||||
# Nothing was polled at all — an egress outage (the tinyproxy-after-reboot incident), a
|
||||
# dead DNS, a read-only database. Returning normally would record the run as `ok` with the
|
||||
# failure hidden inside the result string, i.e. a green pill for a total outage; the count
|
||||
# in the message keeps the summary just as informative.
|
||||
raise RuntimeError(f"RSS poll failed for all {total} channels")
|
||||
# failure hidden inside the result string, i.e. a green pill for a total outage.
|
||||
# The first error TRAVELS WITH the message on purpose: on an instance with one or two
|
||||
# subscriptions "all channels failed" is also what ONE dead feed id looks like, and a red
|
||||
# card the user can't act on is its own kind of dishonest. `channel: HTTP 404` says
|
||||
# "fix this subscription"; `Connection refused` says "the egress is down".
|
||||
raise RuntimeError(f"RSS poll failed for all {total} channels — first: {first_error}")
|
||||
return {"new": new, "failed": failed}
|
||||
|
||||
|
||||
|
||||
@@ -330,11 +330,17 @@ _CANCEL_POLL_INTERVAL_S = 1.0
|
||||
def _rate_limiter(interval_s: float, *, now=time.monotonic):
|
||||
"""`should_run()` → True at most once per `interval_s` (the first call always passes).
|
||||
`now` is injectable so tests drive a clock instead of patching the global one."""
|
||||
state = {"at": 0.0}
|
||||
# `None`, not 0.0: "never ran" has to be a VALUE, not a falsy timestamp. A truthiness test here
|
||||
# reads a legitimate `now() == 0.0` as "never ran" and lets a second call straight through —
|
||||
# unreachable with real `time.monotonic()`, but `now` exists to be injected, and a test clock
|
||||
# starting at 0 is the obvious choice. That would let a throttle test encode the wrong
|
||||
# behaviour and still pass.
|
||||
state: dict[str, float | None] = {"at": None}
|
||||
|
||||
def should_run() -> bool:
|
||||
t = now()
|
||||
if state["at"] and t - state["at"] < interval_s:
|
||||
last = state["at"]
|
||||
if last is not None and t - last < interval_s:
|
||||
return False
|
||||
state["at"] = t
|
||||
return True
|
||||
|
||||
@@ -1,10 +1,16 @@
|
||||
"""The guard in front of a recursive delete.
|
||||
"""The guards in front of a recursive delete.
|
||||
|
||||
`sweep_orphans` rmtree's directories under `_HLS_ROOT`, and that root is admin-configurable
|
||||
(PLEX_HLS_DIR) — it can legitimately be a shared scratch path. `_SESSION_DIR_RE` is the ONLY thing
|
||||
that keeps the sweep off a neighbour's data, so the names it accepts are pinned here."""
|
||||
that keeps the sweep off a neighbour's data, so the names it accepts are pinned here.
|
||||
|
||||
Since the sweep stopped being awaited at startup it also runs CONCURRENTLY with playback, so the
|
||||
second half of this file pins what keeps it off a LIVE session's segments."""
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from app.plex import stream
|
||||
from app.plex.stream import _SESSION_DIR_RE
|
||||
|
||||
|
||||
@@ -42,3 +48,97 @@ def test_our_session_directories_are_swept(name):
|
||||
)
|
||||
def test_foreign_directories_are_left_alone(name):
|
||||
assert not _SESSION_DIR_RE.match(name)
|
||||
|
||||
|
||||
class _FakeSession:
|
||||
def __init__(self, directory):
|
||||
self.dir = directory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hls_root(tmp_path, monkeypatch):
|
||||
"""Point the sweep at a temp root with no sessions registered."""
|
||||
monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path)
|
||||
monkeypatch.setattr(stream, "_sessions", {})
|
||||
return tmp_path
|
||||
|
||||
|
||||
def _session_dir(root, name, *, age_s=0.0):
|
||||
"""A session directory, optionally backdated by `age_s` to look like a previous run's."""
|
||||
d = root / name
|
||||
d.mkdir()
|
||||
(d / "seg_0.ts").write_bytes(b"x")
|
||||
if age_s:
|
||||
t = d.stat().st_mtime - age_s
|
||||
os.utime(d, (t, t))
|
||||
return d
|
||||
|
||||
|
||||
def _cutoff(root):
|
||||
"""A "process start" read from the FILESYSTEM, not from a clock.
|
||||
|
||||
The guard compares a directory's mtime against a timestamp, and the two do not have to come
|
||||
from the same clock: on a bind-mounted dev volume the file times are stamped by the host while
|
||||
`time.time()` runs in the container VM, and the drift between them is real (it made an earlier
|
||||
version of these tests flaky). Anchoring the cutoff to a file this test just created removes
|
||||
the clock from the assertion; that the DEFAULT cutoff is `_PROCESS_START` is pinned separately
|
||||
below."""
|
||||
probe = root / ".cutoff-probe"
|
||||
probe.write_bytes(b"")
|
||||
at = probe.stat().st_mtime
|
||||
probe.unlink()
|
||||
return at
|
||||
|
||||
|
||||
class TestSweepOrphans:
|
||||
def test_a_previous_runs_leftovers_are_deleted(self, hls_root):
|
||||
d = _session_dir(hls_root, "12345_0_None_+0.00", age_s=3600)
|
||||
assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 1
|
||||
assert not d.exists()
|
||||
|
||||
def test_foreign_directories_survive(self, hls_root):
|
||||
keep = hls_root / "backup_2024"
|
||||
keep.mkdir()
|
||||
(keep / "important.tar").write_bytes(b"x")
|
||||
assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
|
||||
assert (keep / "important.tar").exists()
|
||||
|
||||
def test_a_live_sessions_directory_is_never_deleted(self, hls_root):
|
||||
# Registered session, but with an OLD directory — only the live check can save it.
|
||||
d = _session_dir(hls_root, "12345_0_multi_+0.00", age_s=3600)
|
||||
stream._sessions["k"] = _FakeSession(d)
|
||||
assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
|
||||
assert d.exists()
|
||||
|
||||
def test_a_directory_this_run_created_is_never_deleted(self, hls_root):
|
||||
# THE RACE: the sweep is no longer awaited before the app serves, and `start_session`
|
||||
# REUSES the path of a same-key session — so between our listing and our rmtree, a stale
|
||||
# name can become a live directory. A fresh mtime is what tells them apart, and it must
|
||||
# hold even while `_sessions` is still empty (the session registers a beat later).
|
||||
cutoff = _cutoff(hls_root)
|
||||
fresh = _session_dir(hls_root, "12345_0_None_+0.00") # created "after boot"
|
||||
old = _session_dir(hls_root, "999_0_None_+0.00", age_s=3600)
|
||||
assert stream.sweep_orphans(older_than=cutoff) == 1 # only the leftover
|
||||
assert fresh.exists()
|
||||
assert not old.exists()
|
||||
|
||||
def test_the_default_cutoff_is_this_processs_start(self, hls_root, monkeypatch):
|
||||
# The parameter exists for the tests; production calls it with no argument, so pin what
|
||||
# that means — otherwise the guard above could be right and unused.
|
||||
d = _session_dir(hls_root, "12345_0_None_+0.00")
|
||||
monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 1)
|
||||
assert stream.sweep_orphans() == 1 # "older than boot" → swept
|
||||
assert not d.exists()
|
||||
|
||||
def test_a_missing_root_is_not_an_error(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path / "nope")
|
||||
monkeypatch.setattr(stream, "_sessions", {})
|
||||
assert stream.sweep_orphans() == 0
|
||||
|
||||
def test_a_stray_file_with_a_session_name_is_left_alone(self, hls_root):
|
||||
# Only directories are ours; a same-named FILE belongs to someone else.
|
||||
f = hls_root / "12345_0"
|
||||
f.write_bytes(b"x")
|
||||
os.utime(f, (f.stat().st_mtime - 3600,) * 2) # old enough that only is_dir() saves it
|
||||
assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
|
||||
assert f.exists()
|
||||
|
||||
@@ -156,12 +156,22 @@ class TestRssPollResult:
|
||||
with pytest.raises(RuntimeError, match="all 2 channels"):
|
||||
self._run(monkeypatch, fetch=boom, apply_=lambda db, ch, e: 1)
|
||||
|
||||
def test_the_outage_message_carries_the_underlying_error(self, monkeypatch):
|
||||
# "all N channels failed" is ALSO what one dead feed id looks like on a small instance.
|
||||
# The card has to say which: `HTTP 404` = fix the subscription, `Connection refused` =
|
||||
# the egress is down.
|
||||
def dead(cid):
|
||||
raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404")
|
||||
|
||||
with pytest.raises(RuntimeError, match="HTTP 404"):
|
||||
self._run(monkeypatch, fetch=dead, apply_=lambda db, ch, e: 1, channels=("UC_dead",))
|
||||
|
||||
def test_a_read_only_database_is_a_total_outage_too(self, monkeypatch):
|
||||
# Every feed reads fine, nothing can be stored: just as un-polled, just as red.
|
||||
def boom(db, ch, entries):
|
||||
raise RuntimeError("read-only transaction")
|
||||
|
||||
with pytest.raises(RuntimeError, match="all 2 channels"):
|
||||
with pytest.raises(RuntimeError, match="read-only transaction"):
|
||||
self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=boom)
|
||||
|
||||
def test_no_channels_at_all_is_not_an_outage(self, monkeypatch):
|
||||
|
||||
Reference in New Issue
Block a user