fix(registry): free a job cancelled in the pop-to-claim window, and boot with a bad registry

app/core/registry.py

@@ -175,21 +175,33 @@ def restore(jobs_dir: Path) -> None:
             _pending_resume.extend(
                 j.id for j in sorted(resume, key=lambda j: (j.queue_position, j.created_at))
             )
-        except (OSError, json.JSONDecodeError, TypeError, ValueError):
+        except Exception:
+            # Broad on purpose. restore() runs at import time, so anything that
+            # escapes here stops the backend booting with no way for a user to
+            # recover short of deleting the file by hand. A registry.json that
+            # is valid JSON but not an object (a top-level list, null, a bare
+            # string) used to do exactly that: _migrate calls data.get() and
+            # raises AttributeError, which the old tuple did not name (#520).
             logger.warning("failed to load registry from %s", path, exc_info=True)
 
-    with _lock:
-        known = set(_jobs)
-    for job_dir in jobs_dir.iterdir():
-        if not job_dir.is_dir() or not JOB_ID_RE.match(job_dir.name) or job_dir.name in known:
-            continue
-        recovered = _recover_done_job(job_dir)
-        if recovered is not None:
-            with _lock:
-                _jobs[recovered.id] = recovered
-            changed = True
-    if changed:
-        persist(jobs_dir)
+    # Orphan recovery, and the persist that follows it, were outside the guard
+    # above -- an unreadable jobs_dir or a failed write was fatal at startup for
+    # the same reason.
+    try:
+        with _lock:
+            known = set(_jobs)
+        for job_dir in jobs_dir.iterdir():
+            if not job_dir.is_dir() or not JOB_ID_RE.match(job_dir.name) or job_dir.name in known:
+                continue
+            recovered = _recover_done_job(job_dir)
+            if recovered is not None:
+                with _lock:
+                    _jobs[recovered.id] = recovered
+                changed = True
+        if changed:
+            persist(jobs_dir)
+    except Exception:
+        logger.warning("failed to recover jobs from %s", jobs_dir, exc_info=True)
 
 
 def _resume_or_recover(job: Job, job_dir: Path) -> Job | None:

app/pipeline/jobqueue.py

@@ -200,6 +200,21 @@ async def _dispatch(job: Job) -> None:
     await run_pipeline(job, source_url, JOBS_DIR)
 
 
+def _finalise_dropped_job(job: Job) -> None:
+    """Complete a cancellation the API could not finish itself.
+
+    cancel_job can only finalise a job it still finds in the queue. Between
+    _pop_next() and _set_running() a job is in neither the queue nor the
+    running slot, so a cancel arriving there sets cancel_requested and returns.
+    The worker owns the job by then and is the only thing that can close it
+    out (#520)."""
+    if not job.cancel_requested or job.status in ("done", "error", "cancelled"):
+        return
+    _set(job, status="cancelled", stage="Cancelled")
+    cleanup_job_dir(job.id)
+    registry_persist(JOBS_DIR)
+
+
 async def _worker_loop() -> None:
     assert _wake is not None
     while not _stopping:
@@ -220,7 +235,19 @@ async def _worker_loop() -> None:
         if job is None:
             continue
         if job.cancel_requested or job.status in ("done", "error", "cancelled"):
-            # Cancelled or finished while it waited; drop it silently.
+            # Cancelled or finished while it waited.
+            #
+            # Finalising here is not optional. Between _pop_next() above and
+            # _set_running() below the job is in neither the queue nor the
+            # running slot, so a cancel arriving in that window finds
+            # discard() False and running_id() None and returns having only
+            # set cancel_requested. This worker is the sole consumer and owns
+            # the job by now, so if it just dropped it the job would sit at
+            # "queued" forever: absent from the queue view, still counted by
+            # pending_count against the capacity limit, its uploaded source
+            # never freed, and -- because "queued" is persisted -- re-queued
+            # on every restart (#520).
+            _finalise_dropped_job(job)
             continue
 
         # Claim it. No await between the pop and this status write, so a job is

tests/test_registry_resilience.py

@@ -0,0 +1,118 @@
+"""Two ways the registry used to strand state (#520).
+
+A cancel that lands in the window between the queue worker popping a job and
+claiming it left the job at "queued" forever, invisible but still counted
+against the capacity limit. And a registry.json that was valid JSON but not an
+object stopped the backend booting at all.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+
+import pytest
+
+from app.core import registry as _registry
+from app.core.config import JOB_ID_RE
+from app.core.models import Job
+from app.core.registry import registry_path
+
+
+def _job(job_id="a1b2c3d4e5f6", **kw):
+    return Job(id=job_id, **kw)
+
+
+# ─── a registry file we cannot use must not stop the backend ───
+
+
+@pytest.mark.parametrize("body", ["[1, 2, 3]", "null", '"a string"', "42", "[]"])
+def test_a_non_object_registry_does_not_raise(tmp_path, body):
+    # _migrate calls data.get(); anything that is not a dict raises
+    # AttributeError, which restore() ran at import time and never caught.
+    registry_path(tmp_path).write_text(body, encoding="utf-8")
+
+    _registry.restore(tmp_path)  # must not raise
+
+    assert _registry.all_jobs() == {}
+
+
+def test_a_corrupt_registry_does_not_raise(tmp_path):
+    registry_path(tmp_path).write_text("{not json", encoding="utf-8")
+
+    _registry.restore(tmp_path)
+
+    assert _registry.all_jobs() == {}
+
+
+def test_a_good_registry_still_loads(tmp_path):
+    job = _job(status="done", title="Song")
+    registry_path(tmp_path).write_text(json.dumps({"jobs": [job.to_record()]}), encoding="utf-8")
+
+    _registry.restore(tmp_path)
+
+    assert job.id in _registry.all_jobs()
+
+
+def test_an_unreadable_jobs_dir_does_not_raise(tmp_path, monkeypatch):
+    # Orphan recovery sat outside the guard, so an OSError from iterdir() was
+    # fatal at startup too.
+    def _boom(self):
+        raise OSError("permission denied")
+
+    monkeypatch.setattr("pathlib.Path.iterdir", _boom)
+
+    _registry.restore(tmp_path)  # must not raise
+
+
+# ─── a cancel in the pop-to-claim window must not strand the job ───
+
+
+def test_job_id_re_matches_the_ids_we_generate():
+    # The orphan-recovery filter depends on this; a drift here would silently
+    # stop recovery working at all.
+    assert JOB_ID_RE.match("a1b2c3d4e5f6")
+
+
+async def test_cancel_between_pop_and_claim_finalises_the_job(tmp_path, monkeypatch):
+    """Drive the real worker loop, so this also catches the worker simply not
+    calling the finaliser."""
+    import asyncio
+
+    from app.pipeline import jobqueue
+
+    job = _job(status="queued")
+    _registry.register(job)
+    jobqueue.enqueue(job.id)
+    assert _registry.pending_count(uploads=False) == 1
+
+    # The cancel lands in the pop-to-claim window: cancel_job's discard() has
+    # already lost the race, so all it can do is set the flag.
+    job.cancel_requested = True
+
+    task = jobqueue.start_worker()
+    try:
+        for _ in range(50):
+            await asyncio.sleep(0.01)
+            if job.status == "cancelled":
+                break
+    finally:
+        jobqueue.request_stop()
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+    assert job.status == "cancelled", "a stranded job holds a capacity slot forever"
+    assert _registry.pending_count(uploads=False) == 0, "the capacity slot must be released"
+
+
+async def test_a_dropped_job_that_was_not_cancelled_is_left_alone(tmp_path):
+    # Already-terminal jobs reach the same branch; finalising them would
+    # rewrite a real result.
+    job = _job(status="done", title="Song")
+    _registry.register(job)
+
+    jobqueue_mod = __import__("app.pipeline.jobqueue", fromlist=["jobqueue"])
+    jobqueue_mod._finalise_dropped_job(job)
+
+    assert job.status == "done"