prefect-redis: ephemeral consumer loops on `no such key` when trimming a missing Redis stream

Summary

A Prefect server using prefect-redis for messaging can enter an endless reconnect/backoff loop for an ephemeral consumer when the Redis stream does not exist.

The server remains otherwise functional and flow runs continue through Pending -> Submitting -> Running -> Completed, but the ephemeral messaging consumer repeatedly logs misleading connection warnings such as:

WARNING | prefect.prefect_redis.messaging - Redis connection error in consumer ephemeral-<host>-<uuid>, reconnecting in 64.0s (attempt 9): no such key

This does not appear to be a network/connectivity issue. The error is caused by XINFO GROUPS being called on a missing stream from _trim_stream_to_lowest_delivered_id().

Versions

prefect = 3.7.7
prefect_redis = 0.2.14

Backend: AWS Valkey / Redis-compatible endpoint.

Effective messaging configuration

The running Prefect process is configured with:

PREFECT_MESSAGING_BROKER=prefect_redis.messaging
PREFECT_MESSAGING_CACHE=prefect_redis.messaging
PREFECT_SERVER_EVENTS_MESSAGING_BROKER=prefect_redis.messaging
PREFECT_SERVER_EVENTS_MESSAGING_CACHE=prefect_redis.messaging
PREFECT_SERVER_EVENTS_CAUSAL_ORDERING=prefect_redis.ordering
PREFECT_REDIS_MESSAGING_URL=<set>

Reproduction path / root cause

The installed ephemeral_subscription() correctly handles a missing stream:

try:
    stream_info = await redis_client.xinfo_stream(source_stream)
    starting_message_id = stream_info["last-generated-id"]
except ResponseError as exc:
    if "no such key" not in str(exc).lower():
        raise
    starting_message_id = "0-0"

...

"use_consumer_group": False

So the ephemeral consumer takes the non-consumer-group path in Consumer.run():

if not self.use_consumer_group:
    await self._run_without_consumer_group(handler, redis_client)
    return

_run_without_consumer_group() periodically calls:

await self._trim_stream_if_necessary(last_id)

which calls:

await _trim_stream_to_lowest_delivered_id(
    self.stream, latest_delivered_id=latest_delivered_id
)
await _cleanup_empty_consumer_groups(self.stream)

The problem is in _trim_stream_to_lowest_delivered_id():

# Get information about all consumer groups for this stream
groups = await redis_client.xinfo_groups(stream_name)

This call is not protected against the stream being absent. Redis/Valkey returns:

ERR no such key

The exception propagates all the way to Consumer.run(), where it is caught by the broad:

except RedisError as e:

and therefore logged as a Redis connection failure, followed by client cache clearing and exponential backoff.

By contrast, _cleanup_empty_consumer_groups() already handles this situation safely:

try:
    groups = await redis_client.xinfo_groups(stream_name)
except Exception as e:
    logger.debug(...)
    return

Observed behavior

Typical sequence:

1. ephemeral subscription starts
2. stream does not exist yet
3. ephemeral_subscription() handles XINFO STREAM -> no such key and sets starting_message_id = "0-0"
4. consumer uses XREAD (use_consumer_group=False)
5. periodic stream trimming runs
6. _trim_stream_to_lowest_delivered_id() calls XINFO GROUPS on the still-missing stream
7. Redis/Valkey returns no such key
8. Consumer.run() treats it as a connection error and enters reconnect/backoff
9. cycle repeats indefinitely

Meanwhile normal flow runs continue to execute successfully.

Expected behavior

A missing stream should be treated as a normal condition for an ephemeral subscriber, consistent with the handling already present in ephemeral_subscription().

The consumer should continue waiting for live messages rather than entering the Redis reconnect path.

Suggested fix

Handle the missing-stream ResponseError in _trim_stream_to_lowest_delivered_id() before calling xinfo_groups, for example:

try:
    groups = await redis_client.xinfo_groups(stream_name)
except ResponseError as exc:
    if "no such key" in str(exc).lower():
        return
    raise

This would make the trimming behavior consistent with ephemeral_subscription() and avoid misclassifying a missing stream as a Redis connection failure.

Impact

Non-blocking for flow execution in our environment, but it creates persistent warning noise and an unnecessary reconnect/backoff loop for the ephemeral consumer.