diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py index a50bd26..56ff2e2 100644 --- a/backend/app/api/system_activity.py +++ b/backend/app/api/system_activity.py @@ -152,6 +152,8 @@ async def list_runs(): queue= filter to one queue status= filter to one status (running/ok/error/timeout/retry) task= case-insensitive substring match on task_name + celery_task_id= exactly one run — how a page follows a job it + started without having to know its lane limit= default 50, max 200 before_id= cursor for keyset pagination @@ -167,6 +169,7 @@ async def list_runs(): queue = request.args.get("queue") status = request.args.get("status") task = request.args.get("task") + celery_task_id = request.args.get("celery_task_id") before_id_raw = request.args.get("before_id") before_id = int(before_id_raw) if before_id_raw else None @@ -176,6 +179,8 @@ async def list_runs(): stmt = stmt.where(TaskRun.queue == queue) if status: stmt = stmt.where(TaskRun.status == status) + if celery_task_id: + stmt = stmt.where(TaskRun.celery_task_id == celery_task_id) if task: # Task names contain literal underscores (download_source, # vacuum_analyze) — escape LIKE wildcards so a search for diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index a6e007c..dec94ea 100644 --- a/backend/app/celery_signals.py +++ b/backend/app/celery_signals.py @@ -18,6 +18,7 @@ dark for that interval. Monitoring NEVER breaks the thing it's monitoring. """ +import functools import logging from datetime import UTC, datetime @@ -53,42 +54,29 @@ _INT32_MIN = -2_147_483_648 def _queue_for(task) -> str: - """Reverse the task→queue routing from celery_app.task_routes. - Keep in sync if task_routes is reordered. + """The queue Celery routes this task to — asked of the router itself. - Audit 2026-06-02: backup/admin/library_audit prefixes were - missing here even though task_routes sent all three to - 'maintenance'. The TaskRun.queue column then lied for those - rows (claimed 'default') so per-queue dashboard filters and - per-queue threshold overrides silently missed them. + This was a hand-kept copy of `celery_app.task_routes`, and it drifted + twice (the 2026-06-02 audit, then #4432). Long-lane jobs were recorded as + `maintenance`, and translation and gpu_queue runs as `default`, where the + 5-minute stall sweep failed healthy 35-minute translation runs. The router + answers from the same table the broker uses, so the two cannot disagree. """ name = getattr(task, "name", "") or "" - if name.startswith("backend.app.tasks.import_file."): - return "import" - if name.startswith("backend.app.tasks.ml."): - return "ml" - if name.startswith("backend.app.tasks.thumbnail."): - return "thumbnail" - if name.startswith(( - "backend.app.tasks.download.", - # External file-host fetches share the download lane (celery_app - # routes external.* → download). Mirror it here or TaskRun.queue - # lies 'default' for them, so per-queue dashboard filters and the - # per-queue threshold override miss them — the same gap the - # 2026-06-02 audit fixed for backup/admin/library_audit. - "backend.app.tasks.external.", - )): - return "download" - if name.startswith("backend.app.tasks.scan."): - return "scan" - if name.startswith(( - "backend.app.tasks.maintenance.", - "backend.app.tasks.backup.", - "backend.app.tasks.admin.", - "backend.app.tasks.library_audit.", - )): - return "maintenance" - return "default" + app = getattr(task, "app", None) + if app is None: + from .celery_app import celery as app + return _routed_queue(app, name) + + +@functools.lru_cache(maxsize=1024) +def _routed_queue(app, name: str) -> str: + try: + queue = app.amqp.router.route({}, name).get("queue") + except Exception: # noqa: BLE001 — monitoring never breaks the task + log.warning("task_run: could not resolve the queue for %s", name) + return "default" + return getattr(queue, "name", None) or (queue if isinstance(queue, str) else "default") def _target_id_from_args(args) -> int | None: diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index db013ed..31c0fda 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -137,7 +137,13 @@ IMPORT_BATCH_KEEP_DAYS = 30 # (the import queue itself stays at the 5-min default for single # files); time_limit=2100. QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { - "ml": 25, + # ml: the scheduled auto-apply sweeps and refresh_character_prototypes run + # to a 35-min hard limit (2100s); 25 swept them mid-run (#4432). The two + # 65-min jobs have their own entries below. + "ml": 40, + # import: import_media_file's hard limit is 6 min (360s), one past the + # 5-min default this queue fell to (#4432). + "import": 10, # download_source legitimately walks 5-25 min (Patreon/gallery-dl # deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT # (1500s = 25m). The 5-min default flagged healthy in-flight walks as @@ -154,6 +160,12 @@ QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { # overrides below cover the outliers (backups, library audit). "maintenance": 75, "scan": 75, + # The long lane (#4432). Until TaskRun.queue asked the router, nothing was + # recorded here: these runs read as `maintenance` (75) or, for + # translation, `default` (5 — which failed healthy 35-min runs). The + # longest task without its own entry below is the admin family at a + # 40-min hard limit; 45 = 40 + 5. + "maintenance_long": 45, } TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "backend.app.tasks.import_file.import_archive_file": 40, @@ -179,6 +191,10 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { # external-fetch entry above — without an override a healthy in-flight walk # is swept 'RecoverySweep' at the bare 5-min default. 30 = 25 + 5. "backend.app.tasks.admin.reclaim_orphaned_attachments_task": 30, + # Head training and the manual head apply run to 65 min (3900s) — past the + # ml queue's threshold (#4432). 70 = 65 + 5. + "backend.app.tasks.ml.train_heads": 70, + "backend.app.tasks.ml.apply_head_tags": 70, } diff --git a/frontend/src/components/settings/ArchiveReextractCard.vue b/frontend/src/components/settings/ArchiveReextractCard.vue index 897a71e..a69a083 100644 --- a/frontend/src/components/settings/ArchiveReextractCard.vue +++ b/frontend/src/components/settings/ArchiveReextractCard.vue @@ -17,7 +17,7 @@ mdi-folder-zip-outline Re-extract archives now Queued ✓ - + diff --git a/frontend/src/components/settings/MissingFileRepairCard.vue b/frontend/src/components/settings/MissingFileRepairCard.vue index 301c320..e416dd5 100644 --- a/frontend/src/components/settings/MissingFileRepairCard.vue +++ b/frontend/src/components/settings/MissingFileRepairCard.vue @@ -18,7 +18,7 @@ mdi-file-remove-outline Repair missing-file records Queued ✓ - + diff --git a/frontend/src/components/settings/QueuesTable.vue b/frontend/src/components/settings/QueuesTable.vue index 6ccc372..e75a729 100644 --- a/frontend/src/components/settings/QueuesTable.vue +++ b/frontend/src/components/settings/QueuesTable.vue @@ -41,7 +41,7 @@ const props = defineProps({ const QUEUE_NAMES = [ 'default', 'import', 'thumbnail', 'ml', - 'download', 'scan', 'maintenance', + 'download', 'scan', 'maintenance', 'maintenance_long', ] function formatDepth(name) { diff --git a/frontend/src/stores/admin.js b/frontend/src/stores/admin.js index 99743cd..7e1cdb9 100644 --- a/frontend/src/stores/admin.js +++ b/frontend/src/stores/admin.js @@ -119,7 +119,7 @@ export const useAdminStore = defineStore('admin', () => { // --- Task progress polling (taps FC-3i activity dashboard) -------- /** - * Polls /api/system/activity/runs?queue=maintenance every 3s, + * Polls /api/system/activity/runs?celery_task_id= every 3s, * resolves when a task_run row with the given celery task_id * reaches a terminal status (ok / error / timeout). Returns the * row. Times out after 30 min by default. @@ -129,7 +129,9 @@ export const useAdminStore = defineStore('admin', () => { while (Date.now() < deadline) { const body = await api.get( '/api/system/activity/runs', - { params: { queue: 'maintenance', limit: 20 } }, + // By id, not by lane: these jobs run on `maintenance_long`, and a + // lane filter here is one more copy of the routing table (#4432). + { params: { celery_task_id: taskId, limit: 1 } }, ) const row = (body.runs || []).find(r => r.celery_task_id === taskId) if (row && ['ok', 'error', 'timeout'].includes(row.status)) { diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py index da5239d..6379834 100644 --- a/tests/test_api_system_activity.py +++ b/tests/test_api_system_activity.py @@ -295,3 +295,11 @@ async def test_failures_only_within_24h_window(client, _seed_failures): body = await resp.get_json() ids = {r["error_type"] for r in body["recent"]} assert "OldError" not in ids + + +@pytest.mark.asyncio +async def test_runs_filter_by_celery_task_id(client, _seed_runs): + # How a page follows a job it started, without knowing its lane (#4432). + resp = await client.get("/api/system/activity/runs?celery_task_id=tid-3") + body = await resp.get_json() + assert [r["celery_task_id"] for r in body["runs"]] == ["tid-3"] diff --git a/tests/test_celery_routing.py b/tests/test_celery_routing.py index e7a24e6..de28fd9 100644 --- a/tests/test_celery_routing.py +++ b/tests/test_celery_routing.py @@ -2,6 +2,8 @@ `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).""" +import pytest + from backend.app.celery_app import celery @@ -20,23 +22,64 @@ def test_quick_maintenance_stays_on_maintenance(): 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).""" +@pytest.mark.parametrize(("name", "queue"), [ + ("backend.app.tasks.external.fetch_external_link", "download"), + # The rows #4432 found recorded on the wrong lane: + ("backend.app.tasks.translation.translate_posts", "maintenance_long"), + ("backend.app.tasks.gpu_queue.enqueue_gpu_backfill", "maintenance"), + ("backend.app.tasks.maintenance.backfill_phash", "maintenance_long"), + ("backend.app.tasks.admin.normalize_tags_task", "maintenance_long"), + ("backend.app.tasks.backup.backup_db_task", "maintenance_long"), + ("backend.app.tasks.maintenance.vacuum_analyze", "maintenance"), + ("backend.app.tasks.not_routed.anything", "default"), +]) +def test_task_run_records_the_queue_the_router_sends_to(name, queue): + """TaskRun.queue comes from the router, not a copy of task_routes (#4432): + the stall sweep's per-queue thresholds and the System activity filters both + key off it.""" from backend.app.celery_signals import _queue_for class _T: - name = "backend.app.tasks.external.fetch_external_link" + pass - assert _queue_for(_T()) == "download" - assert ( - celery.conf.task_routes["backend.app.tasks.external.*"]["queue"] - == "download" + t = _T() + t.name = name + t.app = celery + assert _queue_for(t) == queue + + +def test_no_task_outlives_its_stall_threshold(): + """A task whose hard time limit is longer than the stall sweep's threshold + for it gets failed 'RecoverySweep' while it is still healthy — the class + #4432 found on the ml, import and long-maintenance lanes. The threshold is + resolved the way recover_stalled_task_runs resolves it: task-name override, + then queue, then the default.""" + from backend.app.celery_signals import _queue_for + from backend.app.tasks.maintenance import ( + QUEUE_STUCK_THRESHOLD_MINUTES, + STUCK_THRESHOLD_MINUTES, + TASK_STUCK_THRESHOLD_MINUTES, ) + celery.loader.import_default_modules() + checked = 0 + too_short = [] + for name, task in sorted(celery.tasks.items()): + if name.startswith("celery."): + continue + limit = getattr(task, "time_limit", None) + if not limit: + continue + threshold = TASK_STUCK_THRESHOLD_MINUTES.get( + name, + QUEUE_STUCK_THRESHOLD_MINUTES.get(_queue_for(task), STUCK_THRESHOLD_MINUTES), + ) + checked += 1 + if limit / 60 > threshold: + too_short.append(f"{name}: limit {limit / 60:.0f} min > sweep {threshold} min") + assert checked > 20, "the task registry did not load — this guard checked nothing" + assert not too_short, "\n".join(too_short) + def test_backfill_phash_runs_on_the_long_lane(): """It lives in maintenance.py, so the quick-lane glob matches it too —