From 5a2507efa53886ef685af3d23fa593900d5c8b3b Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:54:15 +0100 Subject: [PATCH 01/12] fix(storage): close HTTP responses through one logged helper Fifteen `try: response.close() except Exception: pass` blocks in the storage HTTP layer and the Google Drive, Dropbox, OneDrive, WebDAV and S3 providers become http.close_quietly(). It catches only OSError (socket already gone, requests' ConnectionError) and urllib3's HTTPError, logs them at debug level with the provider name, and lets anything else propagate. The transfer result is unchanged: closing only returns the connection to the pool. Co-Authored-By: Claude Opus 5.5 --- core/storage/http.py | 25 +++++++-- core/storage/providers/dropbox.py | 15 ++---- core/storage/providers/google_drive.py | 25 ++------- core/storage/providers/onedrive.py | 15 ++---- core/storage/providers/s3.py | 6 +-- core/storage/providers/webdav.py | 10 +--- tests/test_error_handling.py | 70 ++++++++++++++++++++++++++ 7 files changed, 106 insertions(+), 60 deletions(-) create mode 100644 tests/test_error_handling.py diff --git a/core/storage/http.py b/core/storage/http.py index c1a4e7d..f9b21a7 100644 --- a/core/storage/http.py +++ b/core/storage/http.py @@ -11,6 +11,7 @@ from typing import Any, Optional import requests +from urllib3.exceptions import HTTPError as Urllib3HTTPError from .base import StorageError @@ -50,6 +51,25 @@ def session() -> requests.Session: return current +def close_quietly(resource, *, provider: str = "") -> None: + """Release a response or stream body whose result is no longer needed. + + Closing only hands the connection back to the pool; the transfer result + is already decided by then. A socket that is already gone (OSError, + including requests' ConnectionError) or a urllib3 protocol error while + draining must therefore not replace the caller's result or exception. + Anything else is a programming error and propagates. + """ + try: + resource.close() + except (OSError, Urllib3HTTPError) as exc: + logger.debug( + "%s: closing HTTP response failed: %s", + provider or "storage", + sanitize_error_text(exc), + ) + + def _retry_after_seconds(response: requests.Response, attempt: int) -> float: raw = (response.headers.get("Retry-After") or "").strip() if raw: @@ -127,10 +147,7 @@ def request( if response.status_code not in statuses or attempt + 1 >= retries: return response delay = _retry_after_seconds(response, attempt) - try: - response.close() - except Exception: - pass + close_quietly(response, provider=provider) time.sleep(delay) raise StorageError( diff --git a/core/storage/providers/dropbox.py b/core/storage/providers/dropbox.py index a3adca0..2054063 100644 --- a/core/storage/providers/dropbox.py +++ b/core/storage/providers/dropbox.py @@ -152,10 +152,7 @@ def _rpc(self, endpoint: str, payload: dict, action: str) -> dict: timeout=(10, 60), ) if response.status_code == 401 and attempt == 0: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) self._token(force=True) continue return http.json_response(response, action, provider=self.type_id) @@ -369,10 +366,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -388,10 +382,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: try: diff --git a/core/storage/providers/google_drive.py b/core/storage/providers/google_drive.py index 18083e0..586dad4 100644 --- a/core/storage/providers/google_drive.py +++ b/core/storage/providers/google_drive.py @@ -186,10 +186,7 @@ def _api(self, method: str, path: str, **kwargs): kwargs["headers"] = self._headers(**(kwargs.get("headers") or {})) response = http.request(method, url, provider=self.type_id, **kwargs) if response.status_code == 401 and attempt == 0: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) self._token(force=True) continue return response @@ -345,10 +342,7 @@ def _upload_empty(self, name: str, parent_id: str, existing_id: str, mime: str) ) if response.status_code != 401 or attempt == 1: break - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) data = http.json_response(response, "Google-Upload (leer)", provider=self.type_id) return { "remote_id": str(data.get("id") or existing_id or ""), @@ -410,10 +404,7 @@ def upload_file( ) if response.status_code != 401 or attempt == 1: break - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) http.raise_for_status(response, "Google-Upload starten", provider=self.type_id) location = str(response.headers.get("Location") or "").strip() if not location: @@ -540,10 +531,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -559,10 +547,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: remote_path = normalize_remote_path(remote_path) diff --git a/core/storage/providers/onedrive.py b/core/storage/providers/onedrive.py index fda13da..4be197f 100644 --- a/core/storage/providers/onedrive.py +++ b/core/storage/providers/onedrive.py @@ -153,10 +153,7 @@ def _api(self, method: str, url: str, **kwargs): kwargs["headers"] = self._headers(**(kwargs.get("headers") or {})) response = http.request(method, url, provider=self.type_id, **kwargs) if response.status_code == 401 and attempt == 0: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) self._token(force=True) continue return response @@ -352,10 +349,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -371,10 +365,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: response = self._api("DELETE", self._item_url(remote_path), timeout=(10, 60)) diff --git a/core/storage/providers/s3.py b/core/storage/providers/s3.py index e95dbc7..2752c48 100644 --- a/core/storage/providers/s3.py +++ b/core/storage/providers/s3.py @@ -8,6 +8,7 @@ from flask_babel import gettext, lazy_gettext +from core.storage import http from core.storage.base import ( ProgressCallback, ProviderField, @@ -268,10 +269,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: break yield chunk finally: - try: - body.close() - except Exception: - pass + http.close_quietly(body, provider=self.type_id) def delete(self, remote_path: str) -> None: key = self._key(remote_path) diff --git a/core/storage/providers/webdav.py b/core/storage/providers/webdav.py index e118012..feee9dc 100644 --- a/core/storage/providers/webdav.py +++ b/core/storage/providers/webdav.py @@ -185,10 +185,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -209,10 +206,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - try: - response.close() - except Exception: - pass + http.close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: response = self._request("DELETE", self._url(remote_path), timeout=(10, 60), retries=2) diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py new file mode 100644 index 0000000..518cd3c --- /dev/null +++ b/tests/test_error_handling.py @@ -0,0 +1,70 @@ +"""Former `except Exception: pass` paths: specific error classes, logged. + +Each test pins two things: the expected failure is still absorbed (the caller +keeps its fallback) and it now leaves a log line behind. +The environment bootstrap (test credentials, temp BASE_DIR, fcntl shim on +Windows) lives in test_recorder_reliability and runs on import. +""" + +from __future__ import annotations + +import unittest +from unittest import mock + +try: + from test_recorder_reliability import IsolatedBaseTest # noqa: F401 +except ImportError: # run as tests.test_error_handling + from tests.test_recorder_reliability import IsolatedBaseTest # noqa: F401 + +from core.storage import http as storage_http # noqa: E402 +from core.storage.providers import google_drive as google_module # noqa: E402 +from core.storage.providers.google_drive import GoogleDriveProvider # noqa: E402 + + +class StorageResponseCloseTests(unittest.TestCase): + def test_close_failure_is_logged_with_provider(self): + response = mock.Mock() + response.close.side_effect = ConnectionResetError("reset by peer") + with self.assertLogs("storage.http", level="DEBUG") as logs: + storage_http.close_quietly(response, provider="dropbox") + self.assertTrue(any("dropbox" in line and "reset by peer" in line for line in logs.output)) + + def test_unexpected_close_error_is_not_swallowed(self): + response = mock.Mock() + response.close.side_effect = RuntimeError("programming error") + with self.assertRaises(RuntimeError): + storage_http.close_quietly(response, provider="dropbox") + + def test_google_drive_401_retry_survives_a_failing_close(self): + class Unauthorized: + status_code = 401 + reason = "Unauthorized" + text = "denied" + content = b"" + headers = {} + + def json(self): + raise ValueError("not json") + + def close(self): + raise ConnectionResetError("socket gone") + + provider = GoogleDriveProvider({}) + token_calls = [] + provider._token = lambda force=False: token_calls.append(bool(force)) or "token" + requests = [] + + def fake_request(method, url, **kwargs): + requests.append((method, url)) + return Unauthorized() + + with mock.patch.object(google_module.http, "request", fake_request), \ + self.assertLogs("storage.http", level="DEBUG"): + response = provider._api("GET", "/files") + self.assertEqual(response.status_code, 401) + self.assertEqual(len(requests), 2) + self.assertIn(True, token_calls, "the 401 must still force a token refresh") + + +if __name__ == "__main__": + unittest.main() From 69ce1cb4be43fe7a68506d9ed5b6067d22f866b6 Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:54:15 +0100 Subject: [PATCH 02/12] fix(storage): log failed token persistence and progress bookkeeping - base: a failing config-changed callback (rotated OAuth token not persisted) is caught as StorageError/sqlite3.Error/OSError/ValueError and logged as a warning; the request still succeeds as before. - service: a locked database while saving transfer progress is logged at debug level; a failure to record the provider error after a failed restore is logged as a warning and the restore error is still raised. Co-Authored-By: Claude Opus 5.5 --- core/storage/base.py | 16 +++++++++++++--- core/storage/service.py | 17 +++++++++++++---- tests/test_storage.py | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 7 deletions(-) diff --git a/core/storage/base.py b/core/storage/base.py index 23ca66f..03f14c6 100644 --- a/core/storage/base.py +++ b/core/storage/base.py @@ -8,15 +8,19 @@ from __future__ import annotations +import logging import mimetypes import os import posixpath import re +import sqlite3 from dataclasses import dataclass from typing import Callable, Iterable, Iterator, Optional from flask_babel import gettext, lazy_gettext +logger = logging.getLogger("storage.base") + class StorageError(RuntimeError): """A provider/configuration error with optional retry semantics.""" @@ -210,10 +214,16 @@ def _config_changed(self) -> None: return try: callback(dict(self.config)) - except Exception: + except (StorageError, sqlite3.Error, OSError, ValueError) as exc: # Token rotation must never make a request fail just because the - # optional persistence callback is unavailable. - pass + # optional persistence callback is unavailable. The rotated token + # stays valid in memory; after a restart the stored one is used, + # which a provider with single-use refresh tokens may reject. + logger.warning( + "%s: rotated credentials not persisted: %s", + self.type_id or "storage", + mask_error(exc), + ) # ------------------------------------------------------------------ # Configuration helpers diff --git a/core/storage/service.py b/core/storage/service.py index 1207299..e869d24 100644 --- a/core/storage/service.py +++ b/core/storage/service.py @@ -11,6 +11,7 @@ import logging import os import shutil +import sqlite3 import stat import time from typing import Optional @@ -1132,8 +1133,10 @@ def callback(done: int, size: int): state["last_pct"] = pct try: database.update_storage_object(object_id, progress=round(pct, 1)) - except Exception: - pass + except sqlite3.Error as exc: + # Progress is cosmetic and rewritten every two seconds; a busy + # database must not abort the transfer itself. + logger.debug("storage object %s: progress not saved: %s", object_id, exc) return callback @@ -1396,8 +1399,14 @@ def run_restore(payload: dict) -> dict: database.update_storage_object(object_id, last_error=message) try: database.set_storage_provider_error(obj["provider_id"], message) - except Exception: - pass + except sqlite3.Error as db_exc: + # Keep the restore error for the caller; the provider status line + # is only a convenience for the storage page. + logger.warning( + "provider %s: restore error not recorded: %s", + obj["provider_id"], + db_exc, + ) raise diff --git a/tests/test_storage.py b/tests/test_storage.py index a1298d7..b84442a 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -15,6 +15,7 @@ import io import json import os +import sqlite3 import stat import tempfile import time @@ -2082,5 +2083,36 @@ def test_storage_admin_error_log_omits_traceback_and_secrets(self): self.assertNotIn("supersecret", str(log_error.call_args)) +class StorageBookkeepingFailureTests(StorageTestBase): + """Database hiccups in side bookkeeping are logged, never fatal.""" + + def test_progress_update_on_locked_database_is_logged(self): + callback = service._progress_callback(1, 100) + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(service.database, "update_storage_object", side_effect=locked), self.assertLogs("storage.service", level="DEBUG") as logs: + callback(50, 100) + self.assertTrue(any("database is locked" in line for line in logs.output)) + + def test_restore_error_survives_failing_provider_error_bookkeeping(self): + path, obj = RestoreSafetyTests._restore_object(self, b"new", remote_size=3, local=b"older") + fake = RestoreSafetyTests._FakeDownloadProvider(error=StorageError("remote gone")) + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(service, "build_provider", return_value=fake), mock.patch.object(service.database, "set_storage_provider_error", side_effect=locked), self.assertLogs("storage.service", level="WARNING") as logs: + with self.assertRaises(StorageError) as ctx: + service.run_restore({"object_id": obj["id"]}) + self.assertIn("remote gone", str(ctx.exception)) + self.assertTrue(any("database is locked" in line for line in logs.output)) + self.assertEqual(path.read_bytes(), b"older") + + def test_failing_token_persistence_does_not_fail_the_request(self): + provider = DropboxProvider({}) + provider.set_config_changed_callback( + mock.Mock(side_effect=sqlite3.OperationalError("database is locked")) + ) + with self.assertLogs("storage.base", level="WARNING") as logs: + provider._config_changed() + self.assertTrue(any("dropbox" in line for line in logs.output)) + + if __name__ == "__main__": unittest.main() From 43c27d6c39821ab957af4c2ed23bb76824baced7 Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:56:33 +0100 Subject: [PATCH 03/12] fix(core): log swallowed errors in Twitch, diagnostics, chat and access helpers Ten `except Exception: pass` handlers outside storage and the web UI now catch the specific failure and leave a log line; every fallback value is unchanged: - twitch_api/twitch_eventsub: an unreadable HTTP error body (OSError, http.client.HTTPException) is logged at debug level; the status code still decides the result. reload_env() calls redis_client.delete_state() directly because it already absorbs Redis errors. - live_watch_diag: sqlite3.Error/ValueError reading channel configs is a warning; the diagnosis continues without the database fields. - chat_chunker: an unreadable cached meta.json is a warning and the chunks are rebuilt. - redis_client: failing to remove the dashboard shm cache is a warning. - tube_auth: sqlite3.Error reading VOD access rows is a warning and the seed list is used, as before. Co-Authored-By: Claude Opus 5.5 --- core/chat_chunker.py | 4 +- core/live_watch_diag.py | 12 ++++-- core/redis_client.py | 4 +- core/tube_auth.py | 11 +++-- core/twitch_api.py | 28 +++++++------ core/twitch_eventsub.py | 15 +++---- tests/test_error_handling.py | 80 ++++++++++++++++++++++++++++++++++++ 7 files changed, 124 insertions(+), 30 deletions(-) diff --git a/core/chat_chunker.py b/core/chat_chunker.py index 7ad5bc9..7ff7788 100644 --- a/core/chat_chunker.py +++ b/core/chat_chunker.py @@ -40,8 +40,8 @@ def ensure_chat_chunks( if os.path.getmtime(meta_path) >= os.path.getmtime(log_path): with open(meta_path, "r", encoding="utf-8") as f: return json.load(f) - except Exception: - pass + except (OSError, ValueError) as exc: + logger.warning("chat chunk cache %s unreadable, rebuilding: %s", meta_path, exc) if not os.path.isfile(log_path) or os.path.getsize(log_path) == 0: return {"ok": False, "error": "Log file not found or empty", "total_chunks": 0} diff --git a/core/live_watch_diag.py b/core/live_watch_diag.py index 79bfbb8..de25d68 100644 --- a/core/live_watch_diag.py +++ b/core/live_watch_diag.py @@ -6,8 +6,10 @@ from __future__ import annotations +import logging import os import re +import sqlite3 import subprocess import time from pathlib import Path @@ -15,6 +17,8 @@ from core import config +logger = logging.getLogger(__name__) + LIVE_HLS_TTL_SEC = int(os.environ.get("LIVE_HLS_TTL_SEC", "24")) LIVE_WATCH_DEMAND_SEC = int(os.environ.get("LIVE_WATCH_DEMAND_SEC", "120")) LIVE_STALL_RESTART_SEC = int(os.environ.get("LIVE_STALL_RESTART_SEC", "12")) @@ -84,8 +88,8 @@ def list_candidate_channels(extra: list[str] | None = None) -> list[str]: found.add(name) if int(conf.get("live") or 0) == 1 and name: found.add(name) - except Exception: - pass + except (sqlite3.Error, ValueError) as exc: + logger.warning("live diag: channel configs unavailable: %s", exc) for name in extra or []: if name: found.add(str(name).strip().lower()) @@ -173,8 +177,8 @@ def diagnose_channel(channel: str, *, logs: int = 30, helix: dict | None = None) if conf: db_live = int(conf.get("live") or 0) db_record = int(conf.get("record") or 0) - except Exception: - pass + except (sqlite3.Error, ValueError) as exc: + logger.warning("live diag %s: channel config unavailable: %s", channel, exc) helix_info = (helix or {}).get(channel) if helix is not None else None if helix_info is None: diff --git a/core/redis_client.py b/core/redis_client.py index 28ea239..2aec383 100644 --- a/core/redis_client.py +++ b/core/redis_client.py @@ -143,8 +143,8 @@ def invalidate_dashboard_cache() -> bool: shm_path = "/dev/shm/streamrecorder_dashboard_media.json" if os.path.exists(shm_path): os.remove(shm_path) - except Exception: - pass + except OSError as exc: + logger.warning("dashboard shm cache %s not removed: %s", shm_path, exc) return True diff --git a/core/tube_auth.py b/core/tube_auth.py index eec5512..f101dc6 100644 --- a/core/tube_auth.py +++ b/core/tube_auth.py @@ -8,9 +8,11 @@ from __future__ import annotations +import logging import os import re import secrets +import sqlite3 import string import time from typing import Any @@ -20,6 +22,8 @@ from core import database +logger = logging.getLogger(__name__) + VALID_ROLES = {"admin", "viewer", "user", "member", "live_only"} MEMBER_ROLE = "member" LIVE_ONLY_ROLE = "live_only" @@ -93,7 +97,8 @@ def _expand_access_slugs(raw_slugs) -> set[str]: def list_default_vod_access_slugs() -> list[str]: try: rows = database.list_role_vod_access(MEMBER_ROLE) - except Exception: + except sqlite3.Error as exc: + logger.warning("VOD access defaults unreadable, using seed list: %s", exc) rows = [] if rows: return list(rows) @@ -110,8 +115,8 @@ def effective_vod_access_bases(role: str | None = None) -> list[str]: rows = database.list_role_vod_access(role_key) if rows: return rows - except Exception: - pass + except sqlite3.Error as exc: + logger.warning("VOD access for role %s unreadable, using defaults: %s", role_key, exc) return list_default_vod_access_slugs() diff --git a/core/twitch_api.py b/core/twitch_api.py index 554a443..d640c26 100644 --- a/core/twitch_api.py +++ b/core/twitch_api.py @@ -8,6 +8,7 @@ from __future__ import annotations +import http.client import json import logging import os @@ -236,6 +237,18 @@ def _wait_if_circuit_open(): time.sleep(delay) +def _drain_http_error(exc: urllib.error.HTTPError, path: str) -> None: + """Read and discard an error response body so the socket can close. + + The status code has already decided the outcome; a body that cannot be + read (connection reset, truncated chunk) changes nothing about it. + """ + try: + exc.read() + except (OSError, http.client.HTTPException) as read_exc: + logger.debug("Twitch Helix %s: error body unreadable: %s", path, read_exc) + + def helix_get(path: str, params: list[tuple[str, str]] | dict[str, Any] | None = None) -> dict | None: token = get_app_access_token() if not token or not TWITCH_CLIENT_ID: @@ -280,10 +293,7 @@ def helix_get(path: str, params: list[tuple[str, str]] | dict[str, Any] | None = attempt + 1, max_attempts, ) - try: - exc.read() - except Exception: - pass + _drain_http_error(exc, path) if attempt + 1 < max_attempts: time.sleep(sleep_for + random.uniform(0.0, 0.5)) continue @@ -298,19 +308,13 @@ def helix_get(path: str, params: list[tuple[str, str]] | dict[str, Any] | None = attempt + 1, max_attempts, ) - try: - exc.read() - except Exception: - pass + _drain_http_error(exc, path) if attempt + 1 < max_attempts: time.sleep(sleep_for) continue return None logger.info("Twitch Helix request failed (%s): HTTP %s", path, code) - try: - exc.read() - except Exception: - pass + _drain_http_error(exc, path) return None except Exception as exc: logger.info("Twitch Helix request failed (%s): %s", path, exc) diff --git a/core/twitch_eventsub.py b/core/twitch_eventsub.py index a9a7f00..6515cf3 100644 --- a/core/twitch_eventsub.py +++ b/core/twitch_eventsub.py @@ -8,6 +8,7 @@ import hashlib import hmac +import http.client import json import logging import os @@ -52,11 +53,10 @@ def reload_env() -> None: with _STATE_LOCK: _STATE_CACHE.clear() _STATE_CACHE.update({"by_login": {}, "by_user_id": {}, "updated_at": 0.0}) - try: - from core import redis_client - redis_client.delete_state("sysstate:eventsub_state") - except Exception: - pass + from core import redis_client + + # delete_state() absorbs Redis errors itself and returns False. + redis_client.delete_state("sysstate:eventsub_state") def enabled() -> bool: @@ -224,8 +224,9 @@ def _helix_post(path: str, payload: dict[str, Any]) -> dict | None: detail = "" try: detail = exc.read().decode("utf-8", errors="replace")[:400] - except Exception: - pass + except (OSError, http.client.HTTPException) as read_exc: + # The HTTP status below is the result; the body is only detail. + logger.debug("EventSub Helix POST %s: error body unreadable: %s", path, read_exc) logger.warning("EventSub Helix POST %s failed: HTTP %s %s", path, exc.code, detail) return None except Exception as exc: diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index 518cd3c..dfac662 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -8,7 +8,12 @@ from __future__ import annotations +import io +import json +import os +import sqlite3 import unittest +import urllib.error from unittest import mock try: @@ -16,6 +21,8 @@ except ImportError: # run as tests.test_error_handling from tests.test_recorder_reliability import IsolatedBaseTest # noqa: F401 +from core import chat_chunker, live_watch_diag, redis_client, tube_auth # noqa: E402 +from core import twitch_api, twitch_eventsub # noqa: E402 from core.storage import http as storage_http # noqa: E402 from core.storage.providers import google_drive as google_module # noqa: E402 from core.storage.providers.google_drive import GoogleDriveProvider # noqa: E402 @@ -66,5 +73,78 @@ def fake_request(method, url, **kwargs): self.assertIn(True, token_calls, "the 401 must still force a token refresh") +def _http_error_with_broken_body(code): + exc = urllib.error.HTTPError("https://api.twitch.tv/helix/x", code, "err", {}, io.BytesIO(b"")) + exc.read = mock.Mock(side_effect=ConnectionResetError("body lost")) + return exc + + +class TwitchHttpErrorBodyTests(unittest.TestCase): + def setUp(self): + for name, value in (("get_app_access_token", lambda: "app-token"), + ("TWITCH_CLIENT_ID", "client")): + patcher = mock.patch.object(twitch_api, name, value) + patcher.start() + self.addCleanup(patcher.stop) + + def test_helix_get_logs_unreadable_error_body_and_returns_none(self): + with mock.patch.object(twitch_api.urllib.request, "urlopen", + side_effect=_http_error_with_broken_body(404)), self.assertLogs(twitch_api.logger, level="DEBUG") as logs: + self.assertIsNone(twitch_api.helix_get("streams", {"user_login": "x"})) + self.assertTrue(any("body lost" in line for line in logs.output)) + + def test_eventsub_post_logs_unreadable_error_body_and_returns_none(self): + with mock.patch.object(twitch_eventsub.urllib.request, "urlopen", + side_effect=_http_error_with_broken_body(409)), self.assertLogs(twitch_eventsub.logger, level="DEBUG") as logs: + self.assertIsNone(twitch_eventsub._helix_post("eventsub/subscriptions", {})) + self.assertTrue(any("body lost" in line for line in logs.output)) + + +class CoreDatabaseFallbackTests(IsolatedBaseTest): + def test_vod_access_bases_fall_back_to_defaults_and_log(self): + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(tube_auth.database, "list_role_vod_access", side_effect=locked), self.assertLogs("core.tube_auth", level="WARNING"): + bases = tube_auth.effective_vod_access_bases() + self.assertEqual(bases, sorted(tube_auth._VOD_ACCESS_SEED_BASES)) + + def test_live_diag_candidates_survive_database_error(self): + from core import database + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(database, "list_channel_configs", side_effect=locked), self.assertLogs("core.live_watch_diag", level="WARNING"): + names = live_watch_diag.list_candidate_channels(extra=["Extra"]) + self.assertIn("extra", names) + + def test_live_diag_channel_report_survives_database_error(self): + from core import database + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(database, "get_channel_config", side_effect=locked), mock.patch.object(live_watch_diag, "streamlink_counts", return_value={}), mock.patch.object(live_watch_diag, "ffmpeg_hls_pids", return_value={}), self.assertLogs("core.live_watch_diag", level="WARNING"): + report = live_watch_diag.diagnose_channel("chan", helix={"chan": {}}) + self.assertEqual(report["db"], {"live": None, "record": None}) + + +class ChatChunkCacheTests(IsolatedBaseTest): + def test_corrupt_cached_meta_is_logged_and_rebuilt(self): + log_path = self.base / "chan" / "chat_logs" / "vod.jsonl" + log_path.parent.mkdir(parents=True) + log_path.write_text(json.dumps({"t": 1.0, "msg": "hi"}) + "\n", encoding="utf-8") + chunks_dir = chat_chunker.get_chunks_dir(str(self.base), "chan", "vod.mp4") + os.makedirs(chunks_dir) + meta = os.path.join(chunks_dir, "meta.json") + with open(meta, "w", encoding="utf-8") as handle: + handle.write("{truncated") + os.utime(meta, (os.path.getmtime(log_path) + 60,) * 2) + with self.assertLogs("streamvault.chat_chunker", level="WARNING"): + result = chat_chunker.ensure_chat_chunks(str(self.base), "chan", "vod.mp4", str(log_path)) + self.assertTrue(result.get("ok"), result) + with open(meta, encoding="utf-8") as handle: + self.assertEqual(json.load(handle)["total_chunks"], result["total_chunks"]) + + +class RedisDashboardCacheTests(unittest.TestCase): + def test_shm_cache_removal_failure_is_logged(self): + with mock.patch.object(redis_client, "delete_state", return_value=False), mock.patch.object(redis_client, "publish_event", return_value=False), mock.patch.object(redis_client.os.path, "exists", return_value=True), mock.patch.object(redis_client.os, "remove", side_effect=PermissionError("busy")), self.assertLogs("streamvault.redis", level="WARNING"): + self.assertTrue(redis_client.invalidate_dashboard_cache()) + + if __name__ == "__main__": unittest.main() From 64cad6b2ca737dc642f16d463f76d867662401be Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:02:04 +0100 Subject: [PATCH 04/12] fix(webui): stop hiding the 404 for deleted media behind a broad except abort_deleted_media() called abort(404) inside `try ... except Exception: pass`. werkzeug's NotFound is an Exception, so the handler swallowed it and a tombstoned VOD or clip whose file was still on disk (restore, failed unlink, re-sync) kept opening in the player. The tombstone lookup now catches only sqlite3.Error, logs it and serves as before; abort(404) runs outside the handler. The alias lookup in the same function logs sqlite3.Error instead of passing silently. Co-Authored-By: Claude Opus 5.5 --- WEBUI/webui.py | 16 ++++++++----- tests/test_error_handling.py | 46 ++++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 6 deletions(-) diff --git a/WEBUI/webui.py b/WEBUI/webui.py index ef4c3ff..6b1d98b 100644 --- a/WEBUI/webui.py +++ b/WEBUI/webui.py @@ -4827,18 +4827,22 @@ def abort_deleted_media(streamer, filename, file_id=None): slugs = set() try: slugs.update(related_archive_slugs(streamer) or []) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Tombstone check for %s/%s without aliases (database error): %s", streamer, name, exc) primary = archive_name_for(streamer) or str(streamer or "").strip() if primary: slugs.add(primary) if streamer: slugs.add(str(streamer).strip()) try: - if database.is_deleted_media(filename=name, file_id=file_id, streamers=slugs): - abort(404) - except Exception: - pass + deleted = database.is_deleted_media(filename=name, file_id=file_id, streamers=slugs) + except sqlite3.Error as exc: + # Same outcome as before: without the tombstone table the file + # itself decides (resolve_media_path still 404s a missing file). + logging.warning("Tombstone check for %s/%s failed, serving: %s", streamer, name, exc) + return + if deleted: + abort(404) def visible_streamers(streamers=None): diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index dfac662..9f0eb86 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -12,6 +12,7 @@ import json import os import sqlite3 +import time import unittest import urllib.error from unittest import mock @@ -21,6 +22,7 @@ except ImportError: # run as tests.test_error_handling from tests.test_recorder_reliability import IsolatedBaseTest # noqa: F401 +from core import config # noqa: E402 from core import chat_chunker, live_watch_diag, redis_client, tube_auth # noqa: E402 from core import twitch_api, twitch_eventsub # noqa: E402 from core.storage import http as storage_http # noqa: E402 @@ -146,5 +148,49 @@ def test_shm_cache_removal_failure_is_logged(self): self.assertTrue(redis_client.invalidate_dashboard_cache()) + +def _admin_client(webui): + webui.app.config.update(TESTING=True) + client = webui.app.test_client() + with client.session_transaction() as sess: + now = time.time() + sess.update(logged_in=True, user_role="admin", username=config.VOD_USER, + login_time=now, last_active=now, _csrf_token="t") + return client + + +class DeletedMediaTombstoneTests(IsolatedBaseTest): + """A tombstoned VOD stays hidden even while its file is still on disk.""" + + def setUp(self): + super().setUp() + from core import database + from WEBUI import webui + self.webui = webui + self.database = database + patcher = mock.patch.object(webui, "BASE_DIR", str(self.base)) + patcher.start() + self.addCleanup(patcher.stop) + webui.ensure_admin_user() + archive = self.base / "streamer" + archive.mkdir() + (archive / "vod.mp4").write_bytes(b"\0" * 32) + + def test_tombstoned_vod_page_is_404(self): + client = _admin_client(self.webui) + self.assertEqual(client.get("/vod/streamer/vod.mp4").status_code, 200) + self.database.upsert_deleted_media("streamer", "vod.mp4", deleted_by="admin") + self.assertEqual(client.get("/vod/streamer/vod.mp4").status_code, 404) + + def test_tombstone_lookup_failure_is_logged_and_page_still_served(self): + client = _admin_client(self.webui) + locked = sqlite3.OperationalError("database is locked") + with mock.patch.object(self.webui.database, "is_deleted_media", side_effect=locked), self.assertLogs(level="WARNING") as logs: + response = client.get("/vod/streamer/vod.mp4") + self.assertEqual(response.status_code, 200) + self.assertTrue(any("database is locked" in line for line in logs.output)) + + + if __name__ == "__main__": unittest.main() From eca59f123764acbfb587f5c5e61c42ab167fdfcd Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:02:45 +0100 Subject: [PATCH 05/12] fix(webui): log the remaining swallowed lookup errors The last sixteen pass-only broad handlers in webui.py (plus the adjacent channel-config lookup in twitch_watch) catch the failure they guard against and log a warning; every fallback stays as it was: - channel/alias lookups (can_view_streamer, resolve_media_path, delete_video_file, list_all_streamer_folders, _recording_logins_set, twitch_watch, _vod_access_candidate_slugs): sqlite3.Error, plus OSError where the archive folder or recording state is read. - Twitch GQL profile/user-id lookups: OSError (URLError, timeouts), ValueError (bad JSON) and http.client.HTTPException. Unexpected answer shapes are parsed by _twitch_gql_user() instead of raising AttributeError into the handler. - maintenance bypass check: OSError/ValueError, user stays not exempt. - title from the chat log in read_media_sidecar_metadata: OSError. - can_view_streamer no longer wraps normalize_channel_name(), which is pure string work and cannot fail. AST count of `except Exception/bare: pass` outside tests: 46 -> 0. Co-Authored-By: Claude Opus 5.5 --- WEBUI/webui.py | 97 +++++++++++++++++++---------------- tests/test_error_handling.py | 98 ++++++++++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 44 deletions(-) diff --git a/WEBUI/webui.py b/WEBUI/webui.py index 6b1d98b..21d4ea1 100644 --- a/WEBUI/webui.py +++ b/WEBUI/webui.py @@ -3,6 +3,7 @@ import bisect import glob import hashlib +import http.client import ipaddress import json import logging @@ -1721,8 +1722,8 @@ def current_user_bypasses_maintenance() -> bool: username = session.get("username") if username and site_maintenance.is_user_exempt(username, current_user_role()): return True - except Exception: - pass + except (OSError, ValueError) as exc: + logging.warning("Maintenance bypass check failed, treating user as not exempt: %s", exc) return False @@ -2580,16 +2581,16 @@ def list_all_streamer_folders(): folders.add(name) if arch and not is_system_backup_slug(arch): folders.add(arch) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Folder list without channel configs (database error): %s", exc) rows = [] profiles = load_profiles() configs = {} try: configs = {c.get("name"): c for c in database.list_channel_configs()} - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Folder list cannot mark stream channels (database error): %s", exc) for f in sorted(folders, key=lambda s: s.lower()): p_data = profiles.get(f) or {} @@ -2649,6 +2650,13 @@ def list_system_backups(): return sorted(items.values(), key=lambda x: x["mtime"], reverse=True) +def _twitch_gql_user(data): + """The `data.user` object of a Twitch GQL answer, or None for any other shape.""" + inner = data.get("data") if isinstance(data, dict) else None + user = inner.get("user") if isinstance(inner, dict) else None + return user if isinstance(user, dict) and user else None + + def get_twitch_profile_and_banner(channel): query = { "query": """ @@ -2674,11 +2682,12 @@ def get_twitch_profile_and_banner(channel): ) with urllib.request.urlopen(req, timeout=5) as resp: data = json.loads(resp.read().decode('utf-8')) - if data and "data" in data and data["data"].get("user"): - u = data["data"]["user"] - return u.get("profileImageURL"), u.get("bannerImageURL") - except Exception: - pass + except (OSError, ValueError, http.client.HTTPException) as exc: + logging.warning("Twitch profile lookup for %s failed: %s", channel, exc) + return None, None + user = _twitch_gql_user(data) + if user: + return user.get("profileImageURL"), user.get("bannerImageURL") return None, None def get_twitch_profile_image(channel): @@ -2747,12 +2756,14 @@ def get_twitch_user_id(channel): ) with urllib.request.urlopen(req, timeout=5) as resp: data = json.loads(resp.read().decode("utf-8")) - user_id = data.get("data", {}).get("user", {}).get("id") if isinstance(data, dict) else None - if user_id: - TWITCH_USER_ID_CACHE[login] = {"id": str(user_id), "ts": now} - return str(user_id) - except Exception: - pass + except (OSError, ValueError, http.client.HTTPException) as exc: + logging.warning("Twitch user id lookup for %s failed: %s", login, exc) + data = None + user = _twitch_gql_user(data) + user_id = user.get("id") if user else None + if user_id: + TWITCH_USER_ID_CACHE[login] = {"id": str(user_id), "ts": now} + return str(user_id) TWITCH_USER_ID_CACHE[login] = {"id": None, "ts": now} return None @@ -4756,23 +4767,20 @@ def can_view_streamer(streamer): return False candidates = {str(streamer).strip().lower()} - try: - norm = normalize_channel_name(streamer) - if norm: - candidates.add(norm.lower()) - except Exception: - pass + norm = normalize_channel_name(streamer) + if norm: + candidates.add(norm.lower()) try: arch = archive_name_for(streamer) if arch: candidates.add(arch.lower()) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Access check for %s without archive name (database error): %s", streamer, exc) try: for s in related_archive_slugs(streamer) or []: candidates.add(str(s).strip().lower()) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Access check for %s without archive aliases (database error): %s", streamer, exc) # 1. Check disabled_channels (Folder explicitly shut off / disabled) disabled_set = ctx.get("disabled_channels") or set() @@ -5780,8 +5788,8 @@ def read_media_sidecar_metadata(media_path): m_w = re.search(r"(?:Neuer Status|Stream-Titel):\s*(.+)$", msg, re.IGNORECASE) if m_w and not title: title = m_w.group(1).strip() - except Exception: - pass + except OSError as exc: + logging.warning("Title from chat log of %s unavailable: %s", media_path, exc) return { "title": clean_metadata_label(title, 180), @@ -6228,8 +6236,8 @@ def resolve_media_path(streamer, filename, route): candidates.append(streamer) try: candidates.extend(related_archive_slugs(streamer)) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Media lookup for %s without archive aliases (database error): %s", streamer, exc) seen = set() files_dir = getattr(config, "FILES_DIR", None) or os.path.join(BASE_DIR, "files") for cand in candidates: @@ -7074,8 +7082,8 @@ def delete_video_file(streamer, filename, is_clip): tombstone_slugs = set() try: tombstone_slugs.update(related_archive_slugs(streamer) or []) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Tombstone for %s/%s without archive aliases (database error): %s", streamer, filename, exc) primary = archive_name_for(streamer) or str(streamer or "").strip() if primary: tombstone_slugs.add(primary) @@ -7106,8 +7114,8 @@ def delete_video_file(streamer, filename, is_clip): _td, alt_path, alt_lock = thumb_paths(slug, filename) paths_to_clear.add(alt_path) paths_to_clear.add(alt_lock) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Thumbnails of %s/%s under alias folders kept (database error): %s", streamer, filename, exc) for path in paths_to_clear: if not path: continue @@ -8951,13 +8959,13 @@ def _vod_access_candidate_slugs(): name = str((item or {}).get("name") or "").strip() if name: found.add(database._normalize_vod_access_slug(name) or name.rstrip("_").lower()) - except Exception: - pass + except (OSError, sqlite3.Error) as exc: + logging.warning("VOD access candidates without archive folders: %s", exc) try: for slug in database.list_vod_access_defaults(): found.add(slug) - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("VOD access candidates without stored defaults: %s", exc) return sorted(s for s in found if s) @@ -11862,8 +11870,8 @@ def _recording_logins_set(): login = twitch_api.normalize_login(conf.get("name")) if login and current_recording_info(login) is not None: logins.add(login) - except Exception: - pass + except (sqlite3.Error, OSError) as exc: + logging.warning("Recording channel list unavailable: %s", exc) return logins @@ -11997,7 +12005,8 @@ def twitch_watch(channel): conf = database.get_channel_config(login) if conf: archive_name = normalize_channel_name(conf.get("archive_name")) or login - except Exception: + except sqlite3.Error as exc: + logging.warning("Live watch %s: channel config unavailable: %s", login, exc) archive_name = "" if not archive_name: try: @@ -12005,8 +12014,8 @@ def twitch_watch(channel): if twitch_api.normalize_login(conf.get("name")) == login: archive_name = normalize_channel_name(conf.get("archive_name")) or login break - except Exception: - pass + except sqlite3.Error as exc: + logging.warning("Live watch %s: channel list unavailable: %s", login, exc) # Default: Twitch embed without account cookies (iframe credentialless ≈ Incognito). # Optional ?proxy=1 for server HLS when embed is blocked or unsupported. diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index 9f0eb86..f931cfe 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -192,5 +192,103 @@ def test_tombstone_lookup_failure_is_logged_and_page_still_served(self): +class WebuiLookupFailureTests(IsolatedBaseTest): + """Web UI lookups that used to swallow every error now log specific ones.""" + + def setUp(self): + super().setUp() + from werkzeug.security import generate_password_hash + + from WEBUI import webui + self.webui = webui + patcher = mock.patch.object(webui, "BASE_DIR", str(self.base)) + patcher.start() + self.addCleanup(patcher.stop) + webui.ensure_admin_user() + webui.save_user("plainviewer", generate_password_hash("Role-Test-Pass-1"), role="viewer") + archive = self.base / "streamer" + archive.mkdir() + (archive / "vod.mp4").write_bytes(b"\0" * 32) + self.locked = sqlite3.OperationalError("database is locked") + + def _viewer_request(self): + ctx = self.webui.app.test_request_context("/") + ctx.push() + self.addCleanup(ctx.pop) + now = time.time() + self.webui.session.update(logged_in=True, user_role="viewer", username="plainviewer", + login_time=now, last_active=now) + + def test_maintenance_bypass_check_failure_is_logged_and_denied(self): + from core import site_maintenance + self._viewer_request() + with mock.patch.object(site_maintenance, "is_user_exempt", side_effect=OSError("state unreadable")), \ + self.assertLogs(level="WARNING") as logs: + self.assertFalse(self.webui.current_user_bypasses_maintenance()) + self.assertTrue(any("state unreadable" in line for line in logs.output)) + + def test_viewer_access_check_survives_channel_lookup_failure(self): + self._viewer_request() + expected = self.webui.can_view_streamer("streamer") + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=self.locked), \ + self.assertLogs(level="WARNING") as logs: + self.assertEqual(self.webui.can_view_streamer("streamer"), expected) + self.assertTrue(any("database is locked" in line for line in logs.output)) + + def test_streamer_folders_survive_channel_config_failure(self): + with mock.patch.object(self.webui.database, "list_channel_configs", side_effect=self.locked), \ + self.assertLogs(level="WARNING"): + rows = self.webui.list_all_streamer_folders() + self.assertIn("streamer", [row["name"] for row in rows]) + + def test_media_path_resolves_when_alias_lookup_fails(self): + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=self.locked), \ + self.assertLogs(level="WARNING"): + resolved = self.webui.resolve_media_path("streamer", "vod.mp4", "vod") + self.assertEqual(os.path.normpath(resolved), os.path.normpath(str(self.base / "streamer" / "vod.mp4"))) + + def test_vod_access_candidates_survive_failures(self): + with mock.patch.object(self.webui, "list_streamers", side_effect=OSError("archive unreadable")), \ + mock.patch.object(self.webui.database, "list_vod_access_defaults", side_effect=self.locked), \ + self.assertLogs(level="WARNING") as logs: + self.assertEqual(self.webui._vod_access_candidate_slugs(), []) + self.assertEqual(len(logs.output), 2) + + def test_recording_logins_survive_database_error(self): + with mock.patch.object(self.webui.database, "list_channel_configs", side_effect=self.locked), \ + self.assertLogs(level="WARNING"): + self.assertEqual(self.webui._recording_logins_set(), set()) + + def test_twitch_profile_lookup_failure_is_logged(self): + offline = urllib.error.URLError("offline") + with mock.patch.object(self.webui.urllib.request, "urlopen", side_effect=offline), \ + self.assertLogs(level="WARNING"): + self.assertEqual(self.webui.get_twitch_profile_and_banner("someone"), (None, None)) + + def test_twitch_profile_unexpected_payload_is_no_profile(self): + response = mock.MagicMock() + response.__enter__.return_value.read.return_value = b'{"data": null}' + with mock.patch.object(self.webui.urllib.request, "urlopen", return_value=response): + self.assertEqual(self.webui.get_twitch_profile_and_banner("someone"), (None, None)) + + def test_twitch_user_id_failure_is_logged_and_cached_as_unknown(self): + self.webui.TWITCH_USER_ID_CACHE.pop("someone", None) + self.addCleanup(self.webui.TWITCH_USER_ID_CACHE.pop, "someone", None) + with mock.patch.object(self.webui.urllib.request, "urlopen", side_effect=TimeoutError("slow")), \ + self.assertLogs(level="WARNING"): + self.assertIsNone(self.webui.get_twitch_user_id("someone")) + self.assertIsNone(self.webui.TWITCH_USER_ID_CACHE["someone"]["id"]) + + def test_delete_still_removes_file_and_writes_tombstone_when_alias_lookup_fails(self): + self._viewer_request() + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=self.locked), \ + self.assertLogs(level="WARNING") as logs: + self.assertTrue(self.webui.delete_video_file("streamer", "vod.mp4", False)) + # media lookup, tombstone aliases, thumbnail aliases + self.assertEqual(sum("database is locked" in line for line in logs.output), 3) + self.assertFalse((self.base / "streamer" / "vod.mp4").exists()) + self.assertTrue(self.webui.database.is_deleted_media(streamer="streamer", filename="vod.mp4")) + + if __name__ == "__main__": unittest.main() From 5902de1b3b51ec6271f0b509ce6eed919dafbbbf Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:02:58 +0100 Subject: [PATCH 06/12] docs(changelog): broad exception handlers removed, tombstone 404 fixed Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 70d9959..49c950b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,9 +96,12 @@ All notable changes to this project are documented here. The format follows ### Fixed -- Broad `except Exception: pass` handlers are gone from the database layer, - recorder, worker and clip engine; 46 such handlers remain elsewhere (18 in - the web UI, the rest in storage providers and API clients). +- Broad `except Exception: pass` handlers are gone from the whole code base + (database layer, recorder, worker, clip engine, web UI, storage providers + and API clients). The failures they hid are now caught by class and logged; + fallback values are unchanged. +- A deleted VOD or clip whose file was still on disk opened in the player: + the tombstone check's `abort(404)` was swallowed by its own error handler. - A fresh database no longer logs a bogus `users.role` migration warning. ## [0.1.0] - 2026-11-09 From 625291fff1c3fcc9dcd0405fedea740149a42d0e Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:23:39 +0100 Subject: [PATCH 07/12] style(tests): wrap with-statements that were joined onto one line Thirteen context-manager continuations in the new error-handling tests had lost their backslash-newline and ran on one 200+ character line. Only line breaks change; no assertion or patch target differs. Co-Authored-By: Claude Opus 5.5 --- tests/test_error_handling.py | 20 +++++++++++++++----- tests/test_storage.py | 7 +++++-- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index f931cfe..d370c81 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -105,21 +105,26 @@ def test_eventsub_post_logs_unreadable_error_body_and_returns_none(self): class CoreDatabaseFallbackTests(IsolatedBaseTest): def test_vod_access_bases_fall_back_to_defaults_and_log(self): locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(tube_auth.database, "list_role_vod_access", side_effect=locked), self.assertLogs("core.tube_auth", level="WARNING"): + with mock.patch.object(tube_auth.database, "list_role_vod_access", side_effect=locked), \ + self.assertLogs("core.tube_auth", level="WARNING"): bases = tube_auth.effective_vod_access_bases() self.assertEqual(bases, sorted(tube_auth._VOD_ACCESS_SEED_BASES)) def test_live_diag_candidates_survive_database_error(self): from core import database locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(database, "list_channel_configs", side_effect=locked), self.assertLogs("core.live_watch_diag", level="WARNING"): + with mock.patch.object(database, "list_channel_configs", side_effect=locked), \ + self.assertLogs("core.live_watch_diag", level="WARNING"): names = live_watch_diag.list_candidate_channels(extra=["Extra"]) self.assertIn("extra", names) def test_live_diag_channel_report_survives_database_error(self): from core import database locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(database, "get_channel_config", side_effect=locked), mock.patch.object(live_watch_diag, "streamlink_counts", return_value={}), mock.patch.object(live_watch_diag, "ffmpeg_hls_pids", return_value={}), self.assertLogs("core.live_watch_diag", level="WARNING"): + with mock.patch.object(database, "get_channel_config", side_effect=locked), \ + mock.patch.object(live_watch_diag, "streamlink_counts", return_value={}), \ + mock.patch.object(live_watch_diag, "ffmpeg_hls_pids", return_value={}), \ + self.assertLogs("core.live_watch_diag", level="WARNING"): report = live_watch_diag.diagnose_channel("chan", helix={"chan": {}}) self.assertEqual(report["db"], {"live": None, "record": None}) @@ -144,7 +149,11 @@ def test_corrupt_cached_meta_is_logged_and_rebuilt(self): class RedisDashboardCacheTests(unittest.TestCase): def test_shm_cache_removal_failure_is_logged(self): - with mock.patch.object(redis_client, "delete_state", return_value=False), mock.patch.object(redis_client, "publish_event", return_value=False), mock.patch.object(redis_client.os.path, "exists", return_value=True), mock.patch.object(redis_client.os, "remove", side_effect=PermissionError("busy")), self.assertLogs("streamvault.redis", level="WARNING"): + with mock.patch.object(redis_client, "delete_state", return_value=False), \ + mock.patch.object(redis_client, "publish_event", return_value=False), \ + mock.patch.object(redis_client.os.path, "exists", return_value=True), \ + mock.patch.object(redis_client.os, "remove", side_effect=PermissionError("busy")), \ + self.assertLogs("streamvault.redis", level="WARNING"): self.assertTrue(redis_client.invalidate_dashboard_cache()) @@ -185,7 +194,8 @@ def test_tombstoned_vod_page_is_404(self): def test_tombstone_lookup_failure_is_logged_and_page_still_served(self): client = _admin_client(self.webui) locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(self.webui.database, "is_deleted_media", side_effect=locked), self.assertLogs(level="WARNING") as logs: + with mock.patch.object(self.webui.database, "is_deleted_media", side_effect=locked), \ + self.assertLogs(level="WARNING") as logs: response = client.get("/vod/streamer/vod.mp4") self.assertEqual(response.status_code, 200) self.assertTrue(any("database is locked" in line for line in logs.output)) diff --git a/tests/test_storage.py b/tests/test_storage.py index b84442a..58146b7 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -2089,7 +2089,8 @@ class StorageBookkeepingFailureTests(StorageTestBase): def test_progress_update_on_locked_database_is_logged(self): callback = service._progress_callback(1, 100) locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(service.database, "update_storage_object", side_effect=locked), self.assertLogs("storage.service", level="DEBUG") as logs: + with mock.patch.object(service.database, "update_storage_object", side_effect=locked), \ + self.assertLogs("storage.service", level="DEBUG") as logs: callback(50, 100) self.assertTrue(any("database is locked" in line for line in logs.output)) @@ -2097,7 +2098,9 @@ def test_restore_error_survives_failing_provider_error_bookkeeping(self): path, obj = RestoreSafetyTests._restore_object(self, b"new", remote_size=3, local=b"older") fake = RestoreSafetyTests._FakeDownloadProvider(error=StorageError("remote gone")) locked = sqlite3.OperationalError("database is locked") - with mock.patch.object(service, "build_provider", return_value=fake), mock.patch.object(service.database, "set_storage_provider_error", side_effect=locked), self.assertLogs("storage.service", level="WARNING") as logs: + with mock.patch.object(service, "build_provider", return_value=fake), \ + mock.patch.object(service.database, "set_storage_provider_error", side_effect=locked), \ + self.assertLogs("storage.service", level="WARNING") as logs: with self.assertRaises(StorageError) as ctx: service.run_restore({"object_id": obj["id"]}) self.assertIn("remote gone", str(ctx.exception)) From d5164486497c9142d790f869fd46b6669a1c4f2c Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:27:05 +0100 Subject: [PATCH 08/12] fix: database fallbacks catch DB_ACCESS_ERRORS, per-request warnings are rate-limited Review of #14: handlers that caught only sqlite3.Error let OSError from get_db() (it creates the data directory) and ValueError from _normalize_channel_row() (non-numeric record flag) through as HTTP 500s. - core.database.DB_ACCESS_ERRORS = (sqlite3.Error, OSError, ValueError); every database fallback in webui, tube_auth, live_watch_diag and the storage service catches exactly this tuple. Other classes propagate (negative test). - webui._warn_db_fallback(): can_view_streamer, resolve_media_path and the tombstone check log at most once per minute and call site, with the number of repeats, instead of once per list entry or request. - storage: a failed progress save warns once per transfer, then stays at debug; it never aborts the transfer. - current_user_bypasses_maintenance(): the handler around is_user_exempt() was dead (site_maintenance.load_state() handles unreadable files and logs itself). It is removed. The mock-based test is replaced by a test with a real corrupt maintenance.json. - Tests for twitch_watch with failing channel lookups. Co-Authored-By: Claude Opus 5.5 --- WEBUI/webui.py | 84 +++++++++++++------ core/database.py | 7 ++ core/live_watch_diag.py | 5 +- core/storage/service.py | 17 ++-- core/tube_auth.py | 5 +- tests/test_error_handling.py | 156 +++++++++++++++++++++++++++++++++-- tests/test_storage.py | 15 ++++ 7 files changed, 242 insertions(+), 47 deletions(-) diff --git a/WEBUI/webui.py b/WEBUI/webui.py index 21d4ea1..3bca7b7 100644 --- a/WEBUI/webui.py +++ b/WEBUI/webui.py @@ -1716,14 +1716,12 @@ def maintenance_mode_enabled(scope: str = "global"): def current_user_bypasses_maintenance() -> bool: if current_user_is_admin(): return True - try: - from core import site_maintenance + from core import site_maintenance - username = session.get("username") - if username and site_maintenance.is_user_exempt(username, current_user_role()): - return True - except (OSError, ValueError) as exc: - logging.warning("Maintenance bypass check failed, treating user as not exempt: %s", exc) + # load_state() logs and replaces an unreadable state file itself. + username = session.get("username") + if username and site_maintenance.is_user_exempt(username, current_user_role()): + return True return False @@ -1777,6 +1775,35 @@ def get_user_color(username): hash_val = ord(char) + ((hash_val << 5) - hash_val) return colors[abs(hash_val) % len(colors)] +DB_WARNING_INTERVAL_SEC = 60.0 +_DB_WARNING_LAST = {} +_DB_WARNING_SUPPRESSED = {} +_DB_WARNING_LOCK = threading.Lock() + + +def _warn_db_fallback(site, message, *args): + """Log a database fallback at most once per minute and call site. + + Access checks and media lookups run for every entry of a list and on + every request. During a database outage one line per site and minute, + with the number of repeats since the last one, says as much as thousands. + """ + now = time.monotonic() + with _DB_WARNING_LOCK: + last = _DB_WARNING_LAST.get(site) + if last is None: + _DB_WARNING_SUPPRESSED.pop(site, None) + elif now - last < DB_WARNING_INTERVAL_SEC: + _DB_WARNING_SUPPRESSED[site] = _DB_WARNING_SUPPRESSED.get(site, 0) + 1 + return + _DB_WARNING_LAST[site] = now + repeats = _DB_WARNING_SUPPRESSED.pop(site, 0) + if repeats: + message += " (%d more since the last warning)" + args = args + (repeats,) + logging.warning(message, *args) + + def normalize_channel_name(name): if not name: return None @@ -2581,7 +2608,7 @@ def list_all_streamer_folders(): folders.add(name) if arch and not is_system_backup_slug(arch): folders.add(arch) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Folder list without channel configs (database error): %s", exc) rows = [] @@ -2589,7 +2616,7 @@ def list_all_streamer_folders(): configs = {} try: configs = {c.get("name"): c for c in database.list_channel_configs()} - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Folder list cannot mark stream channels (database error): %s", exc) for f in sorted(folders, key=lambda s: s.lower()): @@ -4774,13 +4801,15 @@ def can_view_streamer(streamer): arch = archive_name_for(streamer) if arch: candidates.add(arch.lower()) - except sqlite3.Error as exc: - logging.warning("Access check for %s without archive name (database error): %s", streamer, exc) + except database.DB_ACCESS_ERRORS as exc: + _warn_db_fallback("can_view_streamer:archive", + "Access check for %s without archive name (database error): %s", streamer, exc) try: for s in related_archive_slugs(streamer) or []: candidates.add(str(s).strip().lower()) - except sqlite3.Error as exc: - logging.warning("Access check for %s without archive aliases (database error): %s", streamer, exc) + except database.DB_ACCESS_ERRORS as exc: + _warn_db_fallback("can_view_streamer:aliases", + "Access check for %s without archive aliases (database error): %s", streamer, exc) # 1. Check disabled_channels (Folder explicitly shut off / disabled) disabled_set = ctx.get("disabled_channels") or set() @@ -4835,8 +4864,9 @@ def abort_deleted_media(streamer, filename, file_id=None): slugs = set() try: slugs.update(related_archive_slugs(streamer) or []) - except sqlite3.Error as exc: - logging.warning("Tombstone check for %s/%s without aliases (database error): %s", streamer, name, exc) + except database.DB_ACCESS_ERRORS as exc: + _warn_db_fallback("abort_deleted_media:aliases", + "Tombstone check for %s/%s without aliases (database error): %s", streamer, name, exc) primary = archive_name_for(streamer) or str(streamer or "").strip() if primary: slugs.add(primary) @@ -4844,10 +4874,11 @@ def abort_deleted_media(streamer, filename, file_id=None): slugs.add(str(streamer).strip()) try: deleted = database.is_deleted_media(filename=name, file_id=file_id, streamers=slugs) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: # Same outcome as before: without the tombstone table the file # itself decides (resolve_media_path still 404s a missing file). - logging.warning("Tombstone check for %s/%s failed, serving: %s", streamer, name, exc) + _warn_db_fallback("abort_deleted_media:lookup", + "Tombstone check for %s/%s failed, serving: %s", streamer, name, exc) return if deleted: abort(404) @@ -6236,8 +6267,9 @@ def resolve_media_path(streamer, filename, route): candidates.append(streamer) try: candidates.extend(related_archive_slugs(streamer)) - except sqlite3.Error as exc: - logging.warning("Media lookup for %s without archive aliases (database error): %s", streamer, exc) + except database.DB_ACCESS_ERRORS as exc: + _warn_db_fallback("resolve_media_path:aliases", + "Media lookup for %s without archive aliases (database error): %s", streamer, exc) seen = set() files_dir = getattr(config, "FILES_DIR", None) or os.path.join(BASE_DIR, "files") for cand in candidates: @@ -7082,7 +7114,7 @@ def delete_video_file(streamer, filename, is_clip): tombstone_slugs = set() try: tombstone_slugs.update(related_archive_slugs(streamer) or []) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Tombstone for %s/%s without archive aliases (database error): %s", streamer, filename, exc) primary = archive_name_for(streamer) or str(streamer or "").strip() if primary: @@ -7114,7 +7146,7 @@ def delete_video_file(streamer, filename, is_clip): _td, alt_path, alt_lock = thumb_paths(slug, filename) paths_to_clear.add(alt_path) paths_to_clear.add(alt_lock) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Thumbnails of %s/%s under alias folders kept (database error): %s", streamer, filename, exc) for path in paths_to_clear: if not path: @@ -8959,12 +8991,12 @@ def _vod_access_candidate_slugs(): name = str((item or {}).get("name") or "").strip() if name: found.add(database._normalize_vod_access_slug(name) or name.rstrip("_").lower()) - except (OSError, sqlite3.Error) as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("VOD access candidates without archive folders: %s", exc) try: for slug in database.list_vod_access_defaults(): found.add(slug) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("VOD access candidates without stored defaults: %s", exc) return sorted(s for s in found if s) @@ -11870,7 +11902,7 @@ def _recording_logins_set(): login = twitch_api.normalize_login(conf.get("name")) if login and current_recording_info(login) is not None: logins.add(login) - except (sqlite3.Error, OSError) as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Recording channel list unavailable: %s", exc) return logins @@ -12005,7 +12037,7 @@ def twitch_watch(channel): conf = database.get_channel_config(login) if conf: archive_name = normalize_channel_name(conf.get("archive_name")) or login - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Live watch %s: channel config unavailable: %s", login, exc) archive_name = "" if not archive_name: @@ -12014,7 +12046,7 @@ def twitch_watch(channel): if twitch_api.normalize_login(conf.get("name")) == login: archive_name = normalize_channel_name(conf.get("archive_name")) or login break - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logging.warning("Live watch %s: channel list unavailable: %s", login, exc) # Default: Twitch embed without account cookies (iframe credentialless ≈ Incognito). diff --git a/core/database.py b/core/database.py index cdeaf81..de34884 100644 --- a/core/database.py +++ b/core/database.py @@ -6,6 +6,13 @@ import time logger = logging.getLogger("database") + +# What a caller of the get_db()-based helpers can see for reasons outside its +# control: SQLite errors (locked, corrupt, read-only), OSError (get_db() creates +# the data directory) and ValueError (a malformed stored value, for instance a +# non-numeric record flag in _normalize_channel_row). Fallback handlers catch +# exactly this tuple; anything else is a programming error and propagates. +DB_ACCESS_ERRORS = (sqlite3.Error, OSError, ValueError) from functools import wraps from contextlib import contextmanager diff --git a/core/live_watch_diag.py b/core/live_watch_diag.py index de25d68..b0b3b66 100644 --- a/core/live_watch_diag.py +++ b/core/live_watch_diag.py @@ -9,7 +9,6 @@ import logging import os import re -import sqlite3 import subprocess import time from pathlib import Path @@ -88,7 +87,7 @@ def list_candidate_channels(extra: list[str] | None = None) -> list[str]: found.add(name) if int(conf.get("live") or 0) == 1 and name: found.add(name) - except (sqlite3.Error, ValueError) as exc: + except database.DB_ACCESS_ERRORS as exc: logger.warning("live diag: channel configs unavailable: %s", exc) for name in extra or []: if name: @@ -177,7 +176,7 @@ def diagnose_channel(channel: str, *, logs: int = 30, helix: dict | None = None) if conf: db_live = int(conf.get("live") or 0) db_record = int(conf.get("record") or 0) - except (sqlite3.Error, ValueError) as exc: + except database.DB_ACCESS_ERRORS as exc: logger.warning("live diag %s: channel config unavailable: %s", channel, exc) helix_info = (helix or {}).get(channel) if helix is not None else None diff --git a/core/storage/service.py b/core/storage/service.py index e869d24..9dd14f0 100644 --- a/core/storage/service.py +++ b/core/storage/service.py @@ -11,7 +11,6 @@ import logging import os import shutil -import sqlite3 import stat import time from typing import Optional @@ -1118,7 +1117,7 @@ def cleanup_local_copies(provider_id=None, *, limit: int = 100) -> dict: # --------------------------------------------------------------------------- def _progress_callback(object_id: int, total: int): - state = {"last": 0.0, "last_pct": -1.0} + state = {"last": 0.0, "last_pct": -1.0, "warned": False} def callback(done: int, size: int): now = time.time() @@ -1133,10 +1132,14 @@ def callback(done: int, size: int): state["last_pct"] = pct try: database.update_storage_object(object_id, progress=round(pct, 1)) - except sqlite3.Error as exc: - # Progress is cosmetic and rewritten every two seconds; a busy - # database must not abort the transfer itself. - logger.debug("storage object %s: progress not saved: %s", object_id, exc) + except database.DB_ACCESS_ERRORS as exc: + # Progress is cosmetic and rewritten every two seconds; a busy or + # read-only database must not abort the transfer itself. Warn + # once per transfer, then stay at debug so a long upload during a + # lock does not write a line every two seconds. + level = logging.DEBUG if state["warned"] else logging.WARNING + state["warned"] = True + logger.log(level, "storage object %s: progress not saved: %s", object_id, exc) return callback @@ -1399,7 +1402,7 @@ def run_restore(payload: dict) -> dict: database.update_storage_object(object_id, last_error=message) try: database.set_storage_provider_error(obj["provider_id"], message) - except sqlite3.Error as db_exc: + except database.DB_ACCESS_ERRORS as db_exc: # Keep the restore error for the caller; the provider status line # is only a convenience for the storage page. logger.warning( diff --git a/core/tube_auth.py b/core/tube_auth.py index f101dc6..3394a34 100644 --- a/core/tube_auth.py +++ b/core/tube_auth.py @@ -12,7 +12,6 @@ import os import re import secrets -import sqlite3 import string import time from typing import Any @@ -97,7 +96,7 @@ def _expand_access_slugs(raw_slugs) -> set[str]: def list_default_vod_access_slugs() -> list[str]: try: rows = database.list_role_vod_access(MEMBER_ROLE) - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logger.warning("VOD access defaults unreadable, using seed list: %s", exc) rows = [] if rows: @@ -115,7 +114,7 @@ def effective_vod_access_bases(role: str | None = None) -> list[str]: rows = database.list_role_vod_access(role_key) if rows: return rows - except sqlite3.Error as exc: + except database.DB_ACCESS_ERRORS as exc: logger.warning("VOD access for role %s unreadable, using defaults: %s", role_key, exc) return list_default_vod_access_slugs() diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index d370c81..c1839ff 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -220,6 +220,8 @@ def setUp(self): archive.mkdir() (archive / "vod.mp4").write_bytes(b"\0" * 32) self.locked = sqlite3.OperationalError("database is locked") + webui._DB_WARNING_LAST.clear() + self.addCleanup(webui._DB_WARNING_LAST.clear) def _viewer_request(self): ctx = self.webui.app.test_request_context("/") @@ -229,14 +231,6 @@ def _viewer_request(self): self.webui.session.update(logged_in=True, user_role="viewer", username="plainviewer", login_time=now, last_active=now) - def test_maintenance_bypass_check_failure_is_logged_and_denied(self): - from core import site_maintenance - self._viewer_request() - with mock.patch.object(site_maintenance, "is_user_exempt", side_effect=OSError("state unreadable")), \ - self.assertLogs(level="WARNING") as logs: - self.assertFalse(self.webui.current_user_bypasses_maintenance()) - self.assertTrue(any("state unreadable" in line for line in logs.output)) - def test_viewer_access_check_survives_channel_lookup_failure(self): self._viewer_request() expected = self.webui.can_view_streamer("streamer") @@ -300,5 +294,151 @@ def test_delete_still_removes_file_and_writes_tombstone_when_alias_lookup_fails( self.assertTrue(self.webui.database.is_deleted_media(streamer="streamer", filename="vod.mp4")) +class DatabaseAccessErrorTests(IsolatedBaseTest): + """Fallbacks around database reads cover every error get_db() callers can see.""" + + def setUp(self): + super().setUp() + from WEBUI import webui + self.webui = webui + patcher = mock.patch.object(webui, "BASE_DIR", str(self.base)) + patcher.start() + self.addCleanup(patcher.stop) + archive = self.base / "streamer" + archive.mkdir() + (archive / "vod.mp4").write_bytes(b"\0" * 32) + self.webui._DB_WARNING_LAST.clear() + + def test_malformed_channel_row_raises_value_error(self): + from core import database + conn = database.get_db() + try: + conn.execute("INSERT INTO channels (name, record) VALUES ('streamer', 'abc')") + conn.commit() + finally: + conn.close() + with self.assertRaises(database.DB_ACCESS_ERRORS): + database.get_channel_config("streamer") + + def test_vod_access_bases_fall_back_when_data_dir_is_unreadable(self): + with mock.patch.object(tube_auth.database, "list_role_vod_access", + side_effect=PermissionError("data dir")), \ + self.assertLogs("core.tube_auth", level="WARNING"): + bases = tube_auth.effective_vod_access_bases() + self.assertEqual(bases, sorted(tube_auth._VOD_ACCESS_SEED_BASES)) + + def test_media_path_resolves_on_os_and_value_errors(self): + for error in (PermissionError("data dir"), ValueError("bad record flag")): + self.webui._DB_WARNING_LAST.clear() + with self.subTest(error=type(error).__name__), \ + mock.patch.object(self.webui, "related_archive_slugs", side_effect=error), \ + self.assertLogs(level="WARNING"): + resolved = self.webui.resolve_media_path("streamer", "vod.mp4", "vod") + self.assertIsNotNone(resolved) + + def test_unexpected_error_in_alias_lookup_is_not_swallowed(self): + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=RuntimeError("bug")): + with self.assertRaises(RuntimeError): + self.webui.resolve_media_path("streamer", "vod.mp4", "vod") + + def test_vod_access_candidates_survive_malformed_defaults(self): + with mock.patch.object(self.webui, "list_streamers", return_value=[{"name": "streamer"}]), \ + mock.patch.object(self.webui.database, "list_vod_access_defaults", + side_effect=ValueError("bad row")), \ + self.assertLogs(level="WARNING"): + self.assertEqual(self.webui._vod_access_candidate_slugs(), ["streamer"]) + + +class DatabaseWarningRateLimitTests(IsolatedBaseTest): + """A database outage leaves one warning per call site and minute, not one per lookup.""" + + def setUp(self): + super().setUp() + from werkzeug.security import generate_password_hash + + from WEBUI import webui + self.webui = webui + patcher = mock.patch.object(webui, "BASE_DIR", str(self.base)) + patcher.start() + self.addCleanup(patcher.stop) + webui.save_user("plainviewer", generate_password_hash("Role-Test-Pass-1"), role="viewer") + archive = self.base / "streamer" + archive.mkdir() + (archive / "vod.mp4").write_bytes(b"\0" * 32) + self.webui._DB_WARNING_LAST.clear() + self.addCleanup(self.webui._DB_WARNING_LAST.clear) + self.locked = sqlite3.OperationalError("database is locked") + + def _locked_lines(self, logs): + return [line for line in logs.output if "database is locked" in line] + + def test_media_lookup_warns_once_per_minute(self): + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=self.locked), \ + self.assertLogs(level="WARNING") as logs: + for _ in range(5): + self.assertIsNotNone(self.webui.resolve_media_path("streamer", "vod.mp4", "vod")) + self.assertEqual(len(self._locked_lines(logs)), 1) + for site in list(self.webui._DB_WARNING_LAST): + self.webui._DB_WARNING_LAST[site] -= 61 + self.webui.resolve_media_path("streamer", "vod.mp4", "vod") + locked = self._locked_lines(logs) + self.assertEqual(len(locked), 2) + self.assertIn("4 more", locked[1]) + + def test_access_check_in_a_list_warns_once(self): + ctx = self.webui.app.test_request_context("/") + ctx.push() + self.addCleanup(ctx.pop) + now = time.time() + self.webui.session.update(logged_in=True, user_role="viewer", username="plainviewer", + login_time=now, last_active=now) + with mock.patch.object(self.webui, "related_archive_slugs", side_effect=self.locked), \ + self.assertLogs(level="WARNING") as logs: + for _ in range(10): + self.webui.can_view_streamer("streamer") + self.assertEqual(len(self._locked_lines(logs)), 1) + + +class MaintenanceBypassTests(IsolatedBaseTest): + def test_corrupt_maintenance_state_keeps_viewer_not_exempt(self): + from werkzeug.security import generate_password_hash + + from core import site_maintenance + from WEBUI import webui + webui.save_user("plainviewer", generate_password_hash("Role-Test-Pass-1"), role="viewer") + with open(site_maintenance.state_path(), "w", encoding="utf-8") as handle: + handle.write("{broken") + ctx = webui.app.test_request_context("/") + ctx.push() + self.addCleanup(ctx.pop) + now = time.time() + webui.session.update(logged_in=True, user_role="viewer", username="plainviewer", + login_time=now, last_active=now) + with self.assertLogs(level="WARNING") as logs: + self.assertFalse(webui.current_user_bypasses_maintenance()) + self.assertTrue(any("maintenance state unreadable" in line for line in logs.output)) + + +class TwitchWatchFallbackTests(IsolatedBaseTest): + def setUp(self): + super().setUp() + from WEBUI import webui + self.webui = webui + webui.ensure_admin_user() + + def test_watch_page_renders_when_channel_lookups_fail(self): + locked = sqlite3.OperationalError("database is locked") + client = _admin_client(self.webui) + with mock.patch.object(self.webui.database, "get_channel_config", side_effect=locked), \ + mock.patch.object(self.webui.database, "list_channel_configs", side_effect=locked), \ + mock.patch.object(self.webui, "add_live_channel"), \ + self.assertLogs(level="WARNING") as logs: + response = client.get("/live/watch/somechannel?proxy=0") + self.assertEqual(response.status_code, 200) + warned = [line for line in logs.output if "Live watch somechannel" in line] + self.assertEqual(len(warned), 2) + + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_storage.py b/tests/test_storage.py index 58146b7..1dac38c 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -2094,6 +2094,21 @@ def test_progress_update_on_locked_database_is_logged(self): callback(50, 100) self.assertTrue(any("database is locked" in line for line in logs.output)) + def test_progress_save_failures_warn_once_per_transfer_and_never_abort(self): + callback = service._progress_callback(1, 100) + clock = [1000.0] + with mock.patch.object(service.time, "time", side_effect=lambda: clock[0]), \ + mock.patch.object(service.database, "update_storage_object", + side_effect=PermissionError("data dir read-only")), \ + self.assertLogs("storage.service", level="DEBUG") as logs: + for done in (10, 30, 50, 70): + callback(done, 100) + clock[0] += 5 + warnings = [line for line in logs.output if line.startswith("WARNING")] + self.assertEqual(len(warnings), 1) + self.assertIn("data dir read-only", warnings[0]) + self.assertEqual(len(logs.output), 4) + def test_restore_error_survives_failing_provider_error_bookkeeping(self): path, obj = RestoreSafetyTests._restore_object(self, b"new", remote_size=3, local=b"older") fake = RestoreSafetyTests._FakeDownloadProvider(error=StorageError("remote gone")) From 669d655a5baef4e8f56c8ae86fa2ad74bf2bd827 Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:27:56 +0100 Subject: [PATCH 09/12] refactor(storage): close_quietly lives in storage.base, debug levels justified - close_quietly() moves from storage.http to storage.base with a Closable protocol type; s3.py no longer imports the HTTP helper just for it. Its log line now comes from the "storage.base" logger (tests follow). - The remaining debug-level lines say in code why they are not warnings: close_quietly (routine connection resets, no outcome change), twitch_api._drain_http_error and the EventSub POST (the HTTP status is already logged at warning/info). Co-Authored-By: Claude Opus 5.5 --- core/storage/base.py | 30 +++++++++++++++++++++++++- core/storage/http.py | 23 +------------------- core/storage/providers/dropbox.py | 7 +++--- core/storage/providers/google_drive.py | 11 +++++----- core/storage/providers/onedrive.py | 7 +++--- core/storage/providers/s3.py | 4 ++-- core/storage/providers/webdav.py | 5 +++-- core/twitch_api.py | 2 ++ core/twitch_eventsub.py | 3 ++- tests/test_error_handling.py | 10 ++++----- 10 files changed, 58 insertions(+), 44 deletions(-) diff --git a/core/storage/base.py b/core/storage/base.py index 03f14c6..28edc2c 100644 --- a/core/storage/base.py +++ b/core/storage/base.py @@ -15,9 +15,10 @@ import re import sqlite3 from dataclasses import dataclass -from typing import Callable, Iterable, Iterator, Optional +from typing import Callable, Iterable, Iterator, Optional, Protocol from flask_babel import gettext, lazy_gettext +from urllib3.exceptions import HTTPError as Urllib3HTTPError logger = logging.getLogger("storage.base") @@ -382,6 +383,33 @@ def mask_text(value) -> str: return text[:500] +class Closable(Protocol): + def close(self) -> None: ... + + +def close_quietly(resource: Closable, *, provider: str = "") -> None: + """Release a response or stream body whose result is no longer needed. + + Closing only hands the connection back to the pool; the transfer result + is already decided by then. A socket that is already gone (OSError, + including requests' ConnectionError) or a urllib3 protocol error while + draining must therefore not replace the caller's result or exception. + Anything else is a programming error and propagates. + + Logged at debug, not warning: a peer that resets an idle or finished + connection is routine for every provider and changes no outcome, so a + warning would only add noise to every transfer log. + """ + try: + resource.close() + except (OSError, Urllib3HTTPError) as exc: + logger.debug( + "%s: closing HTTP response failed: %s", + provider or "storage", + mask_error(exc), + ) + + def mask_error(exc: Exception) -> str: """Never leak credential-looking values into logs/UI.""" return mask_text(exc) or "Storage-Fehler" diff --git a/core/storage/http.py b/core/storage/http.py index f9b21a7..405c2f4 100644 --- a/core/storage/http.py +++ b/core/storage/http.py @@ -11,9 +11,7 @@ from typing import Any, Optional import requests -from urllib3.exceptions import HTTPError as Urllib3HTTPError - -from .base import StorageError +from .base import StorageError, close_quietly logger = logging.getLogger("storage.http") @@ -51,25 +49,6 @@ def session() -> requests.Session: return current -def close_quietly(resource, *, provider: str = "") -> None: - """Release a response or stream body whose result is no longer needed. - - Closing only hands the connection back to the pool; the transfer result - is already decided by then. A socket that is already gone (OSError, - including requests' ConnectionError) or a urllib3 protocol error while - draining must therefore not replace the caller's result or exception. - Anything else is a programming error and propagates. - """ - try: - resource.close() - except (OSError, Urllib3HTTPError) as exc: - logger.debug( - "%s: closing HTTP response failed: %s", - provider or "storage", - sanitize_error_text(exc), - ) - - def _retry_after_seconds(response: requests.Response, attempt: int) -> float: raw = (response.headers.get("Retry-After") or "").strip() if raw: diff --git a/core/storage/providers/dropbox.py b/core/storage/providers/dropbox.py index 2054063..e792d0e 100644 --- a/core/storage/providers/dropbox.py +++ b/core/storage/providers/dropbox.py @@ -12,6 +12,7 @@ from flask_babel import gettext, lazy_gettext from core.storage.base import ( + close_quietly, ProgressCallback, ProviderField, RemoteObject, @@ -152,7 +153,7 @@ def _rpc(self, endpoint: str, payload: dict, action: str) -> dict: timeout=(10, 60), ) if response.status_code == 401 and attempt == 0: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) self._token(force=True) continue return http.json_response(response, action, provider=self.type_id) @@ -366,7 +367,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -382,7 +383,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: try: diff --git a/core/storage/providers/google_drive.py b/core/storage/providers/google_drive.py index 586dad4..eed7ead 100644 --- a/core/storage/providers/google_drive.py +++ b/core/storage/providers/google_drive.py @@ -11,6 +11,7 @@ from flask_babel import lazy_gettext from core.storage.base import ( + close_quietly, ProgressCallback, ProviderField, RemoteObject, @@ -186,7 +187,7 @@ def _api(self, method: str, path: str, **kwargs): kwargs["headers"] = self._headers(**(kwargs.get("headers") or {})) response = http.request(method, url, provider=self.type_id, **kwargs) if response.status_code == 401 and attempt == 0: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) self._token(force=True) continue return response @@ -342,7 +343,7 @@ def _upload_empty(self, name: str, parent_id: str, existing_id: str, mime: str) ) if response.status_code != 401 or attempt == 1: break - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) data = http.json_response(response, "Google-Upload (leer)", provider=self.type_id) return { "remote_id": str(data.get("id") or existing_id or ""), @@ -404,7 +405,7 @@ def upload_file( ) if response.status_code != 401 or attempt == 1: break - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) http.raise_for_status(response, "Google-Upload starten", provider=self.type_id) location = str(response.headers.get("Location") or "").strip() if not location: @@ -531,7 +532,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -547,7 +548,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: remote_path = normalize_remote_path(remote_path) diff --git a/core/storage/providers/onedrive.py b/core/storage/providers/onedrive.py index 4be197f..8714332 100644 --- a/core/storage/providers/onedrive.py +++ b/core/storage/providers/onedrive.py @@ -10,6 +10,7 @@ from flask_babel import lazy_gettext from core.storage.base import ( + close_quietly, ProgressCallback, ProviderField, RemoteObject, @@ -153,7 +154,7 @@ def _api(self, method: str, url: str, **kwargs): kwargs["headers"] = self._headers(**(kwargs.get("headers") or {})) response = http.request(method, url, provider=self.type_id, **kwargs) if response.status_code == 401 and attempt == 0: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) self._token(force=True) continue return response @@ -349,7 +350,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -365,7 +366,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: response = self._api("DELETE", self._item_url(remote_path), timeout=(10, 60)) diff --git a/core/storage/providers/s3.py b/core/storage/providers/s3.py index 2752c48..8387632 100644 --- a/core/storage/providers/s3.py +++ b/core/storage/providers/s3.py @@ -8,8 +8,8 @@ from flask_babel import gettext, lazy_gettext -from core.storage import http from core.storage.base import ( + close_quietly, ProgressCallback, ProviderField, RemoteObject, @@ -269,7 +269,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: break yield chunk finally: - http.close_quietly(body, provider=self.type_id) + close_quietly(body, provider=self.type_id) def delete(self, remote_path: str) -> None: key = self._key(remote_path) diff --git a/core/storage/providers/webdav.py b/core/storage/providers/webdav.py index feee9dc..ab6d2ba 100644 --- a/core/storage/providers/webdav.py +++ b/core/storage/providers/webdav.py @@ -10,6 +10,7 @@ from flask_babel import gettext, lazy_gettext from core.storage.base import ( + close_quietly, ProgressCallback, ProviderField, RemoteObject, @@ -185,7 +186,7 @@ def download_file( os.fsync(handle.fileno()) os.replace(tmp_path, dest_path) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) if os.path.exists(tmp_path): try: os.remove(tmp_path) @@ -206,7 +207,7 @@ def open_range(self, remote_path: str, start: int, end: int) -> Iterable[bytes]: try: yield from iter_response_range(response, start, end) finally: - http.close_quietly(response, provider=self.type_id) + close_quietly(response, provider=self.type_id) def delete(self, remote_path: str) -> None: response = self._request("DELETE", self._url(remote_path), timeout=(10, 60), retries=2) diff --git a/core/twitch_api.py b/core/twitch_api.py index d640c26..213dde6 100644 --- a/core/twitch_api.py +++ b/core/twitch_api.py @@ -242,6 +242,8 @@ def _drain_http_error(exc: urllib.error.HTTPError, path: str) -> None: The status code has already decided the outcome; a body that cannot be read (connection reset, truncated chunk) changes nothing about it. + Debug, not warning: the caller already logs the HTTP status at + warning/info level, and a second line per failed request adds nothing. """ try: exc.read() diff --git a/core/twitch_eventsub.py b/core/twitch_eventsub.py index 6515cf3..db6f148 100644 --- a/core/twitch_eventsub.py +++ b/core/twitch_eventsub.py @@ -225,7 +225,8 @@ def _helix_post(path: str, payload: dict[str, Any]) -> dict | None: try: detail = exc.read().decode("utf-8", errors="replace")[:400] except (OSError, http.client.HTTPException) as read_exc: - # The HTTP status below is the result; the body is only detail. + # The HTTP status below is the result and is logged as a warning + # on the next line; the unreadable body is only detail (debug). logger.debug("EventSub Helix POST %s: error body unreadable: %s", path, read_exc) logger.warning("EventSub Helix POST %s failed: HTTP %s %s", path, exc.code, detail) return None diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index c1839ff..173c8b6 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -25,7 +25,7 @@ from core import config # noqa: E402 from core import chat_chunker, live_watch_diag, redis_client, tube_auth # noqa: E402 from core import twitch_api, twitch_eventsub # noqa: E402 -from core.storage import http as storage_http # noqa: E402 +from core.storage import base as storage_base # noqa: E402 from core.storage.providers import google_drive as google_module # noqa: E402 from core.storage.providers.google_drive import GoogleDriveProvider # noqa: E402 @@ -34,15 +34,15 @@ class StorageResponseCloseTests(unittest.TestCase): def test_close_failure_is_logged_with_provider(self): response = mock.Mock() response.close.side_effect = ConnectionResetError("reset by peer") - with self.assertLogs("storage.http", level="DEBUG") as logs: - storage_http.close_quietly(response, provider="dropbox") + with self.assertLogs("storage.base", level="DEBUG") as logs: + storage_base.close_quietly(response, provider="dropbox") self.assertTrue(any("dropbox" in line and "reset by peer" in line for line in logs.output)) def test_unexpected_close_error_is_not_swallowed(self): response = mock.Mock() response.close.side_effect = RuntimeError("programming error") with self.assertRaises(RuntimeError): - storage_http.close_quietly(response, provider="dropbox") + storage_base.close_quietly(response, provider="dropbox") def test_google_drive_401_retry_survives_a_failing_close(self): class Unauthorized: @@ -68,7 +68,7 @@ def fake_request(method, url, **kwargs): return Unauthorized() with mock.patch.object(google_module.http, "request", fake_request), \ - self.assertLogs("storage.http", level="DEBUG"): + self.assertLogs("storage.base", level="DEBUG"): response = provider._api("GET", "/files") self.assertEqual(response.status_code, 401) self.assertEqual(len(requests), 2) From a4cf2af8c1c13e53b5e07ce77e582ab77e159b3b Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:34:48 +0100 Subject: [PATCH 10/12] fix(core,recorder): log the silent continue/return fallbacks outside the web UI Review of #14, point 3, for everything except webui.py: - recorder: a failed salvage move (OSError) and an unreadable recording pointer (OSError/ValueError, throttled) are warnings. The recording-loop checks (is_process_running, live presence, EventSub, Helix, chat overlay) keep their broad handler on purpose, so a failure there can never stop a recording, but now log a throttled warning with the reason in code. The timestamp-format loop logs at debug, because trying the next format is its job. - clip_engine/worker: ffprobe and frame extraction catch OSError/SubprocessError(/ValueError); recording pointers OSError/ValueError plus a dict check; a candidate file that vanishes between isfile() and getsize() is skipped explicitly; _is_direct_child fails closed with a warning. - chat_chunker: unreadable lines are counted, not swallowed (warning for more than one, debug for a single half-written tail line); non-object lines no longer reach data.get(). - redis_client: REDIS_ERRORS = (RedisError, OSError); ping, delete_state and queue length log a warning. - storage quotas (Dropbox, OneDrive, WebDAV): provider/parse errors are logged; the storage page still shows no quota. Co-Authored-By: Claude Opus 5.5 --- core/chat_chunker.py | 13 +++++++++- core/redis_client.py | 16 +++++++++--- core/storage/providers/dropbox.py | 6 ++++- core/storage/providers/onedrive.py | 6 ++++- core/storage/providers/webdav.py | 6 ++++- recorder/Streamrecorder.py | 40 ++++++++++++++++++++++++------ recorder/clip_engine.py | 33 ++++++++++++++++-------- recorder/worker.py | 7 ++++-- tests/test_error_handling.py | 22 ++++++++++++++++ 9 files changed, 121 insertions(+), 28 deletions(-) diff --git a/core/chat_chunker.py b/core/chat_chunker.py index 7ff7788..d4b43dd 100644 --- a/core/chat_chunker.py +++ b/core/chat_chunker.py @@ -50,6 +50,7 @@ def ensure_chat_chunks( chunks: Dict[int, List[Dict[str, Any]]] = {} total_messages = 0 + skipped = 0 try: with open(log_path, "r", encoding="utf-8", errors="ignore") as f: @@ -59,7 +60,10 @@ def ensure_chat_chunks( continue try: data = json.loads(line) - except Exception: + except ValueError: + data = None + if not isinstance(data, dict): + skipped += 1 continue rel = data.get("rel") @@ -79,6 +83,13 @@ def ensure_chat_chunks( chunks[chunk_idx].append(data) total_messages += 1 + # One bad line is normally the half-written tail of a log that is still + # being appended to (debug); more than one means a damaged file. + if skipped > 1: + logger.warning("chat chunks %s: %d unreadable lines skipped", log_path, skipped) + elif skipped: + logger.debug("chat chunks %s: 1 unreadable line skipped", log_path) + # Write out each chunk as compact static JSON max_chunk = max(chunks.keys()) if chunks else -1 total_chunks = max_chunk + 1 diff --git a/core/redis_client.py b/core/redis_client.py index 2aec383..f81b3e3 100644 --- a/core/redis_client.py +++ b/core/redis_client.py @@ -12,6 +12,13 @@ logger = logging.getLogger("streamvault.redis") +try: + from redis.exceptions import RedisError as _RedisError +except ImportError: # redis is optional; get_redis() then returns None + _RedisError = OSError +# What a call on a connected client raises when Redis goes away. +REDIS_ERRORS = (_RedisError, OSError) + _REDIS_CLIENT = None _REDIS_AVAILABLE = None _LAST_CONNECT_TRY = 0 @@ -93,7 +100,8 @@ def is_available() -> bool: return False try: return bool(r.ping()) - except Exception: + except REDIS_ERRORS as exc: + logger.warning("Redis ping failed: %s", exc) return False @@ -157,7 +165,8 @@ def delete_state(key: str) -> bool: full_key = f"{STATE_KEY_PREFIX}{key}" r.delete(full_key) return True - except Exception: + except REDIS_ERRORS as exc: + logger.warning("Redis state %s not deleted: %s", key, exc) return False @@ -244,7 +253,8 @@ def get_queue_len() -> int: return 0 try: return int(r.llen(TASK_QUEUE_KEY)) - except Exception: + except REDIS_ERRORS as exc: + logger.warning("Redis queue length unavailable: %s", exc) return 0 diff --git a/core/storage/providers/dropbox.py b/core/storage/providers/dropbox.py index e792d0e..97d428b 100644 --- a/core/storage/providers/dropbox.py +++ b/core/storage/providers/dropbox.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import json import os import posixpath @@ -13,6 +14,7 @@ from core.storage.base import ( close_quietly, + mask_error, ProgressCallback, ProviderField, RemoteObject, @@ -25,6 +27,7 @@ ) from core.storage import http +logger = logging.getLogger(__name__) API = "https://api.dropboxapi.com/2" CONTENT = "https://content.dropboxapi.com/2" AUTH_URL = "https://www.dropbox.com/oauth2/authorize" @@ -256,7 +259,8 @@ def quota(self) -> Optional[dict]: "usage": int(data.get("used") or 0), "limit": int((data.get("allocation") or {}).get("allocated") or 0), } - except Exception: + except (StorageError, AttributeError, TypeError, ValueError) as exc: + logger.warning("%s quota unavailable: %s", self.type_id, mask_error(exc)) return None def upload_file( diff --git a/core/storage/providers/onedrive.py b/core/storage/providers/onedrive.py index 8714332..3896671 100644 --- a/core/storage/providers/onedrive.py +++ b/core/storage/providers/onedrive.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import os import time import urllib.parse @@ -11,6 +12,7 @@ from core.storage.base import ( close_quietly, + mask_error, ProgressCallback, ProviderField, RemoteObject, @@ -26,6 +28,7 @@ ) from core.storage import http +logger = logging.getLogger(__name__) GRAPH = "https://graph.microsoft.com/v1.0" AUTH_URL = "https://login.microsoftonline.com/{tenant}/oauth2/v2.0/authorize" TOKEN_URL = "https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token" @@ -201,7 +204,8 @@ def quota(self) -> Optional[dict]: data = http.json_response(response, "OneDrive-Quota", provider=self.type_id) quota = data.get("quota") or {} return {"limit": int(quota.get("total") or 0), "usage": int(quota.get("used") or 0)} - except Exception: + except (StorageError, AttributeError, TypeError, ValueError) as exc: + logger.warning("%s quota unavailable: %s", self.type_id, mask_error(exc)) return None def upload_file( diff --git a/core/storage/providers/webdav.py b/core/storage/providers/webdav.py index ab6d2ba..9c48552 100644 --- a/core/storage/providers/webdav.py +++ b/core/storage/providers/webdav.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import os import urllib.parse import xml.etree.ElementTree as ET @@ -11,6 +12,7 @@ from core.storage.base import ( close_quietly, + mask_error, ProgressCallback, ProviderField, RemoteObject, @@ -25,6 +27,7 @@ ) from core.storage import http +logger = logging.getLogger(__name__) DAV = "{DAV:}" @@ -307,5 +310,6 @@ def quota(self) -> Optional[dict]: available_i = int(available) if available and available.isdigit() else 0 used_i = int(used) if used and used.isdigit() else 0 return {"limit": available_i + used_i, "usage": used_i} - except Exception: + except (StorageError, ET.ParseError, ValueError) as exc: + logger.warning("%s quota unavailable: %s", self.type_id, mask_error(exc)) return None diff --git a/recorder/Streamrecorder.py b/recorder/Streamrecorder.py index af4cdae..c1a27e9 100644 --- a/recorder/Streamrecorder.py +++ b/recorder/Streamrecorder.py @@ -525,7 +525,8 @@ def salvage_temp_recording(temp_path, output_file, reason="", source_channel=Non os.replace(temp_path, dest) final_path = dest break - except Exception: + except OSError as exc: + logger.warning("Salvage move to %s failed: %s", dest, exc) continue if final_path is None: # NEVER silently delete large .part files @@ -811,7 +812,9 @@ def load_current_recording(recordings_dir, source_channel=None): if info_source and info_source != source_channel: return None return data - except Exception: + except (OSError, ValueError) as exc: + throttled_log("warning", f"current-recording:{path}", + f"Recording pointer {path} unreadable: {exc}") return None def save_current_recording(recordings_dir, info, source_channel=None): @@ -977,7 +980,11 @@ def is_process_running(pid): return False except PermissionError: return True - except Exception: + except Exception as exc: + # Kept broad on purpose: this runs inside the recorder's lock logic and + # must never raise; an unusable PID (TypeError, OverflowError) means + # "not running". + logger.warning("PID %r not checkable, treating as not running: %s", pid, exc) return False return True @@ -1261,7 +1268,9 @@ def parse_recording_timestamp(filename, source_channel=None, archive_channel=Non for fmt in (RECORDING_TIMESTAMP_FORMAT, LEGACY_RECORDING_TIMESTAMP_FORMAT): try: return time.mktime(time.strptime(timestamp, fmt)) - except Exception: + except (ValueError, OverflowError) as exc: + # Trying the next format is the point of this loop; only debug. + logger.debug("Recording timestamp %s is not %s: %s", timestamp, fmt, exc) continue return None @@ -1902,7 +1911,11 @@ def should_emit_deferred_live_notify(source_channel, conf, recording_metadata=No """False when the source already flipped offline (leftover playlist).""" try: info = cheap_live_presence(source_channel, conf) - except Exception: + except Exception as exc: + # Kept broad on purpose: a presence-check failure must not stop the + # recording loop; the notification is sent as before. + throttled_log("warning", f"presence:{source_channel}", + f"Live presence check for {source_channel} failed, notifying anyway: {exc}") return True return info.get("live") is not False @@ -1915,7 +1928,11 @@ def source_eventsub_is_live(source_channel, conf=None): try: login = twitch_login_from_source(source_channel, conf) return twitch_eventsub.is_live(login) - except Exception: + except Exception as exc: + # Kept broad on purpose: called from the recording loop's wake check; + # "unknown" (None) is the safe answer. + throttled_log("warning", f"eventsub:{source_channel}", + f"EventSub state for {source_channel} unavailable: {exc}") return None @@ -1974,7 +1991,11 @@ def _helix_live_status(source_channel, conf=None): return None login = twitch_login_from_source(source_channel, conf) stream = twitch_api.get_stream(login, monitored=True) - except Exception: + except Exception as exc: + # Kept broad on purpose: Helix is a fallback inside the recording loop; + # "unknown" (None) keeps the loop running. + throttled_log("warning", f"helix-live:{source_channel}", + f"Helix live status for {source_channel} unavailable: {exc}") return None if not isinstance(stream, dict) or not stream.get("is_live"): return None @@ -2172,7 +2193,10 @@ def format_chat_content(chat_history): line = " " * len(prefix) + word lines.append(line) return "\n".join(lines[-maxlines:]) + "\n" - except Exception: + except Exception as exc: + # Kept broad on purpose: overlay text runs next to a live recording and + # an empty overlay is better than a crash. + throttled_log("warning", "chat-overlay", f"Chat overlay text not built: {exc}") return "" def write_chat_text(chat_history, output_path): diff --git a/recorder/clip_engine.py b/recorder/clip_engine.py index 05ac45a..537d2e8 100644 --- a/recorder/clip_engine.py +++ b/recorder/clip_engine.py @@ -363,7 +363,8 @@ def probe_duration_sec(path: str, http_headers: dict | None = None) -> float: text=True, ) return float((result.stdout or "").strip() or 0.0) - except Exception: + except (OSError, subprocess.SubprocessError, ValueError) as exc: + logger.warning("ffprobe duration of %s failed: %s", path, exc) return 0.0 @@ -743,7 +744,10 @@ def estimate_recording_duration(input_path: str) -> float: try: with open(meta_path, "r", encoding="utf-8") as handle: data = json.load(handle) or {} - except Exception: + except (OSError, ValueError) as exc: + logger.warning("Recording pointer %s unreadable: %s", meta_path, exc) + continue + if not isinstance(data, dict): continue names = [] for key in ("record_file", "filename"): @@ -1025,17 +1029,24 @@ def find_active_recording_file(archive_dir: str, source_channel: str = "") -> st try: with open(meta_path, "r", encoding="utf-8") as handle: data = json.load(handle) or {} - names = [] - for key in ("record_file", "filename"): - value = str(data.get(key) or "").strip() - if value and os.path.basename(value) == value and value not in names: - names.append(value) - for name in names: - for path in _recording_candidate_paths(archive_dir, name): + except (OSError, ValueError) as exc: + logger.warning("Recording pointer %s unreadable: %s", meta_path, exc) + continue + if not isinstance(data, dict): + continue + names = [] + for key in ("record_file", "filename"): + value = str(data.get(key) or "").strip() + if value and os.path.basename(value) == value and value not in names: + names.append(value) + for name in names: + for path in _recording_candidate_paths(archive_dir, name): + try: if os.path.isfile(path) and os.path.getsize(path) > 100_000: return path - except Exception: - continue + except OSError as exc: + # The file vanished between the two checks (finished/renamed). + logger.debug("Recording candidate %s gone: %s", path, exc) try: parts = [ os.path.join(archive_dir, name) diff --git a/recorder/worker.py b/recorder/worker.py index 1100a72..e217d48 100644 --- a/recorder/worker.py +++ b/recorder/worker.py @@ -511,7 +511,8 @@ def read_thumbnail_frame(input_path, timestamp, width=320, height=180, http_head ]) try: result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=60) - except Exception: + except (OSError, subprocess.SubprocessError) as exc: + logger.warning("Frame extraction from %s failed: %s", input_path, exc) return None expected = width * height * 3 if result.returncode != 0 or len(result.stdout) != expected: @@ -780,7 +781,9 @@ def _is_direct_child(parent, path): parent_real = os.path.realpath(parent) path_real = os.path.realpath(path) return os.path.dirname(path_real) == parent_real - except Exception: + except (OSError, TypeError, ValueError) as exc: + # Unresolvable means "not a direct child" (fail closed). + logger.warning("Path check %s under %s failed: %s", path, parent, exc) return False diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index 173c8b6..9375291 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -440,5 +440,27 @@ def test_watch_page_renders_when_channel_lookups_fail(self): + + +class RedisErrorLoggingTests(unittest.TestCase): + def test_delete_state_logs_redis_errors(self): + client = mock.Mock() + client.delete.side_effect = redis_client.REDIS_ERRORS[0]("connection lost") + with mock.patch.object(redis_client, "get_redis", return_value=client), \ + self.assertLogs("streamvault.redis", level="WARNING"): + self.assertFalse(redis_client.delete_state("k")) + + +class StorageQuotaErrorTests(unittest.TestCase): + def test_dropbox_quota_failure_is_logged_and_none(self): + from core.storage.base import StorageError + from core.storage.providers.dropbox import DropboxProvider + provider = DropboxProvider({}) + with mock.patch.object(provider, "_rpc", side_effect=StorageError("quota endpoint down")), \ + self.assertLogs(level="WARNING") as logs: + self.assertIsNone(provider.quota()) + self.assertTrue(any("quota endpoint down" in line for line in logs.output)) + + if __name__ == "__main__": unittest.main() From 8e71c384882f650adcd73731b284571f76ab4f4e Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:35:44 +0100 Subject: [PATCH 11/12] fix(webui): count unreadable chat lines, narrow the remaining constant returns Review of #14, point 3, for webui.py: - _ChatLineSkips replaces ten `except Exception: continue` around chat-log JSON (media title, jump markers, chat index, global search, chat replay, live chat). Lines that are not a JSON object are counted and reported once per read: debug for a single half-written tail line, a rate-limited warning for more. Non-object lines no longer reach data.get(), where build_chat_jump_markers used to raise AttributeError. - current_recording_info: OSError/ValueError, rate-limited warning. - constant returns: rate-limit storage import (ImportError), user-rights check (RuntimeError outside a request at debug, DB_ACCESS_ERRORS as warning), log tail/remux/status/transcript files (OSError/ValueError), ffprobe (OSError/SubprocessError/ValueError; the 1 s per-file probe rate-limited), URL parsing (ValueError, fail closed, debug), sort keys and chat timestamps (TypeError/ValueError, debug), thumbnail URL (BuildError/DB_ACCESS_ERRORS; RuntimeError outside an app context at debug). - _warn_db_fallback is now the general _warn_rate_limited. AST count of broad handlers without log or raise, outside tests/docs: origin/master 183 (pass 46, continue 15, constant return 31, other 91) -> 89 (other 89). Co-Authored-By: Claude Opus 5.5 --- WEBUI/webui.py | 203 ++++++++++++++++++++++++----------- tests/test_error_handling.py | 126 ++++++++++++++++++++-- 2 files changed, 259 insertions(+), 70 deletions(-) diff --git a/WEBUI/webui.py b/WEBUI/webui.py index 3bca7b7..0bf8902 100644 --- a/WEBUI/webui.py +++ b/WEBUI/webui.py @@ -59,6 +59,7 @@ ) from flask_limiter import Limiter from markupsafe import Markup, escape +from werkzeug.routing import BuildError from werkzeug.security import check_password_hash, generate_password_hash from werkzeug.utils import safe_join, secure_filename @@ -209,7 +210,8 @@ def _resolve_ratelimit_storage_uri(raw: str) -> str: scheme = uri.split("://", 1)[0].lower() if scheme not in SCHEMES: return "memory://" - except Exception: + except ImportError as exc: + logging.warning("Rate-limit storage schemes unavailable (%s); using memory://", exc) return "memory://" return uri @@ -1288,7 +1290,12 @@ def current_user_can_manage_users() -> bool: return True perms = ctx.get("permissions") or {} return bool(perms.get("can_manage_users")) - except (RuntimeError, Exception): + except RuntimeError as exc: + # Outside a request (CLI, background thread) there is no user and no right. + logging.debug("User-management check outside a request: %s", exc) + return False + except database.DB_ACCESS_ERRORS as exc: + logging.warning("User-management right unavailable (database error): %s", exc) return False @@ -1775,35 +1782,69 @@ def get_user_color(username): hash_val = ord(char) + ((hash_val << 5) - hash_val) return colors[abs(hash_val) % len(colors)] -DB_WARNING_INTERVAL_SEC = 60.0 -_DB_WARNING_LAST = {} -_DB_WARNING_SUPPRESSED = {} -_DB_WARNING_LOCK = threading.Lock() +RATE_LIMITED_WARNING_SEC = 60.0 +_RATE_LIMITED_LAST = {} +_RATE_LIMITED_SUPPRESSED = {} +_RATE_LIMITED_LOCK = threading.Lock() -def _warn_db_fallback(site, message, *args): - """Log a database fallback at most once per minute and call site. +def _warn_rate_limited(site, message, *args): + """Log a warning at most once per minute and call site. Access checks and media lookups run for every entry of a list and on every request. During a database outage one line per site and minute, with the number of repeats since the last one, says as much as thousands. """ now = time.monotonic() - with _DB_WARNING_LOCK: - last = _DB_WARNING_LAST.get(site) + with _RATE_LIMITED_LOCK: + last = _RATE_LIMITED_LAST.get(site) if last is None: - _DB_WARNING_SUPPRESSED.pop(site, None) - elif now - last < DB_WARNING_INTERVAL_SEC: - _DB_WARNING_SUPPRESSED[site] = _DB_WARNING_SUPPRESSED.get(site, 0) + 1 + _RATE_LIMITED_SUPPRESSED.pop(site, None) + elif now - last < RATE_LIMITED_WARNING_SEC: + _RATE_LIMITED_SUPPRESSED[site] = _RATE_LIMITED_SUPPRESSED.get(site, 0) + 1 return - _DB_WARNING_LAST[site] = now - repeats = _DB_WARNING_SUPPRESSED.pop(site, 0) + _RATE_LIMITED_LAST[site] = now + repeats = _RATE_LIMITED_SUPPRESSED.pop(site, 0) if repeats: message += " (%d more since the last warning)" args = args + (repeats,) logging.warning(message, *args) +class _ChatLineSkips: + """Count chat-log lines that are not a JSON object; report once per read. + + A single bad line is normally the half-written last line of a log the + recorder is still appending to, so it is logged at debug. More than one + means a damaged file: a warning, at most once per minute and reader. + """ + + def __init__(self, reader, source): + self.reader = reader + self.source = source + self.count = 0 + + def parse(self, line): + try: + data = json.loads(line) + except ValueError: + data = None + if isinstance(data, dict): + return data + self.count += 1 + return None + + def report(self): + if self.count == 1: + logging.debug("%s: 1 unreadable chat line skipped in %s", self.reader, self.source) + elif self.count > 1: + _warn_rate_limited( + "chat-lines:" + self.reader, + "%s: %d unreadable chat lines skipped in %s", + self.reader, self.count, self.source, + ) + + def normalize_channel_name(name): if not name: return None @@ -1967,7 +2008,8 @@ def tail_matching_lines(file_path, limit, query=None): if q and q not in line.lower(): continue dq.append(line.rstrip("\n")) - except Exception: + except OSError as exc: + logging.warning("Log file %s unreadable: %s", file_path, exc) return [] return list(dq) @@ -2446,7 +2488,9 @@ def get_streamer_archive_fallback_banner(streamer): def _ts(it): try: return float(it.get("recorded_ts") or it.get("mtime") or 0) - except Exception: + except (TypeError, ValueError) as exc: + # A malformed timestamp sorts as oldest; the card still renders. + logging.debug("Unusable timestamp on %s: %s", it.get("name"), exc) return 0.0 sorted_items = sorted(items, key=_ts, reverse=True) top = sorted_items[0] @@ -2454,7 +2498,12 @@ def _ts(it): thumb_dir, thumb_path, _ = resolve_thumb_paths(streamer, top["name"]) thumb_v = int(os.path.getmtime(thumb_path)) if thumb_path and os.path.exists(thumb_path) else int(top.get("mtime") or 0) return url_for("thumb_view", streamer=streamer, filename=top["name"], v=thumb_v) - except Exception: + except RuntimeError as exc: + # url_for outside an app context (background sync, CLI): no URL to build. + logging.debug("Preview thumbnail for %s outside a request: %s", streamer, exc) + return None + except (BuildError,) + database.DB_ACCESS_ERRORS as exc: + logging.warning("Preview thumbnail for %s unavailable: %s", streamer, exc) return None @@ -2848,7 +2897,9 @@ def twitch_helix_json(url): def is_twitch_badge_image_url(url): try: parsed = urllib.parse.urlparse(str(url or "")) - except Exception: + except ValueError as exc: + # urlparse rejects e.g. a malformed IPv6 host; such a URL is not a badge. + logging.debug("Badge URL rejected: %s", exc) return False return ( parsed.scheme == "https" @@ -3510,7 +3561,8 @@ def load_remux_status(): return None with open(path, "r", encoding="utf-8") as f: return json.load(f) - except Exception: + except (OSError, ValueError) as exc: + logging.warning("Remux status file unreadable: %s", exc) return None def storage_alert_path(): @@ -3523,7 +3575,9 @@ def load_storage_alert_status(): def is_allowed_url(url): try: parsed = urllib.parse.urlparse(url) - except Exception: + except ValueError as exc: + # Unparseable means not allowed (fail closed). + logging.debug("URL rejected: %s", exc) return False if parsed.scheme not in ("http", "https"): return False @@ -3562,7 +3616,9 @@ def is_safe_next(target): return False try: parsed = urllib.parse.urlsplit(target) - except Exception: + except ValueError as exc: + # Unparseable redirect target: refuse it (fail closed). + logging.debug("Redirect target rejected: %s", exc) return False if parsed.scheme or parsed.netloc: return False @@ -4078,7 +4134,8 @@ def get_media_duration(file_path): if result.returncode != 0: return None return float(result.stdout.strip()) - except Exception: + except (OSError, subprocess.SubprocessError, ValueError) as exc: + logging.warning("ffprobe duration failed: %s", exc) return None @@ -4089,7 +4146,8 @@ def load_transcript_status(transcript_path): try: with open(status_path, "r", encoding="utf-8") as f: return json.load(f) - except Exception: + except (OSError, ValueError) as exc: + logging.warning("Status file %s unreadable: %s", status_path, exc) return None @@ -4292,7 +4350,8 @@ def load_transcript(file_path): try: with open(transcript_path, "r", encoding="utf-8") as f: return f.read().strip() - except Exception: + except (OSError, ValueError) as exc: + logging.warning("Transcript %s unreadable: %s", transcript_path, exc) return None return None @@ -4471,7 +4530,9 @@ def get_duration(file_path): # Timeout auf 1s reduziert fuer bessere Performance result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, timeout=1) return float(result.stdout.strip()) - except Exception: + except (OSError, subprocess.SubprocessError, ValueError) as exc: + # Runs per file in listings with a 1 s budget: rate-limited. + _warn_rate_limited("ffprobe-quick-duration", "Quick duration probe failed: %s", exc) return 0.0 @@ -4802,13 +4863,13 @@ def can_view_streamer(streamer): if arch: candidates.add(arch.lower()) except database.DB_ACCESS_ERRORS as exc: - _warn_db_fallback("can_view_streamer:archive", + _warn_rate_limited("can_view_streamer:archive", "Access check for %s without archive name (database error): %s", streamer, exc) try: for s in related_archive_slugs(streamer) or []: candidates.add(str(s).strip().lower()) except database.DB_ACCESS_ERRORS as exc: - _warn_db_fallback("can_view_streamer:aliases", + _warn_rate_limited("can_view_streamer:aliases", "Access check for %s without archive aliases (database error): %s", streamer, exc) # 1. Check disabled_channels (Folder explicitly shut off / disabled) @@ -4865,7 +4926,7 @@ def abort_deleted_media(streamer, filename, file_id=None): try: slugs.update(related_archive_slugs(streamer) or []) except database.DB_ACCESS_ERRORS as exc: - _warn_db_fallback("abort_deleted_media:aliases", + _warn_rate_limited("abort_deleted_media:aliases", "Tombstone check for %s/%s without aliases (database error): %s", streamer, name, exc) primary = archive_name_for(streamer) or str(streamer or "").strip() if primary: @@ -4877,7 +4938,7 @@ def abort_deleted_media(streamer, filename, file_id=None): except database.DB_ACCESS_ERRORS as exc: # Same outcome as before: without the tombstone table the file # itself decides (resolve_media_path still 404s a missing file). - _warn_db_fallback("abort_deleted_media:lookup", + _warn_rate_limited("abort_deleted_media:lookup", "Tombstone check for %s/%s failed, serving: %s", streamer, name, exc) return if deleted: @@ -5801,15 +5862,15 @@ def read_media_sidecar_metadata(media_path): streamer_dir_cand = os.path.dirname(media_path) chat_cand = os.path.join(streamer_dir_cand, "chat_logs", f"{os.path.basename(base_cand)}.jsonl") if os.path.isfile(chat_cand): + skips = _ChatLineSkips("media title", chat_cand) with open(chat_cand, "r", encoding="utf-8", errors="ignore") as cf: for i, line in enumerate(cf): if i > 60: break - msg = "" - try: - msg = str(json.loads(line).get("msg") or "").strip() - except Exception: + row = skips.parse(line) + if row is None: continue + msg = str(row.get("msg") or "").strip() m_se = re.search(r"Streaming\s+([^:]+):\s*(.+)$", msg, re.IGNORECASE) if m_se: if not category: @@ -5819,6 +5880,7 @@ def read_media_sidecar_metadata(media_path): m_w = re.search(r"(?:Neuer Status|Stream-Titel):\s*(.+)$", msg, re.IGNORECASE) if m_w and not title: title = m_w.group(1).strip() + skips.report() except OSError as exc: logging.warning("Title from chat log of %s unavailable: %s", media_path, exc) @@ -6268,7 +6330,7 @@ def resolve_media_path(streamer, filename, route): try: candidates.extend(related_archive_slugs(streamer)) except database.DB_ACCESS_ERRORS as exc: - _warn_db_fallback("resolve_media_path:aliases", + _warn_rate_limited("resolve_media_path:aliases", "Media lookup for %s without archive aliases (database error): %s", streamer, exc) seen = set() files_dir = getattr(config, "FILES_DIR", None) or os.path.join(BASE_DIR, "files") @@ -6336,15 +6398,15 @@ def build_chat_jump_markers(log_path, duration_sec, bucket_sec=30, limit=24): bucket = max(10, int(bucket_sec or 30)) buckets = {} samples = 0 + skips = _ChatLineSkips("chat jump markers", log_path) try: with open(log_path, "r", encoding="utf-8", errors="ignore") as handle: for line in handle: line = line.strip() if not line: continue - try: - data = json.loads(line) - except Exception: + data = skips.parse(line) + if data is None: continue rel = data.get("rel") if rel is None: @@ -6383,6 +6445,7 @@ def build_chat_jump_markers(log_path, duration_sec, bucket_sec=30, limit=24): break except OSError: return [] + skips.report() if not buckets: return [] scored = [] @@ -6492,6 +6555,7 @@ def find_related_recordings(streamer, filename, meta=None, limit=4): def build_chat_index(log_path): index = [(0.0, 0)] line_count = 0 + skips = _ChatLineSkips("chat index", log_path) try: with open(log_path, "rb") as f: while True: @@ -6502,17 +6566,22 @@ def build_chat_index(log_path): line_count += 1 if line_count % CHAT_INDEX_STEP != 0: continue + data = skips.parse(line.decode("utf-8", errors="ignore")) + if data is None: + continue + rel = data.get("rel") + if rel is None: + continue try: - data = json.loads(line.decode("utf-8", errors="ignore")) - rel = data.get("rel") - if rel is None: - continue rel_val = float(rel) - index.append((rel_val, offset)) - except Exception: + except (TypeError, ValueError): + skips.count += 1 continue - except Exception: + index.append((rel_val, offset)) + except OSError as exc: + logging.warning("Chat index for %s not built: %s", log_path, exc) return None + skips.report() return index def get_chat_index(log_path): @@ -6614,7 +6683,10 @@ def chat_item_from_row(data, *, time_key): return None try: time_val = float(raw_time) - except Exception: + except (TypeError, ValueError): + # One chat row with a broken timestamp is dropped; logged per row at + # debug only, the reader reports damaged files as a whole. + logging.debug("Chat row without usable %s: %r", time_key, raw_time) return None user = str(data.get("user", "") or "").strip() msg = str(data.get("msg", "") or "") @@ -6874,14 +6946,14 @@ def global_search(query): if len(results["chat"]) >= GLOBAL_SEARCH_CHAT_LIMIT: break full_path = os.path.join(chat_dir, log_file) + skips = _ChatLineSkips("global search", full_path) try: with open(full_path, "r", encoding="utf-8", errors="ignore") as handle: for line in handle: if len(results["chat"]) >= GLOBAL_SEARCH_CHAT_LIMIT: break - try: - data = json.loads(line) - except Exception: + data = skips.parse(line) + if data is None: continue msg = str(data.get("msg", "") or "") user = str(data.get("user", "") or "") @@ -6896,6 +6968,7 @@ def global_search(query): "msg_html": highlight_chat_message(msg, needle), "link": chat_vod_url_for_log(streamer_name, log_file, rel), }) + skips.report() except OSError: continue return results @@ -6980,7 +7053,8 @@ def current_recording_info(streamer): if state_mtime >= best_mtime: best_mtime = state_mtime best = data - except Exception: + except (OSError, ValueError) as exc: + _warn_rate_limited("current_recording_info", "Recording state %s unreadable: %s", path, exc) continue _RECORDING_INFO_CACHE[streamer] = {"ts": now, "data": best} @@ -9354,7 +9428,9 @@ def _compute_dashboard_media_sync(streamers=None): def _vod_timestamp(x): try: return float(x.get("recorded_ts") or x.get("mtime") or 0) - except Exception: + except (TypeError, ValueError) as exc: + # A malformed timestamp sorts as oldest; the dashboard still renders. + logging.debug("Unusable VOD timestamp on %s: %s", x.get("name"), exc) return 0.0 recent_vods.sort(key=_vod_timestamp, reverse=True) @@ -11169,6 +11245,7 @@ def chat_vod(streamer, filename): items = [] seen = set() recent = {} + skips = _ChatLineSkips("chat replay", log_path) try: index = get_chat_index(log_path) if index: @@ -11181,16 +11258,16 @@ def chat_vod(streamer, filename): line = raw.decode("utf-8", errors="ignore").strip() if not line: continue - try: - data = json.loads(line) - except Exception: + data = skips.parse(line) + if data is None: continue rel = data.get("rel") if rel is None: continue try: rel_val = float(rel) - except Exception: + except (TypeError, ValueError): + skips.count += 1 continue if rel_val < start: continue @@ -11219,16 +11296,16 @@ def chat_vod(streamer, filename): line = line.strip() if not line: continue - try: - data = json.loads(line) - except Exception: + data = skips.parse(line) + if data is None: continue rel = data.get("rel") if rel is None: continue try: rel_val = float(rel) - except Exception: + except (TypeError, ValueError): + skips.count += 1 continue if rel_val < start: continue @@ -11253,6 +11330,7 @@ def chat_vod(streamer, filename): break except Exception: return Response("[]", 200, mimetype="application/json") + skips.report() return Response(json.dumps(items, ensure_ascii=False), 200, mimetype="application/json") @@ -11380,14 +11458,14 @@ def chat_live(streamer): except Exception: return Response("[]", 200, mimetype="application/json") lines = list(tail) + skips = _ChatLineSkips("live chat", log_path) for line in lines: line = line.strip() if not line: continue - try: - data = json.loads(line) - except Exception: + data = skips.parse(line) + if data is None: continue parsed = chat_item_from_row(data, time_key="t") if not parsed: @@ -11406,6 +11484,7 @@ def chat_live(streamer): items.append(item) if len(items) >= 200: break + skips.report() return Response(json.dumps(items, ensure_ascii=False), 200, mimetype="application/json") diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index 9375291..4008c30 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -220,8 +220,8 @@ def setUp(self): archive.mkdir() (archive / "vod.mp4").write_bytes(b"\0" * 32) self.locked = sqlite3.OperationalError("database is locked") - webui._DB_WARNING_LAST.clear() - self.addCleanup(webui._DB_WARNING_LAST.clear) + webui._RATE_LIMITED_LAST.clear() + self.addCleanup(webui._RATE_LIMITED_LAST.clear) def _viewer_request(self): ctx = self.webui.app.test_request_context("/") @@ -307,7 +307,7 @@ def setUp(self): archive = self.base / "streamer" archive.mkdir() (archive / "vod.mp4").write_bytes(b"\0" * 32) - self.webui._DB_WARNING_LAST.clear() + self.webui._RATE_LIMITED_LAST.clear() def test_malformed_channel_row_raises_value_error(self): from core import database @@ -329,7 +329,7 @@ def test_vod_access_bases_fall_back_when_data_dir_is_unreadable(self): def test_media_path_resolves_on_os_and_value_errors(self): for error in (PermissionError("data dir"), ValueError("bad record flag")): - self.webui._DB_WARNING_LAST.clear() + self.webui._RATE_LIMITED_LAST.clear() with self.subTest(error=type(error).__name__), \ mock.patch.object(self.webui, "related_archive_slugs", side_effect=error), \ self.assertLogs(level="WARNING"): @@ -365,8 +365,8 @@ def setUp(self): archive = self.base / "streamer" archive.mkdir() (archive / "vod.mp4").write_bytes(b"\0" * 32) - self.webui._DB_WARNING_LAST.clear() - self.addCleanup(self.webui._DB_WARNING_LAST.clear) + self.webui._RATE_LIMITED_LAST.clear() + self.addCleanup(self.webui._RATE_LIMITED_LAST.clear) self.locked = sqlite3.OperationalError("database is locked") def _locked_lines(self, logs): @@ -378,8 +378,8 @@ def test_media_lookup_warns_once_per_minute(self): for _ in range(5): self.assertIsNotNone(self.webui.resolve_media_path("streamer", "vod.mp4", "vod")) self.assertEqual(len(self._locked_lines(logs)), 1) - for site in list(self.webui._DB_WARNING_LAST): - self.webui._DB_WARNING_LAST[site] -= 61 + for site in list(self.webui._RATE_LIMITED_LAST): + self.webui._RATE_LIMITED_LAST[site] -= 61 self.webui.resolve_media_path("streamer", "vod.mp4", "vod") locked = self._locked_lines(logs) self.assertEqual(len(locked), 2) @@ -462,5 +462,115 @@ def test_dropbox_quota_failure_is_logged_and_none(self): self.assertTrue(any("quota endpoint down" in line for line in logs.output)) + +class ChatLogLineSkipTests(IsolatedBaseTest): + """Unreadable chat-log lines are skipped and reported once per read.""" + + def setUp(self): + super().setUp() + from WEBUI import webui + self.webui = webui + webui._RATE_LIMITED_LAST.clear() + self.addCleanup(webui._RATE_LIMITED_LAST.clear) + self.log = self.base / "chat.jsonl" + + def _write(self, *lines): + self.log.write_text("\n".join(lines) + "\n", encoding="utf-8") + + def test_jump_markers_skip_non_object_lines(self): + rows = [json.dumps({"rel": 40 + i * 0.1, "msg": "hype"}) for i in range(30)] + self._write("{broken", "[1, 2]", "42", *rows) + with self.assertLogs(level="WARNING") as logs: + markers = self.webui.build_chat_jump_markers(str(self.log), 600) + self.assertTrue(markers) + self.assertTrue(any("3 unreadable" in line for line in logs.output)) + + def test_single_partial_tail_line_is_only_debug(self): + self._write(json.dumps({"rel": 1.0, "msg": "a"}), '{"rel": 2.0, "ms') + with self.assertLogs(level="DEBUG") as logs: + self.webui.build_chat_jump_markers(str(self.log), 600) + self.assertEqual([line for line in logs.output if line.startswith("WARNING")], []) + self.assertTrue(any("1 unreadable" in line for line in logs.output)) + + def test_chunker_counts_skipped_lines(self): + self._write("{broken", "{also broken", json.dumps({"t": 1.0, "rel": 1.0, "msg": "hi"})) + with self.assertLogs("streamvault.chat_chunker", level="WARNING") as logs: + result = chat_chunker.ensure_chat_chunks(str(self.base), "chan", "vod.mp4", str(self.log)) + self.assertTrue(result.get("ok"), result) + self.assertTrue(any("2 unreadable" in line for line in logs.output)) + + + +class StateFileFallbackTests(IsolatedBaseTest): + def test_corrupt_remux_status_is_logged_and_none(self): + from WEBUI import webui + with open(webui.remux_status_path(), "w", encoding="utf-8") as handle: + handle.write("{broken") + with self.assertLogs(level="WARNING"): + self.assertIsNone(webui.load_remux_status()) + + def test_corrupt_current_recording_pointer_is_logged_and_none(self): + from recorder import Streamrecorder as recorder + recordings = self.base / "chan" + recordings.mkdir() + with open(recorder.current_recording_path(str(recordings)), "w", encoding="utf-8") as handle: + handle.write("{broken") + recorder.LOG_THROTTLE.clear() + with self.assertLogs(recorder.logger, level="WARNING"): + self.assertIsNone(recorder.load_current_recording(str(recordings))) + + def test_legacy_recording_timestamps_still_parse(self): + from recorder import Streamrecorder as recorder + stamp = time.strftime(recorder.LEGACY_RECORDING_TIMESTAMP_FORMAT, time.localtime(1_700_000_000)) + parsed = recorder.parse_recording_timestamp(f"chan_{stamp}.mp4") + self.assertEqual(parsed, time.mktime(time.strptime(stamp, recorder.LEGACY_RECORDING_TIMESTAMP_FORMAT))) + + def test_clip_engine_skips_corrupt_recording_pointer(self): + from recorder import clip_engine + archive = self.base / "chan" + archive.mkdir() + (archive / ".current_recording.json").write_text("{broken", encoding="utf-8") + with self.assertLogs(clip_engine.logger, level="WARNING"): + clip_engine.find_active_recording_file(str(archive)) + + + +class SidecarTitleFromChatTests(IsolatedBaseTest): + def setUp(self): + super().setUp() + from WEBUI import webui + self.webui = webui + archive = self.base / "streamer" + (archive / "chat_logs").mkdir(parents=True) + self.media = archive / "vod.mp4" + self.media.write_bytes(b"\0" * 32) + (archive / "vod.info.json").write_text(json.dumps({"platform": "twitch"}), encoding="utf-8") + self.chat = archive / "chat_logs" / "vod.jsonl" + + def test_title_comes_from_the_first_status_line_after_broken_lines(self): + self.chat.write_text("{broken\n[1, 2]\n" + json.dumps({"msg": "Streaming Just Chatting: Hello world"}) + "\n", + encoding="utf-8") + with self.assertLogs(level="DEBUG") as logs: + meta = self.webui.read_media_sidecar_metadata(str(self.media)) + self.assertEqual(meta["title"], "Hello world") + self.assertEqual(meta["category"], "Just Chatting") + self.assertTrue(any("2 unreadable" in line for line in logs.output)) + + def test_unreadable_chat_log_leaves_title_empty_and_logs(self): + self.chat.write_text(json.dumps({"msg": "Streaming X: Y"}) + "\n", encoding="utf-8") + real_open = open + + def guarded_open(path, *args, **kwargs): + if str(path).endswith(".jsonl"): + raise PermissionError("chat log locked") + return real_open(path, *args, **kwargs) + + with mock.patch("builtins.open", side_effect=guarded_open), \ + self.assertLogs(level="WARNING") as logs: + meta = self.webui.read_media_sidecar_metadata(str(self.media)) + self.assertEqual(meta["title"], "") + self.assertTrue(any("chat log locked" in line for line in logs.output)) + + if __name__ == "__main__": unittest.main() From 77cc24a54da9ecdffac934762cd27836d961b782 Mon Sep 17 00:00:00 2001 From: Blackwell Code <161588956+Blackwellcode@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:35:57 +0100 Subject: [PATCH 12/12] docs(changelog): state what the narrowed handlers change for operators Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 49c950b..3388bfd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,10 +96,21 @@ All notable changes to this project are documented here. The format follows ### Fixed -- Broad `except Exception: pass` handlers are gone from the whole code base - (database layer, recorder, worker, clip engine, web UI, storage providers - and API clients). The failures they hid are now caught by class and logged; - fallback values are unchanged. +- No broad exception handler that only passes, continues or returns a + constant without logging is left (database layer, recorder, worker, clip + engine, web UI, storage providers, API clients). The expected failures are + caught by class and logged, and fallback values are unchanged. Operator + impact: + - An unexpected error class (for example a `TypeError` from a code bug) is + no longer hidden. It now shows up as an HTTP 500 or a traceback in the + log. + - Database fallbacks catch `sqlite3.Error`, `OSError` and `ValueError` + (`core.database.DB_ACCESS_ERRORS`). Warnings from per-request lookups + are limited to one per minute and call site. + - The recording loop's own checks stay broad on purpose, so they can never + stop a recording, but they now log a throttled warning. + - Broad handlers that do something other than these three patterns still + exist and log nothing. - A deleted VOD or clip whose file was still on disk opened in the player: the tombstone check's `abort(404)` was swallowed by its own error handler. - A fresh database no longer logs a bogus `users.role` migration warning.