agentscope-ai / agentscope-ai/agentscope
fix(app): reconnect index task consumer after Redis Pub/Sub disconnect
- Dominant language
- Python
- Stars
- 31.5k
- Forks
- 3.5k
- Avg merge
- 1d 23h
- Merged PRs (30d)
- 95
Description
## 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.
Contributor guide
Assessment
This issue has not been assessed yet.