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 674e7017d fix(ai): stop the retention purge deleting busy chats (#5810)
674e7017d is described below
commit 674e7017d040652da0aeb1fc36976a82a199087d
Author: 0 <[email protected]>
AuthorDate: Mon Oct 19 19:54:24 2026 +0800
fix(ai): stop the retention purge deleting busy chats (#5810)
Verified against `rocketmq-studio` at fbdc5f22b 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:
- fix(ai): keep the retention purge moving past the conversations it skips
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/AiConversationService.java | 39 +++++++++++++-
.../ops/ai/conversation/AiRunRepository.java | 11 ++++
.../conversation/MybatisPlusAiRunRepository.java | 24 +++++++++
.../AiConversationPersistenceIntegrationTest.java | 16 ++++++
.../ai/conversation/AiConversationServiceTest.java | 61 ++++++++++++++++++++++
5 files changed, 149 insertions(+), 2 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationService.java
index 24175f5c0..4452db8a8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationService.java
@@ -487,15 +487,50 @@ public class AiConversationService implements
ApplicationRunner {
if (ids.isEmpty()) {
break;
}
- int deleted = deleteCascade(ids);
+ List<Long> purgable = skipConversationsWithARunInFlight(ids);
+ if (purgable.isEmpty()) {
+ // The whole page is mid-run, so nothing was deleted and the
next query returns this
+ // same page: no cursor here, and the loop stays bounded by
maxBatches. This is only
+ // reachable while a run is genuinely live —
reapOrphanedRuns() has already turned
+ // every ownerless non-terminal run into FAILED/ORPHANED
earlier in the same pass — so
+ // what is left finishes on its own and a later pass purges it.
+ continue;
+ }
+ int deleted = deleteCascade(purgable);
totalDeleted += deleted;
- if (deleted < batchSize) {
+ // The page size says whether the backlog is gone, not the delete
count: a full page can
+ // still yield fewer deletes than batchSize once the busy
conversations are skipped, and
+ // stopping there would abandon the rest of the backlog for this
pass.
+ if (ids.size() < batchSize) {
break;
}
}
return totalDeleted;
}
+ /**
+ * Drops the ids that still have a run in flight. The on-request {@link
#delete(Long, String)}
+ * stops that run before it removes anything, because the user asked for
the conversation to go;
+ * retention has no such mandate, so a busy conversation is left for a
later pass rather than
+ * having its rows and its workspace removed while a worker is still
streaming into them.
+ *
+ * <p>One batch query for the whole page, not one {@code
findActiveByConversationId} per id: a
+ * retention pass looks at up to {@code cleanup-batch-size} conversations
per batch, so the N+1
+ * would be 500 SELECTs a pass at the default, for an answer one statement
gives.
+ */
+ private List<Long> skipConversationsWithARunInFlight(List<Long> ids) {
+ Set<Long> busy = runRepository.findActiveConversationIds(ids);
+ List<Long> purgable = new ArrayList<>(ids.size());
+ for (Long id : ids) {
+ if (busy.contains(id)) {
+ log.info("skipping the retention purge of conversation {}: it
has a run in flight", id);
+ continue;
+ }
+ purgable.add(id);
+ }
+ return purgable;
+ }
+
/**
* Events, then runs, then conversations: no FK constraints here, so the
cascade is this method.
*
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunRepository.java
index 9597c0664..2b1827488 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunRepository.java
@@ -23,6 +23,7 @@ import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
/** Persistence port for rmq_ai_run. */
public interface AiRunRepository {
@@ -35,6 +36,16 @@ public interface AiRunRepository {
/** The conversation's run in QUEUED or RUNNING, if any. At most one by
admission rule. */
Optional<RmqAiRun> findActiveByConversationId(Long conversationId);
+ /**
+ * The subset of the given conversations that still have a run in QUEUED
or RUNNING.
+ *
+ * <p>A batch for the same reason {@link #findLatestByConversationIds} is
one: the retention pass
+ * asks about a whole page of up to {@code
studio.ai.conversation.cleanup-batch-size} conversations
+ * at once, and answering that through {@link #findActiveByConversationId}
would be one SELECT per
+ * row on every pass. This is one statement, served by {@code
idx_ai_run_conversation}.
+ */
+ Set<Long> findActiveConversationIds(List<Long> conversationIds);
+
List<RmqAiRun> findByConversationId(Long conversationId);
/**
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiRunRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiRunRepository.java
index 84aaf8b8b..6f45ad048 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiRunRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiRunRepository.java
@@ -30,6 +30,8 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
* MySQL-backed AI run repository (rmq_ai_run).
@@ -71,6 +73,28 @@ public class MybatisPlusAiRunRepository implements
AiRunRepository {
return active.stream().findFirst();
}
+ @Override
+ public Set<Long> findActiveConversationIds(List<Long> conversationIds) {
+ if (conversationIds == null || conversationIds.isEmpty()) {
+ return Set.of();
+ }
+ List<Long> ids =
conversationIds.stream().filter(Objects::nonNull).distinct().toList();
+ if (ids.isEmpty()) {
+ return Set.of();
+ }
+ // One statement for a whole retention page instead of one SELECT per
conversation. The
+ // conversation_id predicate is what idx_ai_run_conversation leads
with, so the status filter
+ // only narrows rows the index already selected.
+ return runMapper.selectObjs(new QueryWrapper<RmqAiRun>()
+ .select("DISTINCT conversation_id")
+ .in("conversation_id", ids)
+ .in("status", RunStatus.ACTIVE_STATUSES))
+ .stream()
+ .filter(Objects::nonNull)
+ .map(value -> ((Number) value).longValue())
+ .collect(Collectors.toSet());
+ }
+
@Override
public List<RmqAiRun> findByConversationId(Long conversationId) {
if (conversationId == null) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
index 3bf72e40d..d5cad3686 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
@@ -371,6 +371,22 @@ class AiConversationPersistenceIntegrationTest {
verifyNoCascade();
}
+ @Test
+ void theBatchActiveRunQueryShouldAnswerForAWholeRetentionPageTest() {
+ Seeded busy = seed();
+ markRunning(busy.secondRunId);
+ long idle = seedConversation(OWNER, "an idle conversation of the same
age");
+ seedRun(idle, 1, 1, EVENTS_PER_RUN);
+
+ // A hand-written DISTINCT over an IN list, standing in for one
findActiveByConversationId per
+ // row on a retention page. Only a real database says whether it
answers with the busy subset
+ // and nothing else — which is what the purge then trusts with a
cascade delete.
+
assertThat(runRepository.findActiveConversationIds(List.of(busy.conversationId,
idle, -1L)))
+ .containsExactly(busy.conversationId);
+
assertThat(runRepository.findActiveConversationIds(List.of(idle))).isEmpty();
+
assertThat(runRepository.findActiveConversationIds(List.of())).isEmpty();
+ }
+
// --- harness
------------------------------------------------------------------
private void verifyNoCascade() {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationServiceTest.java
index 926657b1f..892acc505 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationServiceTest.java
@@ -43,6 +43,7 @@ import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -421,6 +422,66 @@ class AiConversationServiceTest {
order.verify(workspace).delete(3L);
}
+ @Test
+ void retentionShouldSkipAConversationWhoseRunIsStillInFlightTest() {
+
when(conversationRepository.findIdsCreatedBefore(any(LocalDateTime.class),
eq(500)))
+ .thenReturn(List.of(1L, 2L));
+ when(runRepository.findActiveConversationIds(List.of(1L,
2L))).thenReturn(Set.of(2L));
+ when(conversationRepository.deleteByIds(anyList())).thenReturn(1);
+
+ service.purgeExpired();
+
+ // The on-request delete stops the active run before it removes
anything; a retention pass
+ // that deletes the rows and the workspace of a conversation mid-run
leaves the worker
+ // writing into a directory that no longer exists. Retention defers to
a later pass instead.
+ verify(eventRepository).deleteByConversationIds(List.of(1L));
+ verify(runRepository).deleteByConversationIds(List.of(1L));
+ verify(conversationRepository).deleteByIds(List.of(1L));
+ verify(workspace, never()).delete(2L);
+ }
+
+ @Test
+ void retentionShouldKeepPagingPastABatchThatHeldABusyConversationTest() {
+ properties.setCleanupBatchSize(2);
+ properties.setCleanupMaxBatches(5);
+ service = newService();
+
when(conversationRepository.findIdsCreatedBefore(any(LocalDateTime.class),
eq(2)))
+ .thenReturn(List.of(1L, 2L), List.of(3L, 4L), List.of());
+ when(runRepository.findActiveConversationIds(List.of(1L,
2L))).thenReturn(Set.of(2L));
+ when(conversationRepository.deleteByIds(anyList())).thenReturn(1, 2);
+
+ service.purgeExpired();
+
+ // The first page came back full, so the backlog is not necessarily
gone even though only one
+ // of its two rows was deletable. Judging that by the delete count
would stop the pass here
+ // and leave conversations 3 and 4 to a later pass for no reason but a
busy neighbour.
+ verify(conversationRepository, times(3)).findIdsCreatedBefore(any(),
anyInt());
+ verify(conversationRepository).deleteByIds(List.of(1L));
+ verify(conversationRepository).deleteByIds(List.of(3L, 4L));
+ verify(workspace, never()).delete(2L);
+ }
+
+ @Test
+ void retentionShouldNotGiveUpWhenAWholeBatchIsBusyTest() {
+ properties.setCleanupBatchSize(2);
+ properties.setCleanupMaxBatches(5);
+ service = newService();
+
when(conversationRepository.findIdsCreatedBefore(any(LocalDateTime.class),
eq(2)))
+ .thenReturn(List.of(1L, 2L), List.of(3L, 4L), List.of());
+ when(runRepository.findActiveConversationIds(List.of(1L,
2L))).thenReturn(Set.of(1L, 2L));
+ when(conversationRepository.deleteByIds(anyList())).thenReturn(2);
+
+ service.purgeExpired();
+
+ // Nothing on the first page is deletable, so the query answers the
same page again: bounded
+ // by cleanupMaxBatches, and only reachable while a run is genuinely
live, because the orphan
+ // sweep already terminated every ownerless one earlier in this same
pass.
+ verify(conversationRepository, times(3)).findIdsCreatedBefore(any(),
anyInt());
+ verify(conversationRepository).deleteByIds(List.of(3L, 4L));
+ verify(workspace, never()).delete(1L);
+ verify(workspace, never()).delete(2L);
+ }
+
@Test
void
retentionShouldKeepBatchingUntilTheBacklogIsGoneOrTheCapIsReachedTest() {
properties.setCleanupBatchSize(2);
