agentscope-ai / agentscope-ai/agentscope-java

[Bug]: Streaming retry re-subscribes the model and duplicates already-delivered chunks

Đang mở
#2,478 1 bình luận 0 reaction 0 người được giao Xem trên GitHub
area/core/agent area/core/memory bug
Ngôn ngữ chính
Java
Star
5.6k
Fork
1.3k
Merge trung bình
4 ngày 12 giờ
Pull request đã merge (30 ngày)
77

Mô tả

## Summary

`ModelUtils.applyTimeoutAndRetry` attaches `retryWhen` directly to the **streaming** `Flux`. `retryWhen` re-subscribes its upstream, so when a retryable error arrives **mid-stream** the model regenerates the entire response and downstream consumers receive the already-delivered chunks a second time.

This is a correctness problem, not just a latency one: the duplicated chunks have already been handed to middlewares, written to memory, and rendered in the UI.

## Affected version

- `2.0.0` (and current `main`, verified below)

## Code path

`agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java`

```java
Retry retrySpec = Retry.backoff(maxAttempts - 1, initialBackoff)
.maxBackoff(maxBackoff)
.jitter(0.5)
.filter(retryOn) // no notion of "already emitted"
.doBeforeRetry(...);

responseFlux = responseFlux.retryWhen(retrySpec); // re-subscribes the stream
```

There is no guard for whether the stream has already emitted anything. `ExecutionConfig.RETRYABLE_ERRORS` treats `IOException` and `TimeoutException` as retryable, which is exactly what a dropped streaming connection produces.

With `MODEL_DEFAULTS` (`maxAttempts=3`) the same content can be delivered **three times**.

This applies to the streaming path of every provider routed through the helper — `OpenAIChatModel.doStream`, `DashScopeChatModel`, `AnthropicChatModel`, `OllamaChatModel`.

## Reproduction (no live model needed)

Upstream emits two chunks, then fails with a retryable `IOException`:

```java
Flux upstream = Flux.defer(() -> {
subscriptions.incrementAndGet();
return Flux.concat(
Flux.just(chunk("Hello"), chunk(" world")),
Flux.error(new IOException("connection reset by peer")));
});

ModelUtils.applyTimeoutAndRetry(upstream, options, options, "m", "p")
.map(this::textOf)
.onErrorResume(e -> Flux.empty())
.collectList()
.block();
```

Observed on current `main`:

```
upstream subscriptions = 3
chunks delivered downstream = [Hello, world, Hello, world, Hello, world]
```

Expected: `[Hello, world]`, upstream subscribed once.

## Impact

1. **Duplicated content** downstream — middlewares, memory writes and UI all see the response twice or three times.
2. **Latency** — each retry costs a full regeneration. For long or thinking-mode responses that is tens of seconds during which no event reaches the client.

Point 2 may explain #2279 (30 s of silence before completion). That report attributes the delay to `Flux.create` / `OverflowStrategy.BUFFER` signal propagation, but `doFinally` is attached to the lifecycle `Mono` — it cannot fire until the `Mono` terminates, and a retry keeps the `Mono` alive across backoff plus a full regeneration. The timeline in that issue (middlewares `onComplete` at `00:15:33`, `POST_CALL` at `00:16:03`) is consistent with "the Mono finished 30 s later", not "the completion signal took 30 s to propagate".

Separately, the fix suggested there — `sink.tryEmitComplete()` — cannot compile: `FluxSink` has no `tryEmit*` method in any Reactor 3.x release (verified with `javap` against 3.1.7.RELEASE, 3.4.16, 3.6.18, 3.7.6); `tryEmitComplete()` belongs to `Sinks.Many`.

## Suggested fix

Stop retrying once anything has been emitted, while keeping retries for failures **before** the first chunk — connection setup failures and HTTP 429 ahead of the first token, where retrying is both safe and valuable:

```java
AtomicBoolean emitted = new AtomicBoolean(false);
final Predicate retryableError = retryOn;

Retry retrySpec = Retry.backoff(maxAttempts - 1, initialBackoff)
.maxBackoff(maxBackoff)
.jitter(0.5)
.filter(error -> !emitted.get() && retryableError.test(error))
.doBeforeRetry(...);

responseFlux = responseFlux.doOnNext(r -> emitted.set(true)).retryWhen(retrySpec);
```

PR incoming with this change plus regression coverage: mid-stream retryable error, failure before the first chunk, non-retryable mid-stream error, and a clean stream. Only the first fails without the change — the other three pass both ways, pinning that retries are not over-disabled.

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.