Skip to content

Commit 520f910

Browse files
AllureCurtainAllureCurtain
committed
fix: avoid document delete chunk race
Fixes #42
1 parent f0f6723 commit 520f910

20 files changed

Lines changed: 1414 additions & 35 deletions

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/config/CleanupTaskProperties.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.config;
19+
20+
import lombok.Data;
21+
import org.springframework.boot.context.properties.ConfigurationProperties;
22+
import org.springframework.context.annotation.Configuration;
23+
import org.springframework.validation.annotation.Validated;
24+
25+
/**
26+
* outbox 清理任务 worker 配置
27+
*/
28+
@Data
29+
@Validated
30+
@Configuration
31+
@ConfigurationProperties(prefix = "rag.knowledge.cleanup")
32+
public class CleanupTaskProperties {
33+
34+
/** worker 扫描间隔（毫秒） */
35+
private Long scanDelayMs = 15000L;
36+
37+
/** 每次扫描批量大小 */
38+
private Integer batchSize = 50;
39+
40+
/** 最大重试次数，超过置为 failed 并告警 */
41+
private Integer maxRetry = 5;
42+
43+
/** 重试退避基数（秒），实际退避 = baseBackoffSeconds * 2^retryCount */
44+
private Long baseBackoffSeconds = 30L;
45+
}

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

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.dao.entity;
19+
20+
import com.baomidou.mybatisplus.annotation.FieldFill;
21+
import com.baomidou.mybatisplus.annotation.IdType;
22+
import com.baomidou.mybatisplus.annotation.TableField;
23+
import com.baomidou.mybatisplus.annotation.TableId;
24+
import com.baomidou.mybatisplus.annotation.TableName;
25+
import lombok.AllArgsConstructor;
26+
import lombok.Builder;
27+
import lombok.Data;
28+
import lombok.NoArgsConstructor;
29+
30+
import java.util.Date;
31+
32+
/**
33+
* 文件清理任务（outbox）。
34+
* 删除主事务内写入，由 CleanupTaskWorker 异步执行文件存储删除并保证最终一致。
35+
*/
36+
@Data
37+
@NoArgsConstructor
38+
@AllArgsConstructor
39+
@Builder
40+
@TableName("t_file_cleanup_task")
41+
public class FileCleanupTaskDO {
42+
43+
@TableId(type = IdType.ASSIGN_ID)
44+
private String id;
45+
46+
/** 文件存储地址（fileUrl） */
47+
private String fileUrl;
48+
49+
/** 任务状态：pending/success/failed，见 CleanupTaskStatus */
50+
private String status;
51+
52+
/** 已重试次数 */
53+
private Integer retryCount;
54+
55+
/** 下次可执行时间（退避重试） */
56+
private Date nextRetryTime;
57+
58+
/** 当前领取者 */
59+
private String lockOwner;
60+
61+
/** 领取租约到期时间 */
62+
private Date lockUntil;
63+
64+
/** 最近一次错误信息 */
65+
private String errorMessage;
66+
67+
@TableField(fill = FieldFill.INSERT)
68+
private Date createTime;
69+
70+
@TableField(fill = FieldFill.INSERT_UPDATE)
71+
private Date updateTime;
72+
}

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

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.dao.entity;
19+
20+
import com.baomidou.mybatisplus.annotation.FieldFill;
21+
import com.baomidou.mybatisplus.annotation.IdType;
22+
import com.baomidou.mybatisplus.annotation.TableField;
23+
import com.baomidou.mybatisplus.annotation.TableId;
24+
import com.baomidou.mybatisplus.annotation.TableName;
25+
import lombok.AllArgsConstructor;
26+
import lombok.Builder;
27+
import lombok.Data;
28+
import lombok.NoArgsConstructor;
29+
30+
import java.util.Date;
31+
32+
/**
33+
* 向量清理任务（outbox）。
34+
* 删除主事务内写入，由 CleanupTaskWorker 异步执行 PgVector 删除并保证最终一致。
35+
*/
36+
@Data
37+
@NoArgsConstructor
38+
@AllArgsConstructor
39+
@Builder
40+
@TableName("t_vector_cleanup_task")
41+
public class VectorCleanupTaskDO {
42+
43+
@TableId(type = IdType.ASSIGN_ID)
44+
private String id;
45+
46+
/** 文档 ID */
47+
private String docId;
48+
49+
/** 向量集合名（collection_name） */
50+
private String collectionName;
51+
52+
/** 任务状态：pending/success/failed，见 CleanupTaskStatus */
53+
private String status;
54+
55+
/** 已重试次数 */
56+
private Integer retryCount;
57+
58+
/** 下次可执行时间（退避重试） */
59+
private Date nextRetryTime;
60+
61+
/** 当前领取者 */
62+
private String lockOwner;
63+
64+
/** 领取租约到期时间 */
65+
private Date lockUntil;
66+
67+
/** 最近一次错误信息 */
68+
private String errorMessage;
69+
70+
@TableField(fill = FieldFill.INSERT)
71+
private Date createTime;
72+
73+
@TableField(fill = FieldFill.INSERT_UPDATE)
74+
private Date updateTime;
75+
}

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

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.dao.mapper;
19+
20+
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
21+
import com.nageoffer.ai.ragent.knowledge.dao.entity.FileCleanupTaskDO;
22+
import org.apache.ibatis.annotations.Param;
23+
import org.apache.ibatis.annotations.Update;
24+
25+
import java.util.Date;
26+
27+
public interface FileCleanupTaskMapper extends BaseMapper<FileCleanupTaskDO> {
28+
29+
@Update("UPDATE t_file_cleanup_task "
30+
+ "SET status = #{running}, lock_owner = #{lockOwner}, lock_until = #{lockUntil}, update_time = NOW() "
31+
+ "WHERE id = #{id} AND status = #{pending} "
32+
+ "AND (next_retry_time IS NULL OR next_retry_time <= #{now})")
33+
int claimProcessing(@Param("id") String id, @Param("lockOwner") String lockOwner,
34+
@Param("lockUntil") Date lockUntil, @Param("now") Date now,
35+
@Param("running") String runningCode, @Param("pending") String pendingCode);
36+
37+
@Update("UPDATE t_file_cleanup_task "
38+
+ "SET status = #{success}, lock_owner = NULL, lock_until = NULL, error_message = NULL, update_time = NOW() "
39+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
40+
int markSuccessIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
41+
@Param("success") String successCode, @Param("running") String runningCode);
42+
43+
@Update("UPDATE t_file_cleanup_task "
44+
+ "SET status = #{pending}, retry_count = #{retryCount}, next_retry_time = #{nextRetryTime}, "
45+
+ "error_message = #{errorMessage}, lock_owner = NULL, lock_until = NULL, update_time = NOW() "
46+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
47+
int markRetryIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
48+
@Param("pending") String pendingCode, @Param("running") String runningCode,
49+
@Param("retryCount") int retryCount, @Param("nextRetryTime") Date nextRetryTime,
50+
@Param("errorMessage") String errorMessage);
51+
52+
@Update("UPDATE t_file_cleanup_task "
53+
+ "SET status = #{failed}, retry_count = #{retryCount}, error_message = #{errorMessage}, "
54+
+ "lock_owner = NULL, lock_until = NULL, update_time = NOW() "
55+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
56+
int markFailedIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
57+
@Param("failed") String failedCode, @Param("running") String runningCode,
58+
@Param("retryCount") int retryCount, @Param("errorMessage") String errorMessage);
59+
60+
@Update("UPDATE t_file_cleanup_task "
61+
+ "SET status = #{pending}, lock_owner = NULL, lock_until = NULL, update_time = NOW() "
62+
+ "WHERE status = #{running} AND lock_until <= #{now}")
63+
int recoverExpiredProcessing(@Param("now") Date now,
64+
@Param("running") String runningCode, @Param("pending") String pendingCode);
65+
}

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,35 @@
1919

2020
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
2121
import com.nageoffer.ai.ragent.knowledge.dao.entity.KnowledgeDocumentDO;
22+
import org.apache.ibatis.annotations.Param;
23+
import org.apache.ibatis.annotations.Select;
24+
import org.apache.ibatis.annotations.Update;
25+
26+
import java.util.List;
2227

2328
public interface KnowledgeDocumentMapper extends BaseMapper<KnowledgeDocumentDO> {
29+
30+
/**
31+
* CAS 抢占文档状态：仅当当前状态命中 fromStatuses 且未删除时，原子置为 toStatus。
32+
*
33+
* @return 受影响行数；0 表示并发冲突（状态已被其它流程改写）
34+
*/
35+
@Update("<script>"
36+
+ "UPDATE t_knowledge_document "
37+
+ "SET status = #{toStatus}, update_time = NOW() "
38+
+ "WHERE id = #{docId} AND deleted = 0 "
39+
+ "AND status IN "
40+
+ "<foreach collection='fromStatuses' item='s' open='(' close=')' separator=','>#{s}</foreach>"
41+
+ "</script>")
42+
int casStatus(@Param("docId") String docId,
43+
@Param("fromStatuses") List<String> fromStatuses,
44+
@Param("toStatus") String toStatus);
45+
46+
/**
47+
* 仅查询文档当前状态（不走逻辑删除过滤，供分块中段二次校验使用）。
48+
*
49+
* @return status 字符串；文档不存在时返回 null
50+
*/
51+
@Select("SELECT status FROM t_knowledge_document WHERE id = #{docId}")
52+
String selectStatusById(@Param("docId") String docId);
2453
}

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

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.dao.mapper;
19+
20+
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
21+
import com.nageoffer.ai.ragent.knowledge.dao.entity.VectorCleanupTaskDO;
22+
import org.apache.ibatis.annotations.Param;
23+
import org.apache.ibatis.annotations.Update;
24+
25+
import java.util.Date;
26+
27+
public interface VectorCleanupTaskMapper extends BaseMapper<VectorCleanupTaskDO> {
28+
29+
@Update("UPDATE t_vector_cleanup_task "
30+
+ "SET status = #{running}, lock_owner = #{lockOwner}, lock_until = #{lockUntil}, update_time = NOW() "
31+
+ "WHERE id = #{id} AND status = #{pending} "
32+
+ "AND (next_retry_time IS NULL OR next_retry_time <= #{now})")
33+
int claimProcessing(@Param("id") String id, @Param("lockOwner") String lockOwner,
34+
@Param("lockUntil") Date lockUntil, @Param("now") Date now,
35+
@Param("running") String runningCode, @Param("pending") String pendingCode);
36+
37+
@Update("UPDATE t_vector_cleanup_task "
38+
+ "SET status = #{success}, lock_owner = NULL, lock_until = NULL, error_message = NULL, update_time = NOW() "
39+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
40+
int markSuccessIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
41+
@Param("success") String successCode, @Param("running") String runningCode);
42+
43+
@Update("UPDATE t_vector_cleanup_task "
44+
+ "SET status = #{pending}, retry_count = #{retryCount}, next_retry_time = #{nextRetryTime}, "
45+
+ "error_message = #{errorMessage}, lock_owner = NULL, lock_until = NULL, update_time = NOW() "
46+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
47+
int markRetryIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
48+
@Param("pending") String pendingCode, @Param("running") String runningCode,
49+
@Param("retryCount") int retryCount, @Param("nextRetryTime") Date nextRetryTime,
50+
@Param("errorMessage") String errorMessage);
51+
52+
@Update("UPDATE t_vector_cleanup_task "
53+
+ "SET status = #{failed}, retry_count = #{retryCount}, error_message = #{errorMessage}, "
54+
+ "lock_owner = NULL, lock_until = NULL, update_time = NOW() "
55+
+ "WHERE id = #{id} AND status = #{running} AND lock_owner = #{lockOwner}")
56+
int markFailedIfOwned(@Param("id") String id, @Param("lockOwner") String lockOwner,
57+
@Param("failed") String failedCode, @Param("running") String runningCode,
58+
@Param("retryCount") int retryCount, @Param("errorMessage") String errorMessage);
59+
60+
@Update("UPDATE t_vector_cleanup_task "
61+
+ "SET status = #{pending}, lock_owner = NULL, lock_until = NULL, update_time = NOW() "
62+
+ "WHERE status = #{running} AND lock_until <= #{now}")
63+
int recoverExpiredProcessing(@Param("now") Date now,
64+
@Param("running") String runningCode, @Param("pending") String pendingCode);
65+
}

0 commit comments

Comments
 (0)