RedisBroker treats `redis.TimeoutError` as an unexpected error instead of a connection loss
- 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
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