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

@@ -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
         """

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:

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

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):