agentscope-ai / agentscope-ai/agentscope
fix(app): reconnect index task consumer after Redis Pub/Sub disconnect
- 主要語言
- Python
- 星號
- 31.6k
- 分支
- 3.5k
- 平均合併
- 1 天 16 小時
- 30 天內合併 PR
- 103
描述
## Problem
`IndexTaskConsumer` subscribes to the Redis Pub/Sub signal once. If the subscription raises a connection error (for example during a Redis restart, failover, or network interruption), `_loop()` logs the exception and exits permanently.
The indexing task itself is stored in the durable `agentscope:index:tasks` queue, but no consumer remains to drain it. The API-side `IndexSweeper` can re-enqueue stale documents, but it cannot restart the exited consumer.
## Reproduction
Using the production `RedisMessageBus`, `IndexTaskConsumer`, and `IndexWorker` code with a Redis-compatible in-memory backend, inject one `redis.exceptions.ConnectionError` at the Pub/Sub `get_message()` boundary:
```ini
subscription_task_done=True
pubsub_calls=1
worker_calls=[]
remaining_queue=[after-reset]
```
After the subscription fails, a later indexing task remains in the durable queue and `IndexWorker.process()` is not called until the worker process is restarted.
## Expected behavior
The long-lived index-task consumer should reconnect after a transient subscription failure and drain tasks that accumulated while the signal subscription was unavailable.
## Root cause
[`IndexTaskConsumer._loop`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/app/_service/_index_task_consumer.py) catches the subscription exception but does not retry. [`RedisMessageBus.subscribe`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/app/message_bus/_redis_message_bus.py) correctly exposes the transport failure; the application-level consumer needs to restore its long-lived subscription.
## Proposed scope
- Keep the generic `MessageBus.subscribe` transient broadcast contract unchanged.
- Add a reconnect loop to `IndexTaskConsumer` with bounded exponential backoff (`1s` to `30s`), while preserving immediate cancellation.
- Drain the durable index queue after each successful re-subscription so a signal lost during the outage is not required.
- Add a regression test that fails the first subscription, enqueues a task without sending another signal, and verifies that the task is dispatched after reconnection.
This is separate from startup backlog draining in [#2515](https://github.com/agentscope-ai/agentscope/pull/2515) and channel reply-stream recovery in [#2531](https://github.com/agentscope-ai/agentscope/pull/2531).
I would like to take this fix myself. If this scope and direction look right, please explicitly confirm that I may implement it, and I will prepare a focused test-first PR linked to this issue.
貢獻指南
研究方向
Start with `src/agentscope/app/_service/_index_task_consumer.py` and read `_loop()` where the Redis Pub/Sub consumer exits after a caught exception. Then inspect `src/agentscope/app/message_bus/_redis_message_bus.py` for how `ConnectionError` from `subscribe/get_message()` is surfaced. Next, locate the existing tests around `IndexTaskConsumer`/`IndexWorker` and add a regression that fails the first subscription, leaves one durable task queued, and verifies it is processed after reconnect and backlog drain. Done is a passing focused test proving reconnect with bounded backoff and no manual worker restart required.
由索引模型根據 Issue 內容生成。
評估
- 技術堆疊
- python
- 領域
- backend
- Issue 類型
- 缺陷
- 難度
- 3/5
- 預估耗時
- 1-2 天
- 活躍度
- 活躍
- 描述清晰度
- 描述清楚
- 新手友好度
- 72/100