unbridled-41 opened a new pull request, #5069:
URL: https://github.com/apache/rocketmq-dashboard/pull/5069
## Problem
A client that reconnects to a run that is still generating can receive the
run's own live frames **interleaved with the replayed ones**, which visibly
shuffles the transcript it renders.
`AgentStreamSession` buffers every live frame that arrives while the
persisted rows are being replayed and drains that buffer in `finishReplay()`.
The drain, however, runs *after* the lock is released:
```java
void finishReplay() {
Deque<LiveFrame> pending;
synchronized (sendLock) {
if (state.get() != State.REPLAYING) { return; }
state.set(State.LIVE);
pending = new ArrayDeque<>(buffered);
buffered.clear();
}
pending.forEach(frame -> deliver(frame.seq(), frame.event())); // <-
outside the lock
}
```
`deliver` re-acquires `sendLock` per frame, and `sendAgent` serialises
*before* it takes the lock, so between two drained frames the lock is free and
a frame published by the run's worker thread (`AiEventSink` →
`AgentRunRegistry.publish` → `deliver`) is written to the socket immediately —
after buffered frame *k* and before buffered frame *k+1*.
The order matters because it is the whole point of the class: the class
javadoc (`AgentStreamSession.java:57-62`) states the invariant — "a frame
cannot slip in between two replayed rows and reorder the transcript" — and the
web client renders arrival order, appending every `text_delta` to the trailing
text block (`web/src/pages/ai/render/blocks.ts:140-146`). A delta that
overtakes buffered frames is a scrambled answer for anyone who reloaded the
page or whose SSE connection dropped mid-run.
## Evidence
Pre-fix, on base `a562601d`, with the new test in place and only the
production file at its base revision:
```
$ cd server && mvn -o test -Dtest=AgentStreamSessionReplayOrderTest
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0
java.lang.AssertionError:
Expecting actual:
"{"type":"text_delta","content":"live-one"}"
to contain:
"replay-two"
at
AgentStreamSessionReplayOrderTest.aFramePublishedDuringTheDrainShouldNotOvertakeTheBufferedFrames(AgentStreamSessionReplayOrderTest.java:89)
```
The live frame is at index 1, the second buffered frame at index 2 — the
order the test asserts is `replay-one, replay-two, live-one`.
The test is not timing-dependent: it parks the drain inside the
*serialisation* of its second buffered frame (`ObjectMapper.writeValueAsString`
hook) and then publishes a frame from another thread. Serialisation is outside
the lock on both sides of the fix, so the pause is precisely the window the
defect leaves open.
Reachability: `AiRunService.attach` (`AiRunService.java:268-288`) — the
reconnect path of `GET /api/ai/runs/{runId}/stream` — calls `finishReplay()`
after `registry.attach(...)`, so the worker can publish from the moment the
observer is registered, i.e. exactly while the buffer is being drained.
## Root cause
`finishReplay()` released `sendLock` between the state flip and the drain,
so the drain was no longer atomic with respect to live frames.
## Fix
Drain the copied buffer while still holding `sendLock`. `deliver` is
reentrant on the same monitor, the dedup predicate is unchanged, and the lock
already serialises every write to the socket (heartbeat, replayed frame, live
frame), so nothing else can now be written until the buffer is empty. The
javadoc records why the lock is held across the drain, since that is the
non-obvious part.
## Score
`PRIORITY = 70` (impact 25: a visibly shuffled answer on every reconnect
that lands inside the drain, which is the documented failure mode of the whole
buffering design; scope 15: the AI streaming surface, every reconnecting
client; reproducibility 15: the regression test decides the race
deterministically; maintenance value 15: the class documents the invariant it
was breaking).
`FIX_CONFIDENCE = 85`.
## Tests
```
$ cd server && mvn -o test -Dtest=AgentStreamSessionReplayOrderTest
# post-fix
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS
$ cd server && mvn -o test
-Dtest='org.apache.rocketmq.studio.ops.ai.**.*Test'
Tests run: 924, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS
```
The pre-fix run of the same test (only `AgentStreamSession.java` reverted to
`a562601d`, test kept) is the failure quoted under Evidence. Checkstyle runs as
part of the `validate` phase of both commands.
## Risk
Low, and bounded to the replay path: the drain now happens under the same
monitor that already guards every socket write, so a concurrent publish waits
for the drain instead of interleaving with it. That is the intended
serialisation; the only cost is a longer lock hold for the length of one buffer
(a replay-length burst of socket writes, previously interleaved with live ones
anyway). No API, payload or dedup behaviour changes. `sendAgent` can still
detach mid-drain on an IO failure, which is how the pre-existing code behaves
too.
## Dedup
- `gh search prs --repo apache/rocketmq-dashboard "AgentStreamSession"` →
only #4948 (drain a backlog longer than one timeline page), which changes the
*replay read* in `AiRunService`/`replayInto` and does not touch `finishReplay`
or `AgentStreamSession.java` at all — the two changes sit in different files
and can merge in either order.
- `gh search issues "AgentStreamSession"` → only #4963 (the same #4948
defect).
- `gh search prs/issues "finishReplay"`, `"replay buffered frames"`,
`"reconnect transcript order"` → no results.
- `git log --oneline a562601d -- .../AgentStreamSession.java` → `5b761df8`
(introduction) only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
