From 261da3b1a6de6c47d6a6942a63c608d19d66b0cd Mon Sep 17 00:00:00 2001 From: Thales Pereira <31625914+thcp@users.noreply.github.com> Date: Mon, 31 Aug 2026 18:07:16 +0100 Subject: [PATCH] fix(registry): free a job cancelled in the pop-to-claim window, and boot with a bad registry Two ways the registry stranded state. A cancel arriving between the worker's _pop_next() and _set_running() found the job in neither the queue nor the running slot. discard() returned False so cancel_job never finalised it, running_id() was None so nothing was terminated, and the worker then dropped it silently. The job stayed at "queued" forever: absent from the queue view, still counted by pending_count against the capacity limit, its uploaded source (up to 400 MB) never freed, and -- because "queued" is persisted -- re-queued on every restart. The worker is the sole consumer and owns the job once it has popped it, so it now closes out the cancellation itself via _finalise_dropped_job(). Extracted rather than inlined so the drop path can be tested directly, and so an already-terminal job is explicitly left alone rather than having a real result rewritten. Separately, restore() caught (OSError, JSONDecodeError, TypeError, ValueError). A registry.json that was valid JSON but not an object -- a top-level list, null, a bare string -- makes _migrate call data.get() and raise AttributeError, which that tuple does not name. restore() runs at import time, so the backend simply never started, with no self-healing path and no way for a user to recover short of deleting the file by hand. Orphan recovery and the persist that follows it sat outside the guard entirely, so an unreadable jobs_dir was fatal at startup for the same reason. Both are now caught broadly and logged. A registry we cannot use should cost the user their job list, not their app. The cancel test drives the real worker loop rather than calling the helper, so it also fails if the worker stops calling it. Verified: reverting the worker's call fails that test, and reverting restore()'s except fails five others. Refs #520 --- app/core/registry.py | 38 ++++++---- app/pipeline/jobqueue.py | 29 +++++++- tests/test_registry_resilience.py | 118 ++++++++++++++++++++++++++++++ 3 files changed, 171 insertions(+), 14 deletions(-) create mode 100644 tests/test_registry_resilience.py 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"