feat: configure Celery with per-feature queue lanes and a smoke task

Routes are pre-declared for FC-2/FC-3 task modules (import, ml, thumbnail,
download, scan, maintenance). Queue lanes match the ImageRepo pattern where
beat+maintenance run on a separate worker so long imports don't starve
periodic tasks. Smoke ping task confirms the wiring in eager mode for CI.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-14 07:34:47 -04:00
parent d6a156dcd2
commit 8773f2aae6
4 changed files with 77 additions and 0 deletions
+45
View File
@@ -0,0 +1,45 @@
"""Celery configuration with separate queue lanes.
Queues:
import — filesystem import path (FC-2)
ml — WD14 + SigLIP inference (FC-2; runs in the ml-worker image)
thumbnail — image and video thumbnail generation (FC-2)
download — gallery-dl tasks (FC-3)
scan — periodic source checks (FC-3) — kept separate so long imports
don't starve the scheduler
maintenance — pHash recomputation, centroid rebuild, etc. (FC-2/FC-3)
default — anything not explicitly routed
"""
from celery import Celery
from .config import get_config
def make_celery() -> Celery:
cfg = get_config()
app = Celery(
"fabledcurator",
broker=cfg.celery_broker_url,
backend=cfg.celery_result_backend,
include=["backend.app.tasks.smoke"], # FC-2/FC-3 extend this list
)
app.conf.update(
task_default_queue="default",
task_routes={
"backend.app.tasks.import_file.*": {"queue": "import"},
"backend.app.tasks.ml.*": {"queue": "ml"},
"backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
"backend.app.tasks.download.*": {"queue": "download"},
"backend.app.tasks.scan.*": {"queue": "scan"},
"backend.app.tasks.maintenance.*": {"queue": "maintenance"},
},
# Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
task_acks_late=True,
worker_prefetch_multiplier=1,
broker_connection_retry_on_startup=True,
)
return app
celery = make_celery()
View File
+8
View File
@@ -0,0 +1,8 @@
"""Smoke task used during FC-1 to confirm Celery routing works end-to-end."""
from ..celery_app import celery
@celery.task(name="backend.app.tasks.smoke.ping")
def ping(value: str = "pong") -> str:
return value