From 5082636f151a0bc13dd5847ba9a7c16698fa577d Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sun, 18 Oct 2026 21:21:19 +0200 Subject: [PATCH] fix(activity): a run nothing here has open reads as interrupted, a rollback sums a stopped run's spend - The scheduler keeps one registry of the runs this process has open, for the interval jobs and the ad-hoc runs alike. A row the table still says is `running` but no run here has open (its outcome failed to write) reads as interrupted, as the next start's sweep will close it: no pulsing dot beside "No run recorded yet". - Migration 0109's downgrade sums a still-running run's quota events into its spend before the link to them is dropped, as the startup sweep would. - A test holds that a dialog closing onto the same address leaves an open log line open. --- backend/alembic/versions/0109_job_runs.py | 10 ++- backend/app/scheduler.py | 69 ++++++++++++++----- backend/tests/test_migration_0109.py | 11 +++ backend/tests/test_scheduler.py | 27 +++++++- .../src/components/AdminLogs.dom.test.tsx | 13 ++++ 5 files changed, 109 insertions(+), 21 deletions(-) diff --git a/backend/alembic/versions/0109_job_runs.py b/backend/alembic/versions/0109_job_runs.py index f75f2c5e..310be464 100644 --- a/backend/alembic/versions/0109_job_runs.py +++ b/backend/alembic/versions/0109_job_runs.py @@ -134,12 +134,18 @@ def downgrade() -> None: THEN jsonb_build_object('result', result) ELSE '{}'::jsonb END - || jsonb_build_object('trigger', trigger, 'quota_units', quota_units, + || jsonb_build_object('trigger', trigger, 'quota_units', 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) - FROM (SELECT *, CASE WHEN status = 'running' THEN 'interrupted' ELSE status END AS st + FROM (SELECT *, CASE WHEN status = 'running' THEN 'interrupted' ELSE status END AS st, + -- A run stopped mid-way has not summed its spend yet: sum it as the startup + -- sweep would, before the link to it is dropped. + CASE WHEN status = 'running' + THEN (SELECT coalesce(sum(q.units), 0) FROM quota_events q + WHERE q.job_run_id = job_runs.id) + ELSE quota_units END AS units FROM job_runs) AS r ORDER BY id """ diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 4e112a56..d7e4c82b 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -45,10 +45,26 @@ _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] = {} +# The 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_runs: dict[int, str] = {} + + +def _opened(run_id: int, job: str) -> None: + with _activity_lock: + _open_runs[run_id] = job + + +def _closed(run_id: int) -> None: + with _activity_lock: + _open_runs.pop(run_id, None) + + +def _open_run_ids() -> dict[int, str]: + with _activity_lock: + return dict(_open_runs) + # 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 @@ -129,6 +145,7 @@ def _job(name: str, fn) -> None: trigger = RunTrigger.SCHEDULED if actor_id is None else RunTrigger.MANUAL _record(name, progress=None) # running already claimed db = SessionLocal() + run_id = None try: try: run_id = jobruns.start(name, trigger, actor_id) @@ -136,11 +153,14 @@ def _job(name: str, fn) -> None: # 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 + _opened(run_id, name) 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". + if run_id is not None: + _closed(run_id) _record(name, running=False, progress=None) db.close() @@ -247,8 +267,12 @@ 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) + # Open before the read or after it: a run that started or ended while the rows were read is + # still counted as the open run it was, never as a stray one. + open_runs = _open_run_ids() recent = jobruns.recent_runs(db, [*JOB_INTERVALS, *AD_HOC_JOBS], RECENT_RUNS) + open_runs |= _open_run_ids() + going_id = max((r for r, j in open_runs.items() if j == CHANNEL_BACKFILL), default=None) jobs = [] for job_id, default_interval in JOB_INTERVALS.items(): a = acts.get(job_id, {}) @@ -259,7 +283,7 @@ def scheduler_snapshot(db) -> dict: "next_run": next_runs.get(job_id), "running": a.get("running", False), "progress": a.get("progress"), - **_last_runs(recent.get(job_id, [])), + **_last_runs(recent.get(job_id, []), open_runs), } ) # Work a user starts on one target: no interval, no "run now" (it needs its target), but its @@ -285,7 +309,7 @@ def scheduler_snapshot(db) -> dict: if shown is not None and shown.target else None ), - **_last_runs(backfills), + **_last_runs(backfills, open_runs), } ] return { @@ -300,16 +324,27 @@ def scheduler_snapshot(db) -> dict: RECENT_RUNS = 3 -def _last_runs(runs: list) -> dict: +def _seen(run, open_runs: dict[int, str]) -> RunStatus: + """A run's status as the page tells it. A row `running` that no run here has open is one whose + outcome failed to write: the next start's sweep closes it as interrupted, so it reads so now — + not as a run still going beside a row that says nothing runs.""" + if run.status == RunStatus.RUNNING and run.id not in open_runs: + return RunStatus.INTERRUPTED + return RunStatus(run.status) + + +def _last_runs(runs: list, open_runs: dict[int, str]) -> dict: """A row's run fields from its newest runs (newest first): how the last one went, and the few before it for the history dots.""" run = runs[0] if runs else None + status = _seen(run, open_runs) if run is not None else None # 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 + ended = run if status is not None and status != RunStatus.RUNNING else None return { - "status": ended.status if ended else None, + "status": 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, + # A stray row has no end of its own: it ended, as far as anyone can tell, where it began. + "last_finished": iso(ended.finished_at or ended.started_at) if ended 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), @@ -317,14 +352,14 @@ def _last_runs(runs: list) -> dict: "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, - "recent": [_run_brief(r) for r in runs], + "recent": [_run_brief(r, open_runs) for r in runs], } -def _run_brief(run) -> dict: +def _run_brief(run, open_runs: dict[int, str]) -> dict: return { "id": run.id, - "status": run.status, + "status": _seen(run, open_runs), "trigger": run.trigger, "started_at": iso(run.started_at), "finished_at": iso(run.finished_at), @@ -607,13 +642,11 @@ 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 + _opened(run_id, job) try: _ad_hoc_work(job, fn, run_id) finally: - with _activity_lock: - _open_ad_hoc.pop(run_id, None) + _closed(run_id) def _ad_hoc_work(job: str, fn: Callable[[], dict], run_id: int) -> None: diff --git a/backend/tests/test_migration_0109.py b/backend/tests/test_migration_0109.py index 9fd08814..495d7af7 100644 --- a/backend/tests/test_migration_0109.py +++ b/backend/tests/test_migration_0109.py @@ -131,6 +131,13 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user): ), {"u": uid}, ) + # What the run caught mid-way had spent: only its quota rows know it yet. + conn.execute( + text( + "INSERT INTO quota_events (action, units, job_run_id)" + " SELECT 'videos.list', 7, id FROM job_runs WHERE job = 'maintenance'" + ) + ) cfg = Config("alembic.ini") command.downgrade(cfg, "0108_folder_links") command.upgrade(cfg, "head") @@ -141,6 +148,9 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user): kept = conn.execute( text("SELECT quota_units, finished_at - started_at FROM job_runs WHERE job = 'enrich'") ).one() + stopped = conn.execute( + text("SELECT quota_units FROM job_runs WHERE job = 'maintenance'") + ).scalar() stranded = conn.execute( text("SELECT count(*) FROM audit_log WHERE action = 'job_run'") ).scalar() @@ -151,4 +161,5 @@ def test_a_down_and_up_round_trip_keeps_every_run(db, make_user): ("enrich", "manual", None, None, "ok"), ] assert tuple(kept) == (42, timedelta(minutes=30)) + assert stopped == 7 assert stranded == 0 diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index cb6f1ef7..32707328 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -460,13 +460,38 @@ def test_a_run_whose_outcome_cannot_be_written_is_not_called_failed(db, make_use assert not scheduler._is_running("autotag") (run,) = _runs(db, "autotag") assert run.status == "running" # left for the next start's sweep, not rewritten as a failure + # Nothing here runs it: the page reads it as the sweep will close it, not as a run going on. + row = _row(db, "autotag") + assert (row["running"], row["status"], row["recent"][0]["status"]) == ( + False, + "interrupted", + "interrupted", + ) + assert row["last_finished"] is not None user = make_user("blip@example.com") 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 + (row,) = scheduler.scheduler_snapshot(db)["ad_hoc"] + assert (row["running"], row["status"], row["recent"][0]["status"]) == ( + False, + "interrupted", + "interrupted", + ) + + +def test_a_run_going_on_reads_as_running_on_its_row(db): + seen = [] + scheduler._job("autotag", lambda _db: seen.append(_row(db, "autotag")) or {"tagged": 1}) + (row,) = seen + assert (row["running"], row["status"], row["recent"][0]["status"]) == (True, None, "running") + assert _row(db, "autotag")["recent"][0]["status"] == "ok" + + +def _row(db, job_id: str) -> dict: + return next(j for j in scheduler.scheduler_snapshot(db)["jobs"] if j["id"] == job_id) def test_a_run_that_cannot_be_recorded_does_not_wedge_its_job(db, monkeypatch): diff --git a/frontend/src/components/AdminLogs.dom.test.tsx b/frontend/src/components/AdminLogs.dom.test.tsx index fde06401..01580322 100644 --- a/frontend/src/components/AdminLogs.dom.test.tsx +++ b/frontend/src/components/AdminLogs.dom.test.tsx @@ -114,6 +114,19 @@ afterEach(() => { }); describe("AdminLogs", () => { + it("a dialog closing onto the same address leaves the open line open", async () => { + logsMock.mockResolvedValue({ items: [row(1)], accounts: {}, next_cursor: null }); + renderConsole(); + await waitFor(() => expect(lines()).toHaveLength(1)); + fireEvent.click(lines()[0]!); + expect(lines()[0]!.getAttribute("aria-expanded")).toBe("true"); + // A closing dialog steps back over its own history entry: a popstate onto the same address. + act(() => { + window.dispatchEvent(new PopStateEvent("popstate")); + }); + expect(lines()[0]!.getAttribute("aria-expanded")).toBe("true"); + }); + it("a request id narrows the list to that request, with every other filter off", async () => { window.history.replaceState(null, "", "/?page=activity&at=logs&lgrp=error&aq=boom"); logsMock.mockResolvedValue({ items: [row(1)], accounts: {}, next_cursor: null });