RedisBroker treats `redis.TimeoutError` as an unexpected error instead of a connection loss
- 主要言語
- Python
- スター
- 5.3k
- フォーク
- 383
- 平均マージ
- 9時間 41分
- マージ済み PR(30日)
- 2
説明
## 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.
コントリビューションガイド
調査の方向性
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.
索引モデルが issue の本文から書いたものです。
評価
- 技術スタック
- python, redis
- 領域
- backend, distributed-systems
- issue の種類
- バグ
- 難易度
- 2/5
- 見積もり時間
- 1〜3時間
- 活発さ
- 活発
- 明瞭さ
- 明確に書かれている
- 初心者へのやさしさ
- 75/100