瀏覽代碼

fix(report): 修复异步采集并发重跑导致同一草稿重复入库

根因:insertReport 执行耗时长(224条食材写入+归一化管道),草稿长时间滞留
collecting 状态且 confirmed_report_id 为空,被 ReportCollectRetryTask 的 10
分钟卡死兜底误判为卡死并重新提交采集;两次采集的读-写幂等守卫存在竞态,
均通过检查后各自调用 insertReport,产生重复报告。

修复:
- insertReport 前守卫改为 CAS 原子抢占(collecting->inserting,仅
  confirmed_report_id 为空时成功),后到者失败跳过,消除并发读-写竞态
- collect() 开头早退守卫补检查 confirmed_report_id
- listPendingRetry 卡死兜底阈值 10 -> 30 分钟,避免正常长解析被误判
Xiaogang Liao 1 周之前
父節點
當前提交
f0240c6a92

+ 11 - 2
cfc-backend/src/main/java/com/etotem/cfc/mapper/HealthReportDraftMapper.java

@@ -37,14 +37,23 @@ public interface HealthReportDraftMapper extends BaseMapper<HealthReportDraft> {
             "WHERE id = #{draftId} AND (next_retry_at IS NULL OR next_retry_at <= #{now})")
     int claimRetry(@Param("draftId") Long draftId, @Param("now") Date now);
 
+    /**
+     * 原子抢占报告插入槽位:仅当草稿仍在采集中且尚未确认入库时,将 parse_status 改为 'inserting',
+     * 防止并发的重跑任务在同一份草稿上重复调用 insertReport 创建多份报告。
+     * 抢到(rows==1)后由调用方执行 insertReport;insertReport 完成会标记 completed。
+     */
+    @Update("UPDATE health_report_drafts SET parse_status = 'inserting' " +
+            "WHERE id = #{draftId} AND parse_status = 'collecting' AND confirmed_report_id IS NULL")
+    int tryClaimInsertSlot(@Param("draftId") Long draftId);
+
     /**
      * 查询待重试草稿:状态 pending + 采集中 + (已到 next_retry_at,或从未有过 next_retry_at 且
-     * 最近一次解析开始超过10分钟——用于兜底卡死/丢失的采集任务),按 updated_at 升序取前 N 条。
+     * 最近一次解析开始超过30分钟——用于兜底卡死/丢失的采集任务),按 updated_at 升序取前 N 条。
      */
     @Select("SELECT * FROM health_report_drafts " +
             "WHERE status = 'pending' AND parse_status = 'collecting' " +
             "AND ((next_retry_at IS NOT NULL AND next_retry_at <= #{now}) " +
-            "OR (next_retry_at IS NULL AND (parse_started_at IS NULL OR parse_started_at < DATE_SUB(#{now}, INTERVAL 10 MINUTE)))) " +
+            "OR (next_retry_at IS NULL AND (parse_started_at IS NULL OR parse_started_at < DATE_SUB(#{now}, INTERVAL 30 MINUTE)))) " +
             "ORDER BY updated_at ASC LIMIT #{limit}")
     List<HealthReportDraft> listPendingRetry(@Param("now") Date now, @Param("limit") int limit);
 }

+ 7 - 0
cfc-backend/src/main/java/com/etotem/cfc/service/HealthReportDraftService.java

@@ -293,6 +293,13 @@ public class HealthReportDraftService {
         return draftMapper.claimRetry(draftId, now) == 1;
     }
 
+    /**
+     * 原子抢占报告插入槽位(见 Mapper.tryClaimInsertSlot),防止并发重跑对同一草稿重复入库。
+     */
+    public boolean tryClaimInsertSlot(Long draftId) {
+        return draftMapper.tryClaimInsertSlot(draftId) == 1;
+    }
+
     /**
      * 人工重试重置:清零重试次数与错误,回到 collecting 待采集状态。
      */

+ 7 - 7
cfc-backend/src/main/java/com/etotem/cfc/service/ReportCollectService.java

@@ -150,7 +150,8 @@ public class ReportCollectService {
             log.warn("异步采集草稿不存在 draftId={}", draftId);
             return;
         }
-        if (STATUS_COMPLETED.equals(draft.getParseStatus())) {
+        if (STATUS_COMPLETED.equals(draft.getParseStatus())
+                || draft.getConfirmedReportId() != null) {
             return;
         }
         String pdfPath = null;
@@ -262,12 +263,11 @@ public class ReportCollectService {
             healthReportDraftService.updateParseMethod(draftId, method);
             healthReportDraftService.updatePayload(draftId, objectMapper.writeValueAsString(payload));
 
-            // 幂等守卫:解析期间若草稿已被确认入库或已完成(并发/外部触发),跳过本次入库
-            HealthReportDraft fresh = healthReportDraftService.getDraftById(draftId);
-            if (fresh != null && (fresh.getConfirmedReportId() != null
-                    || STATUS_COMPLETED.equals(fresh.getParseStatus()))) {
-                log.warn("草稿解析期间已被处理,跳过重复入库 draftId={}, confirmedReportId={}, parseStatus={}",
-                        draftId, fresh.getConfirmedReportId(), fresh.getParseStatus());
+            // 幂等守卫:原子抢占插入槽位(collecting -> inserting 且 confirmed_report_id 为空)。
+            // 并发重跑任务(定时重试驱动器/外部触发)解析完成后到达此处时,只有先到者能抢到槽位,
+            // 后到者因 parse_status 已变为 inserting 或 confirmed_report_id 非空而失败,避免重复入库。
+            if (!healthReportDraftService.tryClaimInsertSlot(draftId)) {
+                log.warn("草稿插入槽位已被抢占或已入库,跳过重复入库 draftId={}", draftId);
                 return;
             }