agentscope-ai / agentscope-ai/agentscope-java

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

Abierto
#2,478 1 comentario 0 reacciones 0 asignados Ver en GitHub
area/core/agent area/core/memory bug
Lenguaje dominante
Java
Estrellas
5.6k
Forks
1.3k
Merge medio
4 d 12 h
PR fusionados (30 d)
77

Descripción

## 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.

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.