Spaces:
Runtime error
Runtime error
File size: 2,556 Bytes
31e8a9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """Celery application (PRD §14.4), tuned for the Upstash free tier.
UPSTASH BUDGET: the free tier allows 10,000 Redis commands/day. Celery's default
0.1 s broker polling would burn ~864K/day, so ``polling_interval`` is 2.0 s.
Even that is ~43K polls/day for ONE worker — above the free quota. Run a single
worker, and expect to need a paid Upstash plan (or local Redis) for sustained
use. Results are ignored (the DB stores outcomes), saving result-backend writes.
Worker start::
cd backend && celery -A app.worker.celery_app worker --loglevel=info
Startup behaviour:
* ``worker_init`` loads the ML pipeline and exits the process if any artifact is
missing (hard-fail policy, DECISIONS.md) — before consuming any message.
* ``worker_ready`` starts the outbox dispatcher thread (:mod:`app.worker.outbox`).
"""
from __future__ import annotations
import logging
from typing import Any
from celery import Celery
from celery.signals import worker_init, worker_ready
from app.config import settings
logger = logging.getLogger(__name__)
celery_app = Celery("vera", broker=settings.broker_url, include=["app.worker.tasks"])
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
timezone="UTC",
enable_utc=True,
task_ignore_result=True,
broker_transport_options={"visibility_timeout": 3600, "polling_interval": 2.0},
broker_connection_timeout=3,
broker_connection_retry_on_startup=True,
worker_prefetch_multiplier=1,
task_acks_late=True,
task_reject_on_worker_lost=True,
# Serialize inference so this worker does not duplicate the loaded FP32
# models. Deployment memory must be sized and verified separately; the
# free-tier 512 MB target is below the documented production requirement.
worker_pool="solo",
worker_concurrency=1,
)
@worker_init.connect
def load_models_or_exit(**_: Any) -> None:
"""Fail fast: a worker without valid artifacts must not consume jobs."""
from app.worker.tasks import get_pipeline
try:
get_pipeline()
except Exception as exc:
logger.critical("VERA model initialisation failed: %s", exc)
# SystemExit (a BaseException) escapes Celery's signal error handling.
raise SystemExit(1) from exc
@worker_ready.connect
def start_outbox(**_: Any) -> None:
"""Run the outbox dispatcher alongside the worker (no extra process needed)."""
from app.worker.outbox import start_background_dispatcher
start_background_dispatcher()
logger.info("Outbox dispatcher thread started")
|