Skip to content

Celery in this repository

How background tasks are configured, discovered, and run in production versus pytest.

Application instance

  • Shared Celery app object: src/tasks/celery_app.py (celery, locks, broker/conf). Job modules import from here (from tasks.celery_app import celery) so they do not pull full task registration.
  • Full worker / beat / Flower: celery -A tasks.celery:celery — registers every job module + listener.tasks and wires RedBeat (PERIODIC_TASKS). Use the module:attr form; bare -A tasks.celery is ambiguous with the tasks package.
  • Listener worker: celery -A tasks.celery_listener:celery — registers only tasks.jobs.match_listeners (no MLB/NHL/NFL refresh, archive/S3, or answer-eval imports). Compose points celery-listener-worker at this entry.
  • Task modules decorate callables with @celery.task:
from tasks.celery_app import celery

@celery.task
def my_task(arg: str) -> None:
    ...
  • Package __init__ files under tasks / tasks.jobs stay empty of job re-exports so import tasks does not load Sportradar/S3 stacks.

Broker and result backend

  • Broker: Redis db2 (REDIS_URL with path /2; see tasks/celery.py). visibility_timeout is 12 hours so long-running match_listeners tasks are not redelivered while still running (task_acks_late).
  • Ack behaviour (src/tasks/celery.py):
  • task_acks_late=True — the broker message is acknowledged after the task body finishes.
  • task_acks_on_failure_or_timeout=True — a raised FAILURE (and soft/hard time limits) still acks the message. The broker does not redeliver that invocation.
  • Re-runs of start_match_listener come only from Beat / manual / force enqueue (pending + attempt-cap keys), not from Celery retry.
  • If a worker is hard-killed before ack, Redis may redeliver after visibility_timeout (12h). That path is distinct from Beat re-enqueue.
  • RedBeat: Redis db1.
  • App cache / Celery locks / live play: Redis db0 (celerylock:{id} via task_lock; play open-questions cache play:open:shared and pub/sub play:match:* — see Live play client and Environment configuration).
  • Result backend:
  • Non-test: db+{database_url} so results are stored in PostgreSQL (Celery’s database backend expects the db+ prefix).
  • APP_ENV=test: Redis only. At import time the test database URL may not be set yet, so the app avoids binding the backend to Postgres during test collection (see the comment in tasks/celery.py).
  • result_expires: 7 days (celery.conf.result_expires). Expired celery_taskmeta rows are removed by the built-in celery.backend_cleanup task, scheduled daily in PERIODIC_TASKS as celery_backend_cleanup_daily.
  • Fire-and-forget jobs (answer eval, archive/sweep, Sportradar refreshes, dispatch) use @celery.task(ignore_result=True) so they do not write result rows. start_match_listener still stores results for Flower / ops status.

Worker process recycle

Compose workers pass Celery --max-tasks-per-child:

  • celery-listener-worker: --max-tasks-per-child=1 so each finished match recycles the prefork child (clears per-game heap / FD residue). Uses -A tasks.celery_listener:celery so idle children do not import archive/S3 or league refresh job modules (base RSS ~20–25 MiB lower per process than the full app).
  • celery-worker: --max-tasks-per-child=200 for short tasks (-A tasks.celery:celery).

Import-smoke coverage: src/lib/tests/test_base_memory_import_smoke.py (API and listener entry must not load lib.s3 / botocore / unrelated jobs).

Worker log lines

Every worker log line carries the active job name and a task-id shortcode:

[2026-07-24 19:20:00,000: INFO/ForkPoolWorker-1] refresh_mlb_league[a1b2c3d4] Refresh complete
  • Format: job_name[task_id_shortcode], where job_name is the bare task function name (not the dotted path). The shortcode is the first 8 characters of the task UUID; custom ids ending in a UUID (such as start-match-listener.{internal_match_id}.{enqueue_uuid}) use that trailing UUID rather than the string prefix. The full task id stays in Flower and DB bookkeeping (MatchListenerRecord.celery_task_id, Match.listener_start_settings['celery_task_id']).
  • Idle worker / beat lines and non-worker CLI runs render both fields as -.
  • Implemented in src/tasks/task_logging.py: task_prerun stashes the job name and shortcode in context vars, and CeleryTaskLogFilter copies them onto every LogRecord. Context vars set in task_prerun remain visible through the async_to_sync body (same worker thread), unlike Celery's stock current_task fields, which are None inside async_to_sync.
  • src/tasks/celery.py sets worker_log_format / worker_task_log_format. Match-listener loggers keep a plain LISTENER_LOG_FORMAT (no job/task fields): under a worker, Celery's worker_redirect_stdouts already wraps stdout with the worker format, so putting those fields in the listener format would double the prefix on console.

Sentry

General Celery workers initialize Sentry on worker_init when SENTRY_DSN_CELERY is set; the listener-queue worker uses SENTRY_DSN_LISTENER. Full behaviour (integrations, listener scope, start-failure fingerprinting): Sentry.

Task autodiscovery

celery.autodiscover_tasks() is configured for:

  • tasks with related_name="jobs" → tasks.jobs
  • api, srsim, listener with the default related_name="tasks"

Put Celery wrappers under src/tasks/jobs/. Async implementations stay in lib (for example Sportradar refresh modules).

Running tasks in production

  • Workers consume tasks from Redis.
  • Beat (RedBeat) can schedule periodic work; see src/tasks/README.md for RedBeat and task_lock usage. Sportradar match rows: 10-minute window jobs (MLB daily / NHL daily / NFL weekly) plus once-daily refresh_active_season_matches (07:00 UTC) for every non-closed season on enabled MLB/NHL/NFL. Manual refresh_*_matches stays on-demand for a full year / first load.
  • Match event metrics (MATCH_EVENT_METRICS_ENABLED, default off): live listener flush is best-effort. Catch-up is ingest_match_event_metrics / sweep_pending_match_event_metrics (every 5 min) plus daily prune_match_event_metrics (14-day retention). Staging yesterday backfill:
make enqueue TASK=sweep_pending_match_event_metrics \
  KW='scheduled_from="2026-08-31T00:00:00Z" scheduled_to="2026-09-01T00:00:00Z"' \
  ENV=staging

Skip is_test matches. Jobs call get_base_environment() only inside the task body. Staging charts: Metabase (collection Match timelines, dashboard Match timeline). Role metabase needs SELECT on match_event_metrics and match_timeline_points (revision b8e4c1a90d27; no-op locally where that role does not exist). - Callers enqueue work with .delay(*args, **kwargs) or .apply_async(...).

Test environment (pytest)

src/tests/conftest.py registers pytest_configure, which sets:

  • celery.conf.task_always_eager = True — tasks run in the same process immediately instead of being sent to a worker.
  • celery.conf.task_eager_propagates = True — exceptions inside the task body propagate to the test (failed assertions and errors fail the test).

So integration tests do not need a running Celery worker; .delay() still executes the full task body.

Database sessions and eager tasks

Eager tasks are real task invocations. Implementations that open a new database session (for example async with db_session() inside a task wrapped with async_to_sync) use the shared engine but a different connection than a test’s db fixture session. Uncommitted work that only lives on the test session (including savepoint-only “commits”) is not visible to that separate connection.

Production: commit the issuance (and any prerequisite rows) before enqueueing so workers see committed data. evaluate_question_answers_batch retries transient failures (exponential backoff + jitter, max 3) — safe because scoring only updates QuestionAnswer rows from already-committed issuance resolution state.

Pytest: enqueue_answer_evaluation_batches in src/listener/lib/stages/answer_evaluation.py checks celery.conf.task_always_eager and, when true, scores answers on the same AsyncSession the listener used (via _execute_question_logic) instead of calling .delay(). That is not because sync_run_async is wrong under eager mode — it is because eager .delay() still runs the task on a different DB connection than the test db fixture, so it would not see rows that only exist inside the fixture’s outer transaction / savepoints. Production workers use .delay() and their own session as usual.

Other tests that call scoring helpers directly should pass the test session into evaluate_question_answers_batch_impl(..., db=db) (see lib/tests/question_engine/conftest.py).

Async database code inside sync Celery tasks

Celery task bodies are synchronous. The usual pattern in this repo is:

  1. Implement async def ..._impl(...) using db_session() from src/lib/db/utils.py.
  2. Wrap it with sync_run_async from src/tasks/async_bridge.py and call that from @celery.task:
from tasks.async_bridge import sync_run_async

_run_impl = sync_run_async(my_async_impl)

@celery.task
def my_task(arg: str) -> None:
    _run_impl(arg)

How sync_run_async behaves

The helper does not read task_always_eager. It only checks whether the current thread already has a running asyncio loop (asyncio.get_running_loop()).

Where the task runs Typical loop on the task thread? What happens
Production worker (task_always_eager false) No — worker threads are plain sync async_to_sync runs in that thread. No extra thread pool. Same as writing async_to_sync(my_async_impl)(...) by hand.
Pytest + eager Celery (task_always_eager true) inside an async test Yes — pytest-asyncio owns the loop The wrapped coroutine runs in a short-lived worker thread so async_to_sync can create/use its own loop (avoids “AsyncToSync in the same thread as an async event loop”).

So task_always_eager being false in production does not disable or change this helper — workers simply take the “no running loop” path every time.

:func:asgiref.sync.sync_to_async is for calling sync code from async code; it does not replace sync_run_async for Celery tasks.

Prefer the same-session eager path in calling code when task_always_eager is set and tests use a shared AsyncSession (see above).

Examples: src/listener/tasks.py, src/lib/sportradar/nhl/refresh_nhl_matches.py.

  • Match listener startup — Beat → dispatch → match_listeners queue → settings merge.
  • Test system — shared pytest plugin, including the Celery row in the pytest_configure table.
  • src/tasks/README.md — periodic tasks, locking, and operational notes.