diff --git a/.github/workflows/repo-guards.yml b/.github/workflows/repo-guards.yml index 77ff061..b266228 100644 --- a/.github/workflows/repo-guards.yml +++ b/.github/workflows/repo-guards.yml @@ -20,3 +20,5 @@ jobs: python-version: "3.12" - name: Run repo guards run: python tests/run_repo_guards.py + - name: Run reliability unit tests + run: python tests/test_reliability.py diff --git a/README.md b/README.md index d0a26e7..e1a33ac 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,13 @@ Environment variables: | `HYPERWALL_NO_RELAUNCH=1` | Skip exe re-launch (script mode) | | `HYPERWALL_ISOLATED=1` | Force G-Sync isolation on (bypass exe-name check) | | `HYPERWALL_AUTO_TRANSCODE=0` | Disable auto-transcode heuristic | +| `HYPERWALL_STALL_TIMEOUT_S` | Stall watchdog: flag a frozen stream after N s of no progress (default 20) | +| `HYPERWALL_WATCHDOG_MS` | Stall watchdog poll interval in ms (default 5000) | +| `HYPERWALL_CRASHLOOP_THRESHOLD` | Failures within the window before a cell is parked (default 5) | +| `HYPERWALL_CRASHLOOP_WINDOW_S` | Rolling window for the crash-loop guard (default 60) | +| `HYPERWALL_CRASHLOOP_COOLDOWN_S` | How long a parked cell waits before resuming (default 120) | +| `HYPERWALL_CACHE_BUDGET_MB` | Aggregate demuxer cache ceiling across all cells (default 3072) | +| `HYPERWALL_DEMUXER_PER_CELL_MB` | Desired per-cell demuxer cache before budget scaling (default 512) | ## Building diff --git a/hyperwall/cell.py b/hyperwall/cell.py index 7e75ae9..6b2c146 100644 --- a/hyperwall/cell.py +++ b/hyperwall/cell.py @@ -51,17 +51,23 @@ AUTOHIDE_MS, CONTROLS_HEIGHT, CONTROLS_OPACITY, + CRASH_LOOP_COOLDOWN_S, + CRASH_LOOP_THRESHOLD, + CRASH_LOOP_WINDOW_S, MAX_RETRIES, MOUSE_IDLE_MS, MPV_LOG_NOISE, MPV_OPTS, OVERLAY_SHOW_MS, + STALL_TIMEOUT_S, STATS_COUNTER_PROPS, STATS_ENABLED, STATS_INFO_PROPS, UI_SCALE, + WATCHDOG_INTERVAL_MS, apply_env_overrides, ) +from .reliability import is_stalled, should_park logger = logging.getLogger("HyperWall") @@ -139,6 +145,15 @@ def __init__(self, controller: Any): self._emby_item_id: str | None = None self._switching = False # set in play(), consumed in _handle_eof + # Reliability / self-healing (Epic 2) + self._last_progress_ts = 0.0 # monotonic ts of last time-pos advance + self._last_seen_pos = -1.0 # last observed time-pos value + self._failure_ts: deque[float] = deque(maxlen=64) # recent failure times + self._parked = False # crash-loop parked → stop retrying + # Budgeted mpv opts, set by the controller once the grid size is known + # (memory-aware demuxer cache). None → fall back to unbudgeted defaults. + self._mpv_opts: dict[str, Any] | None = None + # Stats self._stats_current: dict[str, float] = {} self._stats_total: dict[str, float] = {} @@ -202,6 +217,14 @@ def __init__(self, controller: Any): self.setMouseTracking(True) self._sig_eof.connect(self._handle_eof, Qt.ConnectionType.QueuedConnection) + # Stall watchdog: polls for silent freezes (frozen frame / wedged + # decoder / network hang that survives reconnect) which never emit an + # end-file or error event, so the retry chain would otherwise never fire. + self._watchdog_timer = QTimer(self) + self._watchdog_timer.setInterval(WATCHDOG_INTERVAL_MS) + self._watchdog_timer.timeout.connect(self._check_stall) + self._watchdog_timer.start() + # ── mpv lifecycle ───────────────────────────────────────────────────── def _ensure_mpv(self) -> None: @@ -229,10 +252,11 @@ def _ensure_mpv(self) -> None: _devnull = open(os.devnull, "w") try: sys.stdout = sys.stderr = _devnull + _opts = self._mpv_opts or apply_env_overrides(MPV_OPTS) m = _mpv.MPV( wid=str(wid), log_handler=self._mpv_log, - **apply_env_overrides(MPV_OPTS), + **_opts, ) finally: sys.stdout, sys.stderr = _std_saved @@ -265,6 +289,12 @@ def _on_end_file(ev: Any) -> None: def _on_time(_name: str, value: float | None) -> None: if value is None or gen != self._mpv_gen: return + # Record forward progress for the stall watchdog. mpv can emit the + # same time-pos repeatedly on a frozen stream; only a real advance + # counts as "alive". + if value > self._last_seen_pos: + self._last_seen_pos = value + self._last_progress_ts = _time.monotonic() self._play_pos = value if value > 0.02 and not self._played_anything: self._played_anything = True @@ -399,6 +429,10 @@ def play(self, item: dict[str, Any], url: str) -> None: self.current_item = item self._duration_s = 0.0 self._play_pos = 0.0 + # Reset stall tracking so the freshly-loaded track gets a full grace + # window before the watchdog can flag it. + self._last_seen_pos = -1.0 + self._last_progress_ts = _time.monotonic() title = item.get("Name", "Unknown") self.lbl_title.setText(title) @@ -459,6 +493,10 @@ def play(self, item: dict[str, Any], url: str) -> None: def release(self) -> None: """Clean up and release all resources.""" + try: + self._watchdog_timer.stop() + except Exception: + pass self._destroy_mpv() # ── controls UI ─────────────────────────────────────────────────────── @@ -746,6 +784,11 @@ def _handle_eof(self, gen: int, reason: str) -> None: if gen != self._mpv_gen: return if reason == "error": + # Clear the switching guard here too: it's set in play() and was + # previously only cleared on the first "eof". On the error path it + # would leak True, letting a later genuine EOF be swallowed as if it + # were a stale loadfile end-file. + self._switching = False self._on_error() return if reason == "eof": @@ -782,7 +825,76 @@ def _request_next_throttled(self, is_retry: bool) -> None: self._last_next_request_ts = now self.request_next.emit(self, is_retry) + def _check_stall(self) -> None: + """Watchdog: flag a silently frozen stream as an error. + + Runs every WATCHDOG_INTERVAL_MS. A cell that is playing (mpv alive, not + paused, not being seeked) but whose time-pos hasn't advanced for + STALL_TIMEOUT_S is treated as a playback error, reusing the existing + retry/escalation chain — no new failure semantics. + """ + if self._mpv is None or self._parked or self._switching: + return + if not self._played_anything: + # Startup / EOF-before-first-frame is handled by the normal path. + return + idle_s = _time.monotonic() - self._last_progress_ts + if is_stalled( + idle_s, + paused=self._paused, + dragging=self._dragging, + threshold_s=STALL_TIMEOUT_S, + ): + logger.warning( + "Stall detected — no progress for %.0fs (threshold %ds). " + "Treating as error.", idle_s, STALL_TIMEOUT_S, + ) + # Force a fresh grace window so we don't re-fire before the retry + # has a chance to load new frames. + self._last_progress_ts = _time.monotonic() + self._on_error() + + def _record_failure_and_maybe_park(self) -> bool: + """Record a failure timestamp; park the cell on a crash-loop. + + Returns True if the cell is now parked (caller should stop retrying). + """ + now = _time.monotonic() + self._failure_ts.append(now) + if should_park( + self._failure_ts, now, + window_s=CRASH_LOOP_WINDOW_S, + threshold=CRASH_LOOP_THRESHOLD, + ): + self._parked = True + logger.error( + "Crash-loop guard: %d failures within %ds — parking cell. " + "Retrying in %ds.", + CRASH_LOOP_THRESHOLD, CRASH_LOOP_WINDOW_S, CRASH_LOOP_COOLDOWN_S, + ) + self._show_title_overlay("Media unavailable — retrying soon…") + QTimer.singleShot(CRASH_LOOP_COOLDOWN_S * 1000, self._unpark) + return True + return False + + def _unpark(self) -> None: + """Leave the parked state after the cooldown and try to resume.""" + if not self._parked: + return + self._parked = False + self._failure_ts.clear() + self._retry_count = 0 + self._force_transcode = False + logger.info("Crash-loop cooldown elapsed — resuming cell.") + self._request_next_throttled(False) + def _on_error(self) -> None: + if self._parked: + return + # Crash-loop guard: if failures pile up in a short window, stop + # hammering Emby and park the cell instead. + if self._record_failure_and_maybe_park(): + return self._retry_count += 1 logger.warning( "Playback error (attempt %d/%d)", self._retry_count, MAX_RETRIES diff --git a/hyperwall/constants.py b/hyperwall/constants.py index d16cf13..9aa1426 100644 --- a/hyperwall/constants.py +++ b/hyperwall/constants.py @@ -49,6 +49,34 @@ def _ui_scale() -> float: OVERLAY_SHOW_MS = 3_000 # title overlay before fade MOUSE_IDLE_MS = 3_000 # cursor auto-hide +# ── Reliability / self-healing (Epic 2) ────────────────────────────────────── +# Stall watchdog: if time-pos hasn't advanced for STALL_TIMEOUT_S while a cell +# is actively playing (not paused/seeking), treat it as a silent freeze and run +# the normal error/escalation chain. WATCHDOG_INTERVAL_MS is the poll cadence. + + +def _int_env(name: str, default: int, lo: int, hi: int) -> int: + try: + return max(lo, min(hi, int(os.environ.get(name, default)))) + except (TypeError, ValueError): + return default + + +STALL_TIMEOUT_S = _int_env("HYPERWALL_STALL_TIMEOUT_S", 20, 3, 600) +WATCHDOG_INTERVAL_MS = _int_env("HYPERWALL_WATCHDOG_MS", 5_000, 1_000, 60_000) + +# Crash-loop guard: if a cell records CRASH_LOOP_THRESHOLD failures within +# CRASH_LOOP_WINDOW_S, park it on a "media unavailable" card instead of +# hammering Emby. It re-attempts after CRASH_LOOP_COOLDOWN_S. +CRASH_LOOP_THRESHOLD = _int_env("HYPERWALL_CRASHLOOP_THRESHOLD", 5, 2, 100) +CRASH_LOOP_WINDOW_S = _int_env("HYPERWALL_CRASHLOOP_WINDOW_S", 60, 10, 3_600) +CRASH_LOOP_COOLDOWN_S = _int_env("HYPERWALL_CRASHLOOP_COOLDOWN_S", 120, 10, 7_200) + +# Memory-aware demuxer cache budget. Each cell wants PER_CELL demuxer bytes, but +# the grid total is capped at CACHE_BUDGET_MB so large grids don't blow up RAM. +DEMUXER_PER_CELL_MB = _int_env("HYPERWALL_DEMUXER_PER_CELL_MB", 512, 32, 2_048) +CACHE_BUDGET_MB = _int_env("HYPERWALL_CACHE_BUDGET_MB", 3_072, 128, 65_536) + # ── MPV Options ────────────────────────────────────────────────────────────── MPV_OPTS: dict[str, object] = dict( vo="gpu-next", @@ -125,3 +153,21 @@ def apply_env_overrides(opts: dict) -> dict: except ValueError: pass return out + + +def apply_cache_budget(opts: dict, n_cells: int) -> dict: + """Return a copy of opts with demuxer_max_bytes scaled to the cell count. + + Keeps the aggregate grid demuxer buffer within CACHE_BUDGET_MB so large + grids (e.g. 6x6) don't exhaust RAM. Pure w.r.t. the reliability helper. + """ + from .reliability import scale_demuxer_mb + + out = dict(opts) + mb = scale_demuxer_mb( + n_cells, + per_cell_mb=DEMUXER_PER_CELL_MB, + total_budget_mb=CACHE_BUDGET_MB, + ) + out["demuxer_max_bytes"] = f"{mb}MiB" + return out diff --git a/hyperwall/reliability.py b/hyperwall/reliability.py new file mode 100644 index 0000000..780307b --- /dev/null +++ b/hyperwall/reliability.py @@ -0,0 +1,73 @@ +""" +Hyperwall — pure reliability helpers (no PyQt / mpv / Emby imports). + +The 24/7-wall self-healing logic lives here as small, side-effect-free +functions so it can be unit-tested without a display server, an mpv build, or +a live Emby instance. `cell.py` imports these; the wiring (QTimers, mpv calls, +overlays) stays in the widget while the *decisions* are verifiable in isolation. + +Three concerns (Epic 2 / #7): + - stall detection → is_stalled() + - crash-loop parking → count_recent() / should_park() + - cache-budget scaling → scale_demuxer_mb() +""" + +from __future__ import annotations + +from typing import Iterable + + +def is_stalled( + idle_s: float, + *, + paused: bool, + dragging: bool, + threshold_s: float, +) -> bool: + """Decide whether a cell has silently stalled (frozen mid-stream). + + A stall is "no forward time-pos progress for longer than threshold_s while + the cell is actively trying to play." Paused/seeking cells never count — + their lack of progress is intentional, not a fault. + """ + if paused or dragging: + return False + return idle_s > threshold_s + + +def count_recent(times: Iterable[float], now: float, window_s: float) -> int: + """Count timestamps within the trailing window ending at `now`.""" + return sum(1 for t in times if 0 <= now - t <= window_s) + + +def should_park( + times: Iterable[float], + now: float, + *, + window_s: float, + threshold: int, +) -> bool: + """True when failures within the rolling window reach the park threshold. + + Distinguishes a single bad item (skip and move on) from a systemic outage + (e.g. Emby down) where a cell would otherwise hammer the server forever. + """ + return count_recent(times, now, window_s) >= threshold + + +def scale_demuxer_mb( + n_cells: int, + *, + per_cell_mb: int, + total_budget_mb: int, + floor_mb: int = 32, +) -> int: + """Per-cell demuxer budget (MiB), scaled so the grid total stays bounded. + + Each cell wants `per_cell_mb`, but N cells must not exceed + `total_budget_mb` in aggregate (a 6x6 grid at 512 MiB/cell would reach + ~18 GB). Returns min(per_cell_mb, budget/N) clamped to `floor_mb`. + """ + n = max(1, int(n_cells)) + per = min(per_cell_mb, total_budget_mb / n) + return int(max(floor_mb, per)) diff --git a/hyperwall/wall.py b/hyperwall/wall.py index 24043ce..37880e7 100644 --- a/hyperwall/wall.py +++ b/hyperwall/wall.py @@ -40,6 +40,7 @@ STATS_ENABLED, STATS_COUNTER_PROPS, STATS_INFO_PROPS, + apply_cache_budget, apply_env_overrides, MPV_OPTS, SCRIPT_DIR, @@ -132,6 +133,18 @@ def __init__( QApplication.instance().installEventFilter(self._escape_filter) self._build_displays() + + # Memory-aware demuxer budget: now that every cell exists, scale the + # per-cell demuxer cache so the grid total stays under CACHE_BUDGET_MB. + n_cells = len(self.cells) + budgeted = apply_cache_budget(apply_env_overrides(MPV_OPTS), n_cells) + for cell in self.cells: + cell._mpv_opts = budgeted + logger.info( + "MPV cache budget: %d cells → demuxer_max_bytes=%s each", + n_cells, budgeted.get("demuxer_max_bytes"), + ) + for win in self.windows: win.showFullScreen() logger.info("Display active: %s", win.windowTitle()) diff --git a/tests/test_reliability.py b/tests/test_reliability.py new file mode 100644 index 0000000..5a3e139 --- /dev/null +++ b/tests/test_reliability.py @@ -0,0 +1,139 @@ +""" +Unit tests for hyperwall.reliability (Epic 2) and the constants that drive it. + +Pure logic only — no PyQt / mpv / Emby. Runnable anywhere Python is. +Run: python tests/test_reliability.py +""" + +from __future__ import annotations + +import os +import sys + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, REPO_ROOT) + +from hyperwall.reliability import ( # noqa: E402 + count_recent, + is_stalled, + scale_demuxer_mb, + should_park, +) + + +# ── is_stalled ──────────────────────────────────────────────────────────────── + +def test_stall_fires_after_threshold(): + assert is_stalled(25.0, paused=False, dragging=False, threshold_s=20) + + +def test_no_stall_within_threshold(): + assert not is_stalled(5.0, paused=False, dragging=False, threshold_s=20) + + +def test_paused_never_stalls(): + # A paused cell isn't making progress on purpose — must not be flagged. + assert not is_stalled(999.0, paused=True, dragging=False, threshold_s=20) + + +def test_seeking_never_stalls(): + assert not is_stalled(999.0, paused=False, dragging=True, threshold_s=20) + + +def test_stall_boundary_is_strict(): + # Exactly at threshold is NOT stalled (must exceed). + assert not is_stalled(20.0, paused=False, dragging=False, threshold_s=20) + assert is_stalled(20.001, paused=False, dragging=False, threshold_s=20) + + +# ── count_recent / should_park ──────────────────────────────────────────────── + +def test_count_recent_window(): + times = [0.0, 10.0, 55.0, 58.0, 59.5] + # now=60, window=60 → 0.0 is 60s ago (inclusive), all 5 count + assert count_recent(times, now=60.0, window_s=60.0) == 5 + # window=10 → only 55,58,59.5 within last 10s + assert count_recent(times, now=60.0, window_s=10.0) == 3 + + +def test_should_park_triggers_on_threshold(): + times = [1.0, 2.0, 3.0, 4.0, 5.0] + assert should_park(times, now=6.0, window_s=60, threshold=5) + + +def test_should_not_park_below_threshold(): + times = [1.0, 2.0, 3.0] + assert not should_park(times, now=6.0, window_s=60, threshold=5) + + +def test_old_failures_age_out_of_window(): + # Four ancient failures + one recent → not a loop. + times = [1.0, 2.0, 3.0, 4.0, 500.0] + assert not should_park(times, now=505.0, window_s=60, threshold=5) + + +# ── scale_demuxer_mb ────────────────────────────────────────────────────────── + +def test_single_cell_gets_full_per_cell(): + assert scale_demuxer_mb(1, per_cell_mb=512, total_budget_mb=3072) == 512 + + +def test_small_grid_under_budget_unscaled(): + # 4 cells * 512 = 2048 <= 3072 → no scaling. + assert scale_demuxer_mb(4, per_cell_mb=512, total_budget_mb=3072) == 512 + + +def test_large_grid_scaled_down(): + # 36 cells, 3072 budget → 3072/36 = 85 MiB each (well under 512). + got = scale_demuxer_mb(36, per_cell_mb=512, total_budget_mb=3072) + assert got == 85, got + assert got * 36 <= 3072 + + +def test_floor_respected(): + # Absurd cell count clamps to the floor, never zero. + got = scale_demuxer_mb(10_000, per_cell_mb=512, total_budget_mb=3072, floor_mb=32) + assert got == 32 + + +def test_zero_cells_is_safe(): + # Defensive: n<1 must not divide-by-zero. + assert scale_demuxer_mb(0, per_cell_mb=512, total_budget_mb=3072) == 512 + + +# ── constants env clamping (integration of _int_env) ────────────────────────── + +def test_constants_defaults_load(): + from hyperwall import constants as c + assert c.STALL_TIMEOUT_S == 20 + assert c.WATCHDOG_INTERVAL_MS == 5_000 + assert c.CRASH_LOOP_THRESHOLD == 5 + assert c.DEMUXER_PER_CELL_MB == 512 + assert c.CACHE_BUDGET_MB == 3_072 + + +def test_apply_cache_budget_shape(): + from hyperwall.constants import apply_cache_budget + out = apply_cache_budget({"demuxer_max_bytes": "512MiB", "vo": "gpu-next"}, 36) + assert out["vo"] == "gpu-next" # untouched keys preserved + assert out["demuxer_max_bytes"].endswith("MiB") + assert int(out["demuxer_max_bytes"][:-3]) < 512 # scaled down for 36 cells + + +def run_all() -> int: + tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + passed = failed = 0 + for t in tests: + try: + t() + passed += 1 + print(f" PASS {t.__name__}") + except Exception as e: # noqa: BLE001 + failed += 1 + print(f" FAIL {t.__name__}: {e}") + print(f"\n{passed} passed, {failed} failed out of {len(tests)} tests.") + return failed + + +if __name__ == "__main__": + sys.exit(run_all())