This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new e1577bd31 fix(ai): page the attach replay through the whole backlog
(#6389)
e1577bd31 is described below
commit e1577bd3106603e1ad1fe724a0cbdd0b3ad61257
Author: 0 <[email protected]>
AuthorDate: Mon Oct 19 20:51:51 2026 +0800
fix(ai): page the attach replay through the whole backlog (#6389)
Verified against `rocketmq-studio` at 3d1641bad before merging: the defect
is still present on the
current trunk and the change merges cleanly.
Maintainer follow-ups on top of the contributor commit:
- docs(ai): promise the lowest limit rows from the seq window
Each follow-up commit message records the review finding it addresses.
Locally verified: targeted suites green on the integration branch; a full
backend and frontend
build plus test pass runs on the trunk once this batch lands.
---
.../ops/ai/conversation/AiEventRepository.java | 4 +-
.../studio/ops/ai/conversation/AiRunService.java | 49 +++++++++++++++++-----
.../ops/ai/conversation/AiRunServiceTest.java | 27 ++++++++++++
3 files changed, 68 insertions(+), 12 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
index c35d169c0..f6868e8e1 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
@@ -33,7 +33,9 @@ public interface AiEventRepository {
int maxSeq(Long conversationId);
/**
- * Events with seq strictly greater than afterSeq, capped at limit,
ascending by seq.
+ * The lowest {@code limit} events whose seq is strictly greater than
afterSeq, ascending by
+ * seq — a page of the backlog, not "whatever fits", so a caller that
walks a cursor can rely
+ * on it never skipping a row it has not seen.
*
* <p>Deliberately does NOT use SQL ORDER BY. The payload column is
MEDIUMTEXT, and any
* sort MySQL cannot satisfy from an index materialises the whole value
into
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
index b43d8e176..0d8787516 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
@@ -103,6 +103,13 @@ public class AiRunService {
static final String REFUSED_INVALID_CODE = "ai.request.invalid";
static final String REFUSED_CODE = "ai.run.refused";
+ /**
+ * Ceiling on the pages one attach may replay: {@value #MAX_REPLAY_PAGES}
pages of
+ * {@value AiConversationService#DEFAULT_TIMELINE_LIMIT} rows. Far beyond
any real backlog; exists
+ * so a misbehaving repository that keeps returning full pages cannot pin
the request thread.
+ */
+ private static final int MAX_REPLAY_PAGES = 25;
+
static final String OVERLOADED_MESSAGE = "AI chat capacity is temporarily
exhausted";
static final String OVERLOADED_HINT = "Wait for the active answer to
finish, then retry.";
@@ -295,19 +302,32 @@ public class AiRunService {
// rows the run persisted while the reconnecting client was catching
up.
boolean live = registry.attach(run.getId(), session);
AgentEventProjector projector = new AgentEventProjector(run.getId());
+ // The replay pages through the whole backlog, not just the first
slice: the query is capped at
+ // DEFAULT_TIMELINE_LIMIT rows, and a tool-heavy turn that kept
writing while the client was
+ // away produces more than one page of it. Stopping after the first
page would strand the rows
+ // in between — already persisted, so their live frames are gone, and
beyond the client's own
+ // cursor, so it would never fetch them either.
+ int cursor = Math.max(0, afterSeq);
boolean terminalReplayed = false;
- List<RmqAiEvent> rows;
+ int replayedRows = 0;
try {
- rows =
eventRepository.findByConversationIdAfterSeq(run.getConversationId(),
- Math.max(0, afterSeq),
AiConversationService.DEFAULT_TIMELINE_LIMIT);
- for (RmqAiEvent row : rows) {
- session.noteWatermark(row.getSeq());
- Optional<TimelineEvent> event =
AiEventCodec.read(objectMapper, row);
- if (event.isEmpty()) {
- continue;
+ for (int page = 0; page < MAX_REPLAY_PAGES; page++) {
+ List<RmqAiEvent> rows =
eventRepository.findByConversationIdAfterSeq(run.getConversationId(),
+ cursor, AiConversationService.DEFAULT_TIMELINE_LIMIT);
+ for (RmqAiEvent row : rows) {
+ session.noteWatermark(row.getSeq());
+ cursor = row.getSeq();
+ Optional<TimelineEvent> event =
AiEventCodec.read(objectMapper, row);
+ if (event.isEmpty()) {
+ continue;
+ }
+ terminalReplayed = terminalReplayed || event.get()
instanceof TimelineEvent.RunStatus;
+ projector.replay(event.get(),
run.getDurationMs()).ifPresent(session::sendReplayed);
+ }
+ replayedRows += rows.size();
+ if (rows.size() <
AiConversationService.DEFAULT_TIMELINE_LIMIT) {
+ break;
}
- terminalReplayed = terminalReplayed || event.get() instanceof
TimelineEvent.RunStatus;
- projector.replay(event.get(),
run.getDurationMs()).ifPresent(session::sendReplayed);
}
} catch (RuntimeException exception) {
// The session this method created never reaches a response now,
so no transport
@@ -316,6 +336,13 @@ public class AiRunService {
session.complete();
throw exception;
}
+ // A short page means the backlog is exhausted; running out of pages
with every one of
+ // them full means the replay was truncated. Say so, because a reload
that silently
+ // dropped the older events looks like a complete one.
+ if (replayedRows >= MAX_REPLAY_PAGES *
AiConversationService.DEFAULT_TIMELINE_LIMIT) {
+ log.warn("attach replay of run {} hit the {} page ceiling after {}
rows; older events are not replayed",
+ run.getId(), MAX_REPLAY_PAGES, replayedRows);
+ }
session.finishReplay();
if (!live) {
// Nothing is generating here any more, so no further frame will
ever arrive. A run reaped by
@@ -329,7 +356,7 @@ public class AiRunService {
session.complete();
}
log.debug("attached an observer to agent run {} after seq {} ({}
replayed row(s), live={})",
- run.getId(), afterSeq, rows.size(), live);
+ run.getId(), afterSeq, replayedRows, live);
return session.emitter();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
index 00a1a38f7..1aaf16866 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
@@ -44,6 +44,7 @@ import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -598,6 +599,32 @@ class AiRunServiceTest {
assertThat(emitters.get(0).completed()).isTrue();
}
+ @Test
+ void attachShouldReplayTheWholeBacklogBeyondOnePageTest() {
+ // A run that kept writing while the client was away can hold more
than one page of backlog
+ // (a page is 200 rows and a tool-heavy turn writes several rows per
call). The replay has to
+ // page through all of it: stopping after the first page would strand
the rows in between,
+ // whose live frames were published before this observer existed and
whose rows lie beyond
+ // the cursor the client already holds.
+ RmqAiRun finished = AiRunTestSupport.run(RUN_ID, CONVERSATION_ID, 1,
RunStatus.COMPLETED);
+ when(runRepository.findById(RUN_ID)).thenReturn(Optional.of(finished));
+ List<RmqAiEvent> firstPage = new ArrayList<>();
+ for (int seq = 1; seq <= 200; seq++) {
+ firstPage.add(event(seq, "text",
"{\"type\":\"text\",\"text\":\"row " + seq + "\"}"));
+ }
+ when(eventRepository.findByConversationIdAfterSeq(CONVERSATION_ID, 0,
200)).thenReturn(firstPage);
+ when(eventRepository.findByConversationIdAfterSeq(CONVERSATION_ID,
200, 200)).thenReturn(List.of(
+ event(201, "text", "{\"type\":\"text\",\"text\":\"the last
row\"}"),
+ event(202, "run_status",
"{\"type\":\"run_status\",\"status\":\"COMPLETED\"}")));
+
+ service.attach(RUN_ID, 0);
+
+ String text = emitters.get(0).eventText();
+ assertThat(text).contains("row 1").contains("the last row");
+ assertThat(text).contains("\"type\":\"run_finished\"");
+ assertThat(emitters.get(0).completed()).isTrue();
+ }
+
@Test
void
attachShouldHonourTheCursorAndSynthesiseATerminalFrameForAReapedRunTest() {
// A run reaped at startup or by the orphan sweep has no run_status
row at all — there was no sink
