diff --git a/CHANGELOG.md b/CHANGELOG.md
index 70d9959..3388bfd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -96,9 +96,23 @@ 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).
+- 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.
 
 ## [0.1.0] - 2026-11-09
diff --git a/WEBUI/webui.py b/WEBUI/webui.py
index ef4c3ff..0bf8902 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
@@ -58,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
 
@@ -208,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
 
@@ -1287,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
 
 
@@ -1715,14 +1723,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 Exception:
-        pass
+    # 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
 
 
@@ -1776,6 +1782,69 @@ def get_user_color(username):
         hash_val = ord(char) + ((hash_val << 5) - hash_val)
     return colors[abs(hash_val) % len(colors)]
 
+RATE_LIMITED_WARNING_SEC = 60.0
+_RATE_LIMITED_LAST = {}
+_RATE_LIMITED_SUPPRESSED = {}
+_RATE_LIMITED_LOCK = threading.Lock()
+
+
+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 _RATE_LIMITED_LOCK:
+        last = _RATE_LIMITED_LAST.get(site)
+        if last is None:
+            _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
+        _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
@@ -1939,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)
 
@@ -2418,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]
@@ -2426,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
 
 
@@ -2580,16 +2657,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 database.DB_ACCESS_ERRORS 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 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()):
         p_data = profiles.get(f) or {}
@@ -2649,6 +2726,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 +2758,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 +2832,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
 
@@ -2810,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"
@@ -3472,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():
@@ -3485,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
@@ -3524,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
@@ -4040,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
 
 
@@ -4051,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
 
 
@@ -4254,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
 
@@ -4433,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
 
 
@@ -4756,23 +4855,22 @@ 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 database.DB_ACCESS_ERRORS as exc:
+        _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 Exception:
-        pass
+    except database.DB_ACCESS_ERRORS as exc:
+        _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)
     disabled_set = ctx.get("disabled_channels") or set()
@@ -4827,18 +4925,24 @@ def abort_deleted_media(streamer, filename, file_id=None):
     slugs = set()
     try:
         slugs.update(related_archive_slugs(streamer) or [])
-    except Exception:
-        pass
+    except database.DB_ACCESS_ERRORS as exc:
+        _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:
         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 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_rate_limited("abort_deleted_media:lookup",
+                          "Tombstone check for %s/%s failed, serving: %s", streamer, name, exc)
+        return
+    if deleted:
+        abort(404)
 
 
 def visible_streamers(streamers=None):
@@ -5758,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:
@@ -5776,8 +5880,9 @@ 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
+                    skips.report()
+            except OSError as exc:
+                logging.warning("Title from chat log of %s unavailable: %s", media_path, exc)
 
         return {
             "title": clean_metadata_label(title, 180),
@@ -6224,8 +6329,9 @@ def resolve_media_path(streamer, filename, route):
         candidates.append(streamer)
     try:
         candidates.extend(related_archive_slugs(streamer))
-    except Exception:
-        pass
+    except database.DB_ACCESS_ERRORS as exc:
+        _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")
     for cand in candidates:
@@ -6292,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:
@@ -6339,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 = []
@@ -6448,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:
@@ -6458,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):
@@ -6570,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 "")
@@ -6830,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 "")
@@ -6852,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
@@ -6936,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}
@@ -7070,8 +7188,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 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:
         tombstone_slugs.add(primary)
@@ -7102,8 +7220,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 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:
                 continue
@@ -8947,13 +9065,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 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 Exception:
-        pass
+    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)
 
 
@@ -9310,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)
@@ -11125,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:
@@ -11137,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
@@ -11175,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
@@ -11209,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")
 
 
@@ -11336,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:
@@ -11362,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")
 
 
@@ -11858,8 +11981,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 database.DB_ACCESS_ERRORS as exc:
+        logging.warning("Recording channel list unavailable: %s", exc)
     return logins
 
 
@@ -11993,7 +12116,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 database.DB_ACCESS_ERRORS as exc:
+        logging.warning("Live watch %s: channel config unavailable: %s", login, exc)
         archive_name = ""
     if not archive_name:
         try:
@@ -12001,8 +12125,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 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).
     # Optional ?proxy=1 for server HLS when embed is blocked or unsupported.
diff --git a/core/chat_chunker.py b/core/chat_chunker.py
index 7ad5bc9..d4b43dd 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}
@@ -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/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 79bfbb8..b0b3b66 100644
--- a/core/live_watch_diag.py
+++ b/core/live_watch_diag.py
@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import logging
 import os
 import re
 import subprocess
@@ -15,6 +16,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 +87,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 database.DB_ACCESS_ERRORS 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 +176,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 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
     if helix_info is None:
diff --git a/core/redis_client.py b/core/redis_client.py
index 28ea239..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
 
 
@@ -143,8 +151,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
 
 
@@ -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/base.py b/core/storage/base.py
index 23ca66f..28edc2c 100644
--- a/core/storage/base.py
+++ b/core/storage/base.py
@@ -8,14 +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 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")
 
 
 class StorageError(RuntimeError):
@@ -210,10 +215,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
@@ -372,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 c1a4e7d..405c2f4 100644
--- a/core/storage/http.py
+++ b/core/storage/http.py
@@ -11,8 +11,7 @@
 from typing import Any, Optional
 
 import requests
-
-from .base import StorageError
+from .base import StorageError, close_quietly
 
 logger = logging.getLogger("storage.http")
 
@@ -127,10 +126,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..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
@@ -12,6 +13,8 @@
 from flask_babel import gettext, lazy_gettext
 
 from core.storage.base import (
+    close_quietly,
+    mask_error,
     ProgressCallback,
     ProviderField,
     RemoteObject,
@@ -24,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"
@@ -152,10 +156,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
+                close_quietly(response, provider=self.type_id)
                 self._token(force=True)
                 continue
             return http.json_response(response, action, provider=self.type_id)
@@ -258,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(
@@ -369,10 +371,7 @@ def download_file(
                 os.fsync(handle.fileno())
             os.replace(tmp_path, dest_path)
         finally:
-            try:
-                response.close()
-            except Exception:
-                pass
+            close_quietly(response, provider=self.type_id)
             if os.path.exists(tmp_path):
                 try:
                     os.remove(tmp_path)
@@ -388,10 +387,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
+            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..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,10 +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:
-                try:
-                    response.close()
-                except Exception:
-                    pass
+                close_quietly(response, provider=self.type_id)
                 self._token(force=True)
                 continue
             return response
@@ -345,10 +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
-            try:
-                response.close()
-            except Exception:
-                pass
+            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 +405,7 @@ def upload_file(
             )
             if response.status_code != 401 or attempt == 1:
                 break
-            try:
-                response.close()
-            except Exception:
-                pass
+            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 +532,7 @@ def download_file(
                 os.fsync(handle.fileno())
             os.replace(tmp_path, dest_path)
         finally:
-            try:
-                response.close()
-            except Exception:
-                pass
+            close_quietly(response, provider=self.type_id)
             if os.path.exists(tmp_path):
                 try:
                     os.remove(tmp_path)
@@ -559,10 +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:
-            try:
-                response.close()
-            except Exception:
-                pass
+            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..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
@@ -10,6 +11,8 @@
 from flask_babel import lazy_gettext
 
 from core.storage.base import (
+    close_quietly,
+    mask_error,
     ProgressCallback,
     ProviderField,
     RemoteObject,
@@ -25,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"
@@ -153,10 +157,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
+                close_quietly(response, provider=self.type_id)
                 self._token(force=True)
                 continue
             return response
@@ -203,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(
@@ -352,10 +354,7 @@ def download_file(
                 os.fsync(handle.fileno())
             os.replace(tmp_path, dest_path)
         finally:
-            try:
-                response.close()
-            except Exception:
-                pass
+            close_quietly(response, provider=self.type_id)
             if os.path.exists(tmp_path):
                 try:
                     os.remove(tmp_path)
@@ -371,10 +370,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
+            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..8387632 100644
--- a/core/storage/providers/s3.py
+++ b/core/storage/providers/s3.py
@@ -9,6 +9,7 @@
 from flask_babel import gettext, lazy_gettext
 
 from core.storage.base import (
+    close_quietly,
     ProgressCallback,
     ProviderField,
     RemoteObject,
@@ -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
+            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..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
@@ -10,6 +11,8 @@
 from flask_babel import gettext, lazy_gettext
 
 from core.storage.base import (
+    close_quietly,
+    mask_error,
     ProgressCallback,
     ProviderField,
     RemoteObject,
@@ -24,6 +27,7 @@
 )
 from core.storage import http
 
+logger = logging.getLogger(__name__)
 DAV = "{DAV:}"
 
 
@@ -185,10 +189,7 @@ def download_file(
                 os.fsync(handle.fileno())
             os.replace(tmp_path, dest_path)
         finally:
-            try:
-                response.close()
-            except Exception:
-                pass
+            close_quietly(response, provider=self.type_id)
             if os.path.exists(tmp_path):
                 try:
                     os.remove(tmp_path)
@@ -209,10 +210,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
+            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)
@@ -312,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/core/storage/service.py b/core/storage/service.py
index 1207299..9dd14f0 100644
--- a/core/storage/service.py
+++ b/core/storage/service.py
@@ -1117,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()
@@ -1132,8 +1132,14 @@ 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 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
 
@@ -1396,8 +1402,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 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(
+                "provider %s: restore error not recorded: %s",
+                obj["provider_id"],
+                db_exc,
+            )
         raise
 
 
diff --git a/core/tube_auth.py b/core/tube_auth.py
index eec5512..3394a34 100644
--- a/core/tube_auth.py
+++ b/core/tube_auth.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import logging
 import os
 import re
 import secrets
@@ -20,6 +21,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 +96,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 database.DB_ACCESS_ERRORS as exc:
+        logger.warning("VOD access defaults unreadable, using seed list: %s", exc)
         rows = []
     if rows:
         return list(rows)
@@ -110,8 +114,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 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/core/twitch_api.py b/core/twitch_api.py
index 554a443..213dde6 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,20 @@ 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.
+    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()
+    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 +295,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 +310,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..db6f148 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,10 @@ 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 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
     except Exception as exc:
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
new file mode 100644
index 0000000..4008c30
--- /dev/null
+++ b/tests/test_error_handling.py
@@ -0,0 +1,576 @@
+"""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 io
+import json
+import os
+import sqlite3
+import time
+import unittest
+import urllib.error
+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 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 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
+
+
+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.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_base.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.base", 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")
+
+
+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())
+
+
+
+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))
+
+
+
+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")
+        webui._RATE_LIMITED_LAST.clear()
+        self.addCleanup(webui._RATE_LIMITED_LAST.clear)
+
+    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_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"))
+
+
+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._RATE_LIMITED_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._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"):
+                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._RATE_LIMITED_LAST.clear()
+        self.addCleanup(self.webui._RATE_LIMITED_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._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)
+        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)
+
+
+
+
+
+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))
+
+
+
+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()
diff --git a/tests/test_storage.py b/tests/test_storage.py
index a1298d7..1dac38c 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,54 @@ 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_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"))
+        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()