diff --git a/CLAUDE.md b/CLAUDE.md index 2acf4a0d..dba071cb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,7 +63,7 @@ how it gets built. Concretely: ## Architecture shape (don't fight it) `routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`, -`downloads/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, +`downloads/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`, `sysconfig`, `state`, `audit`, `notifications`, `userscope`, `ratelimit`, `bodylimit`, `progress`, `collation`, `security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.** Frontend: `main.tsx` provider tree → `App.tsx` shell (a shared `/watch` link boots from its own @@ -92,7 +92,7 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory | Upstream failure status | `422` with JSON detail | `502` (reserve 502 for raw byte-proxies) | | User-facing error | structured `detail` (`{code, reason}`) | a bare English string under a bilingual UI | | Mutation response | `{"ok": true, …}` (`{"deleted": id}`), test-held | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok` | -| Transaction commit | the handler commits; helpers `flush` (test-held list) | helpers that commit (only `quota` owns its session) | +| Transaction commit | the handler commits; helpers `flush` (test-held list) | helpers that commit (only `quota` and `jobruns` own their session) | | Admin knob storage | `sysconfig` (values) | env / `app_state` / `scheduler_settings` scatter | | Job status word | the `downloads/status.py` StrEnum | loose `"canceled"`/`"cancelled"` literals | | Current time | inline `datetime.now(timezone.utc)` | new `now_utc`/`_now` helpers | diff --git a/backend/alembic/versions/0109_job_runs.py b/backend/alembic/versions/0109_job_runs.py new file mode 100644 index 00000000..c642a720 --- /dev/null +++ b/backend/alembic/versions/0109_job_runs.py @@ -0,0 +1,139 @@ +"""Background runs get a durable history + +Revision ID: 0109_job_runs +Revises: 0108_folder_links +Create Date: 2026-10-17 + +What a scheduler job last did lived in the API process's memory, gone at every restart; the audit +log kept a row for a run only when it was manual, failed, or changed something. Now every run is a +`job_runs` row, written when it starts and when it ends. A run's quota spend points at it +(`quota_events.job_run_id`), and so do the lines it logged (`log_events.run_id`, no foreign key: the +log writer runs behind the event). + +Runs are not admin actions, so the audit log stops carrying them: the existing `job_run` audit rows +MOVE into `job_runs` (start = end = when the row was written, the only time it kept), and leave the +audit log. The downgrade moves every run back into the audit log as such a row, then drops the table. +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0109_job_runs" +down_revision: str | None = "0108_folder_links" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.create_table( + "job_runs", + sa.Column("id", sa.BigInteger(), primary_key=True), + sa.Column("job", sa.String(40), nullable=False), + sa.Column("trigger", sa.String(16), nullable=False), + sa.Column("actor_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL")), + sa.Column("target", sa.String(128)), + sa.Column("status", sa.String(16), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True)), + sa.Column("summary", sa.Text()), + sa.Column("result", postgresql.JSONB()), + sa.Column("error", sa.Text()), + sa.Column("failure", postgresql.JSONB()), + sa.Column("quota_units", sa.Integer(), nullable=False, server_default="0"), + sa.CheckConstraint( + "status IN ('running', 'ok', 'error', 'skipped', 'interrupted')", + name="ck_job_runs_status", + ), + sa.CheckConstraint( + "trigger IN ('scheduled', 'manual', 'ad_hoc')", name="ck_job_runs_trigger" + ), + sa.CheckConstraint( + "(status = 'running') = (finished_at IS NULL)", name="ck_job_runs_finished" + ), + ) + op.create_index("ix_job_runs_job_started", "job_runs", ["job", "started_at"]) + op.create_index("ix_job_runs_actor_id", "job_runs", ["actor_id"]) + op.create_index("ix_job_runs_started_at", "job_runs", ["started_at"]) + + op.add_column( + "quota_events", + sa.Column("job_run_id", sa.BigInteger(), sa.ForeignKey("job_runs.id", ondelete="SET NULL")), + ) + op.create_index("ix_quota_events_job_run_id", "quota_events", ["job_run_id"]) + + op.add_column("log_events", sa.Column("run_id", sa.BigInteger())) + op.create_index( + "ix_log_events_run", + "log_events", + ["run_id"], + postgresql_where=sa.text("run_id IS NOT NULL"), + ) + + # The audit trail's runs move over. A status the run table does not know (none was ever + # written) would fail the CHECK, so the move is limited to the three the scheduler wrote. + op.execute( + """ + INSERT INTO job_runs (job, trigger, actor_id, status, started_at, finished_at, summary, + result, error, failure) + SELECT target_id, + CASE WHEN actor_id IS NULL THEN 'scheduled' ELSE 'manual' END, + actor_id, + after::jsonb ->> 'status', + created_at, + created_at, + CASE WHEN after::jsonb ->> 'status' <> 'error' THEN after::jsonb ->> 'summary' END, + CASE WHEN after::jsonb ->> 'status' <> 'error' THEN after::jsonb -> 'result' END, + CASE WHEN after::jsonb ->> 'status' = 'error' THEN after::jsonb ->> 'summary' END, + CASE WHEN after::jsonb ->> 'status' = 'error' THEN after::jsonb -> 'result' END + FROM audit_log + WHERE action = 'job_run' + AND target_id IS NOT NULL + AND after::jsonb ->> 'status' IN ('ok', 'error', 'skipped') + ORDER BY id + """ + ) + op.execute( + """ + DELETE FROM audit_log + WHERE action = 'job_run' + AND target_id IS NOT NULL + AND after::jsonb ->> 'status' IN ('ok', 'error', 'skipped') + """ + ) + + +def downgrade() -> None: + op.execute( + """ + INSERT INTO audit_log (actor_id, action, target_type, target_id, summary, after, created_at) + SELECT actor_id, + 'job_run', + 'job', + job, + left(job || ': ' || status + || coalesce(' — ' || CASE WHEN status = 'error' THEN error ELSE summary END, ''), + 255), + (jsonb_build_object( + 'status', status, + 'summary', CASE WHEN status = 'error' THEN error ELSE summary END) + || CASE + WHEN status = 'error' AND failure IS NOT NULL + THEN jsonb_build_object('result', failure) + WHEN status <> 'error' AND result IS NOT NULL + THEN jsonb_build_object('result', result) + ELSE '{}'::jsonb + END)::json, + coalesce(finished_at, started_at) + FROM job_runs + ORDER BY id + """ + ) + op.drop_index("ix_log_events_run", table_name="log_events") + op.drop_column("log_events", "run_id") + op.drop_index("ix_quota_events_job_run_id", table_name="quota_events") + op.drop_column("quota_events", "job_run_id") + op.drop_index("ix_job_runs_started_at", table_name="job_runs") + op.drop_index("ix_job_runs_actor_id", table_name="job_runs") + op.drop_index("ix_job_runs_job_started", table_name="job_runs") + op.drop_table("job_runs") diff --git a/backend/app/README.md b/backend/app/README.md index 6ff33ab1..0dafee5a 100644 --- a/backend/app/README.md +++ b/backend/app/README.md @@ -21,7 +21,8 @@ Three layers, by *who* runs the code — not by feature: `issues/bundle.py` is the one definition of a bug report's files, for the zip export and the tracker. 3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth` (with `apitokens`, a machine's - folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`, + folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`, `jobruns` (every + background run's durable row, its words in `runstatus`), `notifications`, `userscope`, `ratelimit`, `progress`, `logs` (the one logging configuration), `logsink` (its `log_events` table), `diagnostics` (a user's consented log level), `debugsession` (a recording that tags its user's lines, E16) and `collation` (A→Z in the @@ -36,7 +37,7 @@ pre-abstract for a single caller; don't leave shared logic duplicated once there ## Transaction ownership This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush -(only `quota` owns its session)"** — so read it that way, not as an exception to it. The commit belongs +(only `quota` and `jobruns` own their session)"** — so read it that way, not as an exception to it. The commit belongs to the **operation owner**; anything called *inside* another transaction only stages its change (`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction. `audit.record()` is the reference: it adds the audit row and the caller commits it together with the @@ -45,7 +46,7 @@ mutation it records, so an audit entry can never exist without its change (or vi - The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes, - the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes. + `jobruns`' run rows — the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes. - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM attribute — see `service.decr_ref_count`. diff --git a/backend/app/audit.py b/backend/app/audit.py index d014d313..52349979 100644 --- a/backend/app/audit.py +++ b/backend/app/audit.py @@ -48,8 +48,6 @@ class AuditAction: MAINTENANCE_TUNE = "maintenance_tune" SYNC_PAUSE = "sync_pause" SYNC_RESUME = "sync_resume" - # Scheduler job outcome / run-summary (actor_id set = a manual "run now"; NULL = automatic) - JOB_RUN = "job_run" # Per-user download cookie/auth credential set or removed (the value is NEVER recorded) DOWNLOAD_CREDENTIAL_SET = "download_credential_set" DOWNLOAD_CREDENTIAL_DELETE = "download_credential_delete" diff --git a/backend/app/config.py b/backend/app/config.py index fd477645..f5b66200 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -268,6 +268,9 @@ class Settings(BaseSettings): # YouTube quota history (quota_events): days kept (0 = forever). A year, so the admin usage view's # longest window (365 days) stays whole; the same daily cleanup job prunes it. quota_retention_days: int = 365 + # Background-run history (job_runs): days kept (0 = forever). Every run is a row (~1–1.5k a day + # at the default intervals), so a month; the same daily cleanup job prunes it. + job_run_retention_days: int = 30 # On-disk layout: "plex" (Channel/Season YYYY/… [id].ext + .nfo/poster) or "flat". download_layout: DownloadLayout = "plex" # Default SponsorBlock (skip sponsor segments) for new custom profiles. diff --git a/backend/app/jobruns.py b/backend/app/jobruns.py new file mode 100644 index 00000000..993af617 --- /dev/null +++ b/backend/app/jobruns.py @@ -0,0 +1,104 @@ +"""The durable history of background runs — the `job_runs` table. + +A run is written twice: a `running` row when it starts, its outcome when it ends. Each write is its +OWN short session, committed at once, for the reason `quota.record_usage` gives: the run's own work +may roll back (its `except: db.rollback()`), and the history of what happened must not go with it — +nor may this bookkeeping commit the job's half-done session. + +Only the API process runs jobs (the worker never starts the scheduler), so at its startup nothing can +still be running: `interrupt_leftovers` closes what a stopped process left open. +""" + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import delete, func, select, update +from sqlalchemy.engine import CursorResult +from sqlalchemy.orm import Session + +from app.db import SessionLocal +from app.models import JobRun, QuotaEvent +from app.runstatus import RunStatus, RunTrigger + + +def start(job: str, trigger: RunTrigger, actor_id: int | None, target: str | None = None) -> int: + """Open a run's row; returns its id.""" + with SessionLocal() as s: + row = JobRun( + job=job, + trigger=trigger, + actor_id=actor_id, + target=target, + status=RunStatus.RUNNING, + started_at=datetime.now(timezone.utc), + ) + s.add(row) + s.commit() + return row.id + + +def finish( + run_id: int, + status: RunStatus, + *, + summary: str | None = None, + result: dict | None = None, + error: str | None = None, + failure: dict | None = None, +) -> None: + """Close a run with its outcome and what it spent. Conditional on the row still running: a row + the startup sweep already called interrupted is not rewritten by a thread that outlived it.""" + with SessionLocal() as s: + units = s.scalar( + select(func.coalesce(func.sum(QuotaEvent.units), 0)).where( + QuotaEvent.job_run_id == run_id + ) + ) + s.execute( + update(JobRun) + .where(JobRun.id == run_id, JobRun.status == RunStatus.RUNNING) + .values( + status=status, + finished_at=datetime.now(timezone.utc), + summary=summary, + result=result, + error=error, + failure=failure, + quota_units=units or 0, + ) + ) + s.commit() + + +def interrupt_leftovers() -> int: + """Every row a previous process left `running` → `interrupted`. Called once at startup, before + the scheduler can open a new run. Returns how many.""" + with SessionLocal() as s: + res = s.execute( + update(JobRun) + .where(JobRun.status == RunStatus.RUNNING) + .values(status=RunStatus.INTERRUPTED, finished_at=datetime.now(timezone.utc)) + ) + s.commit() + return res.rowcount if isinstance(res, CursorResult) else 0 + + +def last_runs(db: Session) -> dict[str, JobRun]: + """Each job's most recent run (any trigger), by job id.""" + rows = db.execute( + select(JobRun) + .distinct(JobRun.job) + .order_by(JobRun.job, JobRun.started_at.desc(), JobRun.id.desc()) + ).scalars() + return {r.job: r for r in rows} + + +def gc(db: Session, retention_days: int) -> int: + """Delete finished runs older than `retention_days` (0 = keep forever). A run still going is + never reaped. The caller commits.""" + if retention_days <= 0: + return 0 + cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) + res = db.execute( + delete(JobRun).where(JobRun.started_at < cutoff, JobRun.status != RunStatus.RUNNING) + ) + return res.rowcount if isinstance(res, CursorResult) else 0 diff --git a/backend/app/logs.py b/backend/app/logs.py index 62ff3609..fc6551bd 100644 --- a/backend/app/logs.py +++ b/backend/app/logs.py @@ -10,8 +10,8 @@ the logs were both unjoinable and leaky. `configure_logging()` replaces all four Event shape — the OpenTelemetry log data model, without the SDK; one JSON object per line: ts · level · sev · source · logger · event · msg · attrs plus whatever correlation the context carries (`request_id`, `user_id`, `debug_session_id`, -`job_id`) and the app version. That shape is what the admin log console filters on and what a Gitea -issue attachment will carry, so it is deliberately flat and stable. +`job_id`, `run_id`) and the app version. That shape is what the admin log console filters on and +what a Gitea issue attachment will carry, so it is deliberately flat and stable. Call sites stay ordinary stdlib logging: @@ -81,15 +81,23 @@ _debug_session: contextvars.ContextVar[str | None] = contextvars.ContextVar( "log_debug_session", default=None ) _job_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_job_id", default=None) +# The background run (app/jobruns.py) this context is doing; quota's spend is tagged with it too. +_run_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_run_id", default=None) _CONTEXT_VARS = { "request_id": _request_id, "user_id": _user_id, "debug_session_id": _debug_session, "job_id": _job_id, + "run_id": _run_id, } +def current_run_id() -> int | None: + """The background run this context is doing, if any — `bind(run_id=…)` sets it.""" + return _run_id.get() + + class _RequestScope: """What the auth dependency learns about a request, for every line logged after it. diff --git a/backend/app/logsink.py b/backend/app/logsink.py index c63cea8f..2310ebeb 100644 --- a/backend/app/logsink.py +++ b/backend/app/logsink.py @@ -362,6 +362,7 @@ def serialize(row: LogEvent) -> dict: "request_id": row.request_id, "user_id": row.user_id, "job_id": row.job_id, + "run_id": row.run_id, "debug_session_id": row.debug_session_id, "version": row.version, } diff --git a/backend/app/models.py b/backend/app/models.py index f7c4567c..880fa7de 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -24,6 +24,7 @@ from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR from sqlalchemy.orm import Mapped, mapped_column, relationship, validates from app.db import Base +from app.runstatus import RunStatus, RunTrigger, sql_in from app.utils import canonical_email # Every user-supplied display string in this schema shares one width, and it lives HERE so the @@ -563,6 +564,11 @@ class QuotaEvent(Base): created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True ) + # The background run that spent it (app/jobruns.py); NULL = a request's own spend. SET NULL: a + # reaped run leaves the spend in the budget's history. + job_run_id: Mapped[int | None] = mapped_column( + ForeignKey("job_runs.id", ondelete="SET NULL"), index=True + ) class AuditLog(Base): @@ -609,6 +615,8 @@ class LogEvent(Base): "debug_session_id", postgresql_where="debug_session_id IS NOT NULL", ), + # A run's lines are `WHERE run_id = …`; request lines carry none. + Index("ix_log_events_run", "run_id", postgresql_where="run_id IS NOT NULL"), ) id: Mapped[int] = mapped_column(BigInteger, primary_key=True) @@ -624,7 +632,11 @@ class LogEvent(Base): 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)) + # A DOWNLOAD job's id (download_jobs) — not a background run, which is `run_id`. job_id: Mapped[int | None] = mapped_column(Integer) + # The background run (job_runs) that logged it. No foreign key, as `user_id`: the writer runs + # behind the event, and a run reaped meanwhile must not fail the batch its lines travel in. + run_id: Mapped[int | None] = mapped_column(BigInteger) version: Mapped[str | None] = mapped_column(String(64)) @@ -700,6 +712,46 @@ class SchedulerSetting(Base): interval_minutes: Mapped[int] = mapped_column(Integer) +class JobRun(Base): + """One run of a background job — scheduled, an admin's "run now", or a user's ad-hoc work on one + target — written when it starts and again when it ends (app/jobruns.py), each in its own + committed session, so the history survives the job's own rollback and a restart. A row left + `running` by a process that died is set `interrupted` at the next startup. + + `quota_units` is what the run spent of the YouTube budget: the `quota_events` rows carrying its + id, summed when it ends. `summary` is the English one-line fallback; `result` / `failure` are the + same outcome as data, which the client names in the reader's language.""" + + __tablename__ = "job_runs" + __table_args__ = ( + Index("ix_job_runs_job_started", "job", "started_at"), + CheckConstraint(f"status IN ({sql_in(RunStatus)})", name="ck_job_runs_status"), + CheckConstraint(f"trigger IN ({sql_in(RunTrigger)})", name="ck_job_runs_trigger"), + # Only a run still going has no end. + CheckConstraint( + "(status = 'running') = (finished_at IS NULL)", name="ck_job_runs_finished" + ), + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + job: Mapped[str] = mapped_column(String(40)) + trigger: Mapped[str] = mapped_column(String(16)) + # Who started it: the admin of a manual run, the user of an ad-hoc one; NULL = the schedule. + actor_id: Mapped[int | None] = mapped_column( + ForeignKey("users.id", ondelete="SET NULL"), index=True + ) + # What an ad-hoc run worked on (a channel id); NULL for the recurring jobs. + target: Mapped[str | None] = mapped_column(String(128)) + status: Mapped[str] = mapped_column(String(16)) + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + summary: Mapped[str | None] = mapped_column(Text) + result: Mapped[dict | None] = mapped_column(JSONB) + error: Mapped[str | None] = mapped_column(Text) + failure: Mapped[dict | None] = mapped_column(JSONB) + quota_units: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + class SystemConfig(Base): """Admin-set overrides for operational config that otherwise comes from env/config defaults. Generic key/value store (one row per key); absent = use the env/config diff --git a/backend/app/quota.py b/backend/app/quota.py index c0ef1db5..6f2e5c81 100644 --- a/backend/app/quota.py +++ b/backend/app/quota.py @@ -16,7 +16,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.engine import CursorResult from sqlalchemy.orm import Session -from app import sysconfig +from app import logs, sysconfig from app.db import SessionLocal from app.models import ApiQuotaUsage, QuotaEvent @@ -204,6 +204,14 @@ def record_usage(units: int) -> None: ) with SessionLocal() as qs: qs.execute(stmt) - # Audit detail: who/what spent it (per-user attribution; NULL actor = system). - qs.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units)) + # Audit detail: who/what spent it (per-user attribution; NULL actor = system), and the + # background run it belongs to, so the run's row can say what it cost. + qs.add( + QuotaEvent( + user_id=_actor_id.get(), + action=_action.get(), + units=units, + job_run_id=logs.current_run_id(), + ) + ) qs.commit() diff --git a/backend/app/routes/admin_logs.py b/backend/app/routes/admin_logs.py index 8202cafd..60e2d50b 100644 --- a/backend/app/routes/admin_logs.py +++ b/backend/app/routes/admin_logs.py @@ -86,6 +86,7 @@ class LogFilters: event: str | None request_id: str | None job_id: int | None + run_id: int | None q: str | None since: datetime | None until: datetime | None @@ -113,6 +114,8 @@ class LogFilters: out.append(LogEvent.request_id == self.request_id) if self.job_id is not None: out.append(LogEvent.job_id == self.job_id) + if self.run_id is not None: + out.append(LogEvent.run_id == self.run_id) if self.since is not None: out.append(LogEvent.ts >= _utc(self.since)) if self.until is not None: @@ -140,11 +143,14 @@ def log_filters( ), request_id: str | None = Query(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$"), job_id: int | None = Query(default=None, ge=1), + run_id: int | None = Query(default=None, ge=1), q: str | None = Query(default=None, max_length=200), since: datetime | None = None, until: datetime | None = None, ) -> LogFilters: - return LogFilters(list(levels), list(sources), user, event, request_id, job_id, q, since, until) + return LogFilters( + list(levels), list(sources), user, event, request_id, job_id, run_id, q, since, until + ) @router.get("") diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index 579a691f..c61f3305 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -117,7 +117,7 @@ def set_maintenance_settings( @router.get("") def get_scheduler(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: - snap = scheduler_snapshot() + snap = scheduler_snapshot(db) # Work still queued (shared catalog, so these are instance-wide, not per-user). deep_requested = ( diff --git a/backend/app/runstatus.py b/backend/app/runstatus.py new file mode 100644 index 00000000..497a22b1 --- /dev/null +++ b/backend/app/runstatus.py @@ -0,0 +1,35 @@ +"""The words a background run's row is written with — the `job_runs` table's status and trigger. + +One source for the literals: the scheduler writes them, the startup sweep rewrites `running`, the +model's CHECK constraints are built from these sets, and the dashboard reads them back. A run's +status is never typed as a string anywhere else (the download pipeline's `cancelled`/`canceled` +drift, `downloads/status.py`, is why). +""" + +from enum import StrEnum + + +class RunStatus(StrEnum): + RUNNING = "running" + OK = "ok" + ERROR = "error" + # Sync was paused: the run was due and deliberately did nothing. + SKIPPED = "skipped" + # The process stopped while the run was going (a deploy, a crash): set at the next startup, so a + # row never claims to be running in a process that no longer exists. + INTERRUPTED = "interrupted" + + +class RunTrigger(StrEnum): + # The run's interval came round. + SCHEDULED = "scheduled" + # An admin pressed "run now" (or "run all"). + MANUAL = "manual" + # A user's own action started work on one target (a channel's full history), outside the + # recurring jobs. + AD_HOC = "ad_hoc" + + +def sql_in(values: type[StrEnum]) -> str: + """`'a', 'b'` — the IN-list a CHECK constraint names these values with.""" + return ", ".join(f"'{v.value}'" for v in values) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 8aa3f651..a007f02b 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -6,11 +6,12 @@ import threading from datetime import datetime, timezone from typing import Callable +from apscheduler.executors.pool import ThreadPoolExecutor +from apscheduler.jobstores.base import ConflictingIdError from apscheduler.schedulers.background import BackgroundScheduler from sqlalchemy import select -from app import audit, debugsession, logs, logsink, progress, quota -from app.audit import AuditAction +from app import audit, debugsession, jobruns, logs, logsink, progress, quota from app.config import settings from app.db import SessionLocal from app.downloads.gc import run_download_gc @@ -20,6 +21,7 @@ from app.plex.tmdb import run_enrich as run_tmdb_enrich from app.plex.tmdb import run_people_enrich from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync from app.notifications import create_notification +from app.runstatus import RunStatus, RunTrigger from app.state import is_sync_paused from app.sync.autotag import run_autotag_all from app.sync.explore import purge_ephemeral @@ -37,10 +39,9 @@ from app.sync.runner import ( log = logs.get_logger(__name__) _scheduler: BackgroundScheduler | None = None -# Per-job run activity, kept in-memory for the admin Scheduler dashboard to read (the -# scheduler lives in the same single-worker process as the API, so a request can read it -# directly). Ephemeral by design — it reflects "what the scheduler is doing right now and -# how the last runs went" since this process started; durable history isn't the goal here. +# What is running RIGHT NOW, per job: the single-run claim and the live progress. In memory on +# purpose — the scheduler lives in the API's one process, and progress changes every second. What a +# run did is durable instead: its `job_runs` row (app/jobruns.py), which outlives a restart. _activity: dict[str, dict] = {} _activity_lock = threading.Lock() @@ -130,49 +131,6 @@ def _notify_done( log.exception("failed to write completion notification for job %s", job_id) -def _run_changed(result) -> bool: - """Did an automatic job run actually do something worth an audit row? (No-op polls don't.) - A "change" = a truthy NUMERIC count; status-string dicts like {"skipped": "disabled"} (Plex - off — the default) are no-ops, not changes, so they don't flood the trail every interval.""" - if result is None: - return False - if isinstance(result, dict): - # A phased job (backfill's recent/deep, people_enrich's bios/portraits) nests a result per - # phase: a count inside a phase is a change too. - return any( - _run_changed(v) if isinstance(v, dict) else isinstance(v, (int, float)) and bool(v) - for v in result.values() - ) - if isinstance(result, (int, float)): - return result != 0 - return bool(result) - - -def _audit_run(db, actor_id, job_id, status, summary, result) -> None: - """Write ONE run-summary audit row per job run (never per-item). Manual runs (an admin - clicked "run now" → actor_id set) and errors are always logged; automatic runs only when - they changed something, so routine no-op polls don't spam the trail. Best-effort.""" - if actor_id is None and status == "ok" and not _run_changed(result): - return - # The result as data too: the audit log names it in the reader's language, as the inbox - # does; `summary` stays the English line of the row's own summary column. - data = _result_data(result) - try: - audit.record( - db, - actor_id, - AuditAction.JOB_RUN, - target_type="job", - target_id=job_id, - summary=f"{job_id}: {status}" + (f" — {summary}" if summary else ""), - after={"status": status, "summary": summary, **({"result": data} if data else {})}, - ) - db.commit() - except Exception: - db.rollback() - log.exception("failed to write audit row for job %s", job_id) - - def _job(name: str, fn) -> None: # The one place both run paths converge (manual trigger AND scheduled fire), so claim the run # HERE, atomically: whichever reaches this first wins, the other skips. This is the real @@ -180,11 +138,29 @@ def _job(name: str, fn) -> None: if not _claim_running(name): log.info("job %s already running — skipping this trigger", name) return - db = SessionLocal() actor_id = _manual_actor.get() # set only for a manual "run now" trigger - _record( - name, last_started=datetime.now(timezone.utc).isoformat(), last_error=None, progress=None - ) # running already claimed + trigger = RunTrigger.SCHEDULED if actor_id is None else RunTrigger.MANUAL + _record(name, progress=None) # running already claimed + db = SessionLocal() + try: + try: + run_id = jobruns.start(name, trigger, actor_id) + except Exception: + # No row to tell it in: the database is what failed. The log is the record. + log.exception("job %s not started: its run could not be recorded", name) + return + with logs.bind(run_id=run_id): + _run(db, name, fn, run_id, actor_id) + finally: + # ALWAYS release the claim, even if the body raised before its own status update — so a + # failure can never wedge the job as permanently "running". + _record(name, running=False, progress=None) + db.close() + + +def _run(db, name: str, fn, run_id: int, actor_id: int | None) -> None: + """The run itself, inside its `job_runs` row: the pause-skip, the work, and its outcome — on + the row, and in the triggering admin's inbox for a manual run.""" def sink(current, total, phase): _record(name, progress={"current": current, "total": total, "phase": phase}) @@ -192,34 +168,18 @@ def _job(name: str, fn) -> None: try: if is_sync_paused(db): log.info("job %s skipped (sync paused)", name) - _record( - name, - status="skipped", - last_finished=datetime.now(timezone.utc).isoformat(), - last_result="paused", - result=_PAUSED, - progress=None, - ) + jobruns.finish(run_id, RunStatus.SKIPPED, summary="sync paused", result=_PAUSED) if actor_id is not None: _notify_done(db, actor_id, name, "skipped", "sync paused", _PAUSED) - _audit_run(db, actor_id, name, "skipped", "sync paused", _PAUSED) return with progress.bind(sink): result = fn(db) summary = _summarize(result) data = _result_data(result) log.info("job %s -> %s", name, result) - _record( - name, - status="ok", - last_finished=datetime.now(timezone.utc).isoformat(), - last_result=summary, - result=data, - progress=None, - ) + jobruns.finish(run_id, RunStatus.OK, summary=summary, result=data) if actor_id is not None: _notify_done(db, actor_id, name, "ok", summary, data) - _audit_run(db, actor_id, name, "ok", summary, result) except Exception as exc: db.rollback() log.exception("job %s failed", name, extra=logs.event("scheduler.job.failed", job=name)) @@ -229,22 +189,9 @@ def _job(name: str, fn) -> None: failure = ( {"failure": exc.reason, **exc.params} if isinstance(exc, progress.JobFailed) else None ) - _record( - name, - status="error", - last_finished=datetime.now(timezone.utc).isoformat(), - last_error=err, - result=failure, - progress=None, - ) + jobruns.finish(run_id, RunStatus.ERROR, error=err, failure=failure) if actor_id is not None: _notify_done(db, actor_id, name, "error", err, failure) - _audit_run(db, actor_id, name, "error", err, failure) - finally: - # ALWAYS release the claim, even if the body raised before its own status update — so a - # failure can never wedge the job as permanently "running". - _record(name, running=False) - db.close() # A skipped run's result, in the same shape as a job's own `{"skipped": …}`. @@ -271,11 +218,17 @@ def _summarize(result) -> str | None: return str(result) -def scheduler_snapshot() -> dict: - """What the scheduler is doing, for the admin dashboard. `running_here` is False in - environments where the scheduler is disabled (e.g. local dev shares the prod-ish DB but - runs no scheduler), so the UI can say so while still showing DB-derived queue/quota.""" - running_here = _scheduler is not None +def _iso(ts: datetime | None) -> str | None: + return ts.astimezone(timezone.utc).isoformat() if ts else None + + +def scheduler_snapshot(db) -> dict: + """What the scheduler is doing, for the admin dashboard: each job's live state from this process + and its last run from `job_runs`, so a restart no longer blanks "when did it last run, and how + did it go". `running_here` is False where the scheduler runs no interval jobs + (SCHEDULER_ENABLED=false), so the UI can say so while still showing the DB-derived history, + queue and quota.""" + running_here = _scheduler is not None and settings.scheduler_enabled next_runs: dict[str, str | None] = {} live_intervals: dict[str, int] = {} if _scheduler is not None: @@ -287,27 +240,41 @@ def scheduler_snapshot() -> dict: live_intervals[job.id] = round(iv.total_seconds() / 60) with _activity_lock: acts = {k: dict(v) for k, v in _activity.items()} + last = jobruns.last_runs(db) jobs = [] for job_id, default_interval in JOB_INTERVALS.items(): a = acts.get(job_id, {}) + run = last.get(job_id) + # A run still going has no outcome yet: only an ended run says how it went. + ended = run if run is not None and run.status != RunStatus.RUNNING else None jobs.append( { "id": job_id, "interval_minutes": live_intervals.get(job_id, default_interval), "next_run": next_runs.get(job_id), "running": a.get("running", False), - "status": a.get("status"), - "last_started": a.get("last_started"), - "last_finished": a.get("last_finished"), - "last_result": a.get("last_result"), - "result": a.get("result"), - "last_error": a.get("last_error"), + "status": ended.status if ended else None, + "last_started": _iso(run.started_at) if run else None, + "last_finished": _iso(run.finished_at) if run else None, + "last_result": ended.summary if ended else None, + # The outcome as data: the result, or on a failed run the failure the job named. + "result": _outcome_data(ended), + "last_error": ended.error if ended else None, + "last_trigger": run.trigger if run else None, + "last_run_id": run.id if run else None, + "quota_units": ended.quota_units if ended else None, "progress": a.get("progress"), } ) return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs} +def _outcome_data(run) -> dict | None: + if run is None: + return None + return run.failure if run.status == RunStatus.ERROR else run.result + + def _rss_job() -> None: _job("rss_poll", run_rss_poll) @@ -408,15 +375,17 @@ def _people_enrich_job() -> None: def _audit_gc_job() -> None: - # The history cleanup: audit-log rows past audit_retention_days and YouTube quota history past - # quota_retention_days (0 = keep forever, each). Pure DB, no quota. + # The history cleanup: audit-log rows past audit_retention_days, YouTube quota history past + # quota_retention_days and finished runs past job_run_retention_days (0 = keep forever, each). + # Pure DB, no quota. def work(db): from app import sysconfig reaped = audit.gc(db, sysconfig.get_int(db, "audit_retention_days")) quota_reaped = quota.gc(db, sysconfig.get_int(db, "quota_retention_days")) + runs_reaped = jobruns.gc(db, sysconfig.get_int(db, "job_run_retention_days")) db.commit() - return {"reaped": reaped, "quota_reaped": quota_reaped} + return {"reaped": reaped, "quota_reaped": quota_reaped, "runs_reaped": runs_reaped} _job("audit_gc", work) @@ -485,15 +454,29 @@ def running_job_ids() -> set[str]: return {jid for jid, a in _activity.items() if a.get("running")} -def trigger_job(job_id: str, actor_id: int | None = None) -> str: - """Run a job immediately in a background thread, independent of its interval schedule. - The wrapper handles its own DB session, activity tracking and pause-skip, so the live - dashboard reflects the run. Returns "started", "already_running", or raises KeyError for - an unknown job. Refusing a concurrent run keeps a manual trigger from overlapping a - scheduled run (APScheduler enforces max_instances=1 for the scheduled side). +def _as_actor(fn: Callable[[], None], actor_id: int | None) -> Callable[[], None]: + """`fn` run as `actor_id`'s manual run. The actor is RESET afterwards: the executor's pool + threads are reused and keep their context from task to task, so an actor left set would make + the next scheduled run that thread picks up look like this admin's manual run.""" - `actor_id` (the admin who clicked "run now") is stashed in a contextvar inside the new - thread so the run posts a completion notification to that admin's inbox.""" + def run() -> None: + token = _manual_actor.set(actor_id) + try: + fn() + finally: + _manual_actor.reset(token) + + return run + + +def trigger_job(job_id: str, actor_id: int | None = None) -> str: + """Run a job now, independent of its interval schedule: a one-shot job on the same scheduler, + so a manual run shares the scheduled runs' executor and shutdown — no thread of its own. The + wrapper handles its own DB session, run row and pause-skip. Returns "started", + "already_running", or raises KeyError for an unknown job. + + `actor_id` (the admin who clicked "run now") is set in a contextvar inside the run's thread, so + the run is recorded as theirs and posts a completion notification to their inbox.""" fn = JOB_FUNCS.get(job_id) if fn is None: raise KeyError(job_id) @@ -502,13 +485,18 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str: # _job claims, sees it's taken, and skips — no double run. if _is_running(job_id): return "already_running" + sched = _scheduler + if sched is None: + # Only between shutdown and exit: start_scheduler runs at every API startup. + raise RuntimeError("the scheduler is not running") - def run() -> None: - if actor_id is not None: - _manual_actor.set(actor_id) - fn() - - threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start() + try: + # One id per job: a second click before the first one-shot has even started is the same + # run, not a queued second one. misfire_grace_time=None: a busy pool delays it, never drops it. + sched.add_job(_as_actor(fn, actor_id), "date", id=f"{job_id}@now", misfire_grace_time=None) + # silent-ok: not a failure — the answer is the 409 the admin sees ("already running"). + except ConflictingIdError: + return "already_running" return "started" @@ -521,22 +509,36 @@ def trigger_all(actor_id: int | None = None) -> list[str]: return started +# Room for every job at once (a "run all") beside a few ad-hoc runs: APScheduler's default pool of +# 10 would queue a scheduled fire behind long manual runs. +_POOL_SIZE = len(JOB_FUNCS) + 8 + + def start_scheduler() -> None: + """Start this process's scheduler. It always starts, so "run now" works on every install; only + with SCHEDULER_ENABLED does it also register the recurring interval jobs. Before anything can + run, every run a previous process left open is closed as interrupted.""" global _scheduler - if not settings.scheduler_enabled or _scheduler is not None: + if _scheduler is not None: return - # Effective intervals = env defaults overlaid with any admin overrides from the DB. - db = SessionLocal() - try: - iv = load_intervals(db) - finally: - db.close() - scheduler = BackgroundScheduler(timezone="UTC") - for job_id, fn in JOB_FUNCS.items(): - scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id) + left = jobruns.interrupt_leftovers() + if left: + log.warning("%s run(s) left running by the previous process marked interrupted", left) + scheduler = BackgroundScheduler( + timezone="UTC", executors={"default": ThreadPoolExecutor(_POOL_SIZE)} + ) + if settings.scheduler_enabled: + # Effective intervals = env defaults overlaid with any admin overrides from the DB. + db = SessionLocal() + try: + iv = load_intervals(db) + finally: + db.close() + for job_id, fn in JOB_FUNCS.items(): + scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id) scheduler.start() _scheduler = scheduler - log.info("scheduler started") + log.info("scheduler started (interval jobs %s)", "on" if settings.scheduler_enabled else "off") def shutdown_scheduler() -> None: diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index daf8b1ce..4d91b865 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -181,6 +181,8 @@ SPECS: tuple[ConfigSpec, ...] = ( # --- Audit log --- # Days to keep admin audit-log rows before the audit_gc job prunes them (0 = keep forever). ConfigSpec("audit_retention_days", "int", "audit", "audit_retention_days", min=0, max=3650), + # Days the job_runs table keeps a finished background run before the audit_gc job prunes it. + ConfigSpec("job_run_retention_days", "int", "audit", "job_run_retention_days", min=0, max=3650), # --- Diagnostics --- # Runtime log level for the API process. Applied without a restart — see routes/config.py and # main.py's lifespan. diff --git a/backend/tests/test_admin_logs.py b/backend/tests/test_admin_logs.py index 7277108c..6677d67d 100644 --- a/backend/tests/test_admin_logs.py +++ b/backend/tests/test_admin_logs.py @@ -24,6 +24,7 @@ def _row(**kw) -> dict: "request_id": None, "user_id": None, "job_id": None, + "run_id": None, "version": "0.90.0", } row.update(kw) @@ -80,6 +81,7 @@ def test_newest_first_and_every_row_once_across_pages_even_on_equal_timestamps(d ({"event": "none"}, {"warn"}), ({"request_id": "req-abcdef12"}, {"mine"}), ({"job_id": 42}, {"err"}), + ({"run_id": 7}, {"warn"}), # a background run's lines, not a download job's ({"q": "FORBIDDEN"}, {"err"}), # in attrs, any case ({"q": "KeyError"}, {"warn"}), # in the stack ({"q": "100%"}, {"plain"}), # a literal, not a wildcard @@ -102,7 +104,14 @@ def test_each_filter_selects_exactly_its_rows(db, admin, make_user, params, expe job_id=42, attrs={"code": "forbidden"}, ), - _row(msg="warn", level="WARNING", sev=13, event=None, exc="Traceback …\nKeyError: 'x'"), + _row( + msg="warn", + level="WARNING", + sev=13, + event=None, + exc="Traceback …\nKeyError: 'x'", + run_id=7, + ), _row(msg="plain", ts=T0 - timedelta(hours=2), attrs={"note": "100% done"}), # Matched by no filter above but `until`: another account, and `100%` read as a pattern. _row(msg="decoy", ts=T0 - timedelta(hours=3), attrs={"note": "1000 done"}, user_id=99), diff --git a/backend/tests/test_migration_0109.py b/backend/tests/test_migration_0109.py new file mode 100644 index 00000000..9b19b05c --- /dev/null +++ b/backend/tests/test_migration_0109.py @@ -0,0 +1,98 @@ +"""Migration 0109's data step, DB-lane: the audit log's run rows move into `job_runs` with their +outcome intact, and leave the audit log; an admin action stays where it was. The downgrade moves +them back.""" + +import json + +from alembic import command +from alembic.config import Config +from sqlalchemy import text + +from app.db import engine + + +def _audit(conn, actor, action, target, after): + conn.execute( + text( + "INSERT INTO audit_log (actor_id, action, target_type, target_id, summary, after," + " created_at) VALUES (:a, :act, 'job', :t, 's', CAST(:after AS json)," + " CAST('2026-10-01T03:00:00+00:00' AS timestamptz))" + ), + {"a": actor, "act": action, "t": target, "after": json.dumps(after)}, + ) + + +def test_the_audit_logs_runs_move_into_the_run_table_and_back(db, make_user): + uid = make_user().id + cfg = Config("alembic.ini") + command.downgrade(cfg, "0108_folder_links") + try: + with engine.begin() as conn: + _audit( + conn, + None, + "job_run", + "rss_poll", + {"status": "ok", "summary": "new=3", "result": {"new": 3}}, + ) + _audit( + conn, + uid, + "job_run", + "rss_poll", + {"status": "error", "summary": "boom", "result": {"failure": "rss_all_failed"}}, + ) + _audit( + conn, uid, "job_run", "backfill", {"status": "skipped", "summary": "sync paused"} + ) + _audit(conn, uid, "config_set", None, {"key": "x"}) + finally: + command.upgrade(cfg, "head") + + with engine.connect() as conn: + runs = conn.execute( + text( + "SELECT job, trigger, actor_id, status, summary, result, error, failure," + " to_char(started_at AT TIME ZONE 'UTC', 'MM-DD HH24'), finished_at = started_at" + " FROM job_runs ORDER BY id" + ) + ).all() + left = conn.execute(text("SELECT action FROM audit_log")).scalars().all() + assert [tuple(r) for r in runs] == [ + ("rss_poll", "scheduled", None, "ok", "new=3", {"new": 3}, None, None, "10-01 03", True), + ( + "rss_poll", + "manual", + uid, + "error", + None, + None, + "boom", + {"failure": "rss_all_failed"}, + "10-01 03", + True, + ), + ("backfill", "manual", uid, "skipped", "sync paused", None, None, None, "10-01 03", True), + ] + assert left == ["config_set"] + + command.downgrade(cfg, "0108_folder_links") + try: + with engine.connect() as conn: + back = conn.execute( + text( + "SELECT actor_id, target_id, after::jsonb FROM audit_log" + " WHERE action = 'job_run' ORDER BY id" + ) + ).all() + finally: + command.upgrade(cfg, "head") + assert [tuple(r) for r in back] == [ + (None, "rss_poll", {"status": "ok", "summary": "new=3", "result": {"new": 3}}), + ( + uid, + "rss_poll", + {"status": "error", "summary": "boom", "result": {"failure": "rss_all_failed"}}, + ), + (uid, "backfill", {"status": "skipped", "summary": "sync paused"}), + ] diff --git a/backend/tests/test_plex_people_enrich.py b/backend/tests/test_plex_people_enrich.py index d10909d2..307f85ac 100644 --- a/backend/tests/test_plex_people_enrich.py +++ b/backend/tests/test_plex_people_enrich.py @@ -9,7 +9,6 @@ from datetime import datetime, timezone import httpx import pytest -from app import scheduler from app.models import PlexItem, PlexLibrary, PlexPerson, User from app.plex import imgcache from app.plex import people as plex_people @@ -340,12 +339,6 @@ def test_the_people_job_runs_its_halves_without_a_tmdb_key(db, cache_dir, plex_o } -def test_a_phase_count_is_a_change_worth_an_audit_row(): - assert scheduler._run_changed({"bios": {"enriched": 3}, "portraits": {"cached": 0}}) - assert scheduler._run_changed({"recent": {"videos_added": 2}, "deep": {}}) - assert not scheduler._run_changed({"bios": {"skipped": "x"}, "portraits": {"cached": 0}}) - - def test_the_person_page_says_a_bio_is_on_its_way(db, monkeypatch, override_dep): from fastapi.testclient import TestClient diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index 9c47d3c9..acd52f76 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -1,11 +1,20 @@ -"""_claim_running is the single-run guard that both the scheduled fire and the manual trigger go +"""The scheduler's run path, DB-lane: the single-run guard, and every run as a durable `job_runs` row +— its outcome as data, what it spent, the lines it logged — that outlives the process. + +_claim_running is the single-run guard that both the scheduled fire and the manual trigger go through (at the top of _job): holding _activity_lock across the check-and-set means exactly one caller can hold a job's "running" claim at a time, and it's claimable again once released.""" +import logging +import time +from datetime import datetime, timedelta, timezone + +import pytest from sqlalchemy import select -from app import scheduler -from app.models import Notification +from app import jobruns, logs, quota, scheduler, state +from app.models import AuditLog, JobRun, Notification, QuotaEvent +from app.runstatus import RunStatus, RunTrigger _JOB = "unit_test_claim" # not a real job id — exercises the primitive without touching live jobs @@ -31,26 +40,38 @@ def test_claim_is_exclusive_until_released(): # ---- the result as data -------------------------------------------------------------------------- -def _snapshot_job(job_id): - return next(j for j in scheduler.scheduler_snapshot()["jobs"] if j["id"] == job_id) +def _snapshot_job(db, job_id): + return next(j for j in scheduler.scheduler_snapshot(db)["jobs"] if j["id"] == job_id) + + +def _runs(db, job=None) -> list[JobRun]: + db.expire_all() + q = select(JobRun).order_by(JobRun.id) + if job is not None: + q = q.where(JobRun.job == job) + return list(db.execute(q).scalars()) + + +def _manual(actor_id, name, fn): + token = scheduler._manual_actor.set(actor_id) + try: + scheduler._job(name, fn) + finally: + scheduler._manual_actor.reset(token) def test_a_run_keeps_its_result_as_data_for_the_readers_language(db, make_user): """The dashboard and the manual-run notice showed `_summarize`'s English one-liner ("new=3, failed=0", a Python repr for a nested phase). The result rides along as data, so the client names - it; the one-liner stays as the fallback and the audit text.""" + it; the one-liner stays as the fallback.""" admin = make_user("sched@example.com", role="admin") - token = scheduler._manual_actor.set(admin.id) - try: - scheduler._job("rss_poll", lambda _db: {"new": 3, "failed": 0, "fallback": 1}) - scheduler._job("enrich", lambda _db: 5) # enrich returns a bare count - finally: - scheduler._manual_actor.reset(token) + _manual(admin.id, "rss_poll", lambda _db: {"new": 3, "failed": 0, "fallback": 1}) + _manual(admin.id, "enrich", lambda _db: 5) # enrich returns a bare count - rss = _snapshot_job("rss_poll") + rss = _snapshot_job(db, "rss_poll") assert rss["result"] == {"new": 3, "failed": 0, "fallback": 1} assert rss["last_result"] == "new=3, failed=0, fallback=1" - assert _snapshot_job("enrich")["result"] == {"enriched": 5} + assert _snapshot_job(db, "enrich")["result"] == {"enriched": 5} notices = db.execute( select(Notification).where(Notification.user_id == admin.id).order_by(Notification.id) @@ -64,53 +85,227 @@ def test_a_run_keeps_its_result_as_data_for_the_readers_language(db, make_user): def test_a_failed_run_drops_the_previous_result(db): scheduler._job("rss_poll", lambda _db: {"new": 1}) scheduler._job("rss_poll", lambda _db: (_ for _ in ()).throw(RuntimeError("boom"))) - rss = _snapshot_job("rss_poll") + rss = _snapshot_job(db, "rss_poll") + assert rss["status"] == "error" assert rss["last_error"] == "boom" assert rss["result"] is None -def test_a_job_runs_audit_row_carries_its_result_as_data(db): - """The audit log names a run's result in the reader's language, as the inbox does — - from data, not from the English one-liner (a nested phase there was a Python dict repr).""" - from app.models import AuditLog, User - - admin = User(email="auditrun@example.com") - db.add(admin) - db.commit() - scheduler._audit_run(db, None, "people_enrich", "ok", "x", {"bios": {"enriched": 2}}) - scheduler._audit_run(db, admin.id, "rss_poll", "skipped", "sync paused", scheduler._PAUSED) - scheduler._audit_run(db, admin.id, "rss_poll", "error", "boom", None) - rows = db.execute(select(AuditLog).order_by(AuditLog.id)).scalars().all() - assert [r.after for r in rows] == [ - {"status": "ok", "summary": "x", "result": {"bios": {"enriched": 2}}}, - {"status": "skipped", "summary": "sync paused", "result": {"skipped": "sync paused"}}, - {"status": "error", "summary": "boom"}, - ] - - def test_a_failure_the_job_names_rides_along_as_data(db, make_user): - """A failed RSS poll reached the Hungarian audit log as the server's English sentence. A - failure the job names (`progress.JobFailed`) is stored as data beside that sentence — on the - dashboard, the manual-run notice and the audit row — so the client says it in the reader's - language. Any other failure keeps only its own text (the test above).""" + """A failed RSS poll reached the Hungarian UI as the server's English sentence. A failure the job + names (`progress.JobFailed`) is stored as data beside that sentence — on the run's row, the + dashboard and the manual-run notice — so the client says it in the reader's language.""" from app import progress - from app.models import AuditLog def outage(_db): raise progress.JobFailed("rss_all_failed", "RSS poll failed", total=2, cause="transport") admin = make_user("failure@example.com", role="admin") - token = scheduler._manual_actor.set(admin.id) - try: - scheduler._job("rss_poll", outage) - finally: - scheduler._manual_actor.reset(token) + _manual(admin.id, "rss_poll", outage) named = {"failure": "rss_all_failed", "total": 2, "cause": "transport"} - rss = _snapshot_job("rss_poll") + rss = _snapshot_job(db, "rss_poll") assert rss["last_error"] == "RSS poll failed" assert rss["result"] == named notice = db.execute(select(Notification).where(Notification.user_id == admin.id)).scalar_one() assert notice.data["result"] == named - row = db.execute(select(AuditLog).order_by(AuditLog.id.desc())).scalars().first() - assert row is not None and row.after["result"] == named + (run,) = _runs(db, "rss_poll") + assert (run.status, run.error, run.failure) == ("error", "RSS poll failed", named) + + +# ---- the durable row ----------------------------------------------------------------------------- + + +def test_every_run_is_a_row_with_who_started_it_and_none_in_the_audit_log(db, make_user): + """A no-op scheduled poll was not written anywhere (the audit rule kept only changes), so "did + it run at 03:00?" had no answer. Every run is a row now; the audit log keeps admin actions only.""" + admin = make_user("runs@example.com", role="admin") + scheduler._job("rss_poll", lambda _db: {"new": 0}) + _manual(admin.id, "rss_poll", lambda _db: {"new": 2}) + + scheduled, manual = _runs(db, "rss_poll") + assert (scheduled.trigger, scheduled.actor_id, scheduled.status) == ("scheduled", None, "ok") + assert scheduled.result == {"new": 0} + assert (manual.trigger, manual.actor_id, manual.summary) == ("manual", admin.id, "new=2") + assert manual.finished_at is not None and manual.finished_at >= manual.started_at + assert db.execute(select(AuditLog)).scalars().all() == [] + + +def test_the_last_run_survives_a_restart(db): + """The dashboard's "last run" lived in the process's memory: a deploy blanked every job.""" + scheduler._job("plex_sync", lambda _db: {"items": 4}) + with scheduler._activity_lock: + scheduler._activity.clear() # what a new process starts with + + plex = _snapshot_job(db, "plex_sync") + assert (plex["status"], plex["result"], plex["last_trigger"]) == ( + "ok", + {"items": 4}, + "scheduled", + ) + assert plex["last_finished"] is not None + + +def test_a_paused_sync_is_a_skipped_run(db): + state.set_sync_paused(db, True) + db.commit() + scheduler._job("rss_poll", lambda _db: pytest.fail("a paused run must not do its work")) + (run,) = _runs(db, "rss_poll") + assert (run.status, run.result) == ("skipped", {"skipped": "sync paused"}) + + +def test_a_run_left_running_by_a_stopped_process_is_interrupted_and_stays_so(db): + """A deploy mid-run left nothing behind before; now the row would say `running` forever. The + next startup closes it, and the stale thread's own finish cannot rewrite it.""" + run_id = jobruns.start("backfill", RunTrigger.SCHEDULED, None) + assert jobruns.interrupt_leftovers() == 1 + + (run,) = _runs(db, "backfill") + assert run.status == "interrupted" and run.finished_at is not None + jobruns.finish(run_id, RunStatus.OK, summary="late") + (run,) = _runs(db, "backfill") + assert (run.status, run.summary) == ("interrupted", None) + + +def test_a_run_knows_what_it_spent_and_a_request_spend_is_nobodys_run(db): + def spends(_db): + quota.record_usage(5) + quota.record_usage(3) + return {"enriched": 2} + + scheduler._job("enrich", spends) + quota.record_usage(11) # a request's own spend, outside any run + + (run,) = _runs(db, "enrich") + assert run.quota_units == 8 + events = db.execute(select(QuotaEvent.units, QuotaEvent.job_run_id)).all() + assert sorted(events) == [(3, run.id), (5, run.id), (11, None)] + assert _snapshot_job(db, "enrich")["quota_units"] == 8 + + +class _Collect(logging.Handler): + def __init__(self): + super().__init__() + self.addFilter(logs.ContextFilter()) + self.records: list[logging.LogRecord] = [] + + def emit(self, record): + self.records.append(record) + + +def test_a_runs_log_lines_carry_its_id(db): + """So a run's row can open exactly the lines it logged (the log table's `run_id`).""" + probe = logging.getLogger("siftlode.test.run") + handler = _Collect() + probe.addHandler(handler) + probe.setLevel(logging.INFO) + try: + scheduler._job("autotag", lambda _db: probe.info("tagging") or {"tagged": 1}) + probe.info("after the run") + finally: + probe.removeHandler(handler) + + (run,) = _runs(db, "autotag") + inside, after = (logs.record_fields(r).get("run_id") for r in handler.records) + assert (inside, after) == (run.id, None) + + +def test_retention_reaps_finished_runs_only(db): + old = datetime.now(timezone.utc) - timedelta(days=40) + db.add_all( + [ + JobRun( + job="rss_poll", + trigger="scheduled", + status="ok", + started_at=old, + finished_at=old, + ), + JobRun(job="backfill", trigger="scheduled", status="running", started_at=old), + JobRun( + job="rss_poll", + trigger="scheduled", + status="ok", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + ), + ] + ) + db.commit() + assert jobruns.gc(db, 0) == 0 # 0 = keep forever + assert jobruns.gc(db, 30) == 1 + db.commit() + assert sorted((r.job, r.status) for r in _runs(db)) == [ + ("backfill", "running"), + ("rss_poll", "ok"), + ] + + +# ---- "run now" through the scheduler -------------------------------------------------------------- + + +@pytest.fixture +def live_scheduler(db, monkeypatch): + """This process's scheduler, started as the API starts it (on the migrated schema `db` brings), + with interval jobs off.""" + monkeypatch.setattr(scheduler.settings, "scheduler_enabled", False) + scheduler.start_scheduler() + try: + yield scheduler._scheduler + finally: + scheduler.shutdown_scheduler() + + +def _wait_finished(db, job, timeout=10.0) -> JobRun: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + runs = [r for r in _runs(db, job) if r.status != "running"] + if runs: + return runs[-1] + time.sleep(0.05) + raise AssertionError(f"{job} did not finish") + + +def test_run_now_goes_through_the_scheduler_even_with_interval_jobs_off( + db, make_user, live_scheduler, monkeypatch +): + admin = make_user("now@example.com", role="admin") + monkeypatch.setitem( + scheduler.JOB_FUNCS, "shorts", lambda: scheduler._job("shorts", lambda _db: {"probed": 3}) + ) + assert live_scheduler.get_jobs() == [] # SCHEDULER_ENABLED=false: no interval job + live_scheduler.pause() # hold the one-shot so the second click finds it pending + assert scheduler.trigger_job("shorts", admin.id) == "started" + assert scheduler.trigger_job("shorts", admin.id) == "already_running" + assert [j.id for j in live_scheduler.get_jobs()] == ["shorts@now"] + live_scheduler.resume() + + run = _wait_finished(db, "shorts") + assert (run.trigger, run.actor_id, run.result) == ("manual", admin.id, {"probed": 3}) + + +def test_a_manual_run_leaves_no_actor_behind_on_its_thread(db, make_user): + """The scheduler's pool threads are reused: an actor a manual run left set made the next + scheduled run on that thread an admin's "manual" run, with a notice in their inbox.""" + admin = make_user("pool@example.com", role="admin") + seen = [] + scheduler._as_actor(lambda: seen.append(scheduler._manual_actor.get()), admin.id)() + assert seen == [admin.id] + assert scheduler._manual_actor.get() is None # this same thread's next task runs as nobody + + +def test_run_now_refuses_a_job_already_running(live_scheduler): + assert scheduler._claim_running("maintenance") is True + try: + assert scheduler.trigger_job("maintenance") == "already_running" + finally: + scheduler._record("maintenance", running=False) + + +def test_startup_closes_what_the_previous_process_left_running(db, monkeypatch): + jobruns.start("plex_sync", RunTrigger.MANUAL, None) + monkeypatch.setattr(scheduler.settings, "scheduler_enabled", False) + scheduler.start_scheduler() + scheduler.shutdown_scheduler() + (run,) = _runs(db, "plex_sync") + assert run.status == "interrupted" diff --git a/frontend/src/components/AuditEarlierRuns.tsx b/frontend/src/components/AuditEarlierRuns.tsx deleted file mode 100644 index a5433ac6..00000000 --- a/frontend/src/components/AuditEarlierRuns.tsx +++ /dev/null @@ -1,48 +0,0 @@ -import { useState } from "react"; -import { useTranslation } from "react-i18next"; -import clsx from "clsx"; -import { ChevronRight } from "lucide-react"; -import type { AuditRow } from "../lib/api"; -import { Button } from "./ui/Button"; -import { RelativeTime } from "./ui/RelativeTime"; - -/** A folded run of one job's successful runs (lib/auditGroups): the row shows the newest, this - * opens the earlier ones, each with its moment and its result (`detailOf`, the row's own wording). */ -export default function AuditEarlierRuns({ - runs, - detailOf, -}: { - runs: AuditRow[]; - detailOf: (run: AuditRow) => string | null; -}) { - const { t } = useTranslation(); - const [open, setOpen] = useState(false); - const earlier = runs.slice(1); - return ( -
- - {open && ( - - )} -
- ); -} diff --git a/frontend/src/components/AuditLog.tsx b/frontend/src/components/AuditLog.tsx index 95d402df..94bb5fc8 100644 --- a/frontend/src/components/AuditLog.tsx +++ b/frontend/src/components/AuditLog.tsx @@ -14,10 +14,9 @@ import { LoadingState, StateMessage } from "./ui/QueryState"; import { auditColumns } from "./auditColumns"; import { notifyActionError } from "../lib/actionErrors"; import { compareText } from "../lib/collate"; -import { foldJobRuns } from "../lib/auditGroups"; -// Admin-only audit trail: an append-only who/what/when log of admin actions + scheduler run -// summaries (never per-row). Read-only apart from Clear. Client-side sort/filter/paginate via +// Admin-only audit trail: an append-only who/what/when log of admin actions (never per-row; a +// background run is not one — runs have their own history). Read-only apart from Clear. Client-side sort/filter/paginate via // DataTable; the API returns the most recent window (capped) — the log stays small by design. export default function AuditLog() { const { t, i18n } = useTranslation(); @@ -26,8 +25,7 @@ export default function AuditLog() { const confirm = useConfirm(); const q = useQuery({ queryKey: qk.adminAudit(), queryFn: () => api.adminAudit() }); const data = q.data; - // Back-to-back successful runs of one job fold into one row (lib/auditGroups). - const rows = useMemo(() => foldJobRuns(data?.items ?? []), [data]); + const rows = useMemo(() => data?.items ?? [], [data]); const actionOptions = useMemo(() => { // By what the reader sees — the named action, in their alphabet — not by the code. diff --git a/frontend/src/components/auditColumns.dom.test.tsx b/frontend/src/components/auditColumns.dom.test.tsx index 43001585..34a8a0e8 100644 --- a/frontend/src/components/auditColumns.dom.test.tsx +++ b/frontend/src/components/auditColumns.dom.test.tsx @@ -1,10 +1,10 @@ // @vitest-environment jsdom // -// An audit row whose server summary is a fixed English line ("rss_poll: ok — new=5, failed=0", -// "log_level = DEBUG", "Background sync paused") reads in the reader's language, rebuilt from the row's +// An audit row whose server summary is a fixed English line ("log_level = DEBUG", "rss_poll → 15 +// min/run", "Background sync paused") reads in the reader's language, rebuilt from the row's // data through the REAL catalog. A row whose summary is data (an e-mail, a URL) keeps it. import { afterEach, describe, expect, it } from "vitest"; -import { cleanup, fireEvent, render } from "@testing-library/react"; +import { cleanup, render } from "@testing-library/react"; import i18n, { setLanguage } from "../i18n"; import type { AuditRow } from "../lib/api"; import { auditColumns } from "./auditColumns"; @@ -20,7 +20,7 @@ const row = (over: Partial): AuditRow => ({ id: 1, created_at: null, actor: null, - action: "job_run", + action: "sync_pause", target_type: null, target_id: null, summary: null, @@ -38,27 +38,6 @@ function details(r: AuditRow): string { } describe("an audit row's details", () => { - it("name a job run and its result in the reader's language", () => { - const r = row({ - target_type: "job", - target_id: "rss_poll", - summary: "rss_poll: ok — new=5, failed=0", - after: { status: "ok", summary: "new=5, failed=0" }, - }); - expect(details(r)).toBe("RSS-lekérdezés (új feltöltések) kész | új videó: 5 · sikertelen: 0"); - }); - - it("keep a failed run's error text, which is diagnostic", () => { - const r = row({ - target_type: "job", - target_id: "enrich", - after: { status: "error", summary: "HTTP 403: quotaExceeded" }, - }); - expect(details(r)).toBe( - "Nem sikerült lefuttatni: Adatdúsítás + élő frissítés | HTTP 403: quotaExceeded", - ); - }); - it("name a config change's field and values", () => { const set = row({ action: "config_set", @@ -87,26 +66,6 @@ describe("an audit row's details", () => { expect(details(secret)).toMatch(/ \| titok frissítve/); }); - it("name a nested result and a skipped run from the row's data", () => { - const nested = row({ - target_type: "job", - target_id: "people_enrich", - after: { - status: "ok", - summary: "bios={'enriched': 12}, portraits={'cached': 40}", - result: { bios: { enriched: 12 }, portraits: { cached: 40 } }, - }, - }); - expect(details(nested)).not.toContain("{"); - expect(details(nested)).toMatch(/^Személyadatok és portrék \(Plex\) kész \| /); - const skipped = row({ - target_type: "job", - target_id: "rss_poll", - after: { status: "skipped", summary: "sync paused" }, - }); - expect(details(skipped)).toBe("RSS-lekérdezés (új feltöltések) kihagyva | a szinkron szünetel"); - }); - it("tell a secret's reset from its update, and name a choice", () => { const reset = row({ action: "config_reset", @@ -157,49 +116,3 @@ describe("an audit row's details", () => { ); }); }); - -// A folded run (lib/auditGroups) shows its newest run and opens to the earlier ones; a failed run -// is the line to see in a page of routine ones. -describe("a folded run of one job", () => { - const ok = (id: number, enriched: number) => - row({ - id, - target_type: "job", - target_id: "enrich", - after: { status: "ok", result: { enriched } }, - }); - - it("names how many earlier runs it holds, and opens to each", () => { - setLanguage("hu"); - const t = i18n.getFixedT("hu"); - const col = auditColumns(t, "hu", []).find((c) => c.key === "details")!; - const runs = [ok(3, 50), ok(2, 40), ok(1, 30)]; - const { getByRole, container } = render(<>{col.render!({ ...runs[0]!, runs })}); - const toggle = getByRole("button", { name: "2 korábbi futás" }); - expect(toggle.getAttribute("aria-expanded")).toBe("false"); - expect(container.querySelector("li")).toBeNull(); - fireEvent.click(toggle); - expect(toggle.getAttribute("aria-expanded")).toBe("true"); - expect([...container.querySelectorAll("li")].map((li) => li.textContent)).toEqual([ - "—dúsítva: 40", - "—dúsítva: 30", - ]); - }); - - it("has no toggle for a lone run", () => { - const t = i18n.getFixedT("hu"); - const col = auditColumns(t, "hu", []).find((c) => c.key === "details")!; - const { container } = render(<>{col.render!(ok(1, 5))}); - expect(container.querySelector("button")).toBeNull(); - }); - - it("paints a failed run's line in the danger colour", () => { - const t = i18n.getFixedT("hu"); - const col = auditColumns(t, "hu", []).find((c) => c.key === "details")!; - const failed = row({ target_type: "job", target_id: "enrich", after: { status: "error" } }); - const { container } = render(<>{col.render!(failed)}); - expect(container.querySelector(".wrap-break-word")!.className).toContain("text-danger"); - const { container: fine } = render(<>{col.render!(ok(2, 5))}); - expect(fine.querySelector(".wrap-break-word")!.className).not.toContain("text-danger"); - }); -}); diff --git a/frontend/src/components/auditColumns.tsx b/frontend/src/components/auditColumns.tsx index 782979f9..a11631a7 100644 --- a/frontend/src/components/auditColumns.tsx +++ b/frontend/src/components/auditColumns.tsx @@ -1,11 +1,8 @@ import type { TFunction } from "i18next"; -import clsx from "clsx"; import type { AuditRow } from "../lib/api"; -import type { AuditListRow } from "../lib/auditGroups"; -import AuditEarlierRuns from "./AuditEarlierRuns"; import { formatNumber } from "../lib/format"; import type { JobResult } from "../lib/api"; -import { jobResultText, jobRunText } from "../lib/jobResult"; +import { jobResultText } from "../lib/jobResult"; import type { Column } from "./ui/DataTable"; import { RelativeTime } from "./ui/RelativeTime"; @@ -35,8 +32,7 @@ function counts(after: Record): JobResult | null { } /** The details of the rows whose server `summary` is a fixed English line, rebuilt from the row's - * data in the reader's language: a job run names the job and its result (as the inbox does), a config - * change names the field and its values, an interval or batch change its new value, a count-bearing + * data in the reader's language: a config change names the field and its values, an interval or batch change its new value, a count-bearing * action its counts; a pause/resume says nothing the action has not. `undefined` = not one of these (or * an older row without the data); the generic summary + change rendering applies. */ function localDetails( @@ -46,10 +42,6 @@ function localDetails( ): { head: string | null; sub: string | null } | undefined { const after = row.after ?? {}; const before = row.before ?? {}; - if (row.action === "job_run" && row.target_type === "job" && row.target_id) { - const run = jobRunText(t, { job_id: row.target_id, ...after }, lang); - return { head: run.title, sub: run.body }; - } if ((row.action === "config_set" || row.action === "config_reset") && row.target_id) { const key = row.target_id; const head = t(`config.fields.${key}.label`, key); @@ -104,7 +96,7 @@ export function auditColumns( t: TFunction, lang: string, actionOptions: { value: string; label: string }[], -): Column[] { +): Column[] { return [ { key: "when", @@ -146,23 +138,13 @@ export function auditColumns( const local = localDetails(t, r, lang); const summary = local ? local.head : r.summary; const change = local ? local.sub : changeText(r); - // A failed run is the row to see in a page of routine ones: its line in the danger colour. - const failed = r.action === "job_run" && r.after?.status === "error"; return ( // wrap-break-word (wrap), NOT truncate: in the auto-layout table a `truncate` (nowrap) cell // sizes to its longest row, so one long summary/change stretched the whole table past the // card (the row rules bled into the page). Wrapping keeps every row inside the container.
- {summary && ( -
{summary}
- )} + {summary &&
{summary}
} {change &&
{change}
} - {r.runs && r.runs.length > 1 && ( - localDetails(t, run, lang)?.sub ?? run.summary} - /> - )}
); }, diff --git a/frontend/src/i18n/locales/en/audit.json b/frontend/src/i18n/locales/en/audit.json index 454f58ff..6e6eb7c6 100644 --- a/frontend/src/i18n/locales/en/audit.json +++ b/frontend/src/i18n/locales/en/audit.json @@ -1,6 +1,6 @@ { "title": "Audit log", - "intro": "An append-only record of admin actions and scheduler run summaries — who did what, and when. Logged at action granularity, never per item.", + "intro": "An append-only record of admin actions — who did what, and when. Logged at action granularity, never per item.", "empty": "No audit entries yet.", "system": "System", "clear": "Clear audit log", @@ -20,7 +20,5 @@ "actor": "Who", "action": "Action", "details": "Details" - }, - "earlierRuns_one": "{{count}} earlier run", - "earlierRuns_other": "{{count}} earlier runs" + } } diff --git a/frontend/src/i18n/locales/en/auditActions.json b/frontend/src/i18n/locales/en/auditActions.json index 1be63672..4c87fa0b 100644 --- a/frontend/src/i18n/locales/en/auditActions.json +++ b/frontend/src/i18n/locales/en/auditActions.json @@ -16,7 +16,6 @@ "maintenance_tune": "Maintenance tuned", "sync_pause": "Sync paused", "sync_resume": "Sync resumed", - "job_run": "Scheduler job run", "audit_clear": "Audit log cleared", "download_credential_set": "Download cookies set", "download_credential_delete": "Download cookies deleted", diff --git a/frontend/src/i18n/locales/en/config.json b/frontend/src/i18n/locales/en/config.json index eed76f55..2a73c870 100644 --- a/frontend/src/i18n/locales/en/config.json +++ b/frontend/src/i18n/locales/en/config.json @@ -170,6 +170,10 @@ "label": "Retention (days)", "hint": "How long admin audit-log entries are kept before the audit-log cleanup job prunes them. 0 = keep forever. Default 180." }, + "job_run_retention_days": { + "label": "Run history (days)", + "hint": "How long the history of background runs (scheduled, run now, a channel's full history) is kept before the cleanup job prunes it. 0 = keep forever. Default 30." + }, "log_level": { "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." diff --git a/frontend/src/i18n/locales/hu/audit.json b/frontend/src/i18n/locales/hu/audit.json index 60b56a83..0ec91b4d 100644 --- a/frontend/src/i18n/locales/hu/audit.json +++ b/frontend/src/i18n/locales/hu/audit.json @@ -1,6 +1,6 @@ { "title": "Eseménynapló", - "intro": "Az admin-műveletek és az ütemező futás-összegzéseinek csak-hozzáfűző naplója — ki, mit, mikor. Művelet-szinten naplózva, soha nem elemenként.", + "intro": "Az admin-műveletek csak-hozzáfűző naplója — ki, mit, mikor. Művelet-szinten naplózva, soha nem elemenként.", "empty": "Még nincs naplóbejegyzés.", "system": "Rendszer", "clear": "Eseménynapló ürítése", @@ -20,7 +20,5 @@ "actor": "Ki", "action": "Művelet", "details": "Részletek" - }, - "earlierRuns_one": "{{count}} korábbi futás", - "earlierRuns_other": "{{count}} korábbi futás" + } } diff --git a/frontend/src/i18n/locales/hu/auditActions.json b/frontend/src/i18n/locales/hu/auditActions.json index 6e9fff00..a7e4671b 100644 --- a/frontend/src/i18n/locales/hu/auditActions.json +++ b/frontend/src/i18n/locales/hu/auditActions.json @@ -16,7 +16,6 @@ "maintenance_tune": "Karbantartás hangolva", "sync_pause": "Szinkron szüneteltetve", "sync_resume": "Szinkron folytatva", - "job_run": "Ütemezett feladat lefutott", "audit_clear": "Eseménynapló ürítve", "download_credential_set": "Letöltési cookie-k beállítva", "download_credential_delete": "Letöltési cookie-k törölve", diff --git a/frontend/src/i18n/locales/hu/config.json b/frontend/src/i18n/locales/hu/config.json index a3e3fc5c..f2db4b2a 100644 --- a/frontend/src/i18n/locales/hu/config.json +++ b/frontend/src/i18n/locales/hu/config.json @@ -170,6 +170,10 @@ "label": "Megőrzés (nap)", "hint": "Meddig maradnak meg az admin-naplóbejegyzések, mielőtt az eseménynapló-takarító feladat törli őket. 0 = örökre. Alapérték 180." }, + "job_run_retention_days": { + "label": "Futási előzmények (nap)", + "hint": "Meddig maradnak meg a háttérfutások (ütemezett, azonnali, egy csatorna teljes előzménye) adatai, mielőtt a takarító feladat törli őket. 0 = örökre. Alapérték 30." + }, "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." diff --git a/frontend/src/lib/auditGroups.test.ts b/frontend/src/lib/auditGroups.test.ts deleted file mode 100644 index 85ff1f18..00000000 --- a/frontend/src/lib/auditGroups.test.ts +++ /dev/null @@ -1,72 +0,0 @@ -// The audit log's first screen was one job's "done" ten times over. A job's successful runs fold -// into one row whatever ran between them (the owner: a fold broken by every other job read as the same -// row five times); only that job's own failure or skip breaks it. -import { describe, expect, it } from "vitest"; -import type { AuditRow } from "./api"; -import { foldJobRuns } from "./auditGroups"; - -let nextId = 1; -const run = (job: string, status = "ok"): AuditRow => ({ - id: nextId++, - created_at: null, - actor: null, - action: "job_run", - target_type: "job", - target_id: job, - summary: null, - before: null, - after: { status }, -}); -const admin = (): AuditRow => ({ ...run("x"), action: "config_set", target_type: "config" }); - -const shape = (rows: ReturnType) => - rows.map((r) => `${r.target_id}:${String(r.after?.status)}×${r.runs?.length ?? 1}`); - -describe("folding the job runs", () => { - it("folds a job's successful runs into one row, newest first", () => { - const rows = [run("enrich"), run("enrich"), run("enrich"), run("people")]; - const out = foldJobRuns(rows); - expect(shape(out)).toEqual(["enrich:ok×3", "people:ok×1"]); - expect(out[0]!.id).toBe(rows[0]!.id); - expect(out[0]!.runs!.map((r) => r.id)).toEqual(rows.slice(0, 3).map((r) => r.id)); - }); - - it("never folds a failed or skipped run, and lets it break the run", () => { - const rows = [run("rss"), run("rss", "error"), run("rss", "error"), run("rss"), run("rss")]; - expect(shape(foldJobRuns(rows))).toEqual([ - "rss:ok×1", - "rss:error×1", - "rss:error×1", - "rss:ok×2", - ]); - expect(shape(foldJobRuns([run("rss", "skipped"), run("rss", "skipped")]))).toEqual([ - "rss:skipped×1", - "rss:skipped×1", - ]); - }); - - it("folds across other jobs and an admin's own action, each at its own place", () => { - const rows = [run("enrich"), run("plex"), run("enrich"), admin(), run("shorts"), run("enrich")]; - rows.push(run("plex")); - const out = foldJobRuns(rows); - expect(shape(out)).toEqual(["enrich:ok×3", "plex:ok×2", "x:ok×1", "shorts:ok×1"]); - expect(out[0]!.runs!.map((r) => r.id)).toEqual([rows[0]!.id, rows[2]!.id, rows[5]!.id]); - }); - - it("starts a new fold below the job's own failure, which another job's does not", () => { - const rows = [run("enrich"), run("enrich", "error"), run("enrich"), run("rss", "error")]; - rows.push(run("enrich")); - expect(shape(foldJobRuns(rows))).toEqual([ - "enrich:ok×1", - "enrich:error×1", - "enrich:ok×2", - "rss:error×1", - ]); - }); - - it("leaves the rows it was given as they were", () => { - const rows = [run("enrich"), run("enrich")]; - foldJobRuns(rows); - expect("runs" in rows[0]!).toBe(false); - }); -}); diff --git a/frontend/src/lib/auditGroups.ts b/frontend/src/lib/auditGroups.ts deleted file mode 100644 index 4cb380a7..00000000 --- a/frontend/src/lib/auditGroups.ts +++ /dev/null @@ -1,48 +0,0 @@ -import type { AuditRow } from "./api"; - -// The audit log's scheduler noise, folded (the owner's pick): the background jobs write a row per -// run that changed something, so an hour of "enrichment done" filled the first screen and pushed the -// admins' own actions down. A job's SUCCESSFUL runs become one row that names how many there were and -// opens to each — whatever ran between them (another job, an admin's action): the owner read a fold -// that broke at every other job as the same row five times. Only that job's own failed or skipped run -// breaks it: that row stays on its own, at its moment, where it is seen, and the job's older successes -// start a new fold below it. Rows arrive newest first; a fold stands at its newest run's place. - -/** An audit row as the table lists it: a lone row, or the newest of a job's folded runs (`runs`, the - * whole fold newest first, the row itself included). */ -export type AuditListRow = AuditRow & { runs?: AuditRow[] }; - -/** The job a row is a run of, or null for any other row. */ -function jobOf(row: AuditRow): string | null { - return row.action === "job_run" && row.target_type === "job" && row.target_id - ? row.target_id - : null; -} - -export function foldJobRuns(rows: readonly AuditRow[]): AuditListRow[] { - const out: AuditListRow[] = []; - // Each job's open fold: the row its next older success joins. - const open = new Map(); - for (const row of rows) { - const job = jobOf(row); - if (job === null) { - out.push(row); - continue; - } - if (row.after?.status !== "ok") { - // A failure or a skip stands alone and closes the job's fold. - open.delete(job); - out.push(row); - continue; - } - const head = open.get(job); - if (head) { - head.runs = [...(head.runs ?? [head]), row]; - continue; - } - const fresh: AuditListRow = { ...row }; - open.set(job, fresh); - out.push(fresh); - } - return out; -}