fix(r5): address the second /code-review high round — 7 findings
- the volume keys no longer defer to a scrollable card while in fullscreen
(mirrored into a ref: the keydown handler is bound once and state would be stale)
- the debounced volume flush writes straight to storage; persisting inside a
setState updater is not guaranteed to run for an unmounting component
- the HLS sweep also matches the original {key}_{start} directory naming, which
is what the oldest leftovers on disk actually use
- run_rss_poll returns {new, failed} so an egress outage stops rendering as
"ok — 0" in the scheduler card and the audit log; /api/sync/rss reports it too
- drop the unused poll_rss_channel (it silently gained a new exception)
- _Breaker.run is generic, so a call site keeps its item-id type
- test helper instead of the generator-throw idiom
This commit is contained in:
@@ -34,11 +34,13 @@ log = logging.getLogger("siftlode.plex")
|
||||
_HLS_ROOT = Path(settings.plex_hls_dir) if settings.plex_hls_dir else Path(settings.download_root) / ".plex-hls"
|
||||
_SEG_SECONDS = 6
|
||||
_SESSION_IDLE_S = 600 # reap a session with no access for this long
|
||||
# A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session).
|
||||
# `sweep_orphans` deletes ONLY names matching this shape: `_HLS_ROOT` is admin-configurable
|
||||
# A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session);
|
||||
# before multi-audio it was just `{rating_key}_{int(start_s)}`, so the tail is OPTIONAL here — the
|
||||
# oldest leftovers on disk carry the old shape and are exactly what the sweep exists to remove.
|
||||
# `sweep_orphans` deletes ONLY names matching this: `_HLS_ROOT` is admin-configurable
|
||||
# (PLEX_HLS_DIR), so it may legitimately point at a shared scratch path we do not own — a blanket
|
||||
# "delete every directory under the root" would take the neighbours with it.
|
||||
_SESSION_DIR_RE = re.compile(r"^.+_\d+_[^_]+_[+-]\d+\.\d{2}$")
|
||||
_SESSION_DIR_RE = re.compile(r"^.+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
|
||||
|
||||
_lock = threading.Lock()
|
||||
_sessions: dict[str, "HlsSession"] = {}
|
||||
|
||||
@@ -52,8 +52,10 @@ def sync_subscriptions(
|
||||
def sync_rss(
|
||||
user: User = Depends(require_human), db: Session = Depends(get_db)
|
||||
) -> dict:
|
||||
new = run_rss_poll(db, _user_channels(db, user))
|
||||
return {"new_videos": new}
|
||||
result = run_rss_poll(db, _user_channels(db, user))
|
||||
# `failed` = feeds that could not be read at all (see run_rss_poll); surfaced so a manual poll
|
||||
# reports an outage instead of an innocent-looking "0 new".
|
||||
return {"new_videos": result["new"], "failed_feeds": result["failed"]}
|
||||
|
||||
|
||||
@router.post("/backfill")
|
||||
|
||||
@@ -9,6 +9,7 @@ import hashlib
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime, timezone
|
||||
from typing import TypeVar
|
||||
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -272,8 +273,15 @@ def plan_push(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
|
||||
_MAX_CONSECUTIVE_FAILURES = 5
|
||||
|
||||
|
||||
class _Failed:
|
||||
"""Type of the `FAILED` sentinel, so `run()` can stay generic in the op's return type: a call
|
||||
site gets back `str | _Failed` (not a bare `object`) and keeps its item id typed."""
|
||||
|
||||
|
||||
#: Returned by `_Breaker.run` when the write did not happen (refused or failed).
|
||||
FAILED = object()
|
||||
FAILED = _Failed()
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class _Breaker:
|
||||
@@ -305,7 +313,7 @@ class _Breaker:
|
||||
self.limit,
|
||||
)
|
||||
|
||||
def run(self, op: Callable[[], object], on_fail: Callable[[], None]) -> object:
|
||||
def run(self, op: Callable[[], T], on_fail: Callable[[], None]) -> T | _Failed:
|
||||
"""Attempt one YouTube write. Returns the op's result, or `FAILED` when the breaker is
|
||||
already tripped (op not called) or the op raised YouTubeError. `on_fail` records the item
|
||||
in the caller's failure list either way."""
|
||||
@@ -387,7 +395,7 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
|
||||
lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
|
||||
lambda v=vid: failures.append(v),
|
||||
)
|
||||
if item_id is not FAILED:
|
||||
if not isinstance(item_id, _Failed):
|
||||
item_id_by_vid[vid] = item_id
|
||||
inserted += 1
|
||||
# Reorder to match `desired` via insertion-sort moves over a local model.
|
||||
|
||||
@@ -48,7 +48,11 @@ def get_service_user(db: Session) -> User | None:
|
||||
return users[0] if users else None
|
||||
|
||||
|
||||
def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
|
||||
def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict:
|
||||
"""Poll every channel's free RSS feed. Returns `{"new": inserted, "failed": unreadable_feeds}`
|
||||
— the failure count is part of the RESULT, not just a log line, so the scheduler card and the
|
||||
audit summary can tell "no new uploads" apart from "every feed was unreachable" (an egress
|
||||
outage otherwise renders identically to a quiet hour)."""
|
||||
if channels is None:
|
||||
channels = db.execute(select(Channel)).scalars().all()
|
||||
total = len(channels)
|
||||
@@ -89,7 +93,7 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
|
||||
progress.report(done, total, "rss_poll")
|
||||
if failed:
|
||||
log.warning("RSS poll: %d/%d channel feeds could not be read", failed, total)
|
||||
return new
|
||||
return {"new": new, "failed": failed}
|
||||
|
||||
|
||||
def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
|
||||
|
||||
@@ -14,7 +14,6 @@ from app import progress, sysconfig
|
||||
from app.models import LIVE_OR_UPCOMING, Channel, Video
|
||||
from app.titles import normalize_title
|
||||
from app.youtube.client import YouTubeClient, best_thumbnail
|
||||
from app.youtube.rss import fetch_channel_feed
|
||||
from app.youtube.shorts import make_client, probe_is_short
|
||||
|
||||
_DURATION_RE = re.compile(
|
||||
@@ -84,10 +83,6 @@ def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int:
|
||||
return inserted
|
||||
|
||||
|
||||
def poll_rss_channel(db: Session, channel: Channel) -> int:
|
||||
return apply_rss_feed(db, channel, fetch_channel_feed(channel.id))
|
||||
|
||||
|
||||
# --- Backfill (uploads playlist; costs quota) ---
|
||||
def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
|
||||
content = item.get("contentDetails", {})
|
||||
|
||||
@@ -51,6 +51,11 @@ class TestBreaker:
|
||||
assert not b.open()
|
||||
|
||||
|
||||
def _quota_dead():
|
||||
"""A write op that fails the way a quota-exhausted / scope-revoked account fails."""
|
||||
raise YouTubeError("quota exceeded")
|
||||
|
||||
|
||||
class TestBreakerRun:
|
||||
def test_returns_the_ops_result_and_leaves_failures_alone(self):
|
||||
b, failed = _Breaker(), []
|
||||
@@ -59,22 +64,25 @@ class TestBreakerRun:
|
||||
|
||||
def test_a_youtube_error_records_the_failure_and_counts_towards_the_trip(self):
|
||||
b, failed = _Breaker(limit=2), []
|
||||
assert b.run(lambda: (_ for _ in ()).throw(YouTubeError("quota")), lambda: failed.append("v1")) is FAILED
|
||||
assert b.run(_quota_dead, lambda: failed.append("v1")) is FAILED
|
||||
assert failed == ["v1"] and b.open()
|
||||
b.run(lambda: (_ for _ in ()).throw(YouTubeError("quota")), lambda: failed.append("v2"))
|
||||
b.run(_quota_dead, lambda: failed.append("v2"))
|
||||
assert not b.open()
|
||||
|
||||
def test_a_tripped_breaker_never_calls_the_op(self):
|
||||
b, failed, calls = _Breaker(limit=1), [], []
|
||||
b.run(lambda: (_ for _ in ()).throw(YouTubeError("quota")), lambda: failed.append("v1"))
|
||||
b.run(_quota_dead, lambda: failed.append("v1"))
|
||||
assert b.run(lambda: calls.append("called"), lambda: failed.append("v2")) is FAILED
|
||||
assert calls == [] # the whole point: no more quota spent
|
||||
assert failed == ["v1", "v2"] # …but the item is still reported
|
||||
|
||||
def test_other_exceptions_are_not_swallowed(self):
|
||||
# Only YouTubeError is a "write failed" signal; a bug must still surface.
|
||||
def boom():
|
||||
raise ValueError("bug")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
_Breaker().run(lambda: (_ for _ in ()).throw(ValueError("bug")), lambda: None)
|
||||
_Breaker().run(boom, lambda: None)
|
||||
|
||||
|
||||
class TestRssErrorVsEmpty:
|
||||
|
||||
Reference in New Issue
Block a user