feat(任务恢复): 外观专利接上「补传恢复」；合并冲突保留兜底对象并可诊断

28459 的两个遗留项： 1) 外观专利缺「补传恢复」入口——分片缺失致组装 job 重试耗尽后，客户端补传缺口也无法自动重跑，只能人工重置 job。TaskFileJobService 早有 resetTerminalFailedForRecovery，但只被删除品牌模块接了。现按同一口径在分片提交成功后检查并恢复；best-effort，恢复失败不影响补传本身。 2) 合并 CAS 冲突不可诊断且会删掉唯一的兜底对象——原实现每次冲突都删掉刚写入的版本化对象，重试耗尽即抛异常、行仍指向旧指针；一旦旧对象也不在，该分片永久读不到（28459 的 chunk-462/473/484 正是这个形态）。现在冲突时读回行上当前哈希并写进异常与日志；终局失败保留最后一个对象，作为读路径「同槽位兄弟对象」兜底的恢复源。外观专利与相似ASIN 同一口径。测试：新增 17 个用例（补传恢复 8 / 外观专利合并冲突 5 / 相似ASIN 合并冲突 4）， RED 均已确认；全量 mvn test 3099 个 0 失败。
2026-09-17 11:04:38 +08:00
parent 3137299bfe
commit ddefcbed56
5 changed files with 659 additions and 8 deletions
@@ -424,6 +424,53 @@ public class AppearancePatentTaskService {

    public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) {
        submitResultLocked(taskId, request);
+        // 分片补传后自动恢复终态失败的组装 job —— 走与删除品牌同一套口径，
+        // 此前外观专利没有接该入口，分片补齐后只能人工重置 job（线上任务 28459 即如此）。
+        maybeRecoverTerminalFailedAssemble(taskId);
+    }
+
+    /**
+     * 补传恢复：此前因分片缺失导致组装 job 重试耗尽（终态失败），客户端补传缺口后
+     * 把失败的组装 job 重置为 PENDING 重新派发（{@code resetTerminalFailedForRecovery} 自行补发 dispatch 事件）。
+     *
+     * <p>best-effort：恢复失败不得影响补传本身——分片已经落库，恢复只是让后续组装继续推进。
+     * 常态（无终态失败 job）下只查两次即返回，不触发分片扫描。
+     */
+    private void maybeRecoverTerminalFailedAssemble(Long taskId) {
+        if (taskId == null || taskId <= 0) {
+            return;
+        }
+        try {
+            FileResultEntity result = findResultRecord(taskId);
+            if (result == null) {
+                return;
+            }
+            if (taskFileJobService.hasSuccessfulAssembleJob(taskId, MODULE_TYPE, result.getId())) {
+                return;
+            }
+            if (!taskFileJobService.isTerminalFailedAssembleJob(taskId, MODULE_TYPE, result.getId())) {
+                return;
+            }
+            if (!isResultSubmissionComplete(taskId)) {
+                log.info("[appearance-patent] 组装 job 终态失败但分片仍未补传完整，暂不恢复 taskId={} resultId={}",
+                        taskId, result.getId());
+                return;
+            }
+            log.info("[appearance-patent] 分片已补传完整，恢复终态失败的组装 job taskId={} resultId={}",
+                    taskId, result.getId());
+            taskFileJobService.resetTerminalFailedForRecovery(taskId, MODULE_TYPE, result.getId());
+        } catch (Exception ex) {
+            log.warn("[appearance-patent] 补传恢复检查失败（不影响本次补传）taskId={} err={}", taskId, ex.getMessage(), ex);
+        }
+    }
+
+    /** 按 task 取结果行（不创建）；不存在返回 null。 */
+    private FileResultEntity findResultRecord(Long taskId) {
+        List<FileResultEntity> rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
+                .eq(FileResultEntity::getTaskId, taskId)
+                .eq(FileResultEntity::getModuleType, MODULE_TYPE)
+                .last("limit 1"));
+        return rows == null || rows.isEmpty() ? null : rows.getFirst();
    }

    private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
@@ -1223,6 +1270,7 @@ public class AppearancePatentTaskService {
        if (rows == null || rows.isEmpty()) {
            return;
        }
+        String conflictDetail = "未发生冲突";
        for (int attempt = 1; attempt <= CHUNK_PAYLOAD_MERGE_RETRY_LIMIT; attempt++) {
            TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
                    .eq(TaskChunkEntity::getTaskId, taskId)
@@ -1264,13 +1312,35 @@ public class AppearancePatentTaskService {
                transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
                return;
            }
-            transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
+            // CAS 冲突：读回行上的当前哈希，否则只有一句 conflict 无从判断是谁改的（线上 28459 至今未能定位）
+            String currentHash = currentPayloadHash(chunk.getId());
+            conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash;
            if (attempt < CHUNK_PAYLOAD_MERGE_RETRY_LIMIT) {
-                log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}",
-                        taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT);
+                log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}",
+                        taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT, oldPayloadHash, currentHash);
+                // 还要重试：这次写的对象会被下次重写，先删掉避免堆积
+                transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
+            } else {
+                // 终局失败：保留这次写入的版本化对象，作为读路径「同槽位兄弟对象」兜底的恢复源。
+                // 行没指过去不该让该分片永久判死——删掉它才是线上 28459 丢数据的形态。
+                log.error("[appearance-patent] chunk payload 合并冲突重试耗尽，保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}",
+                        taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload);
            }
        }
-        throw new IllegalStateException("appearance patent chunk payload update conflict");
+        throw new IllegalStateException("appearance patent chunk payload update conflict " + conflictDetail);
+    }
+
+    /** 读回 chunk 行当前的 payload_hash，用于 CAS 冲突定位（行已不存在/读取失败时返回可读标记）。 */
+    private String currentPayloadHash(Long chunkId) {
+        if (chunkId == null) {
+            return "chunkId 为空";
+        }
+        try {
+            TaskChunkEntity latest = taskChunkMapper.selectById(chunkId);
+            return latest == null ? "行已不存在" : latest.getPayloadHash();
+        } catch (Exception ex) {
+            return "读取失败:" + ex.getMessage();
+        }
    }

    private List<AppearancePatentResultRowDto> expandRows(List<AppearancePatentResultRowDto> representatives,
@@ -214,6 +214,7 @@ public class SimilarAsinPipelineSupport {
            return;
        }
        int maxAttempts = 3;
+        String conflictDetail = "未发生冲突";
        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
                    .eq(TaskChunkEntity::getTaskId, taskId)
@@ -275,13 +276,35 @@ public class SimilarAsinPipelineSupport {
                transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
                return;
            }
-            transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
+            // CAS 冲突：读回行上的当前哈希，否则只有一句 conflict 无从判断是谁改的
+            String currentHash = currentPayloadHash(chunk.getId());
+            conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash;
            if (attempt < maxAttempts) {
-                log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}",
-                        taskId, scopeHash, chunkIndex, attempt, maxAttempts);
+                log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}",
+                        taskId, scopeHash, chunkIndex, attempt, maxAttempts, oldPayloadHash, currentHash);
+                // 还要重试：这次写的对象会被下次重写，先删掉避免堆积
+                transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
+            } else {
+                // 终局失败：保留这次写入的版本化对象，作为读路径「同槽位兄弟对象」兜底的恢复源。
+                // 与外观专利同一口径——行没指过去不该让该分片永久判死。
+                log.error("[similar-asin] chunk payload 合并冲突重试耗尽，保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}",
+                        taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload);
            }
        }
-        throw new IllegalStateException("相似ASIN分片载荷更新失败");
+        throw new IllegalStateException("相似ASIN分片载荷更新失败 " + conflictDetail);
+    }
+
+    /** 读回 chunk 行当前的 payload_hash，用于 CAS 冲突定位（行已不存在/读取失败时返回可读标记）。 */
+    private String currentPayloadHash(Long chunkId) {
+        if (chunkId == null) {
+            return "chunkId 为空";
+        }
+        try {
+            TaskChunkEntity latest = taskChunkMapper.selectById(chunkId);
+            return latest == null ? "行已不存在" : latest.getPayloadHash();
+        } catch (Exception ex) {
+            return "读取失败:" + ex.getMessage();
+        }
    }