diff --git a/app/core/registry.py b/app/core/registry.py index dc70c39b..84467c40 100644 --- a/app/core/registry.py +++ b/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: diff --git a/app/pipeline/jobqueue.py b/app/pipeline/jobqueue.py index ae1b61b0..ae872b23 100644 --- a/app/pipeline/jobqueue.py +++ b/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 diff --git a/tests/test_registry_resilience.py b/tests/test_registry_resilience.py new file mode 100644 index 00000000..ace3c062 --- /dev/null +++ b/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"