bvandeusen
25e1e098fb
celery_signals._queue_for is a hand-maintained mirror of task_routes that stamps TaskRun.queue in the prerun signal. It was missing the backend.app.tasks.external. prefix, so external fetches recorded queue='default' even though celery routes external.* → download and runs them on the download worker. The dashboard's per-queue filters and the per-queue recovery-sweep threshold therefore missed them — the same 'queue column lies default' gap the 2026-06-02 audit fixed for backup/admin/library_audit. Map external.* → download in _queue_for. Composes with the fetch_external_link task-name sweep override (#883), which wins by precedence regardless of the recorded queue. Pinned test asserts the mirror agrees with the actual route. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
"""Queue routing — the long one-shot maintenance tasks run on a dedicated
|
|
`maintenance_long` lane so a 30-min backup or a multi-chunk audit can't starve
|
|
the quick recovery sweeps / vacuum on the concurrency-1 `maintenance` lane
|
|
(operator-flagged 2026-06-07)."""
|
|
from backend.app.celery_app import celery
|
|
|
|
|
|
def test_long_one_shots_route_to_maintenance_long():
|
|
routes = celery.conf.task_routes
|
|
for prefix in (
|
|
"backend.app.tasks.backup.*",
|
|
"backend.app.tasks.admin.*",
|
|
"backend.app.tasks.library_audit.*",
|
|
):
|
|
assert routes[prefix]["queue"] == "maintenance_long"
|
|
|
|
|
|
def test_quick_maintenance_stays_on_maintenance():
|
|
routes = celery.conf.task_routes
|
|
assert routes["backend.app.tasks.maintenance.*"]["queue"] == "maintenance"
|
|
|
|
|
|
def test_queue_for_mirrors_external_to_download():
|
|
"""celery_signals._queue_for is a hand-maintained mirror of task_routes
|
|
that stamps TaskRun.queue. external.* routes to the download lane, so the
|
|
mirror must agree — else TaskRun.queue lies 'default' for external fetches
|
|
and per-queue dashboard filters / threshold overrides miss them
|
|
(operator-flagged 2026-06-17)."""
|
|
from backend.app.celery_signals import _queue_for
|
|
|
|
class _T:
|
|
name = "backend.app.tasks.external.fetch_external_link"
|
|
|
|
assert _queue_for(_T()) == "download"
|
|
assert (
|
|
celery.conf.task_routes["backend.app.tasks.external.*"]["queue"]
|
|
== "download"
|
|
)
|