Skip to content

Match listener startup (Celery Beat)

How a match listener is scheduled by Beat, enqueued on the match_listeners queue, configured, and run. For status badges, failures, hard kill vs Stop, and whether Beat will restart, see Lifecycle cases. For in-process event / question behavior after startup, see Match listener system. For Pydantic env-file precedence alone, see Environment configuration.

Processes involved

Process Role
celery-beat RedBeat scheduler; every minute may enqueue dispatch_upcoming_match_listeners
celery-worker Default queues (refresh jobs, the dispatch task itself)
celery-listener-worker Consumes only match_listeners; runs long-lived start_match_listener
srsim For test clones: continuous session producer (feed keeps advancing without a listener; --workers 1)

Operator Stop ends the listener only. It does not rewind the srsim session clock — see continuous sessions what-ifs.

Local compose (docker-compose.yml) includes celery-listener-worker with --concurrency=1 (one match at a time). Staging uses higher concurrency. Raise local concurrency (or run more listener-worker containers) to listen to several matches in parallel — each start_match_listener task occupies one pool slot for the whole game.

Staging Compose wires these in docker-compose.staging.yml. Beat registers periodic entries when the Celery app configures (tasks.beat.setup_periodic_tasks ← tasks.beat_schedules.PERIODIC_TASKS).

When Beat schedules listeners

In src/tasks/beat_schedules.py:

RedBeat name When enabled Kwargs
dispatch_upcoming_match_listeners APP_ENV != develop none (live + test)
dispatch_upcoming_test_match_listeners APP_ENV=develop is_test=True

Both call the same Celery task every 1 minute. Develop is test-clones only so local Beat never auto-starts live Sportradar matches.

Manual one-shot (any env):

make dispatch_upcoming_match_listeners
# defaults to within_minutes=0; optional: KW='within_minutes=30 is_test=true'

Other periodic refresh tasks (league / teams / seasons / MLB and NHL daily matches / NFL weekly matches / daily active-season matches) are enabled whenever APP_ENV != develop. They keep match rows current so dispatch has something to find; they do not start listeners themselves.

End-to-end flow

sequenceDiagram
    participant Beat as "celery-beat (RedBeat)"
    participant Broker as "Redis broker db2"
    participant Worker as celery-worker
    participant LWorker as celery-listener-worker
    participant DB as PostgreSQL
    participant SR as Sportradar stream

    Beat->>Broker: enqueue dispatch_upcoming_match_listeners
    Worker->>DB: find matches needing listeners
    Worker->>Broker: apply_async start_match_listener (queue=match_listeners)
    LWorker->>DB: load Match + league and merge settings
    LWorker->>DB: MatchListenerRecord startup (unless dry-run)
    LWorker->>SR: MatchListener.run() (blocks for the game)
    LWorker->>DB: MatchListenerRecord shutdown

1. RedBeat fires the dispatch task

On Celery configure, PERIODIC_TASKS is written into RedBeat (Redis db1). Beat reads those entries and, on each schedule tick, sends the Celery task name to the broker (Redis db2).

Relevant code:

  • src/tasks/beat.py — setup_periodic_tasks
  • src/tasks/beat_schedules.py — dispatch RedBeat entries (all-match vs develop test-only)
  • src/tasks/celery.py — RedBeat URL, broker, imports tasks.beat

2. dispatch_upcoming_match_listeners selects matches

Celery wrapper: src/tasks/jobs/match_listeners.py. Async query: listener.lib.start.find_upcoming_matches_needing_listeners.

Under a short Redis task_lock("dispatch_upcoming_match_listeners"), the task:

  1. Selects enabled-league matches with Match.auto_dispatch_listener (default True) that still need a listener:
  2. Status inprogress: always eligible (until the match ends)
  3. Status scheduled / created / delayed: scheduled in [now - ACTIVE_LISTENER_MAX_AGE, now + within_minutes] (ACTIVE_LISTENER_MAX_AGE = 12 hours, within_minutes default 10)
  4. auto_dispatch_listener=False skips auto-dispatch only; manual / force start is unchanged
  5. Optional is_test kwarg: True / False filters test vs live; None (default, non-develop Beat) includes both. Develop Beat passes is_test=True.
  6. Drops matches that already have a recent open MatchListenerRecord (ended_at is null and created_at within 12 hours), unless forced (dispatch always uses force=False). Test clones with listener_dry_run still write lifecycle rows so this skip applies.
  7. For each remaining match, SET NX a Redis pending key (celerylock:pending_start_match_listener_{id}), then enqueues with a fresh Celery task id (unique per enqueue so stop + restart does not collide with Postgres celery_taskmeta):
# Dot-separated: both UUIDs contain dashes, so a dash-joined id cannot be split
# back into internal match id and enqueue id.
task_id = f"start-match-listener.{match.id}.{uuid4()}"
# Persist on Match.listener_start_settings['celery_task_id'] for stop-while-queued
start_match_listener.apply_async(
    kwargs={"match_id": str(match.id), "overrides": None, "force": False},
    queue=MATCH_LISTENER_QUEUE,  # "match_listeners"
    task_id=task_id,
)

If the pending key already exists, the match is counted as skipped (already queued or running). Manual / force starts use the same unique-id helper.

match_id here is the internal match id (Match.id), not the Sportradar match id (Match.sr_id).

Stop clears the stored enqueue id after revoke so the next start can issue a new task id.

3. start_match_listener runs on the listener worker

Same module; bound to queue match_listeners, no soft/hard time limits (game length). Before running:

  1. If force: clear Redis start barriers (pending, attempts/exhausted, start task_lock) — does not revoke Celery or close lifecycle rows
  2. Validate optional overrides keys against MatchListenerEnvironment fields
  3. If force is false and an active open lifecycle row exists → skip (active = open + heartbeat within 3 minutes + created within 12h; stale open rows are closed as abandoned exit 14 before the check)
  4. Acquire Redis task_lock(f"start_match_listener_{match_id}", timeout=12h)
  5. Re-check the open-row guard after the lock (unless force)
  6. Call run_match_listener(match_id, overrides) via sync_run_async
  7. Clear the dispatch pending key when finished (or when skipping after checks)

To stop a live listener, use stop_match_listener (Redis stop flag + Celery revoke with SIGKILL + close rows + clear barriers), not force.

4. Settings merge and listener process

run_match_listener (src/listener/lib/start.py):

  1. Loads Match and refreshes league
  2. Builds MatchListenerEnvironment via resolve_listener_environment
  3. Reads whether the match is is_test (no Match.status write — test status is owned by the continuous srsim session)
  4. Records a MatchListenerRecord startup row (skipped in dry-run except for is_test, so dispatch does not re-enqueue); starts a 30s DB heartbeat loop on that row. The Celery task id is captured in the sync worker context and passed in (current_task is unavailable inside async_to_sync).
  5. Instantiates MatchListener(config=...) and await listener.run(), with a 1s poll for the Redis operator-stop flag that cancels the run (exit 15)
  6. On exit: cancel stop-watch + heartbeat; then records shutdown on the lifecycle row when one was written

Recordings / logs: event file is always {shortcode}.json (reopened and appended across restarts after JSON repair; no epoch rotation). Text logs are {shortcode}.{MatchListenerRecord.id[:8]}.log when a lifecycle row exists.

For continuous is_test sessions, srsim owns Match.status (scheduled → inprogress when the session producer starts; delayed / inprogress around game_delay_at_event; closed when the stream completes). Operator stop of the listener does not close the match.

Live (non-test) Match.status is written from the push stream game.status and from Beat schedule refresh. See Match.status: freeze, heal, and stream.

The listener then opens the Sportradar push stream and runs the question pipeline documented in the shared listener context.

Match.status: freeze, heal, and stream

Two writers, two feeds. Do not conflate them.

Writer Feed What it may persist
Listener (live) Push stream payload.game.status (then payload.status / game.status / status) inprogress, delayed, terminal. Never pregame. Never reopens terminal.
Beat schedule refresh (daily / weekly) Sportradar schedule API Incoming status, after Match.refreshed_status

The stream is the live source of truth. The schedule API often still says scheduled after first event. That lag is normal.

Freeze

Once the row is inprogress or terminal, a later schedule refresh must not rewind it to scheduled / created. Otherwise a finished or live game would look upcoming again the next morning.

Heal (leftover the freeze cannot see)

Freeze only protects rows that are already inprogress or terminal. The heal covers the leftover: local status is still scheduled (or created / delayed), the schedule API still says scheduled / created, we already issued in-game questions, and Match.scheduled is in the past.

Why it exists: A game was played and we issued props, but the row never left scheduled (listener died, or status was never written). A week later Beat still sees SR scheduled. Without a heal the match looks like an upcoming game forever. Heal promotes it to inprogress. It becomes closed / complete only when SR actually sends that, or a later listener applies stream complete.

Heal must not invent closed. “We issued a question” + “clock past scheduled start” is also true minutes after the first live event, while the game is still underway.

Counterexample (do not regress): the stream already had payload.game.status = inprogress on the first live event. A schedule refresh minutes later still saw scheduled. Heal-to-closed award-alled the winner question while the stream continued. The listener process stayed Running because it only reloads Match.status every 50 idle heartbeats.

Examples

Schedule refresh (Match.refreshed_status). “Qs” means at least one in-game question exists. “Start past” means now is after Match.scheduled.

Local Incoming (schedule API) Qs Start past Persist Rule
scheduled scheduled no yes or no scheduled Trust SR; nothing to heal
scheduled scheduled yes no scheduled Heal waits until scheduled start
scheduled scheduled yes yes inprogress Heal — leftover still looks upcoming (week-later Tuesday game)
created / delayed scheduled / created yes yes inprogress Heal — same leftover, local still pre-terminal
scheduled inprogress either either inprogress Trust SR; no heal needed
scheduled closed either either closed Trust SR; heal never invents closed
inprogress scheduled either either inprogress Freeze — do not rewind a live game
inprogress delayed either either delayed Trust SR delay
closed scheduled either either closed Freeze — do not reopen a finished game
complete closed either either closed Trust SR; both terminal

Live stream (Match.apply_stream_status). No heal; no scheduled-start clock.

Local Incoming (stream game.status) Persist Rule
scheduled inprogress inprogress First live event on the stream
inprogress delayed delayed Delay from the feed
delayed inprogress inprogress Delay ended
inprogress complete complete Game over on the feed
scheduled scheduled scheduled Stream must not write pregame
closed inprogress closed Never reopen terminal

Settings / config hierarchy at startup

Two layers stack: Pydantic Settings (env / files / defaults), then start-time merge (match row + call kwargs). The start path always goes through resolve_listener_environment.

Layer A — MatchListenerEnvironment (Pydantic)

Same rules as Environment configuration:

  1. Keyword arguments to get_listener_environment(**…) / MatchListenerEnvironment(**…)
  2. Process environment (os.environ)
  3. Env file: repo-root .env
  4. Field defaults on BaseEnvironment + MatchListenerEnvironment

Shared fields (database_url, redis_url, sportradar_api_key, …) come from BaseEnvironment. Listener-only fields (listener_dry_run, listener_sr_url_league, …) are declared on the subclass.

Layer B — start-time merge (resolve_listener_environment)

When starting for a known Match, kwargs passed into Pydantic are assembled as:

strongest → weakest among start inputs

  call overrides          (Celery task / CLI kwargs)
  match.listener_start_settings   (JSON on Match row)
  then Layer A for any key not set above

Additionally, unless the call overrides already set them:

Injected field Source
listener_sr_request_match_uuid str(match.sr_id) (Sportradar match id)
listener_sr_url_league match.league.name

So Beat-driven starts (overrides=None) typically get the Sportradar match id and league from the database, optional per-match tweaks from Match.listener_start_settings, and everything else from the worker’s process env / .env.

flowchart TB
    subgraph start["resolve_listener_environment"]
        C["1 Call overrides"]
        M["2 Match.listener_start_settings"]
        I["Inject sr_id + league.name if not in call overrides"]
    end

    subgraph pydantic["get_listener_environment merged kwargs"]
        P1["kwargs from merge above"]
        P2["process environment"]
        P3["repo-root .env"]
        P4["field defaults"]
    end

    C --> M --> I --> P1
    P1 --> P2 --> P3 --> P4

Match.listener_start_settings may also hold reserved metadata keys that are not env fields (today: celery_task_id for revoke before a lifecycle row exists). Those keys are stripped before override validation / env merge.

Override keys (call kwargs and non-metadata settings) must be names of MatchListenerEnvironment model fields; unknown keys raise ListenerStartSettingsError.

Exit codes → Celery task outcome

start_match_listener maps listener SystemExit codes as follows (defaults):

Code Meaning Celery
0 / None Clean return SUCCESS
10 Max consecutive heartbeats SUCCESS
13 Match no longer in progress SUCCESS
14 Abandoned (stale heartbeat; often after hard stop) Recorded on row by dispatch/start/sweeper cleanup, not a live exit
15 Stopped via stop_match_listener Recorded on row by stop ops (Stopped badge)
16 Unexpected cancel (CancelledError) FAILURE
9 Unexpected exception FAILURE
11 Data / HTTP / JSON error FAILURE
12 Connection / timeout FAILURE
Other non-zero Unexpected FAILURE

Settings / resolve errors (ListenerStartSettingsError) also raise → Celery FAILURE. Intentional skips (active listener, lock not acquired, exhausted attempts) return a SUCCESS dict with skipped_* flags.

Failed starts increment a Redis attempt counter (TTL ≈ 12h). After 15 failures the match is marked exhausted: dispatch skips further enqueues. force=True clears attempt / exhausted state. Each failed attempt (1…15) is a Celery FAILURE (and a Sentry event); fingerprint is per internal match id. Details: Sentry — match listeners.

Ops testing: set listener_force_exit_code (env, clone body, or listener_start_settings) to force SystemExit before stream work.

See also Celery ack behaviour.

Lifecycle cases (status, failures, recovery)

Canonical catalog for operators and agents. Admin badges are derived from the lifecycle row (MatchListenerRecord); Redis barriers and Match.auto_dispatch_listener decide whether Beat will start another run.

Admin badges

Badge Row state Meaning
Running ended_at null, heartbeat within ~3 minutes Process is alive (DB heartbeat loop)
Stale ended_at null, heartbeat older than ~3 minutes Until sweeper/dispatch abandon; often a hard-dead worker
Abandoned Closed with exit 14 Hard fail / orphan: heartbeat went stale; barriers cleared at abandon when no active sibling remains
Stopped Closed with exit 15 Operator Stop; barriers cleared; auto-dispatch off
Ended Closed with any other exit Graceful stream end, failure exit, or unexpected cancel (16)

Matches board and Match Listeners use the same derived statuses (listener_status / status).

Case catalog

Case How it happens Admin / exit Redis start barriers auto_dispatch_listener What happens next
Normal start (Beat) Dispatch enqueues start_match_listener New Running row Pending + start lock held while starting/running Must be on to be selected Runs for the game; graceful end → Ended
Manual Start Admin / API force=true New Running row Cleared then re-acquired Unchanged Same as a normal run; does not revoke a live task; admin disables Start while Running
Graceful end Stream done, max heartbeats (10), match not in progress (13), clean 0 Ended (0 / 10 / 13) Released on task unwind; enqueue id cleared Unchanged No auto re-enqueue while status/schedule no longer eligible (or open active row)
Rain delay (live) Stream or Beat sets Match.status=delayed; listener stays up, logs status checks, writes synthetic delay markers, and resets the consecutive-heartbeat counter on each delayed check so the idle max does not fire Stays Running Held Unchanged On resume (inprogress) continues same process/archive; postpone/cancel → exit 13
Stream / process FAILURE Exit 9 / 11 / 12 / other non-success Ended (failure code); Celery FAILURE Pending cleared on failure path; attempts incr Unchanged Dispatch may retry until attempt cap (15) / exhausted
Unexpected cancel CancelledError not from operator stop Ended (16); Celery FAILURE Cleared on failure path Unchanged Treat like other failures
Start skipped (already active) Open row with fresh heartbeat, or pending/start lock held Unchanged Unchanged — Celery SUCCESS with skipped_*; no new row
Start exhausted 15 failed starts — Exhausted flag set — Dispatch skips until force=true / barrier clear
Hard kill (worker crash, compose stop) Process dies without shutdown write Running → Stale → Abandoned (14) after ~3m Orphaned until abandon; cleared when abandon persists if no active sibling Stays on Beat stale sweeper (~1m non-develop) and/or dispatch may re-enqueue once abandoned. Staging deploy skips the wait: post-up sweep_stale_match_listeners(heartbeat_stale_seconds=0) then dispatch_upcoming_match_listeners (still exit 14, not operator Stop)
Operator Stop Admin Stop listener / POST .../stop-listener Stopped (15, stopped by operator) Cleared Set to off Beat will not re-enqueue; use Start or turn dispatch back on
Dispatch off Matches toggle or after Stop — — off Match excluded from find_upcoming_* until re-enabled
Revoke UniqueViolation noise Reusing Celery task ids (legacy) or result-backend races — — — Bookkeeping warning; prefer unique per-enqueue task ids (current code)

Operator quick actions

You want… Do this
Listener back after hard fail Wait for Abandoned + next dispatch (staging Beat ~1m), or Start listener now
Listener back after staging deploy Deploy workflow already sweeps with heartbeat_stale_seconds=0 then dispatches; or enqueue the same manually. Do not use Stop
Listener down and stay down Stop listener (turns auto-dispatch off)
Listener now despite barriers Start listener (force=true) — prefer Stop first if still Running
Auto-start again after Stop Turn Listener dispatch on (Matches) and/or Start listener
Local re-dispatch make dispatch_upcoming_match_listeners (or wait for develop Beat test-only schedule)

Step-by-step hard-fail recovery: Correcting a hard-failed listener.

Practical examples

Goal How
Staging / main auto-start Beat → dispatch_upcoming_match_listeners → start_match_listener (live + test)
Local develop auto-start Beat → dispatch_upcoming_test_match_listeners (is_test=True only)
Local develop one-shot make dispatch_upcoming_match_listeners (KW='is_test=true' to mirror Beat)
Dry-run one match Set listener_dry_run in Match.listener_start_settings or pass overrides={"listener_dry_run": true}
Force a test exit code Set listener_force_exit_code (clone UI / env override)
Manual start (clear barriers) Admin Start listener or start_match_listener(..., force=True) — does not revoke a live task
Stop and block auto re-enqueue Admin Stop listener — exit 15 + auto_dispatch_listener=False
Recover after hard stop See Lifecycle cases and Correcting a hard-failed listener
CLI without Celery listener.listen / run_match_listener_from_config — builds env from env files only, then looks up match by Sportradar match id for lifecycle logging
Local live record (Compose) make listen SR_ID=<sportradar-match-id> — one-off listener container against live Sportradar (not srsim); record-only by default (SKIP_PROCESSING=true). Writes ./data/listener/{league}/{shortcode}.json. Optional LEAGUE=nfl (default), SR_ENV=trial, LISTEN_MATCH_ID.

Manual enqueue (worker container already has broker + env):

make dispatch_upcoming_match_listeners
# defaults to within_minutes=0; optional: KW='within_minutes=30 is_test=true'

make enqueue TASK=start_match_listener \
  KW='match_id="…" force=true' QUEUE=match_listeners ENV=staging
  # match_id = internal Match.id, not Match.sr_id

See Makefile and src/tasks/enqueue.py.

Deduping and locks

Locks live on Redis db0 with key prefix celerylock: (SET NX EX, token-safe release). See tasks.celery.task_lock.

Guard Purpose
Open MatchListenerRecord with fresh heartbeat (~3m) Skip start if a listener already looks active; stale heartbeats are closed as abandoned (exit 14) and allow restart when no active sibling remains
celerylock:pending_start_match_listener_{id} Dispatch / admin / clone cannot double-enqueue the same match
celerylock:start_match_listener_attempts_{id} Count failed starts (max 15)
celerylock:start_match_listener_exhausted_{id} After 15 failures, dispatch skips enqueue
task_lock("dispatch_upcoming_match_listeners") One dispatch at a time
task_lock("sweep_stale_match_listeners") One stale sweeper at a time; optional heartbeat_stale_seconds (Beat omits; staging deploy uses 0)
task_lock("start_match_listener_{match_id}", 12h) One starter per match for the game duration (match_id = internal match id)

Stop vs force vs abandon — see the Lifecycle cases catalog for the full matrix. Short form:

Action Barriers Auto-dispatch after
Abandoned (14) Cleared when abandon persists Stays on
Operator Stop (15) Cleared Off
Manual Start (force=True) Cleared then re-acquired Unchanged

Hard stop implications

A graceful exit releases Redis locks on task unwind and writes ended_at / exit code. A hard kill does not — the row stays open until heartbeat staleness (~3 minutes), then Abandoned (14) clears barriers so dispatch can recover. Details and the full case list: Lifecycle cases.

Correcting a hard-failed listener

Usually nothing: wait for Abandoned on the next dispatch/start check, then Beat / dispatch_upcoming_match_listeners re-enqueues (if auto-dispatch is still on). See Lifecycle cases for the case matrix.

Use the steps below when you need a listener now, or auto-dispatch was turned off.

match_id below is always the internal Match.id (UUID), not the Sportradar id / shortcode.

Path What you do Who starts the next listener
Wait for abandon + dispatch Confirm Abandoned (14); leave auto-dispatch on Dispatcher on the next run
Start listener now Admin Start listener / force=true enqueue You start immediately
Do not auto-restart Stop listener Turns off auto_dispatch_listener; Start manually or re-enable dispatch later
  1. Confirm nothing is actually running
  2. Admin Match Listeners: badge is Stale or Abandoned (not Running with a fresh heartbeat).

  3. celery-listener-worker logs: no active stream subscribe for that shortcode.

  4. Note the internal match id from the admin card.

  5. Path A — wait for dispatch (preferred after hard fail)

  6. After abandon, Redis barriers are cleared automatically.
  7. Staging: Celery Beat runs dispatch about every minute.
  8. Local develop: Beat typically does not schedule this task (APP_ENV=develop). Run:
make dispatch_upcoming_match_listeners
  1. Path B — start now:
make enqueue TASK=start_match_listener \
  KW='match_id="<internal-match-id>" force=true' \
  QUEUE=match_listeners

Or Start listener on Matches / Match Listeners (same force=true). force=true clears pending / attempts / start lock and skips the open-row guard. It does not revoke a live worker — only use it after step 1.

  1. Path C — operator stop (no auto re-enqueue)
  2. Stop listener revokes tasks, closes rows (exit 15), clears barriers, and sets auto_dispatch_listener=False.

  3. Beat will not start another listener until you turn dispatch back on (Matches settings) or use Start listener.

  4. Verify

  5. Listener worker logs a new subscribe / Event received: … (not already running / skipped_active).
  6. Admin shows a new Running row with a fresh heartbeat.

Redis broker visibility_timeout is 12h (tasks/celery.py) so late-acked listener tasks are not redelivered while still running. Raised FAILURE also acks (task_acks_on_failure_or_timeout); Beat re-enqueue after abandon is the hard-fail recovery path (see Celery). Layout: db0 app/locks, db1 RedBeat, db2 Celery broker.

Code map

Concern Location
Periodic registry src/tasks/beat_schedules.py
RedBeat registration src/tasks/beat.py
Celery app / queues / locks src/tasks/celery.py
Dispatch + start tasks src/tasks/jobs/match_listeners.py
Upcoming query, settings merge, run src/listener/lib/start.py
Listener loop src/listener/listen.py
Settings classes src/lib/base_environment.py, src/listener/listener_environment.py
Lifecycle rows src/lib/db/models/match_listener.py, src/listener/lib/match_listener_db_log.py
Per-match start JSON Match.listener_start_settings in src/lib/db/models/match.py