You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@rocketmq.apache.org by "unbridled-41 (via GitHub)" <gi...@apache.org> on 2026/09/24 11:19:33 UTC

[PR] fix(ai): keep a reconnecting run's buffered frames ahead of live ones [rocketmq-dashboard]

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: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org