Task Run Recorder#

The Task Run Recorder (TaskRunRecorder) is a server-side service that consumes client-emitted task run events from the events messaging topic and persists them as task runs and states in the database. It is defined in src/prefect/server/services/task_run_recorder.py and extends RunInEphemeralServers, Service 1.

The service buffers incoming messages into an in-memory asyncio.Queue, then flushes them in bulk — either when the queue reaches write_batch_size or on a periodic flush_interval timer 2. Only events with event.event.startswith("prefect.task-run") and prefect.orchestration == "client" are processed 3.

Bulk Upsert Pipeline#

The core write path is _record_bulk_task_run_events, which runs through four stages for every flush:

1. Parse events → task runs

Each ReceivedEvent is converted to a TaskRun via task_run_from_event. The state timestamp is set from event.occurred — the moment the event was emitted — not the server's wall clock 4. State IDs are derived directly from event.id 5.

2. In-batch deduplication via union-find

A single flush can contain multiple events for the same task run (via either id or natural key (flow_run_id, task_key, dynamic_key)). Before any DB write, events are sorted by state.timestamp and coalesced using a union-find structure so only the latest-state entry survives per conflict group 6. The _task_run_conflict_keys function exposes both conflict axes 7.

3. Deterministic sort → contiguous batching

Deduplicated rows are sorted by their upsert key so all concurrent recorder instances acquire row-level locks in the same order — preventing deadlocks 8. Rows are then split into contiguous batches sharing the same (column_key_set, conflict_target) signature. Contiguous-only batching (an ordered list rather than a dict) is critical: grouping by signature non-contiguously would re-order rows and undo the sort-order guarantee 9 10.

4. Upsert with timestamp guard

Each batch is issued as a single INSERT ... ON CONFLICT DO UPDATE ... WHERE task_run.state_timestamp < excluded.state_timestamp, so a row is only updated when the incoming state is newer 11. The conflict target is either id or the natural-key unique index (task_run_unique_upsert_columns), chosen per batch 12.

After all task run upserts, every state for every event (not just survivors) is inserted via _insert_task_run_states with ON CONFLICT DO NOTHING on id, so duplicate state rows are silently dropped 13.

Conflict Target Resolution & Canonical ID Fallback#

The conflict_target helper determines which unique index to use for each row. It queries existing rows before the upsert loop 14 and applies this logic:

1. If the natural key (flow_run_id, task_key, dynamic_key) already exists in the DB → use "natural-key".
2. Else if task_run.id already exists → use "id".
3. Canonical ID fallback: If neither direct key matches but another key in the same conflict group does (e.g., a batch contains both the old ID and new natural-key variant of the same run), it rewrites task_run.id to the already-persisted canonical ID and uses "id" 15. This prevents IntegrityErrors from cross-key conflicts within a batch.
4. Defaults: natural-key for flow-run-bound tasks, "id" for autonomous tasks 16.

After all batches, canonical_task_run_id is propagated back to every original event — including deduplicated aliases — so state rows reference the correct persisted ID 17.

Retry & Error Handling#

record_bulk_task_run_events retries the entire batch once on IntegrityError to handle TOCTOU races between concurrent recorder instances — when two batches' existence-check SELECTs race ahead of each other's INSERTs 18. A second failure re-raises the error.

The consumer layer (flush) adds a second tier: individual events are re-queued up to max_persist_retries (default 5) times before being dropped and logged at ERROR level 19. Periodic flush errors are caught and logged without killing the flush loop 20.

Key Entry Points#