agentscope-ai / agentscope-ai/agentscope

fix(app): reconnect index task consumer after Redis Pub/Sub disconnect

未關閉
#2,537 0 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
主要語言
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

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。