diff --git a/CLAUDE.md b/CLAUDE.md index c273a7e8..7a95b229 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -58,7 +58,7 @@ how it gets built. Concretely: `routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`, `downloads/`, `youtube/` (shared by routes + scheduler + worker) → flat core (`auth`, `quota`, `sysconfig`, `state`, `audit`, `notifications`, `userscope`, `realtime`, `ratelimit`, `progress`, -`security`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.** +`security`, `logs`, `logsink`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.** Frontend: `main.tsx` provider tree → `App.tsx` shell → `moduleRegistry` route-chunks → module components → the **platform layer `components/ui/`** (domain-free primitives: the popover engine and its menus, `Modal`/`Overlay`, `Button`/`form`, `DataTable`/`VirtualGrid`, `QueryState`, …). That diff --git a/backend/alembic/versions/0086_log_events.py b/backend/alembic/versions/0086_log_events.py new file mode 100644 index 00000000..258c1d39 --- /dev/null +++ b/backend/alembic/versions/0086_log_events.py @@ -0,0 +1,48 @@ +"""add log_events — stored log events for diagnostics (E22 S3) + +Revision ID: 0086_log_events +Revises: 0085_users_last_login_at +Create Date: 2026-09-17 + +Written by app/logsink.py's background writer (WARNING and above, request lines), pruned by the log_gc +job after `log_retention_days`. Sized against prod on 2026-09-16: ~2k rows on a normal day, ~14k on a +YouTube-RSS-outage day before flood control — a few MB over the 7-day window. `user_id` carries no +foreign key on purpose (see models.LogEvent). +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0086_log_events" +down_revision: str | None = "0085_users_last_login_at" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.create_table( + "log_events", + sa.Column("id", sa.BigInteger(), primary_key=True), + sa.Column("ts", sa.DateTime(timezone=True), nullable=False), + sa.Column("level", sa.String(10), nullable=False), + sa.Column("sev", sa.SmallInteger(), nullable=False), + sa.Column("source", sa.String(16), nullable=False), + sa.Column("logger", sa.String(128), nullable=False), + sa.Column("event", sa.String(64), nullable=True), + sa.Column("msg", sa.Text(), nullable=False), + sa.Column("attrs", postgresql.JSONB(), nullable=True), + sa.Column("exc", sa.Text(), nullable=True), + sa.Column("request_id", sa.String(64), nullable=True), + sa.Column("user_id", sa.Integer(), nullable=True), + sa.Column("debug_session_id", sa.String(64), nullable=True), + sa.Column("job_id", sa.Integer(), nullable=True), + sa.Column("version", sa.String(64), nullable=True), + ) + op.create_index("ix_log_events_ts", "log_events", ["ts"]) + op.create_index("ix_log_events_request_id", "log_events", ["request_id"]) + op.create_index("ix_log_events_user_ts", "log_events", ["user_id", "ts"]) + + +def downgrade() -> None: + op.drop_table("log_events") diff --git a/backend/app/README.md b/backend/app/README.md index db130a7e..bb8f8585 100644 --- a/backend/app/README.md +++ b/backend/app/README.md @@ -20,7 +20,8 @@ Three layers, by *who* runs the code — not by feature: run by both the admin route and the scheduler; `downloads/pipeline.py` is the worker's child pipeline. 3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth`, `sysconfig`, `quota`, `audit`, - `notifications`, `userscope`, `realtime`, `ratelimit`, `progress`, … are single-file primitives many + `notifications`, `userscope`, `realtime`, `ratelimit`, `progress`, `logs` (the one logging + configuration) and `logsink` (its `log_events` table), … are single-file primitives many layers import. They depend on nothing above them (a core module must never import a `routes/…` module). diff --git a/backend/app/auth.py b/backend/app/auth.py index e2f6390d..37515c70 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -12,7 +12,7 @@ from app import mailtext from app import sysconfig from app.config import settings from app.db import get_db -from app.models import Invite, User +from app.models import Invite, LogEvent, User from app.security import decrypt_optional from app.utils import canonical_email @@ -246,6 +246,9 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None: ) # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table. db.execute(delete(User).where(User.id == user_id)) + # Stored log events name the account by id with no foreign key to cascade from (models.LogEvent), + # so they go explicitly — they are request history, i.e. personal data. + db.execute(delete(LogEvent).where(LogEvent.user_id == user_id)) # Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the # canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose # account was just erased — and prod really does carry approved pairs for one mailbox. diff --git a/backend/app/config.py b/backend/app/config.py index 5ba056ef..24b58dc5 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -254,6 +254,13 @@ class Settings(BaseSettings): # "json" (one object per line — what Loki and the log_events sink ingest) or "text" (a readable # line for a local terminal). Redaction applies to both. log_format: LogFormat = "json" + # Also write events to the `log_events` table (app/logsink.py): WARNING and above, the request + # lines, and — once they exist — client batches. Off only where no table is wanted (the tests). + log_db_sink: bool = True + # Days a `log_events` row is kept (the privacy policy promises this), and how often the log_gc job + # enforces it — hourly, so a row never outlives the promise by more than an hour. + log_retention_days: int = 7 + log_gc_minutes: int = 60 # --- Plex integration (optional) --- # The Plex module plays the LOCAL physical media file directly (our own ffmpeg/direct-serve); diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py index 9c69c56e..0e1c70cd 100644 --- a/backend/app/downloads/runner.py +++ b/backend/app/downloads/runner.py @@ -18,6 +18,7 @@ import subprocess import sys from app import logs +from app import logsink from app.config import settings from app.db import SessionLocal from app.downloads import edit as editmod @@ -62,6 +63,7 @@ def main() -> None: # This whole process exists to run ONE job — tag every line it prints with it, so the child's # output joins up with the worker's and the API's view of the same download. logs.set_job(job_id) + logsink.start() # flushed at exit (atexit) — this process has no shutdown hook of its own with SessionLocal() as db: job = db.get(DownloadJob, job_id) diff --git a/backend/app/logs.py b/backend/app/logs.py index 708a7fb5..5beafb23 100644 --- a/backend/app/logs.py +++ b/backend/app/logs.py @@ -41,6 +41,7 @@ from datetime import datetime, timezone from typing import Any from app.config import settings +from app.ratelimit import RateLimiter, client_key # One level below DEBUG for the "Teljes / TRACE" diagnostics level (search strings, redacted request # bodies). Registered on import so `logging.getLogger(...).log(logs.TRACE, …)` and a `LOG_LEVEL=TRACE` @@ -157,6 +158,9 @@ _SECRET_PARAM_RE = re.compile( _EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") REDACTED = "[redacted]" +# `extra={PERSIST: False}` keeps a line out of the `log_events` table while stdout still gets it — +# how `log_requests` caps anonymous traffic per client (app/logsink.py reads it). +PERSIST = "persist" REDACTED_EMAIL = "[email]" @@ -189,7 +193,7 @@ def redact(value: Any, *, key: str = "") -> Any: # attribute and is merged into `attrs` (so `extra={"job_id": 7}` works as well as `event(...)`). _RECORD_STANDARD = frozenset( vars(logging.LogRecord("", 0, "", 0, "", None, None)).keys() - | {"message", "asctime", "taskName", "event", "attrs"} + | {"message", "asctime", "taskName", "event", "attrs", PERSIST} ) @@ -222,33 +226,43 @@ def _safe_message(record: logging.LogRecord) -> str: return f"{record.msg!r} (log formatting failed: {exc})" +def record_fields(record: logging.LogRecord) -> dict[str, Any]: + """The event as structured fields, REDACTED — the one definition both the stdout line and the + `log_events` row are built from, so the two sinks can never disagree about what is safe to keep. + Absent optional fields are left out; `ts` is an aware datetime.""" + fields: dict[str, Any] = { + "ts": datetime.fromtimestamp(record.created, timezone.utc), + "level": record.levelname, + "sev": SEVERITY.get(record.levelname, 9), + "source": _source, + "logger": record.name, + "msg": redact_text(_safe_message(record)), + } + code = getattr(record, "event", None) + if code: + fields["event"] = code + for name in _CONTEXT_VARS: + value = getattr(record, name, None) + if value is not None: + fields[name] = value + attrs = _record_attrs(record) + if attrs: + fields["attrs"] = redact(attrs) + if record.exc_info: + fields["exc"] = redact_text(logging.Formatter().formatException(record.exc_info)) + elif record.exc_text: + # A QueueHandler-style hand-off formats the traceback into exc_text and drops exc_info. + fields["exc"] = redact_text(record.exc_text) + fields["version"] = settings.app_version + return fields + + class JsonFormatter(logging.Formatter): - """One JSON object per line — the shape Loki/the `log_events` table both ingest.""" + """One JSON object per line — the shape Loki and the `log_events` table both ingest.""" def format(self, record: logging.LogRecord) -> str: - payload: dict[str, Any] = { - "ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat( - timespec="milliseconds" - ), - "level": record.levelname, - "sev": SEVERITY.get(record.levelname, 9), - "source": _source, - "logger": record.name, - "msg": redact_text(_safe_message(record)), - } - code = getattr(record, "event", None) - if code: - payload["event"] = code - for name in _CONTEXT_VARS: - value = getattr(record, name, None) - if value is not None: - payload[name] = value - attrs = _record_attrs(record) - if attrs: - payload["attrs"] = redact(attrs) - if record.exc_info: - payload["exc"] = redact_text(self.formatException(record.exc_info)) - payload["version"] = settings.app_version + payload = record_fields(record) + payload["ts"] = payload["ts"].isoformat(timespec="milliseconds") return json.dumps(payload, ensure_ascii=False, default=str) @@ -352,6 +366,11 @@ def current_level() -> str: return logging.getLevelName(logging.getLogger().getEffectiveLevel()) +def current_source() -> str: + """The process name `configure_logging` was given (api, worker, runner, alembic).""" + return _source + + # --- request logging ----------------------------------------------------------------------- # A client may supply its own request id so a browser log line and a server line join up (E22 S4). # Accepted only in this shape: it ends up in a log field, and an unbounded header would be a way to @@ -388,13 +407,15 @@ async def log_requests(request, call_next): # access-log entry at all. _request_log.exception( "request failed", - extra=event( - "http.request", - method=request.method, - path=_route_template(request), - duration_ms=round((time.perf_counter() - started) * 1000, 1), - user_id=getattr(request.state, "user_id", None), - ), + extra={ + **event( + "http.request", + method=request.method, + path=_route_template(request), + duration_ms=round((time.perf_counter() - started) * 1000, 1), + ), + **_request_owner(request, logging.ERROR), + }, ) _request_id.reset(token) raise @@ -413,20 +434,43 @@ async def log_requests(request, call_next): request.method, _route_template(request), status, - extra=event( - "http.request", - method=request.method, - path=_route_template(request), - status=status, - duration_ms=duration_ms, - user_id=getattr(request.state, "user_id", None), - ), + extra={ + **event( + "http.request", + method=request.method, + path=_route_template(request), + status=status, + duration_ms=duration_ms, + ), + **_request_owner(request, level), + }, ) response.headers["X-Request-Id"] = rid _request_id.reset(token) return response +# Anonymous request lines kept per client per minute. Signed-in traffic is not capped: it is a known +# account and exactly what an admin needs to see. Anonymous traffic is where the volume can come from +# nobody in particular — a scanner walking `/wp-login.php`, a bot on a share link — and each of its +# 404s is a WARNING that would otherwise land in `log_events` without limit. The client key is an +# in-memory counter key only; no address is ever written to a log line or a row. +ANONYMOUS_LINES_PER_MINUTE = 60 +_anonymous_lines = RateLimiter(ANONYMOUS_LINES_PER_MINUTE, 60.0) + + +def _request_owner(request, level: int) -> dict[str, Any]: + """`extra=` fields naming who the request belongs to, and whether its line may be stored. The + account comes from `request.state` (filled by the auth dependency), so it lands in the line's + top-level `user_id` — the correlation field — rather than in `attrs`.""" + user_id = getattr(request.state, "user_id", None) + if user_id is not None: + return {"user_id": user_id} + if level < logging.INFO: + return {} # a quiet line is never stored anyway — don't spend the client's allowance on it + return {} if _anonymous_lines.allow(client_key(request)) else {PERSIST: False} + + def _route_template(request) -> str: """The matched route's template (`/api/videos/{video_id}`), else the raw path for a 404 or a static file. Never the query string.""" diff --git a/backend/app/logsink.py b/backend/app/logsink.py new file mode 100644 index 00000000..4b2d22f7 --- /dev/null +++ b/backend/app/logsink.py @@ -0,0 +1,279 @@ +"""The `log_events` sink (E22 S3): the stored copy of the log the diagnostics console reads. + +stdout stays the complete log (docker keeps 3×50 MB of it); this table keeps the part worth querying +for `log_retention_days`: + + * every WARNING and above, + * the `http.request` line of every request at INFO or above (not the DEBUG-level health checks and + static files), + * nothing a caller marked `extra={logs.PERSIST: False}` — how the request log caps anonymous + traffic per client. + +What the table must never do is slow down or break the code that logs. So a line is redacted and +turned into a row on the CALLER's thread (the same `logs.record_fields` the stdout line is built +from), put on a bounded queue without blocking, and written in batches by one background thread +on its own connection — never inside a request's transaction. A full queue drops the row and +counts it; a failed write drops the batch and says so on stderr. Both are reported rather than +retried: a log that retries into a sick database is how a small outage becomes a big one. + +Flood control is the second half. On 2026-09-16 a YouTube RSS outage made prod log 11k near-identical +warnings in four hours. stdout keeps them all; the table keeps `REPEATS_PER_MINUTE` of each message +template per minute and one `log.repeats_suppressed` row saying how many more there were. + +Core layer: imports `logs`, `db`, `models`, `config` — nothing above them. +""" + +import atexit +import json +import logging +import queue +import sys +import threading +import time +from datetime import datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import delete, insert +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session + +from app import logs +from app.config import settings +from app.db import engine +from app.models import LogEvent + +QUEUE_MAX = 10_000 +BATCH_MAX = 500 +FLUSH_SECONDS = 1.0 +REPEATS_PER_MINUTE = 30 +REPEAT_WINDOW_SECONDS = 60.0 +# Distinct templates tracked per window. Past this many, a new template is stored without counting +# — the limiter must not become the unbounded thing it exists to prevent. +_REPEAT_KEYS_MAX = 2_000 + +# Column widths from models.LogEvent. A value past its column would fail the WHOLE batch it rides in. +_LIMITS = { + "level": 10, + "source": 16, + "logger": 128, + "event": 64, + "request_id": 64, + "debug_session_id": 64, + "version": 64, + "msg": 8_000, + "exc": 32_000, +} + +_log = logging.getLogger("siftlode.logsink") + + +def should_store(record: logging.LogRecord) -> bool: + """Whether this record belongs in `log_events` (see the module docstring).""" + if getattr(record, logs.PERSIST, True) is False: + return False + if record.levelno >= logging.WARNING: + return True + return getattr(record, "event", None) == "http.request" and record.levelno >= logging.INFO + + +def to_row(fields: dict[str, Any]) -> dict[str, Any]: + """`logs.record_fields` output → a `log_events` row: known columns only, text clipped to its + column, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout).""" + row = { + column: fields.get(column) for column in LogEvent.__table__.columns.keys() if column != "id" + } + for column, limit in _LIMITS.items(): + value = row.get(column) + if isinstance(value, str) and len(value) > limit: + row[column] = value[: limit - 1] + "…" + if row["attrs"] is not None: + row["attrs"] = json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str)) + return row + + +def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]: + """A row the sink writes about ITSELF. Built directly, not logged: logging it would route it back + through this sink (and the writer thread's own records are deliberately ignored).""" + return to_row( + { + "ts": datetime.now(timezone.utc), + "level": "WARNING", + "sev": logs.SEVERITY["WARNING"], + "source": logs.current_source(), + "logger": _log.name, + "event": code, + "msg": msg, + "attrs": attrs, + "version": settings.app_version, + } + ) + + +class _Repeats: + """Per-template admission for one minute at a time. Used by the writer thread only (unlocked).""" + + def __init__(self, per_window: int, window_seconds: float): + self.per_window = per_window + self.window = window_seconds + self.started = time.monotonic() + self.seen: dict[tuple[str, str, int], int] = {} + + def admit(self, key: tuple[str, str, int] | None) -> bool: + if key is None: + return True + count = self.seen.get(key) + if count is None and len(self.seen) >= _REPEAT_KEYS_MAX: + return True + self.seen[key] = (count or 0) + 1 + return self.seen[key] <= self.per_window + + def roll(self, now: float) -> list[dict[str, Any]]: + """Close the window if it is over: one notice per template that went past the limit.""" + if now - self.started < self.window: + return [] + notices = [ + _notice( + "log.repeats_suppressed", + f"{count - self.per_window} more like this were not stored", + logger=logger, + template=logs.redact_text(template), + suppressed=count - self.per_window, + window_seconds=int(self.window), + ) + for (logger, template, _level), count in self.seen.items() + if count > self.per_window + ] + self.seen.clear() + self.started = now + return notices + + +class LogSink(logging.Handler): + """The handler plus its writer. `start()` runs the writer thread; a test drives `write_pending()` + directly instead, on its own thread, with no timing involved.""" + + def __init__(self, bind: Engine = engine): + super().__init__(level=logging.NOTSET) + # Correlation fields are contextvars: capture them now, on the thread that logged. + self.addFilter(logs.ContextFilter()) + self._bind = bind + self._queue: queue.Queue[tuple[tuple[str, str, int] | None, dict[str, Any]]] = queue.Queue( + QUEUE_MAX + ) + self._dropped = 0 + self._dropped_lock = threading.Lock() + self._repeats = _Repeats(REPEATS_PER_MINUTE, REPEAT_WINDOW_SECONDS) + self._stopping = threading.Event() + self._writer: threading.Thread | None = None + + # --- caller side --------------------------------------------------------------------------- + def emit(self, record: logging.LogRecord) -> None: + if threading.current_thread() is self._writer: + return # the writer's own trouble (a driver warning mid-insert) must not feed itself + if not should_store(record): + return + try: + row = to_row(logs.record_fields(record)) + except Exception: # noqa: BLE001 — a record we cannot turn into a row is reported, not raised + self.handleError(record) + return + # Request lines are one per request by nature; only the repeated SAME message is a flood. + key = ( + None + if row["event"] == "http.request" + else (record.name, str(record.msg), record.levelno) + ) + try: + self._queue.put_nowait((key, row)) + except queue.Full: + with self._dropped_lock: + self._dropped += 1 + + # --- writer side --------------------------------------------------------------------------- + def write_pending(self, wait: float = 0.0) -> int: + """Write one batch: whatever is queued (up to BATCH_MAX), waiting up to `wait` seconds for the + first item. Returns the number of rows written, notices included.""" + rows: list[dict[str, Any]] = [] + try: + item = self._queue.get(timeout=wait) if wait > 0 else self._queue.get_nowait() + except queue.Empty: + item = None + while item is not None: + key, row = item + if self._repeats.admit(key): + rows.append(row) + if len(rows) >= BATCH_MAX: + break + try: + item = self._queue.get_nowait() + except queue.Empty: + item = None + rows.extend(self._repeats.roll(time.monotonic())) + with self._dropped_lock: + dropped, self._dropped = self._dropped, 0 + if dropped: + rows.append( + _notice("log.sink.dropped", f"{dropped} log events were dropped", dropped=dropped) + ) + if not rows: + return 0 + try: + with self._bind.begin() as conn: + conn.execute(insert(LogEvent), rows) + except Exception as exc: # noqa: BLE001 — the database is down or refused the batch + print( + f"log sink: could not store {len(rows)} log events: {logs.redact_text(str(exc))[:300]}", + file=sys.stderr, + ) + return 0 + return len(rows) + + def _run(self) -> None: + while not self._stopping.is_set(): + self.write_pending(wait=FLUSH_SECONDS) + while self.write_pending(): # drain what arrived before the stop + pass + + def start(self) -> None: + self._writer = threading.Thread(target=self._run, name="log-sink", daemon=True) + self._writer.start() + + def stop(self, timeout: float = 5.0) -> None: + self._stopping.set() + if self._writer is not None: + self._writer.join(timeout) + + +_sink: LogSink | None = None + + +def start() -> None: + """Attach the sink to the root logger and start its writer. Call after `logs.configure_logging` + (which replaces the root handlers) and once the schema exists. A no-op when `LOG_DB_SINK` is off + or it is already running.""" + global _sink + if _sink is not None or not settings.log_db_sink: + return + _sink = LogSink() + logging.getLogger().addHandler(_sink) + _sink.start() + # A short-lived process (the download child) exits without a lifespan: flush on the way out. + atexit.register(stop) + + +def stop() -> None: + """Detach the sink and write what is still queued.""" + global _sink + if _sink is None: + return + logging.getLogger().removeHandler(_sink) + _sink.stop() + _sink = None + + +def gc(db: Session, retention_days: int) -> int: + """Stage the deletion of rows older than `retention_days` (at least one day — stored logs always + expire). The caller commits. Returns the number of rows removed.""" + cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days)) + result = db.execute(delete(LogEvent).where(LogEvent.ts < cutoff)) + return result.rowcount or 0 diff --git a/backend/app/main.py b/backend/app/main.py index 8a1423dd..dbebb992 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -23,6 +23,7 @@ from fastapi.staticfiles import StaticFiles from sqlalchemy.exc import DataError from starlette.middleware.sessions import SessionMiddleware +from app import logsink from app import state from app import sysconfig from app.config import settings @@ -60,6 +61,8 @@ from app.scheduler import shutdown_scheduler, start_scheduler @asynccontextmanager async def lifespan(app: FastAPI): log.info("Siftlode starting up") + # The stored copy of the log (log_events). The entrypoint has already migrated the schema. + logsink.start() # First-run: if the instance isn't configured yet, mint a fresh one-time setup token and print # the wizard URL so the operator can open it (the app runs in setup mode until they finish). with SessionLocal() as db: @@ -107,6 +110,7 @@ async def lifespan(app: FastAPI): # Kills a running subtitle extraction: its worker would otherwise hold interpreter exit for # the rest of the read (see plex_subtitles.shutdown). plex_subtitles.shutdown() + logsink.stop() # last: write what the shutdown itself just logged # How often idle HLS sessions are reaped (see plex/stream.py `reap_idle`). diff --git a/backend/app/models.py b/backend/app/models.py index 94203c9b..f414f92a 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -12,6 +12,7 @@ from sqlalchemy import ( Index, Integer, JSON, + SmallInteger, String, Text, UniqueConstraint, @@ -543,6 +544,36 @@ class AuditLog(Base): ) +class LogEvent(Base): + """One stored log event (E22 S3) — the `log_events` table the diagnostics console reads. Rows are + written by app/logsink.py's background writer, never inside a request's transaction, already + redacted (the same `logs.record_fields` the stdout line is built from). + + `user_id` has NO foreign key on purpose: the writer runs behind the event, so a line about an + account can arrive after that account was purged, and a constraint would make it fail the whole + batch it travels in. `auth.purge_user` deletes an account's rows explicitly instead, and the + retention job (`log_retention_days`) bounds whatever a late write leaves behind.""" + + __tablename__ = "log_events" + __table_args__ = (Index("ix_log_events_user_ts", "user_id", "ts"),) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + level: Mapped[str] = mapped_column(String(10)) + sev: Mapped[int] = mapped_column(SmallInteger) + source: Mapped[str] = mapped_column(String(16)) + logger: Mapped[str] = mapped_column(String(128)) + event: Mapped[str | None] = mapped_column(String(64)) + msg: Mapped[str] = mapped_column(Text) + attrs: Mapped[dict | None] = mapped_column(JSONB) + exc: Mapped[str | None] = mapped_column(Text) + request_id: Mapped[str | None] = mapped_column(String(64), index=True) + user_id: Mapped[int | None] = mapped_column(Integer) + debug_session_id: Mapped[str | None] = mapped_column(String(64)) + job_id: Mapped[int | None] = mapped_column(Integer) + version: Mapped[str | None] = mapped_column(String(64)) + + class AppState(Base): """Single-row global app state (admin-controlled).""" diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 04373bb9..08683592 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -10,7 +10,7 @@ from typing import Callable from apscheduler.schedulers.background import BackgroundScheduler from sqlalchemy import select -from app import audit, progress, quota +from app import audit, logsink, progress, quota from app.audit import AuditAction from app.config import settings from app.db import SessionLocal @@ -72,6 +72,7 @@ JOB_INTERVALS: dict[str, int] = { "plex_watch_reconcile": settings.plex_watch_reconcile_interval_min, "tmdb_enrich": settings.tmdb_enrich_interval_min, "audit_gc": settings.audit_gc_minutes, + "log_gc": settings.log_gc_minutes, } # Sane bounds for an admin-set interval (minutes). @@ -381,6 +382,19 @@ def _audit_gc_job() -> None: _job("audit_gc", work) +def _log_gc_job() -> None: + # Prune log_events rows older than log_retention_days — the limit the privacy policy states, so + # this runs hourly rather than daily. Pure DB, no quota. + def work(db): + from app import sysconfig + + reaped = logsink.gc(db, sysconfig.get_int(db, "log_retention_days")) + db.commit() + return {"reaped": reaped} + + _job("log_gc", work) + + # job_id -> wrapper. The single source of truth for which jobs exist and how to run one, # shared by start_scheduler (recurring registration) and trigger_job (manual "run now"). JOB_FUNCS: dict[str, Callable[[], None]] = { @@ -400,6 +414,7 @@ JOB_FUNCS: dict[str, Callable[[], None]] = { "plex_watch_reconcile": _plex_watch_reconcile_job, "tmdb_enrich": _tmdb_enrich_job, "audit_gc": _audit_gc_job, + "log_gc": _log_gc_job, } diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index cbad249d..e64af62e 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -7,12 +7,13 @@ public reads. import logging import threading +from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed from sqlalchemy import select from sqlalchemy.orm import Session -from app import progress, quota, sysconfig +from app import logs, progress, quota, sysconfig log = logging.getLogger("siftlode.sync") from app.db import SessionLocal @@ -32,6 +33,8 @@ from app.youtube.rss import RssError, fetch_channel_feed # RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded. RSS_POLL_WORKERS = 16 +# Failed channels named in a poll's summary warning; the rest are counted by reason. +RSS_FAILURE_EXAMPLES = 5 def token_users(db: Session) -> list[User]: @@ -76,6 +79,27 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # outage from one permanently dead feed id — which is the whole difference on an instance with # a single subscription, where BOTH render as 100% failed. first_error: str | None = None + # What failed, as counts per reason (`http_404`, `transport`, `apply`, …) plus a few examples — + # the ONE warning a poll writes. One line per failed channel was what turned YouTube's RSS outage + # of 2026-09-16 into 11k warnings in four hours; each channel's own line is DEBUG now. + reasons: Counter[str] = Counter() + examples: list[str] = [] + + def fail(reason: str, detail: str, *, unexpected: str | None = None) -> None: + """Count a channel that did not get polled. `unexpected` is the message for a failure that + is a bug on our side: its FIRST occurrence per reason gets a traceback at ERROR, the rest + DEBUG — a read-only database fails every channel the same way, and 979 identical tracebacks + tell nobody anything the first one didn't. Call it from inside the `except` block.""" + nonlocal failed, first_error + if unexpected is not None: + level = logging.ERROR if reasons[reason] == 0 else logging.DEBUG + log.log(level, unexpected, exc_info=True) + failed += 1 + reasons[reason] += 1 + first_error = first_error or detail + if len(examples) < RSS_FAILURE_EXAMPLES: + examples.append(detail) + # Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as # each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a # plain str) to the workers; the ORM instance is touched only here on the main thread. @@ -90,16 +114,15 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where it # was (stamping it made a permanently-broken feed look freshly polled). The shared # counter/progress tail lives after the statement, so the two outcomes can stay two - # clauses: an RssError is EXPECTED (404, timeout) and gets one warning line; anything - # else is a bug on our side and earns a traceback. + # clauses: an RssError is EXPECTED (404, timeout) and is only counted; anything else is a + # bug on our side and earns a traceback (the first of its kind per poll). except RssError as exc: - log.warning("RSS poll skipped for channel %s: %s", channel.id, exc) - failed += 1 - first_error = first_error or f"{channel.id}: {exc}" + log.debug("RSS poll skipped for channel %s: %s", channel.id, exc) + fail(exc.reason, f"{channel.id}: {exc}") except Exception as exc: - log.exception("RSS fetch failed for channel %s", channel.id) - failed += 1 - first_error = first_error or f"{channel.id}: {exc}" + fail( + "error", f"{channel.id}: {exc}", unexpected=f"RSS fetch failed for {channel.id}" + ) else: try: new += apply_rss_feed(db, channel, entries) @@ -107,14 +130,27 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # A write failure (read-only DB, failover) leaves the channel just as un-polled # as an unreachable feed — count it, or a wholly broken run reports "failed=0". db.rollback() - failed += 1 - first_error = first_error or f"{channel.id} (apply): {exc}" - log.exception("RSS apply failed for channel %s", channel.id) + fail( + "apply", + f"{channel.id} (apply): {exc}", + unexpected=f"RSS apply failed for {channel.id}", + ) # One place that advances the counter, whatever happened above. done += 1 progress.report(done, total, "rss_poll") if failed: - log.warning("RSS poll: %d/%d channels could not be polled", failed, total) + log.warning( + "RSS poll: %d/%d channels could not be polled", + failed, + total, + extra=logs.event( + "sync.rss.failed", + failed=failed, + total=total, + reasons=dict(reasons), + examples=examples, + ), + ) if total and failed == total: # Nothing was polled at all — an egress outage (the tinyproxy-after-reboot incident), a # dead DNS, a read-only database. Returning normally would record the run as `ok` with the diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index b9cc7417..20549187 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -161,6 +161,9 @@ SPECS: tuple[ConfigSpec, ...] = ( # Runtime log level for the API process. Applied without a restart — see routes/config.py and # main.py's lifespan. ConfigSpec("log_level", "choice", "diagnostics", "log_level"), + # Days the log_events table keeps a row (the log_gc job prunes). No "0 = forever": stored logs + # carry account-linked request history, and the privacy policy names a limit. + ConfigSpec("log_retention_days", "int", "diagnostics", "log_retention_days", min=1, max=90), # --- Plex integration (optional; local-file playback + Plex metadata) --- ConfigSpec("plex_enabled", "bool", "plex", "plex_enabled"), ConfigSpec("plex_server_url", "str", "plex", "plex_server_url"), diff --git a/backend/app/worker.py b/backend/app/worker.py index a90b5fd9..cecc2ac6 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -29,6 +29,7 @@ from datetime import datetime, timezone from sqlalchemy import text from app import logs +from app import logsink from app import sysconfig from app.config import settings from app.db import SessionLocal @@ -519,6 +520,10 @@ def main() -> None: if _stop.is_set(): return # asked to shut down while waiting — that IS a clean exit raise SystemExit(1) + # After the wait, not before: the DB is reachable now. The wait checks `download_jobs`, not + # `log_events`, so on the deploy that first adds the table the worker may start before the API + # has migrated — those few seconds of batches fail and say so on stderr; nothing else breaks. + logsink.start() _recover_orphans() n = max(1, settings.download_worker_concurrency) log.info("Download worker started (concurrency=%d, root=%s).", n, settings.download_root) diff --git a/backend/app/youtube/rss.py b/backend/app/youtube/rss.py index 5ae93f74..e4624e76 100644 --- a/backend/app/youtube/rss.py +++ b/backend/app/youtube/rss.py @@ -14,7 +14,14 @@ class RssError(Exception): Deliberately NOT the same as an empty feed: a failed poll used to return `[]`, which the caller stamped as a successful `last_rss_at` — so a channel whose feed 404s permanently - (deleted, renamed id) looked like it was being polled fine forever, and nothing surfaced it.""" + (deleted, renamed id) looked like it was being polled fine forever, and nothing surfaced it. + + `reason` is a short stable code (`http_404`, `transport`) so a poll over many channels can report + WHAT failed as counts instead of one line per channel.""" + + def __init__(self, message: str, reason: str): + super().__init__(message) + self.reason = reason def fetch_channel_feed(channel_id: str) -> list[dict]: @@ -24,9 +31,12 @@ def fetch_channel_feed(channel_id: str) -> list[dict]: try: resp = httpx.get(url, timeout=20.0, headers={"User-Agent": "Siftlode/1.0"}) except httpx.HTTPError as exc: - raise RssError(f"RSS fetch failed for {channel_id}: {exc}") from exc + raise RssError(f"RSS fetch failed for {channel_id}: {exc}", "transport") from exc if resp.status_code != 200: - raise RssError(f"RSS fetch for {channel_id} returned HTTP {resp.status_code}") + raise RssError( + f"RSS fetch for {channel_id} returned HTTP {resp.status_code}", + f"http_{resp.status_code}", + ) parsed = feedparser.parse(resp.content) out: list[dict] = [] diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 0dde097b..10c811af 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -57,6 +57,10 @@ def _db_reachable() -> bool: _DB_REACHABLE = _db_reachable() +# No background log writer in tests: every TestClient would otherwise start one on app startup and +# write into tables the fixtures truncate between tests. The sink's own tests drive it explicitly. +settings.log_db_sink = False + @pytest.fixture(scope="session") def _schema(): diff --git a/backend/tests/test_logs.py b/backend/tests/test_logs.py index 62c79e54..e135acfe 100644 --- a/backend/tests/test_logs.py +++ b/backend/tests/test_logs.py @@ -13,7 +13,7 @@ import json import logging import pytest -from fastapi import FastAPI +from fastapi import FastAPI, Request from fastapi.testclient import TestClient from app import logs @@ -27,11 +27,13 @@ class _Sink(logging.Handler): def __init__(self): super().__init__(level=logging.NOTSET) self.lines: list[str] = [] + self.records: list[logging.LogRecord] = [] self.setFormatter(logs.JsonFormatter()) self.addFilter(logs.ContextFilter()) def emit(self, record): self.lines.append(self.format(record)) + self.records.append(record) def objects(self) -> list[dict]: return [json.loads(line) for line in self.lines] @@ -220,6 +222,40 @@ def test_the_request_line_has_the_route_template_and_no_query_string(sink): assert isinstance(payload["attrs"]["duration_ms"], float) +def test_anonymous_request_lines_are_capped_per_client_for_storage_but_still_logged(sink): + # A scanner's 404s are WARNINGs, and WARNINGs are stored. Past the cap they stay on stdout but are + # marked not-for-storage; a signed-in request is never capped. No address appears in any line. + logs._anonymous_lines.reset() + client = _client() + for _ in range(logs.ANONYMOUS_LINES_PER_MINUTE + 3): + client.get("/wp-login.php") + lines = sink.objects()[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :] + assert len(lines) == logs.ANONYMOUS_LINES_PER_MINUTE + 3 # stdout keeps every one + records = sink.records[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :] + stored = [getattr(r, logs.PERSIST, True) for r in records] + assert stored.count(False) == 3 and all(stored[: logs.ANONYMOUS_LINES_PER_MINUTE]) + assert "testclient" not in "".join(sink.lines) # the client key never reaches the line + logs._anonymous_lines.reset() + + +def test_a_signed_in_request_is_never_capped_and_names_its_account(sink): + logs._anonymous_lines.reset() + app = FastAPI() + app.middleware("http")(logs.log_requests) + + @app.get("/api/me") + def _me(request: Request): + request.state.user_id = 42 # what the auth dependency does + return {} + + client = TestClient(app) + for _ in range(logs.ANONYMOUS_LINES_PER_MINUTE + 3): + client.get("/api/me") + records = sink.records[-(logs.ANONYMOUS_LINES_PER_MINUTE + 3) :] + assert all(getattr(r, logs.PERSIST, True) for r in records) + assert sink.objects()[-1]["user_id"] == 42 and "user_id" not in sink.objects()[-1]["attrs"] + + def test_the_request_id_is_echoed_and_shared_by_every_line_of_that_request(sink): client = _client() r = client.get("/api/videos/abc123") diff --git a/backend/tests/test_logsink.py b/backend/tests/test_logsink.py new file mode 100644 index 00000000..7dc13ddc --- /dev/null +++ b/backend/tests/test_logsink.py @@ -0,0 +1,175 @@ +"""E22 S3 — the `log_events` sink: what reaches the table, in what shape, and what it refuses to do. + +The sink is driven synchronously (`write_pending()` on the test's own thread) so nothing here waits on +a timer. It is attached to a private logger rather than the root, for the same reason test_logs.py's +capture handler is: a `configure_logging` call elsewhere replaces the root handlers. +DB lane (`db` fixture): the rows are read back from a real Postgres. +""" + +import logging +import threading +from datetime import datetime, timedelta, timezone + +import pytest +from fastapi import BackgroundTasks +from sqlalchemy import create_engine, select + +from app import logs, logsink, scheduler +from app.auth import purge_user +from app.db import engine +from app.models import LogEvent + + +@pytest.fixture +def stored(db): + """A LogSink on the logger `siftlode.sinktest`, and a reader for the rows it wrote.""" + sink = logsink.LogSink(engine) + logger = logging.getLogger("siftlode.sinktest") + logger.addHandler(sink) + logger.setLevel(logs.TRACE) + + def rows() -> list[LogEvent]: + sink.write_pending() + db.expire_all() + return list(db.execute(select(LogEvent).order_by(LogEvent.id)).scalars()) + + try: + yield sink, logger, rows + finally: + logger.removeHandler(sink) + + +def test_warnings_and_request_lines_are_stored_and_everything_else_is_not(stored): + _sink, log, rows = stored + log.info("an ordinary info line") + log.debug("GET /healthz 200", extra=logs.event("http.request", path="/healthz", status=200)) + log.info("GET /api/feed 200", extra=logs.event("http.request", path="/api/feed", status=200)) + log.warning("scanner", extra={**logs.event("http.request", status=404), logs.PERSIST: False}) + log.warning("disk nearly full") + assert [r.msg for r in rows()] == ["GET /api/feed 200", "disk nearly full"] + + +def test_a_row_is_redacted_exactly_like_the_stdout_line(stored): + _sink, log, rows = stored + try: + raise ValueError("reset link https://x.test/#reset=abc123") + except ValueError: + log.exception( + "sign-in failed for someone@example.com", + extra=logs.event("auth.failed", password="hunter2", reason="bad_password"), + ) + (row,) = rows() + assert row.level == "ERROR" and row.sev == logs.SEVERITY["ERROR"] + assert row.event == "auth.failed" + assert "someone@example.com" not in row.msg and logs.REDACTED_EMAIL in row.msg + assert row.attrs == {"password": logs.REDACTED, "reason": "bad_password"} + assert "abc123" not in row.exc and "ValueError" in row.exc + assert row.source == logs.current_source() and row.version + + +def test_correlation_is_captured_when_the_line_is_logged_not_when_it_is_written(stored): + _sink, log, rows = stored + with logs.bind(job_id=7, request_id="req-0123456789"): + log.warning("inside the job") + # The writer runs later, outside the bind — the row still belongs to job 7. + (row,) = rows() + assert (row.job_id, row.request_id) == (7, "req-0123456789") + + +def test_a_flood_of_one_message_is_stored_up_to_the_limit_and_then_summarised(stored): + sink, log, rows = stored + limit = logsink.REPEATS_PER_MINUTE + for i in range(limit + 70): + log.warning("RSS poll skipped for channel %s: HTTP 404", f"UC{i}") + log.warning("a different message") + assert len(rows()) == limit + 1 # the flood capped; the other message untouched + sink._repeats.started -= logsink.REPEAT_WINDOW_SECONDS # the minute is over + notice = rows()[-1] + assert notice.event == "log.repeats_suppressed" + assert notice.attrs["suppressed"] == 70 + assert notice.attrs["template"] == "RSS poll skipped for channel %s: HTTP 404" + + +def test_request_lines_are_never_treated_as_a_flood(stored): + _sink, log, rows = stored + for _ in range(logsink.REPEATS_PER_MINUTE + 5): + log.info("%s %s %s", "GET", "/api/feed", 200, extra=logs.event("http.request")) + assert len(rows()) == logsink.REPEATS_PER_MINUTE + 5 + + +def test_a_full_queue_drops_and_says_how_many(db, monkeypatch): + monkeypatch.setattr(logsink, "QUEUE_MAX", 3) + sink = logsink.LogSink(engine) + log = logging.getLogger("siftlode.sinktest.small") + log.addHandler(sink) + try: + for i in range(5): + log.warning("message number %d", i) + finally: + log.removeHandler(sink) + assert sink.write_pending() == 4 + notice = db.execute(select(LogEvent).where(LogEvent.event == "log.sink.dropped")).scalar_one() + assert notice.attrs == {"dropped": 2} + + +def test_an_unreachable_database_costs_the_batch_not_the_caller(capsys): + dead = create_engine( + "postgresql+psycopg://x:y@127.0.0.1:1/none", connect_args={"connect_timeout": 1} + ) + sink = logsink.LogSink(dead) + log = logging.getLogger("siftlode.sinktest.dead") + log.addHandler(sink) + try: + log.error("this has nowhere to go") # must not raise + finally: + log.removeHandler(sink) + assert sink.write_pending() == 0 + assert "could not store 1 log events" in capsys.readouterr().err + + +def test_the_writer_thread_does_not_store_its_own_records(stored): + sink, log, rows = stored + sink._writer = threading.current_thread() # as if this line came from inside the writer + log.error("a driver warning raised while inserting") + sink._writer = None + assert rows() == [] + + +def test_gc_keeps_the_retention_window_and_never_keeps_forever(db): + now = datetime.now(timezone.utc) + + def add(age: timedelta) -> None: + db.add( + LogEvent(ts=now - age, level="WARNING", sev=13, source="api", logger="t", msg=str(age)) + ) + + for age in (timedelta(hours=1), timedelta(days=2), timedelta(days=8)): + add(age) + db.commit() + assert logsink.gc(db, 7) == 1 + assert logsink.gc(db, 0) == 1 # 0 is not "forever": stored logs always expire (floor: 1 day) + db.commit() + assert [r.msg for r in db.execute(select(LogEvent)).scalars()] == [str(timedelta(hours=1))] + + +def test_log_gc_is_a_scheduled_job(): + assert "log_gc" in scheduler.JOB_FUNCS and scheduler.JOB_INTERVALS["log_gc"] >= 1 + + +def test_deleting_an_account_deletes_its_stored_log_events(db, make_user): + gone, kept = make_user(), make_user() + for uid in (gone.id, kept.id, gone.id): + db.add( + LogEvent( + ts=datetime.now(timezone.utc), + level="INFO", + sev=9, + source="api", + logger="t", + msg="x", + user_id=uid, + ) + ) + db.commit() + purge_user(db, gone, BackgroundTasks()) + assert [r.user_id for r in db.execute(select(LogEvent)).scalars()] == [kept.id] diff --git a/backend/tests/test_sync_guards.py b/backend/tests/test_sync_guards.py index 4c89458a..ef688baf 100644 --- a/backend/tests/test_sync_guards.py +++ b/backend/tests/test_sync_guards.py @@ -1,6 +1,8 @@ """Guards that stop one bad YouTube answer from cascading: the subscription prune guard, the playlist-push circuit breaker, and RSS telling a failure apart from an empty feed.""" +import logging + import httpx import pytest @@ -120,7 +122,7 @@ class TestRssPollResult: # One dead channel among healthy ones: counted, but the run itself is still a run. def fetch(cid): if cid == "UC_dead": - raise rss.RssError("404") + raise rss.RssError("404", "http_404") return [{"id": "v"}] out = self._run( @@ -143,6 +145,54 @@ class TestRssPollResult: ) assert out == {"new": 1, "failed": 1}, "an unstored channel must not read as ok" + def test_many_failures_write_one_warning_with_counts_not_one_per_channel( + self, monkeypatch, caplog + ): + # YouTube's RSS outage of 2026-09-16: ~900 of 979 feeds failed on every 20-minute poll and + # each wrote its own WARNING — 11k lines in four hours, which the log_events table would + # have stored too. One summary per poll, per-channel detail at DEBUG. + channels = [f"UC{i}" for i in range(40)] + + def fetch(cid): + n = int(cid[2:]) + if n < 30: + raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404") + if n < 35: + raise rss.RssError(f"RSS fetch failed for {cid}: timed out", "transport") + return [{"id": "v"}] + + caplog.set_level(logging.DEBUG, logger="siftlode.sync") + out = self._run(monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=channels) + assert out == {"new": 5, "failed": 35} + warnings = [r for r in caplog.records if r.levelno >= logging.WARNING] + assert len(warnings) == 1 + assert warnings[0].event == "sync.rss.failed" + assert warnings[0].attrs["reasons"] == {"http_404": 30, "transport": 5} + assert len(warnings[0].attrs["examples"]) == runner.RSS_FAILURE_EXAMPLES + assert sum(r.levelno == logging.DEBUG for r in caplog.records) == 35 + + def test_a_failure_on_our_side_gets_one_traceback_per_poll_not_one_per_channel( + self, monkeypatch, caplog + ): + # A read-only database fails every channel's apply identically. The first gets a traceback + # at ERROR; repeating it for each of 979 channels would be the RSS flood all over again. + def apply_(db, ch, entries): + raise RuntimeError("read-only transaction") + + caplog.set_level(logging.DEBUG, logger="siftlode.sync") + with pytest.raises(RuntimeError): + self._run( + monkeypatch, + fetch=lambda cid: [{"id": "v"}], + apply_=apply_, + channels=("A", "B", "C"), + ) + errors = [r for r in caplog.records if r.levelno == logging.ERROR] + assert len(errors) == 1 and errors[0].exc_info is not None + assert ( + sum(r.levelno == logging.DEBUG and r.exc_info is not None for r in caplog.records) == 2 + ) + def test_a_quiet_hour_is_distinguishable_from_an_outage(self, monkeypatch): quiet = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0) assert quiet == {"new": 0, "failed": 0} # same "new", different "failed" — that's the point @@ -152,7 +202,7 @@ class TestRssPollResult: # here recorded the scheduler run as "ok", with the outage only visible inside the summary # text — a green pill for a run that did nothing at all. def boom(cid): - raise rss.RssError("Connection refused") + raise rss.RssError("Connection refused", "transport") with pytest.raises(RuntimeError, match="all 2 channels"): self._run(monkeypatch, fetch=boom, apply_=lambda db, ch, e: 1) @@ -162,7 +212,7 @@ class TestRssPollResult: # The card has to say which: `HTTP 404` = fix the subscription, `Connection refused` = # the egress is down. def dead(cid): - raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404") + raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404") with pytest.raises(RuntimeError, match="HTTP 404"): self._run(monkeypatch, fetch=dead, apply_=lambda db, ch, e: 1, channels=("UC_dead",)) diff --git a/frontend/src/i18n/locales/en/config.json b/frontend/src/i18n/locales/en/config.json index 80cd251c..abf6d62c 100644 --- a/frontend/src/i18n/locales/en/config.json +++ b/frontend/src/i18n/locales/en/config.json @@ -159,6 +159,10 @@ "label": "Log level", "hint": "How much the server writes to its log. INFO is the normal setting; DEBUG and TRACE are for reproducing a problem and are far more detailed. Takes effect immediately, no restart." }, + "log_retention_days": { + "label": "Log retention (days)", + "hint": "How long stored log events are kept before the log cleanup job deletes them. They include account-linked request history, so there is no \"keep forever\". Default 7." + }, "plex_enabled": { "label": "Enable Plex", "hint": "Expose the Plex library as a Feed source with a rich player. Requires the Plex server to be reachable from this host and its media on a local mount (see the path map)." diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index 2ff04775..bb17b854 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -74,6 +74,7 @@ "plex_watch_sync": "Pulls recent Plex watch changes (finished episodes, resume positions) into Siftlode and re-pushes any local watch changes that didn't reach Plex. The cheap two-way keep-in-sync between full reconciles. If it stops, watched/resume made in the Plex app takes longer to show here.", "plex_watch_reconcile": "A full watch-state reconcile that rescans every library — catches what the incremental pass can't, notably an episode un-marked as watched on the Plex side. Heavier, so it runs about once a day. If it stops, un-watches done in Plex aren't mirrored back.", "audit_gc": "Prunes admin audit-log entries older than the configured retention window (Configuration → Audit log). If it stops, the audit log just keeps growing — harmless but unbounded.", + "log_gc": "Deletes stored log events older than the retention window (Configuration → Diagnostics). Runs hourly so nothing outlives the promised limit by more than an hour.", "tmdb_enrich": "Enriches the mirrored Plex catalogue from TMDB (keywords, crew, budget/revenue, TMDb score, recommendations, artwork), matched by each title's canonical id and served from the app's own database. Optional — a no-op until a TMDB API key is set. If it stops, new Plex titles just miss their TMDB extras until it runs again." }, "jobs": { @@ -92,7 +93,8 @@ "plex_watch_sync": "Plex watch sync (incremental)", "plex_watch_reconcile": "Plex watch reconcile (full)", "tmdb_enrich": "TMDB enrichment (Plex)", - "audit_gc": "Audit-log cleanup" + "audit_gc": "Audit-log cleanup", + "log_gc": "Log cleanup" }, "queue": { "recentPending": "Channels to sync", diff --git a/frontend/src/i18n/locales/hu/config.json b/frontend/src/i18n/locales/hu/config.json index b14cb60c..c32c69b4 100644 --- a/frontend/src/i18n/locales/hu/config.json +++ b/frontend/src/i18n/locales/hu/config.json @@ -153,12 +153,16 @@ }, "audit_retention_days": { "label": "Megőrzés (nap)", - "hint": "Meddig maradnak meg az admin-naplóbejegyzések, mielőtt a napló-karbantartó feladat törli őket. 0 = örökre. Alapérték 180." + "hint": "Meddig maradnak meg az admin-naplóbejegyzések, mielőtt az auditnapló-takarító feladat törli őket. 0 = örökre. Alapérték 180." }, "log_level": { "label": "Naplózási szint", "hint": "Mennyit naplózzon a szerver. Az INFO a szokásos; a DEBUG és a TRACE hibakereséshez való, sokkal részletesebb. Azonnal érvénybe lép, újraindítás nélkül." }, + "log_retention_days": { + "label": "Naplómegőrzés (nap)", + "hint": "Meddig maradnak meg a tárolt naplóesemények, mielőtt a naplótakarító feladat törli őket. Fiókhoz kötött kéréstörténetet is tartalmaznak, ezért nincs „örökre” lehetőség. Alapérték 7." + }, "plex_enabled": { "label": "Plex engedélyezése", "hint": "A Plex-könyvtár megjelenítése forrásként, gazdag lejátszóval. Ehhez a Plex-szervernek elérhetőnek kell lennie erről a gépről, a média pedig helyi mounton (lásd az útvonal-leképezést)." diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index 800822a9..9de4ee15 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -74,6 +74,7 @@ "plex_watch_sync": "Áthozza a Plex-oldali friss megtekintés-változásokat (befejezett epizódok, folytatás-pozíciók) a Siftlode-ba, és felnyomja a Plexet el nem érő helyi változásokat. Ez az olcsó kétirányú szinkron a teljes reconcile-ok között. Ha leáll, a Plex-appban jelölt megtekintés/folytatás lassabban jelenik meg itt.", "plex_watch_reconcile": "Teljes megtekintés-állapot reconcile, ami minden könyvtárat újraszken­nel — elkapja, amit az inkrementális menet nem, főleg ha egy epizódról a Plex oldalán levették a megtekintve jelet. Nehezebb, ezért kb. naponta fut. Ha leáll, a Plexben végzett un-watch nem tükröződik vissza.", "audit_gc": "Törli a beállított megőrzési időnél (Konfiguráció → Auditnapló) régebbi admin-naplóbejegyzéseket. Ha leáll, a napló csak nő — ártalmatlan, de korlátlanul.", + "log_gc": "Törli a megőrzési időnél (Konfiguráció → Diagnosztika) régebbi tárolt naplóeseményeket. Óránként fut, így semmi sem marad meg egy óránál tovább az ígért határidőn túl.", "tmdb_enrich": "A tükrözött Plex-katalógust dúsítja a TMDB-ből (kulcsszavak, stáb, költségvetés/bevétel, TMDb-pontszám, ajánlások, artwork), a címek kanonikus azonosítója alapján párosítva, és az app saját adatbázisából szolgálja ki. Opcionális — TMDB API-kulcs nélkül nem csinál semmit. Ha leáll, az új Plex-címek csak a következő futásig maradnak TMDB-extrák nélkül." }, "jobs": { @@ -92,7 +93,8 @@ "plex_watch_sync": "Plex megtekintés-szinkron (inkrementális)", "plex_watch_reconcile": "Plex megtekintés-reconcile (teljes)", "tmdb_enrich": "TMDB-dúsítás (Plex)", - "audit_gc": "Napló-karbantartás" + "audit_gc": "Auditnapló-takarítás", + "log_gc": "Naplótakarítás" }, "queue": { "recentPending": "Szinkronra váró csatorna",