Redis Stream Consumers#

Overview#

Redis stream consumers in Prefect are implemented in prefect_redis/messaging.py and provide two distinct subscription patterns:

• Durable (XREADGROUP) — uses Redis consumer groups for at-least-once delivery, offset tracking, and dead-letter queuing. Used for long-lived service consumers.
• Ephemeral (XREAD) — stateless, reads only live messages from the stream tail, no consumer group state. Used for short-lived or process-scoped subscribers.

The Consumer class handles both modes via the use_consumer_group flag.

Consumer Initialization#

Consumer.__init__() resolves all settings from RedisMessagingConsumerSettings, which can be overridden via environment variables (e.g., PREFECT_REDIS_MESSAGING_CONSUMER_BLOCK, PREFECT_REDIS_MESSAGING_CONSUMER_MAX_RETRIES). Key defaults:

For durable consumers, _ensure_stream_and_group() creates the consumer group and stream via XGROUP CREATE ... MKSTREAM, silently ignoring BUSYGROUP errors if the group already exists.

Subscription Patterns#

Durable: XREADGROUP#

Consumer.run() with use_consumer_group=True (the default):

1. Calls _ensure_stream_and_group() to create or verify the consumer group.
2. Optionally processes pending messages via process_pending_messages(), which uses XAUTOCLAIM to reclaim messages idle beyond min_idle_time.
3. Calls XREADGROUP with ">" to read only new, undelivered messages 1.
4. Acknowledges processed messages in batch via _batch_ack().

Ephemeral: XREAD#

ephemeral_subscription() is an async context manager that yields constructor kwargs for a Consumer with use_consumer_group=False. It:

1. Reads last-generated-id from the stream at startup to skip historical messages 2.
2. Passes use_consumer_group=False so no consumer group is created.

_run_without_consumer_group() then uses XREAD with the current last_id, advancing it on each message. Acknowledgment is a no-op (_noop_ack) since XREAD has no consumer group state.

Why XREAD, not XREADGROUP: Ephemeral subscribers only need live messages; they have no need for durable offset tracking. The original XREADGROUP-based design leaked consumer groups when processes were killed before cleanup, since groups persisted in Redis even after the process died. Switching to stateless XREAD eliminates that state entirely — there is nothing to leak 3.

Error Handling and Reconnection#

Consumer.run() wraps the inner read loop in an outer while True that catches RedisError 4:

1. Computes delay via exponential_backoff_with_jitter — base 1.0s, max 60.0s.
2. Calls clear_cached_clients() to evict stale connections.
3. Sleeps, then on the next iteration calls get_async_redis_client() for a fresh connection.
4. Resets attempt counter and logs reconnection on success 5.

ResponseError from xreadgroup is logged and re-raised to the outer loop, triggering the same backoff path 6.

redis-py 8 compatibility: redis-py 8 introduced a default socket_timeout=5s, which caused idle XREADGROUP BLOCK calls to time out before Redis returned. RedisMessagingSettings now defaults both socket_timeout and socket_connect_timeout to None to preserve unlimited-blocking behavior 7.

Stream Trimming#

_trim_stream_if_necessary() is called after each read batch. It throttles via trim_every (default 60s), then calls:

• _trim_stream_to_lowest_delivered_id(): Finds the minimum last-delivered-id across all active consumer groups. Groups where all consumers have been idle beyond trim_idle_threshold are excluded from the calculation to prevent stale groups from blocking trimming. Then runs XTRIM ... MINID to remove messages below that ID 8. For XREAD consumers, the current last_id is passed directly since there is no consumer group to query.
• _cleanup_empty_consumer_groups(): Deletes consumer groups that are empty (no consumers) and have consumed at least one message (last-delivered-id != "0-0"). Groups with last-delivered-id == "0-0" are preserved to avoid a race condition where a group is deleted before its consumers can join 9.

Dead Letter Queue (DLQ)#

Message failures are tracked in _retry_counts (an in-memory dict keyed by message ID). After exceeding max_retries, _send_to_dlq() stores the message as a Redis hash and adds its key to a Redis set at _dlq_key(topic). The message is then acknowledged to remove it from the pending entries list 10.

Affected Services#

ReactiveTriggers, Actions, EventPersister, EventLogger, TaskRunRecorder all use the Consumer class for durable stream consumption 7.

Key References#