feat: 展示分块索引同步重试状态

- 返回当前版本任务的真实尝试次数与重试原因，区分模型服务和索引写入失败 - 分离提示与按钮布局，保持操作位置稳定，补充分块状态回归测试
2026-09-08 15:05:29 +08:00
parent b5a355247b
commit d506d06f77
10 changed files with 408 additions and 14 deletions
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppService.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppService.java
@@ -5,6 +5,7 @@ import com.easyagents.core.model.embedding.EmbeddingOptions;
 import com.easyagents.core.store.DocumentStore;
 import com.easyagents.core.store.StoreOptions;
 import com.easyagents.core.store.StoreResult;
+import com.easyagents.core.store.VectorData;
 import com.easyagents.search.engine.service.DocumentSearcher;
 import com.easyagents.search.engine.service.KeywordSearchMetadataKeys;
 import com.easyagents.store.milvus.MilvusVectorStore;
@@ -22,6 +23,7 @@ import tech.easyflow.ai.entity.DocumentChunk;
 import tech.easyflow.ai.entity.DocumentChunkSyncTask;
 import tech.easyflow.ai.entity.DocumentCollection;
 import tech.easyflow.ai.entity.Model;
+import tech.easyflow.ai.dto.DocumentChunkSyncStatus;
 import tech.easyflow.ai.mapper.DocumentChunkMapper;
 import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper;
 import tech.easyflow.ai.service.DocumentCollectionService;
@@ -33,7 +35,12 @@ import java.math.BigInteger;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.UUID;
+import java.util.function.Function;
+import java.util.stream.Collectors;

 /**
 * 持久化分块索引同步任务的投递、执行和恢复。
@@ -116,6 +123,31 @@ public class DocumentChunkSyncTaskAppService {
        }
    }

+    public List<DocumentChunkSyncStatus> listSyncStatuses(List<DocumentChunk> chunks) {
+        if (chunks.isEmpty()) {
+            return List.of();
+        }
+        Map<BigInteger, DocumentChunkSyncTask> tasks = taskMapper.selectCurrentForChunks(
+            chunks.stream().map(DocumentChunk::getId).toList()
+        ).stream().collect(Collectors.toMap(DocumentChunkSyncTask::getChunkId, Function.identity()));
+        return chunks.stream().map(chunk -> {
+            DocumentChunkSyncTask task = tasks.get(chunk.getId());
+            boolean current = task != null
+                && Objects.equals(chunk.getIndexSyncVersion(), task.getSyncVersion());
+            // 正文状态可能在两次查询间推进，只有同版本仍在重试的任务提供失败原因。
+            boolean retrying = current && DocumentChunkSyncState.PENDING.equals(chunk.getIndexSyncStatus())
+                && (DocumentChunkSyncState.PENDING.equals(task.getStatus())
+                    || DocumentChunkSyncState.TASK_RUNNING.equals(task.getStatus()));
+            return new DocumentChunkSyncStatus(
+                chunk.getId(), chunk.getIndexSyncStatus(), chunk.getIndexSyncVersion(),
+                retrying ? task.getErrorCode() : chunk.getIndexSyncErrorCode(),
+                retrying ? task.getErrorMessage() : chunk.getIndexSyncErrorMessage(),
+                current ? task.getAttemptCount() : null,
+                MAX_ATTEMPTS
+            );
+        }).toList();
+    }
+
    public void dispatchPendingTasks() {
        Date now = new Date();
        taskMapper.recoverExpired(now);
@@ -228,6 +260,7 @@ public class DocumentChunkSyncTaskAppService {
        StoreContext context = prepareUpsertContext(collection, task.getVectorCollection());
        try {
            com.easyagents.core.document.Document document = toSearchDocument(chunk, task.getDocumentCollectionId());
+            embedDocument(context, document);
            StoreResult vectorResult = context.documentStore.update(
                Collections.singletonList(document),
                context.storeOptions
@@ -248,6 +281,21 @@ public class DocumentChunkSyncTaskAppService {
        }
    }

+    private void embedDocument(StoreContext context, com.easyagents.core.document.Document document) {
+        try {
+            VectorData vectorData = context.documentStore.getEmbeddingModel().embed(
+                document, context.storeOptions.getEmbeddingOptions()
+            );
+            if (vectorData == null || vectorData.getVector() == null || vectorData.getVector().length == 0) {
+                throw new IllegalStateException("向量模型未返回有效向量");
+            }
+            // update 复用已生成的向量，不重复调用模型；分开捕获以区分模型与索引写入失败。
+            document.setVector(vectorData.getVector());
+        } catch (RuntimeException exception) {
+            throw new IndexSyncException("EMBEDDING_REQUEST_FAILED", "向量模型服务调用失败", exception);
+        }
+    }
+
    private void synchronizeDelete(DocumentChunkSyncTask task) {
        MilvusVectorStoreConfig storeConfig = milvusConfig.copyForCollection(task.getVectorCollection());
        AiMilvusClientManager clientManager = milvusClientManagerProvider.getObject();
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatus.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatus.java
@@ -10,6 +10,8 @@ public record DocumentChunkSyncStatus(
    String indexSyncStatus,
    Long indexSyncVersion,
    String indexSyncErrorCode,
-    String indexSyncErrorMessage
+    String indexSyncErrorMessage,
+    Integer indexSyncAttemptCount,
+    int indexSyncMaxAttempts
 ) {
 }
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkSyncTaskMapper.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkSyncTaskMapper.java
@@ -44,6 +44,13 @@ public interface DocumentChunkSyncTaskMapper extends BaseMapper<DocumentChunkSyn
        + "ORDER BY sync_version, id FOR UPDATE")
    List<BigInteger> lockChunkTasks(@Param("chunkId") BigInteger chunkId);

+    @Select("<script>SELECT " + SELECT_COLUMNS + " FROM tb_document_chunk_sync_task "
+        + "WHERE operation='UPSERT' AND (chunk_id, sync_version) IN "
+        + "(SELECT id, index_sync_version FROM tb_document_chunk WHERE id IN "
+        + "<foreach collection='ids' item='id' open='(' separator=',' close=')'>#{id}</foreach>)"
+        + "</script>")
+    List<DocumentChunkSyncTask> selectCurrentForChunks(@Param("ids") List<BigInteger> ids);
+
    @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_chunk_sync_task "
        + "WHERE status='PENDING' AND next_retry_at <= #{now} "
        + "AND (last_dispatched_at IS NULL OR last_dispatched_at <= #{redispatchBefore}) "
@@ -67,7 +74,7 @@ public interface DocumentChunkSyncTaskMapper extends BaseMapper<DocumentChunkSyn

    @Update("UPDATE tb_document_chunk_sync_task SET status='RUNNING', "
        + "attempt_count=attempt_count + 1, execution_token=#{token}, "
-        + "lease_until=#{leaseUntil}, error_code=NULL, error_message=NULL, modified=#{now} "
+        + "lease_until=#{leaseUntil}, modified=#{now} "
        + "WHERE id=#{id} AND status='PENDING' AND next_retry_at <= #{now}")
    int claim(@Param("id") BigInteger id,
              @Param("token") String token,
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java
@@ -263,13 +263,7 @@ public class DocumentChunkServiceImpl
                throw new BusinessException("分块不存在");
            }
        }
-        return chunks.stream().map(chunk -> new DocumentChunkSyncStatus(
-            chunk.getId(),
-            chunk.getIndexSyncStatus(),
-            chunk.getIndexSyncVersion(),
-            chunk.getIndexSyncErrorCode(),
-            chunk.getIndexSyncErrorMessage()
-        )).toList();
+        return syncTaskAppService.listSyncStatuses(chunks);
    }

    private DocumentChunk requireChunk(BigInteger knowledgeId, BigInteger chunkId) {
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppServiceTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppServiceTest.java
@@ -1,6 +1,14 @@
 package tech.easyflow.ai.documentchunk;

 import org.junit.Test;
+import org.junit.Assert;
+import com.easyagents.core.document.Document;
+import com.easyagents.core.model.embedding.EmbeddingModel;
+import com.easyagents.core.model.exception.ModelException;
+import com.easyagents.core.store.DocumentStore;
+import com.easyagents.core.store.StoreResult;
+import com.easyagents.core.store.VectorData;
+import com.easyagents.search.engine.service.DocumentSearcher;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -10,6 +18,8 @@ import tech.easyflow.ai.config.AiMilvusConfig;
 import tech.easyflow.ai.config.SearcherFactory;
 import tech.easyflow.ai.entity.DocumentChunk;
 import tech.easyflow.ai.entity.DocumentChunkSyncTask;
+import tech.easyflow.ai.entity.DocumentCollection;
+import tech.easyflow.ai.entity.Model;
 import tech.easyflow.ai.mapper.DocumentChunkMapper;
 import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper;
 import tech.easyflow.ai.service.DocumentCollectionService;
@@ -25,6 +35,144 @@ import java.util.function.Supplier;
 */
 public class DocumentChunkSyncTaskAppServiceTest {

+    @Test
+    public void pollingShouldExposeCurrentRetryReasonWhileWaitingAndRunning() {
+        Fixture fixture = fixture(1);
+        fixture.task.setErrorCode("EMBEDDING_REQUEST_FAILED");
+        fixture.task.setErrorMessage("向量模型服务调用失败");
+        Mockito.when(fixture.taskMapper.selectCurrentForChunks(List.of(fixture.chunkId)))
+            .thenReturn(List.of(fixture.task));
+
+        var pending = fixture.service.listSyncStatuses(List.of(fixture.chunk)).get(0);
+        Assert.assertEquals("PENDING", pending.indexSyncStatus());
+        Assert.assertEquals("EMBEDDING_REQUEST_FAILED", pending.indexSyncErrorCode());
+        Assert.assertEquals(Integer.valueOf(1), pending.indexSyncAttemptCount());
+        Assert.assertEquals(5, pending.indexSyncMaxAttempts());
+
+        fixture.task.setStatus(DocumentChunkSyncState.TASK_RUNNING);
+        fixture.task.setAttemptCount(2);
+        var running = fixture.service.listSyncStatuses(List.of(fixture.chunk)).get(0);
+        Assert.assertEquals("EMBEDDING_REQUEST_FAILED", running.indexSyncErrorCode());
+        Assert.assertEquals(Integer.valueOf(2), running.indexSyncAttemptCount());
+    }
+
+    @Test
+    public void pollingShouldNotExposeAnotherVersionsOrFinishedTasksFailure() {
+        Fixture fixture = fixture(2);
+        fixture.task.setErrorCode("EMBEDDING_REQUEST_FAILED");
+        Mockito.when(fixture.taskMapper.selectCurrentForChunks(List.of(fixture.chunkId)))
+            .thenReturn(List.of(fixture.task));
+        fixture.task.setSyncVersion(2L);
+        var changed = fixture.service.listSyncStatuses(List.of(fixture.chunk)).get(0);
+        Assert.assertNull(changed.indexSyncErrorCode());
+        Assert.assertNull(changed.indexSyncAttemptCount());
+
+        fixture.task.setSyncVersion(1L);
+        fixture.task.setStatus(DocumentChunkSyncState.TASK_SUCCEEDED);
+        var finished = fixture.service.listSyncStatuses(List.of(fixture.chunk)).get(0);
+        Assert.assertNull(finished.indexSyncErrorCode());
+    }
+
+    @Test
+    public void pollingShouldHandleChunksWithoutTasksAndSkipEmptyBatch() {
+        Fixture fixture = fixture(1);
+        Assert.assertTrue(fixture.service.listSyncStatuses(List.of()).isEmpty());
+        Mockito.verify(fixture.taskMapper, Mockito.never()).selectCurrentForChunks(Mockito.anyList());
+        Mockito.when(fixture.taskMapper.selectCurrentForChunks(List.of(fixture.chunkId)))
+            .thenReturn(List.of());
+        var state = fixture.service.listSyncStatuses(List.of(fixture.chunk)).get(0);
+        Assert.assertEquals("PENDING", state.indexSyncStatus());
+        Assert.assertNull(state.indexSyncAttemptCount());
+        Assert.assertNull(state.indexSyncErrorCode());
+    }
+
+    @Test
+    public void embeddingFailureShouldBeReportedWithoutCallingIndexes() {
+        Fixture fixture = fixture(1);
+        IndexFixture indexes = prepareIndexes(fixture);
+        Mockito.when(indexes.embeddingModel.embed(Mockito.any(Document.class), Mockito.any()))
+            .thenThrow(new ModelException("response is null or empty."));
+
+        fixture.service.handleTask(fixture.taskId);
+
+        Mockito.verify(fixture.taskMapper).failOrRetryOwned(
+            Mockito.eq(fixture.taskId), Mockito.anyString(), Mockito.eq("PENDING"), Mockito.any(),
+            Mockito.eq("EMBEDDING_REQUEST_FAILED"), Mockito.eq("向量模型服务调用失败"), Mockito.any()
+        );
+        Mockito.verify(indexes.store, Mockito.never()).doUpdate(Mockito.anyList(), Mockito.any());
+        Mockito.verifyNoInteractions(indexes.searcher);
+    }
+
+    @Test
+    public void vectorFailureShouldHaveItsOwnReasonAndEmbedOnlyOnce() {
+        Fixture fixture = fixture(1);
+        IndexFixture indexes = prepareIndexes(fixture);
+        Mockito.when(indexes.store.doUpdate(Mockito.anyList(), Mockito.any()))
+            .thenReturn(StoreResult.fail("vector write failed"));
+
+        fixture.service.handleTask(fixture.taskId);
+
+        Mockito.verify(fixture.taskMapper).failOrRetryOwned(
+            Mockito.eq(fixture.taskId), Mockito.anyString(), Mockito.eq("PENDING"), Mockito.any(),
+            Mockito.eq("VECTOR_UPSERT_FAILED"), Mockito.anyString(), Mockito.any()
+        );
+        Mockito.verify(indexes.embeddingModel).embed(Mockito.any(Document.class), Mockito.any());
+        Mockito.verifyNoInteractions(indexes.searcher);
+    }
+
+    @Test
+    public void keywordFailureShouldHaveItsOwnReason() {
+        Fixture fixture = fixture(1);
+        IndexFixture indexes = prepareIndexes(fixture);
+        Mockito.when(indexes.searcher.addDocuments(Mockito.anyList())).thenReturn(false);
+
+        fixture.service.handleTask(fixture.taskId);
+
+        Mockito.verify(fixture.taskMapper).failOrRetryOwned(
+            Mockito.eq(fixture.taskId), Mockito.anyString(), Mockito.eq("PENDING"), Mockito.any(),
+            Mockito.eq("KEYWORD_UPSERT_FAILED"), Mockito.anyString(), Mockito.any()
+        );
+    }
+
+    @Test
+    public void successfulIndexUpdateShouldClearFailureAndMarkChunkSynced() {
+        Fixture fixture = fixture(2);
+        fixture.task.setErrorCode("EMBEDDING_REQUEST_FAILED");
+        IndexFixture indexes = prepareIndexes(fixture);
+        Mockito.when(indexes.searcher.addDocuments(Mockito.anyList())).thenReturn(true);
+
+        fixture.service.handleTask(fixture.taskId);
+
+        Mockito.verify(fixture.taskMapper).finishOwned(
+            Mockito.eq(fixture.taskId), Mockito.anyString(), Mockito.eq("SUCCEEDED"),
+            Mockito.isNull(), Mockito.isNull(), Mockito.any()
+        );
+        Mockito.verify(fixture.chunkMapper).updateSyncState(fixture.chunkId, 1L, "SYNCED", null, null);
+        Mockito.verify(indexes.embeddingModel).embed(Mockito.any(Document.class), Mockito.any());
+    }
+
+    private static IndexFixture prepareIndexes(Fixture fixture) {
+        DocumentCollection collection = Mockito.mock(DocumentCollection.class);
+        DocumentStore store = Mockito.mock(DocumentStore.class, Mockito.CALLS_REAL_METHODS);
+        Model model = Mockito.mock(Model.class);
+        EmbeddingModel embeddingModel = Mockito.mock(EmbeddingModel.class);
+        DocumentSearcher searcher = Mockito.mock(DocumentSearcher.class);
+        Mockito.when(fixture.collectionService.getById(fixture.task.getDocumentCollectionId()))
+            .thenReturn(collection);
+        Mockito.when(collection.toDocumentStore()).thenReturn(store);
+        Mockito.when(fixture.modelService.getModelInstance(Mockito.any())).thenReturn(model);
+        Mockito.when(model.toEmbeddingModel()).thenReturn(embeddingModel);
+        Mockito.when(fixture.searcherFactory.getSearcher()).thenReturn(searcher);
+        VectorData vector = new VectorData();
+        vector.setVector(new float[] { 0.1f, 0.2f });
+        Mockito.when(embeddingModel.embed(Mockito.any(Document.class), Mockito.any())).thenReturn(vector);
+        Mockito.when(store.doUpdate(Mockito.anyList(), Mockito.any())).thenReturn(StoreResult.success());
+        return new IndexFixture(store, embeddingModel, searcher);
+    }
+
+    private record IndexFixture(DocumentStore store, EmbeddingModel embeddingModel, DocumentSearcher searcher) {
+    }
+
    @Test
    public void dispatchPendingShouldRecoverExpiredTasksAndSendClaimedRows() {
        Fixture fixture = fixture(1);
@@ -215,7 +363,7 @@ public class DocumentChunkSyncTaskAppServiceTest {
            Mockito.mock(ObjectProvider.class)
        );
        return new Fixture(service, taskMapper, chunkMapper, collectionService,
-            producer, task, chunk, taskId, chunkId);
+            producer, task, chunk, taskId, chunkId, modelService, searcherFactory);
    }

    private record Fixture(
@@ -227,7 +375,9 @@ public class DocumentChunkSyncTaskAppServiceTest {
        DocumentChunkSyncTask task,
        DocumentChunk chunk,
        BigInteger taskId,
-        BigInteger chunkId
+        BigInteger chunkId,
+        ModelService modelService,
+        SearcherFactory searcherFactory
    ) {
    }
 }
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImplTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImplTest.java
@@ -154,6 +154,11 @@ public class DocumentChunkServiceImplTest {
        Mockito.when(fixture.chunkMapper.selectSyncStates(
            fixture.documentId, List.of(fixture.chunkId)
        )).thenReturn(List.of(state));
+        Mockito.when(fixture.syncTaskAppService.listSyncStatuses(List.of(state)))
+            .thenReturn(List.of(new tech.easyflow.ai.dto.DocumentChunkSyncStatus(
+                state.getId(), state.getIndexSyncStatus(), state.getIndexSyncVersion(),
+                null, null, 0, 5
+            )));

        List<tech.easyflow.ai.dto.DocumentChunkSyncStatus> result =
            fixture.service.listIndexSyncStatus(