fix(activity): an ad-hoc run's row is one run's, a rollback keeps a run whole, Back reads one entry's filters

- The channel-backfill row says "running" from the runs this process has open, not from the table:
  a row whose outcome failed to write no longer reads as running until a restart. The run still
  going leads the row, so a newer run's outcome never sits under the older one's channel.
- Migration 0109's downgrade carries every run's trigger, spend and start in the audit row, and the
  upgrade restores them: a manual run whose admin was deleted stays manual.
- Back/Forward is read by the page's one popstate listener, both halves in one render; the server
  log adopts its half during that render, so no read goes out with one entry's filters and the
  other's range.
- The server log's failed facets use the other tabs' words (the menus may be short); the accounts
  message shows only when the menu really lacks them.
This commit is contained in:
2026-10-18 17:32:22 +02:00
parent fad656988a
commit 13b7577d9f
12 changed files with 188 additions and 72 deletions
+13 -11
View File
@@ -13,7 +13,7 @@ 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 (one still running as
interrupted, a run's trigger and target in its `after`), then drops the table.
interrupted, what the audit row has no column for in its `after`), then drops the table.
"""
import sqlalchemy as sa
@@ -72,25 +72,27 @@ def upgrade() -> None:
)
# The audit trail's runs move over: the three statuses the scheduler wrote, and `interrupted`,
# which a downgrade writes back (as it does a run's `trigger` and `target`, which the audit row
# has no column for), so a rollback and a redeploy keep every run. A status the run table does
# which a downgrade writes back (as it does a run's trigger, target, spend and start, which the
# audit row has no column for: its actor alone cannot say "manual" once that admin is deleted),
# so a rollback and a redeploy keep every run as it was. A status the run table does
# not know would fail the CHECK.
op.execute(
"""
INSERT INTO job_runs (job, trigger, actor_id, target, status, started_at, finished_at,
summary, result, error, failure)
summary, result, error, failure, quota_units)
SELECT target_id,
coalesce(after::jsonb ->> 'trigger',
CASE WHEN actor_id IS NULL THEN 'scheduled' ELSE 'manual' END),
actor_id,
after::jsonb ->> 'target',
after::jsonb ->> 'status',
created_at,
coalesce((after::jsonb ->> 'started_at')::timestamptz, 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
CASE WHEN after::jsonb ->> 'status' = 'error' THEN after::jsonb -> 'result' END,
coalesce((after::jsonb ->> 'quota_units')::int, 0)
FROM audit_log
WHERE action = 'job_run'
AND target_id IS NOT NULL
@@ -109,9 +111,9 @@ def upgrade() -> None:
def downgrade() -> None:
# A run still `running` is one the rollback stopped: it goes back as interrupted. Its trigger and
# target ride in `after` when the audit row's own columns cannot say them (an ad-hoc run of one
# channel), so the upgrade restores them.
# A run still `running` is one the rollback stopped: it goes back as interrupted. What the audit
# row's own columns cannot say rides in `after` (its trigger, target, spend and start), so the
# upgrade restores them.
op.execute(
"""
INSERT INTO audit_log (actor_id, action, target_type, target_id, summary, after, created_at)
@@ -132,8 +134,8 @@ def downgrade() -> None:
THEN jsonb_build_object('result', result)
ELSE '{}'::jsonb
END
|| CASE WHEN trigger = 'ad_hoc'
THEN jsonb_build_object('trigger', trigger) ELSE '{}'::jsonb END
|| jsonb_build_object('trigger', trigger, 'quota_units', quota_units,
'started_at', started_at)
|| CASE WHEN target IS NOT NULL
THEN jsonb_build_object('target', target) ELSE '{}'::jsonb END)::json,
coalesce(finished_at, started_at)
-10
View File
@@ -119,16 +119,6 @@ def recent_runs(db: Session, jobs: list[str], per_job: int) -> dict[str, list[Jo
return out
def running_one(db: Session, job: str) -> JobRun | None:
"""The newest run of `job` still going, however many runs ended after it started."""
return db.scalars(
select(JobRun)
.where(JobRun.job == job, JobRun.status == RunStatus.RUNNING)
.order_by(JobRun.started_at.desc(), JobRun.id.desc())
.limit(1)
).first()
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."""
+26 -5
View File
@@ -15,7 +15,7 @@ from app import audit, debugsession, jobruns, logs, logsink, progress, quota, sy
from app.config import settings
from app.db import SessionLocal
from app.downloads.gc import run_download_gc
from app.models import Channel
from app.models import Channel, JobRun
from app.plex.sync import sync as run_plex_sync
from app.plex.tmdb import run_enrich as run_tmdb_enrich
from app.plex.tmdb import run_people_enrich
@@ -45,6 +45,10 @@ _scheduler: BackgroundScheduler | None = None
# run did is durable instead: its `job_runs` row (app/jobruns.py), which outlives a restart.
_activity: dict[str, dict] = {}
_activity_lock = threading.Lock()
# The ad-hoc runs this process has open (run id -> job), under the same lock. A row can stay
# `running` in the table after its outcome failed to write (_close), so the table cannot say what is
# going; this can, as the one process per database (app/jobruns.py) is the one running them.
_open_ad_hoc: dict[int, str] = {}
# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
# run — and only that run, not the recurring scheduled ones — posts a completion notification
@@ -243,6 +247,7 @@ def scheduler_snapshot(db) -> dict:
live_intervals[job.id] = round(iv.total_seconds() / 60)
with _activity_lock:
acts = {k: dict(v) for k, v in _activity.items()}
going_id = max((r for r, j in _open_ad_hoc.items() if j == CHANNEL_BACKFILL), default=None)
recent = jobruns.recent_runs(db, [*JOB_INTERVALS, *AD_HOC_JOBS], RECENT_RUNS)
jobs = []
for job_id, default_interval in JOB_INTERVALS.items():
@@ -259,11 +264,17 @@ def scheduler_snapshot(db) -> dict:
)
# Work a user starts on one target: no interval, no "run now" (it needs its target), but its
# runs are shown beside the jobs' — it spends the same quota. Its one in-memory guard is the
# target's own (sync/runner), so "running" is what its rows say: two channels can backfill at
# once, so a run still going counts even when a newer one already ended, and the row names it.
# target's own (sync/runner), so "running" is the runs this process has open: two channels can
# backfill at once, so a run still going counts even when a newer one already ended. That run then
# leads the row, so its channel, its "no outcome yet" and its dot are one run's, never the
# newer run's outcome under the older one's channel.
backfills = recent.get(CHANNEL_BACKFILL, [])
going = jobruns.running_one(db, CHANNEL_BACKFILL)
shown = going or (backfills[0] if backfills else None)
going = db.get(JobRun, going_id) if going_id is not None else None
if going is not None and going.status != RunStatus.RUNNING:
going = None # its outcome is written; only the bookkeeping has not caught up
if going is not None:
backfills = [going, *(r for r in backfills if r.id != going.id)][:RECENT_RUNS]
shown = backfills[0] if backfills else None
ad_hoc = [
{
"id": CHANNEL_BACKFILL,
@@ -596,6 +607,16 @@ def _ad_hoc(job: str, fn: Callable[[], dict], actor_id: int, target: str) -> Non
# No row to tell it in: the database is what failed. The log is the record.
log.exception("ad-hoc %s for %s not started: its run could not be recorded", job, target)
return
with _activity_lock:
_open_ad_hoc[run_id] = job
try:
_ad_hoc_work(job, fn, run_id)
finally:
with _activity_lock:
_open_ad_hoc.pop(run_id, None)
def _ad_hoc_work(job: str, fn: Callable[[], dict], run_id: int) -> None:
with logs.bind(run_id=run_id):
# As in _run: only the work is inside the try.
try:
+30 -7
View File
@@ -3,6 +3,7 @@ outcome intact, and leave the audit log; an admin action stays where it was. The
them back."""
import json
from datetime import timedelta
from alembic import command
from alembic.config import Config
@@ -87,14 +88,28 @@ def test_the_audit_logs_runs_move_into_the_run_table_and_back(db, make_user):
).all()
finally:
command.upgrade(cfg, "head")
# What the audit row has no column for rides along, so the next upgrade restores it.
def run(trigger: str) -> dict:
return {"trigger": trigger, "quota_units": 0, "started_at": "2026-10-01T03:00:00+00:00"}
assert [tuple(r) for r in back] == [
(None, "rss_poll", {"status": "ok", "summary": "new=3", "result": {"new": 3}}),
(
None,
"rss_poll",
{"status": "ok", "summary": "new=3", "result": {"new": 3}, **run("scheduled")},
),
(
uid,
"rss_poll",
{"status": "error", "summary": "boom", "result": {"failure": "rss_all_failed"}},
{
"status": "error",
"summary": "boom",
"result": {"failure": "rss_all_failed"},
**run("manual"),
},
),
(uid, "backfill", {"status": "skipped", "summary": "sync paused"}),
(uid, "backfill", {"status": "skipped", "summary": "sync paused", **run("manual")}),
]
@@ -106,10 +121,13 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user):
conn.execute(
text(
"INSERT INTO job_runs (job, trigger, actor_id, target, status, started_at,"
" finished_at) VALUES"
" ('backfill', 'scheduled', NULL, NULL, 'interrupted', now(), now()),"
" ('maintenance', 'manual', :u, NULL, 'running', now(), NULL),"
" ('channel_backfill', 'ad_hoc', :u, 'UC_x', 'ok', now(), now())"
" finished_at, quota_units) VALUES"
" ('backfill', 'scheduled', NULL, NULL, 'interrupted', now(), now(), 0),"
" ('maintenance', 'manual', :u, NULL, 'running', now(), NULL, 0),"
" ('channel_backfill', 'ad_hoc', :u, 'UC_x', 'ok', now(), now(), 0),"
# Its admin is gone (SET NULL): only the row itself still says it was manual.
" ('enrich', 'manual', NULL, NULL, 'ok',"
" '2026-10-01 10:00+00', '2026-10-01 10:30+00', 42)"
),
{"u": uid},
)
@@ -120,6 +138,9 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user):
runs = conn.execute(
text("SELECT job, trigger, actor_id, target, status FROM job_runs ORDER BY id")
).all()
kept = conn.execute(
text("SELECT quota_units, finished_at - started_at FROM job_runs WHERE job = 'enrich'")
).one()
stranded = conn.execute(
text("SELECT count(*) FROM audit_log WHERE action = 'job_run'")
).scalar()
@@ -127,5 +148,7 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user):
("backfill", "scheduled", None, None, "interrupted"),
("maintenance", "manual", uid, None, "interrupted"), # the rollback stopped it
("channel_backfill", "ad_hoc", uid, "UC_x", "ok"),
("enrich", "manual", None, None, "ok"),
]
assert tuple(kept) == (42, timedelta(minutes=30))
assert stranded == 0
+25 -5
View File
@@ -363,8 +363,15 @@ def test_the_page_shows_the_ad_hoc_runs_beside_the_jobs_with_their_channel(db, m
assert len(row["recent"]) == 2
assert all(j["id"] != "channel_backfill" for j in scheduler.scheduler_snapshot(db)["jobs"])
jobruns.start("channel_backfill", RunTrigger.AD_HOC, user.id, "UC_shown")
assert scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"] is True
seen = []
scheduler._ad_hoc(
"channel_backfill",
lambda: seen.append(scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"]) or {},
user.id,
"UC_shown",
)
assert seen == [True]
assert scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"] is False
def _finished(db, job) -> list[JobRun]:
@@ -458,6 +465,8 @@ def test_a_run_whose_outcome_cannot_be_written_is_not_called_failed(db, make_use
calls.clear()
scheduler._ad_hoc("channel_backfill", lambda: {"videos_added": 1}, user.id, "UC_blip")
assert calls == ["ok"]
# Its row is still `running`, but nothing here is: the page does not call it going.
assert scheduler.scheduler_snapshot(db)["ad_hoc"][0]["running"] is False
def test_a_run_that_cannot_be_recorded_does_not_wedge_its_job(db, monkeypatch):
@@ -484,10 +493,21 @@ def test_an_ad_hoc_run_still_going_shows_although_a_newer_one_ended(db, make_use
user = make_user("two@example.com")
db.add(Channel(id="UC_long", title="Long one"))
db.commit()
jobruns.start("channel_backfill", RunTrigger.AD_HOC, user.id, "UC_long")
scheduler._ad_hoc("channel_backfill", lambda: {"skipped": "at capacity"}, user.id, "UC_short")
(row,) = scheduler.scheduler_snapshot(db)["ad_hoc"]
rows = []
def long_one():
scheduler._ad_hoc(
"channel_backfill", lambda: {"skipped": "at capacity"}, user.id, "UC_short"
)
rows.extend(scheduler.scheduler_snapshot(db)["ad_hoc"])
return {}
scheduler._ad_hoc("channel_backfill", long_one, user.id, "UC_long")
(row,) = rows
assert (row["running"], row["target"], row["target_title"]) == (True, "UC_long", "Long one")
# The outcome beside it is the long one's (none yet), never the short one's skip.
assert (row["status"], row["result"], row["last_error"]) == (None, None, None)
assert [r["status"] for r in row["recent"]] == ["running", "ok"]
def test_an_interval_change_lands_with_its_audit_row_or_not_at_all(