Bogdanp / Bogdanp/dramatiq

RedisBroker treats `redis.TimeoutError` as an unexpected error instead of a connection loss

Abierto Apto para principiantes
#894 0 comentarios 0 reacciones 0 asignados Ver en GitHub
redis
Lenguaje dominante
Python
Estrellas
5.3k
Forks
383
Merge medio
9 h 41 min
PR fusionados (30 d)
2

Descripción

## Checklist

* [x] Does your title concisely summarize the problem?
* [x] Did you include a minimal, reproducible example?
* [x] What OS are you using?
* [x] What version of Python are you using?
* [x] What version of Dramatiq are you using?
* [x] What Broker are you using?
* [x] What did you do?
* [x] What did you expect would happen?
* [x] What happened?

## What OS are you using?

macOS (also seen on Linux in production)

## What version of Python are you using?

3.12.13

## What version of Dramatiq are you using?

2.2.0 and 2.2.1, same code on master

## What Broker are you using?

RedisBroker, redis-py 8.1.0 (also 7.4.1)

## What did you do?

Ran a worker against a managed Redis that occasionally stalls a socket. `_RedisConsumer.__next__` and the enqueue path map `redis.ConnectionError` to `ConnectionClosed`, which the worker treats as a connection loss (logs it, keeps the consumer, retries). `redis.TimeoutError` is not a subclass of `redis.ConnectionError` in redis-py, so a socket timeout on the `EVALSHA` fetch escapes raw, hits the worker's generic `except Exception`, is logged CRITICAL as "Consumer encountered an unexpected error." with a full traceback, and the consumer is closed and rebuilt a second later.

Repro, nothing listening on the URL and no command sent, `do_fetch` is stubbed:

```python
import importlib.metadata as md

import redis
from dramatiq.brokers.redis import RedisBroker
from dramatiq.errors import ConnectionClosed

broker = RedisBroker(url="redis://127.0.0.1:1/0") # nothing listens there; no command is sent
consumer = broker.consume("default", prefetch=1, timeout=100)
print(f"dramatiq {md.version('dramatiq')}, redis {md.version('redis')}; TimeoutError is a ConnectionError: {issubclass(redis.TimeoutError, redis.ConnectionError)}")
for exc in (redis.ConnectionError("Connection refused"), redis.TimeoutError("Timeout reading from socket")):
def do_fetch(*args, _exc=exc, **kwargs):
raise _exc
broker.do_fetch = do_fetch
try:
next(consumer)
except ConnectionClosed as e:
print(f" {type(exc).__name__:16} -> ConnectionClosed (worker keeps the consumer, logs a connection error)")
except redis.RedisError as e:
print(f" {type(exc).__name__:16} -> raw {type(e).__module__}.{type(e).__name__} (worker's generic except: CRITICAL 'unexpected error', consumer closed)")
```

## What did you expect would happen?

A socket timeout is the same operational event as a dropped connection: `ConnectionClosed`, the connection-error branch, no CRITICAL line. Adding `redis.TimeoutError` to the three `except redis.ConnectionError` clauses in `dramatiq/brokers/redis.py` would do it; can send the PR.

## What happened?

```
dramatiq 2.2.1, redis 8.1.0; TimeoutError is a ConnectionError: False
ConnectionError -> ConnectionClosed (worker keeps the consumer, logs a connection error)
TimeoutError -> raw redis.exceptions.TimeoutError (worker's generic except: CRITICAL 'unexpected error', consumer closed)
```

In production that is `redis.exceptions.TimeoutError: Timeout reading from socket` from `dramatiq/brokers/redis.py` `__next__` -> `do_fetch`, logged at CRITICAL and paging on every stall.

Guía de contribución

Abrir la guía de contribución

Línea de trabajo

The issue is in dramatiq/brokers/redis.py, where redis.TimeoutError is not caught alongside redis.ConnectionError. Look at the _RedisConsumer.__next__ method and the enqueue path. The fix is to add redis.TimeoutError to the except clauses. Run the provided repro script to verify the behavior before and after the change. Ensure the change doesn't break existing tests.

Escrito por el modelo de indexación a partir del texto del issue.

Evaluación

Stack tecnológico
python, redis
Área
backend, distributed-systems
Tipo de issue
Error
Dificultad
2/5
Tiempo estimado
1-3 horas
Estado de actividad
Activo
Claridad
Bien especificado
Aptitud para principiantes
75/100

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.