fix: prevent document lifecycle races and add async cleanup

bootstrap/src/main/java/com/nageoffer/ai/ragent/core/ingest/DefaultIngestionKernel.java

@@ -36,6 +36,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiConsumer;
 
 /**
  * 摄取内核默认实现：固定五步骨架，全文唯一一条摄取执行序列
@@ -66,7 +67,9 @@ public class DefaultIngestionKernel implements IngestionKernel {
     public IngestionOutcome run(DocumentRef doc,
                                 byte[] bytes,
                                 IngestionSpec spec,
-                                VectorTarget target) {
+                                VectorTarget target,
+                                Runnable beforeWrite,
+                                BiConsumer<String, Integer> beforeCommit) {
         if (bytes == null || bytes.length == 0) {
             throw new ClientException("文件内容为空：docId=" + doc.docId());
         }
@@ -103,7 +106,8 @@ public IngestionOutcome run(DocumentRef doc,
 
         // ⑤ index：扇出到全部落点，事务边界在写入器内
         long indexStart = System.currentTimeMillis();
-        chunkIndexWriter.replaceDocument(target, doc, embedded);
+        chunkIndexWriter.replaceDocument(target, doc, embedded, beforeWrite,
+                () -> beforeCommit.accept(mimeType, embedded.size()));
         long indexMillis = System.currentTimeMillis() - indexStart;
 
         return new IngestionOutcome(mimeType, parser.getParserType(), blocks.size(), chunks,

bootstrap/src/main/java/com/nageoffer/ai/ragent/core/ingest/DocumentRef.java

@@ -25,8 +25,9 @@
  * @param docId    文档 ID，决定资产归属与落库归属
  * @param kbId     所属知识库 ID，决定关系库归属
  * @param filename 原始文件名，供类型识别与溯源，可为空（删除路径不需要）
+ * @param documentVersion 当前操作持有的文档版本
  */
-public record DocumentRef(String docId, String kbId, String filename) {
+public record DocumentRef(String docId, String kbId, String filename, String documentVersion) {
 
     public DocumentRef {
         if (docId == null || docId.isBlank()) {
@@ -35,12 +36,15 @@ public record DocumentRef(String docId, String kbId, String filename) {
         if (kbId == null || kbId.isBlank()) {
             throw new IllegalArgumentException("kbId 不能为空，docId=" + docId);
         }
+        if (documentVersion == null || documentVersion.isBlank()) {
+            throw new IllegalArgumentException("documentVersion 不能为空，docId=" + docId);
+        }
     }
 
     /**
      * 删除路径用：不需要文件名
      */
-    public static DocumentRef of(String docId, String kbId) {
-        return new DocumentRef(docId, kbId, null);
+    public static DocumentRef of(String docId, String kbId, String documentVersion) {
+        return new DocumentRef(docId, kbId, null, documentVersion);
     }
 }

bootstrap/src/main/java/com/nageoffer/ai/ragent/core/ingest/IngestionKernel.java

@@ -17,6 +17,8 @@
 
 package com.nageoffer.ai.ragent.core.ingest;
 
+import java.util.function.BiConsumer;
+
 /**
  * 摄取内核：固定五步骨架，调用方不可跳过、不可换序、不可替换
  * <pre>
@@ -42,5 +44,7 @@ public interface IngestionKernel {
     IngestionOutcome run(DocumentRef doc,
                          byte[] bytes,
                          IngestionSpec spec,
-                         VectorTarget target);
+                         VectorTarget target,
+                         Runnable beforeWrite,
+                         BiConsumer<String, Integer> beforeCommit);
 }

bootstrap/src/main/java/com/nageoffer/ai/ragent/core/ingest/sink/ChunkIndexWriter.java

@@ -43,9 +43,13 @@ public class ChunkIndexWriter {
     /**
      * 整体替换该文档的块：全部落点在同一个事务里
      */
-    public void replaceDocument(VectorTarget target, DocumentRef doc, List<EmbeddedChunk> chunks) {
-        transactionOperations.executeWithoutResult(status ->
-                sinks.forEach(sink -> sink.replaceDocument(target, doc, chunks)));
+    public void replaceDocument(VectorTarget target, DocumentRef doc, List<EmbeddedChunk> chunks,
+                                Runnable beforeWrite, Runnable beforeCommit) {
+        transactionOperations.executeWithoutResult(status -> {
+            beforeWrite.run();
+            sinks.forEach(sink -> sink.replaceDocument(target, doc, chunks));
+            beforeCommit.run();
+        });
         log.info("块索引写入完成 docId={} 分区={} 块数={} 落点数={}",
                 doc.docId(), target.partition(), chunks.size(), sinks.size());
     }

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/dao/entity/KnowledgeDocumentDO.java

@@ -136,9 +136,15 @@ public class KnowledgeDocumentDO {
      * - running：向量化中
      * - failed：向量化失败
      * - success：向量化完成
+     * - deleting：删除中
      */
     private String status;
 
+    /**
+     * 当前文档操作版本及写入 fencing token
+     */
+    private String documentVersion;
+
     /**
      * 创建人
      */

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/dao/mapper/KnowledgeDocumentMapper.java

@@ -19,6 +19,20 @@
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.nageoffer.ai.ragent.knowledge.dao.entity.KnowledgeDocumentDO;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
 
 public interface KnowledgeDocumentMapper extends BaseMapper<KnowledgeDocumentDO> {
+
+    @Select("""
+            SELECT *
+            FROM t_knowledge_document
+            WHERE id = #{docId}
+              AND deleted = 0
+              AND status = 'running'
+              AND document_version = #{ownerVersion}
+            FOR UPDATE
+            """)
+    KnowledgeDocumentDO selectRunningForUpdate(@Param("docId") String docId,
+                                                @Param("ownerVersion") String ownerVersion);
 }

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/enums/DocumentStatus.java

@@ -47,7 +47,12 @@ public enum DocumentStatus {
     /**
      * 文档处理成功
      */
-    SUCCESS("success");
+    SUCCESS("success"),
+
+    /**
+     * 文档删除中
+     */
+    DELETING("deleting");
 
     /**
      * 状态码

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeBaseCleanupConsumer.java

@@ -93,7 +93,7 @@ public void onMessage(MessageWrapper<KnowledgeBaseCleanupEvent> message) {
         LightRagClient lightRagClient = lightRagClientProvider.getIfAvailable();
         if (lightRagClient != null) {
             try {
-                lightRagClient.deleteByCollection(collectionName);
+                lightRagClient.deleteByCollectionOrThrow(collectionName);
             } catch (Exception e) {
                 allSucceeded = false;
                 log.error("删除 LightRAG 图谱数据失败，collectionName={}", collectionName, e);

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentChunkConsumer.java

@@ -51,7 +51,7 @@ public void onMessage(MessageWrapper<KnowledgeDocumentChunkEvent> message) {
 
         UserContext.set(LoginUser.builder().username(event.getOperator()).build());
         try {
-            documentService.executeChunk(event.getDocId());
+            documentService.executeChunk(event.getDocId(), event.getDocumentVersion());
         } finally {
             UserContext.clear();
         }

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentChunkTransactionChecker.java

@@ -61,6 +61,8 @@ public boolean check(MessageWrapper<?> message) {
         KnowledgeDocumentDO documentDO = documentMapper.selectById(docId);
 
         return documentDO != null
-                && DocumentStatus.RUNNING.getCode().equals(documentDO.getStatus());
+                && DocumentStatus.RUNNING.getCode().equals(documentDO.getStatus())
+                && event.getDocumentVersion() != null
+                && event.getDocumentVersion().equals(documentDO.getDocumentVersion());
     }
 }

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentCleanupConsumer.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.knowledge.mq;
+
+import com.nageoffer.ai.ragent.framework.exception.ServiceException;
+import com.nageoffer.ai.ragent.framework.mq.MessageWrapper;
+import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentCleanupEvent;
+import com.nageoffer.ai.ragent.rag.core.graph.LightRagClient;
+import com.nageoffer.ai.ragent.rag.core.keyword.KeywordIndexService;
+import com.nageoffer.ai.ragent.rag.core.vector.VectorStoreService;
+import com.nageoffer.ai.ragent.rag.service.FileStorageService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+/**
+ * 文档删除清理消费者
+ * <p>
+ * 每个外部资源独立尝试，全部尝试结束后只要存在真实失败就抛错触发 MQ 重试。所有删除操作均须幂等
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@RocketMQMessageListener(
+        topic = "knowledge-document-cleanup_topic${unique-name:}",
+        consumerGroup = "knowledge-document-cleanup_cg${unique-name:}"
+)
+public class KnowledgeDocumentCleanupConsumer
+        implements RocketMQListener<MessageWrapper<KnowledgeDocumentCleanupEvent>> {
+
+    private final VectorStoreService vectorStoreService;
+    private final FileStorageService fileStorageService;
+    private final ObjectProvider<KeywordIndexService> keywordIndexServiceProvider;
+    private final ObjectProvider<LightRagClient> lightRagClientProvider;
+
+    @Override
+    public void onMessage(MessageWrapper<KnowledgeDocumentCleanupEvent> message) {
+        KnowledgeDocumentCleanupEvent event = message.getBody();
+        String docId = event.getDocId();
+        String collectionName = event.getCollectionName();
+
+        log.info("[消费者] 开始清理文档外部资源，docId={}, collectionName={}", docId, collectionName);
+        boolean allSucceeded = true;
+
+        try {
+            // 此阶段仅会让外部向量数据库 Milvus 执行清理(若使用)
+            vectorStoreService.deleteDocumentVectorsAfterCommit(collectionName, docId);
+        } catch (Exception e) {
+            allSucceeded = false;
+            log.error("删除文档主向量失败，collectionName={}, docId={}", collectionName, docId, e);
+        }
+
+        KeywordIndexService keywordIndexService = keywordIndexServiceProvider.getIfAvailable();
+        if (keywordIndexService != null) {
+            try {
+                keywordIndexService.deleteDocumentIndex(collectionName, docId);
+            } catch (Exception e) {
+                allSucceeded = false;
+                log.error("删除文档 ES 索引失败，collectionName={}, docId={}", collectionName, docId, e);
+            }
+        }
+
+        LightRagClient lightRagClient = lightRagClientProvider.getIfAvailable();
+        if (lightRagClient != null) {
+            try {
+                lightRagClient.deleteByDocOrThrow(docId);
+            } catch (Exception e) {
+                allSucceeded = false;
+                log.error("删除文档 LightRAG 数据失败，docId={}", docId, e);
+            }
+        }
+
+        if (StringUtils.hasText(event.getFileUrl())) {
+            try {
+                fileStorageService.deleteByUrl(event.getFileUrl());
+            } catch (Exception e) {
+                allSucceeded = false;
+                log.error("删除文档对象文件失败，docId={}, fileUrl={}", docId, event.getFileUrl(), e);
+            }
+        }
+
+        if (!allSucceeded) {
+            throw new ServiceException("文档外部资源清理存在失败项，触发重试");
+        }
+    }
+}

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentCleanupTransactionChecker.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nageoffer.ai.ragent.knowledge.mq;
+
+import cn.hutool.json.JSONUtil;
+import com.nageoffer.ai.ragent.framework.mq.MessageWrapper;
+import com.nageoffer.ai.ragent.framework.mq.producer.DelegatingTransactionListener;
+import com.nageoffer.ai.ragent.framework.mq.producer.TransactionChecker;
+import com.nageoffer.ai.ragent.knowledge.dao.entity.KnowledgeDocumentDO;
+import com.nageoffer.ai.ragent.knowledge.dao.mapper.KnowledgeDocumentMapper;
+import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentCleanupEvent;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+/**
+ * 文档删除清理事务消息回查器
+ * <p>
+ * 只以文档是否已逻辑删除作为本地事务提交凭据
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class KnowledgeDocumentCleanupTransactionChecker implements TransactionChecker {
+
+    private final KnowledgeDocumentMapper documentMapper;
+    private final DelegatingTransactionListener transactionListener;
+
+    @Value("knowledge-document-cleanup_topic${unique-name:}")
+    private String cleanupTopic;
+
+    @PostConstruct
+    public void init() {
+        transactionListener.registerChecker(cleanupTopic, this);
+    }
+
+    @Override
+    public boolean check(MessageWrapper<?> message) {
+        log.info("[事务回查] 文档删除清理，消息体：{}", JSONUtil.toJsonStr(message));
+
+        KnowledgeDocumentCleanupEvent event = (KnowledgeDocumentCleanupEvent) message.getBody();
+        // @TableLogic 会让已删除行对 selectById 不可见,删除事务消息只会在确认文档存在后发送
+        KnowledgeDocumentDO document = documentMapper.selectById(event.getDocId());
+        return document == null || Integer.valueOf(1).equals(document.getDeleted());
+    }
+}

bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/event/KnowledgeDocumentChunkEvent.java

@@ -47,6 +47,11 @@ public class KnowledgeDocumentChunkEvent implements Serializable {
      */
     private String kbId;
 
+    /**
+     * 本次分块持有的文档版本
+     */
+    private String documentVersion;
+
     /**
      * 操作人
      */