fix(scheduler): move the single-run claim into _job (cover both paths)

The atomic claim added in trigger_job only covered the manual path; a scheduled
fire sets running via _job without claiming, so a manual trigger racing it could
still double-run, and the claim (set outside _job) could leak if _job threw
before its own cleanup. Move _claim_running to the top of _job — the one point
both the scheduled fire and the manual trigger pass through — and release it in
_job's finally, so exactly one run wins whichever path it comes from and a
failure can never wedge the job "running". trigger_job's _is_running check is now
just a best-effort fast answer for the UI. Unit-test _claim_running (and
is_messageable_user).

backend/app/scheduler.py

@@ -162,9 +162,15 @@ def _audit_run(db, actor_id, job_id, status, summary, result) -> None:
 
 
 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
+    # single-run guard — trigger_job's pre-check is only a fast best-effort answer for the UI.
+    if not _claim_running(name):
+        logger.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, running=True, last_started=_now_iso(), last_error=None, progress=None)
+    _record(name, last_started=_now_iso(), last_error=None, progress=None)  # running already claimed
 
     def sink(current, total, phase):
         _record(name, progress={"current": current, "total": total, "phase": phase})
@@ -172,7 +178,7 @@ def _job(name: str, fn) -> None:
     try:
         if is_sync_paused(db):
             logger.info("job %s skipped (sync paused)", name)
-            _record(name, running=False, status="skipped", last_finished=_now_iso(),
+            _record(name, status="skipped", last_finished=_now_iso(),
                      last_result="paused", progress=None)
             if actor_id is not None:
                 _notify_done(db, actor_id, name, "skipped", "sync paused")
@@ -182,7 +188,7 @@ def _job(name: str, fn) -> None:
             result = fn(db)
         summary = _summarize(result)
         logger.info("job %s -> %s", name, result)
-        _record(name, running=False, status="ok", last_finished=_now_iso(),
+        _record(name, status="ok", last_finished=_now_iso(),
                 last_result=summary, progress=None)
         if actor_id is not None:
             _notify_done(db, actor_id, name, "ok", summary)
@@ -191,12 +197,15 @@ def _job(name: str, fn) -> None:
         db.rollback()
         logger.exception("job %s failed", name)
         err = str(exc) or exc.__class__.__name__
-        _record(name, running=False, status="error", last_finished=_now_iso(),
+        _record(name, status="error", last_finished=_now_iso(),
                 last_error=err, progress=None)
         if actor_id is not None:
             _notify_done(db, actor_id, name, "error", err)
         _audit_run(db, actor_id, name, "error", err, None)
     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()
 
 
@@ -382,9 +391,9 @@ def _is_running(job_id: str) -> bool:
 
 def _claim_running(job_id: str) -> bool:
     """Atomically claim a job as running: True if it was idle (and marks it running now), False if
-    it was already running. Holding the lock across BOTH the check and the set closes the TOCTOU
-    window where two rapid manual triggers each saw 'idle' (the running flag is otherwise only set
-    later inside the spawned thread) and both started a run."""
+    it was already running. Holding the lock across BOTH the check and the set is what makes the
+    single-run guard race-free — called once at the top of _job so a manual trigger and a scheduled
+    fire (or two rapid triggers) converge here and exactly one wins. Released in _job's finally."""
     with _activity_lock:
         if _activity.get(job_id, {}).get("running"):
             return False
@@ -411,9 +420,10 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str:
     fn = JOB_FUNCS.get(job_id)
     if fn is None:
         raise KeyError(job_id)
-    # Atomic claim (not a bare _is_running check) so two rapid clicks can't both start — the loser
-    # gets "already_running". _job re-affirms running=True and clears it when the run ends.
-    if not _claim_running(job_id):
+    # Best-effort fast answer for the UI; the authoritative single-run guard is _job's atomic claim
+    # (so a manual trigger and a scheduled fire can't both run). A race here just means the loser's
+    # _job claims, sees it's taken, and skips — no double run.
+    if _is_running(job_id):
         return "already_running"
 
     def run() -> None:
@@ -421,12 +431,7 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str:
             _manual_actor.set(actor_id)
         fn()
 
-    try:
-        threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
-    except BaseException:
-        # Never got the worker going — release the claim so the job isn't wedged "running" forever.
-        _record(job_id, running=False)
-        raise
+    threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
     return "started"