From 188aedec84d30465fdb17bd34fdd9cf025196cdf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Thu, 17 Sep 2026 01:19:14 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E4=BB=BB=E5=8A=A1=E6=94=B6=E5=B0=BE):=20?= =?UTF-8?q?=E6=89=93=E7=A0=B4=E3=80=8C=E6=95=B0=E6=8D=AE=E6=B0=B8=E4=B9=85?= =?UTF-8?q?=E7=BC=BA=E5=A4=B1=20=E2=86=92=20=E5=8D=A1=E6=AD=BB=E6=81=A2?= =?UTF-8?q?=E5=A4=8D=E6=97=A0=E9=99=90=E9=87=8D=E5=BB=BA=20job=E3=80=8D?= =?UTF-8?q?=E7=9A=84=E6=AD=BB=E5=BE=AA=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 线上任务 28459（appearance_patent）卡在 RUNNING：chunk 462/473/484 的 payload 对象已不存在，组装失败 → stale recovery 每 30 秒重建一次 assemble job 再失败， 而恢复过程又会刷新任务心跳、任务永远判不出「卡死」，48 分钟空转近百轮。 - appearance-patent：读 chunk 时若 payload 对象已不存在（RustFS NoSuchKey）， 跳过该分片让任务按已有分片出部分结果，不再抛异常把组装永久拖死； similar-asin 在原有 typeMismatch 跳过分支旁补同款处理 - TaskFileJobService.hasExhaustedAssembleJob + stale recovery：已有「重试耗尽且 已终态收尾」的 assemble job 时放弃恢复，交给 finalizeStaleTask 按失败收尾 —— 用户看到明确失败，而不是无限 RUNNING 实测：部署后两节点 28459 相关日志、rustfs 确定性错误、stale recovery 重建全部归零。 --- .../service/AppearancePatentTaskService.java | 31 +++++++++++++++++++ .../SimilarAsinChunkPayloadSupport.java | 9 ++++++ .../task/service/TaskFileJobService.java | 19 ++++++++++++ 3 files changed, 59 insertions(+) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index 75619146..a0e492e0 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -1065,6 +1065,14 @@ public class AppearancePatentTaskService { long activeAssembleJobs = taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE); log.info("[appearance-patent] stale recovery probe taskId={} uploadComplete={} activeAssembleJobs={} persistedRows={}", taskId, uploadComplete, activeAssembleJobs, hasPersistedResultRows(taskId)); + // 已有「重试耗尽且已终态收尾」的 assemble job：说明恢复已经试过、缺失是永久的。 + // 再重建只会每 30 秒空转一轮，而且恢复过程刷新任务心跳会让任务永远 RUNNING + // （线上任务 28459 实测：48 分钟里每隔 30 秒重建一次 job）。返回 false 交给 + // finalizeStaleTask 按失败收尾，用户看到明确失败而不是无限等待。 + if (taskFileJobService.hasExhaustedAssembleJob(taskId, MODULE_TYPE)) { + log.warn("[appearance-patent] stale recovery 放弃：已有重试耗尽的 assemble job，按失败收尾 taskId={}", taskId); + return false; + } if (!hasPersistedResultRows(taskId)) { log.info("[appearance-patent] stale recovery aborted because no persisted rows taskId={}", taskId); return false; @@ -2915,12 +2923,35 @@ public class AppearancePatentTaskService { } catch (Exception ex) { log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}", chunk.getTaskId(), chunk.getChunkIndex(), ex.getMessage()); + if (isPayloadMissing(ex)) { + // payload 对象已不在（被清理或从未写入）：重试多少次都读不回来。继续抛会让 + // ASSEMBLE_RESULT job 的终态回调每轮重跑兜底组装 → 再读同一个缺失对象 → 无限循环 + // （线上任务 28459 每 10~30 秒重试一次）。跳过该分片，让任务按已有分片出部分结果， + // 与品牌/相似ASIN「失败也产出可下载的部分结果」同一口径。 + log.warn("[appearance-patent] chunk payload 已不存在，跳过该分片（任务按已有分片出结果）" + + " taskId={} chunk={}", chunk.getTaskId(), chunk.getChunkIndex()); + return rows; + } throw new BusinessException("appearance patent chunk payload read failed chunk=" + chunk.getChunkIndex() + ": " + ex.getMessage(), ex); } return rows; } + /** payload 对象已不存在（RustFS 返回 NoSuchKey：message 为 "The specified key does not exist."）。 + * 只有这种"重试也没用"的缺失才允许跳过；网络类失败仍照旧抛出以便重试。 */ + private static boolean isPayloadMissing(Throwable error) { + Throwable cursor = error; + while (cursor != null) { + String message = cursor.getMessage(); + if (message != null && message.contains("does not exist")) { + return true; + } + cursor = cursor.getCause(); + } + return false; + } + private String rowKey(AppearancePatentParsedRowVo row) { if (row == null) { return ""; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinChunkPayloadSupport.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinChunkPayloadSupport.java index 63f8d5c7..d69ce81c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinChunkPayloadSupport.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinChunkPayloadSupport.java @@ -87,6 +87,15 @@ public class SimilarAsinChunkPayloadSupport { recordChunkReadFailure(chunk, false, msg); return rows; } + // payload 对象已不存在（RustFS 返回 NoSuchKey）：重试多少次都读不回来，跳过而不是 + // 把组装/收尾永久拖死——与上面的 typeMismatch 分支、以及 collect-data 的降级口径一致。 + // 线上 appearance-patent 28459 就因同类场景每 10~30 秒重试一次（见同批修复）。 + if (msg.contains("does not exist")) { + log.warn("[similar-asin] chunk payload 已不存在，跳过该分片 taskId={} chunk={} err={}", + chunk.getTaskId(), chunk.getChunkIndex(), msg); + recordChunkReadFailure(chunk, false, msg); + return rows; + } log.warn("[similar-asin] read chunk payload failed taskId={} chunk={} crossInstance={} err={}", chunk.getTaskId(), chunk.getChunkIndex(), crossInstance, msg); recordChunkReadFailure(chunk, crossInstance, msg); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java index 8066b283..5aa93a52 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java @@ -643,6 +643,25 @@ public class TaskFileJobService { .set(TaskFileJobEntity::getTerminalCallbackAt, LocalDateTime.now())); } + /** + * 该任务是否已有「重试耗尽且已走完终态回调」的组装 job。 + * + *

用于卡死恢复（stale recovery）判断"再重建一次还有没有意义"：数据永久缺失时， + * 每轮重建只会再失败一次，而恢复过程又会刷新任务心跳，导致任务永远 RUNNING、 + * 恢复每 30 秒空转一轮（线上任务 28459 实测）。 + */ + public boolean hasExhaustedAssembleJob(Long taskId, String moduleType) { + if (taskId == null || taskId <= 0 || moduleType == null || moduleType.isBlank()) { + return false; + } + return taskFileJobMapper.selectCount(new LambdaQueryWrapper() + .eq(TaskFileJobEntity::getTaskId, taskId) + .eq(TaskFileJobEntity::getModuleType, moduleType) + .eq(TaskFileJobEntity::getJobType, JOB_TYPE_ASSEMBLE_RESULT) + .ge(TaskFileJobEntity::getRetryCount, MAX_RETRY_COUNT) + .isNotNull(TaskFileJobEntity::getTerminalCallbackAt)) > 0; + } + public TaskFileJobEntity findAssembleJob(Long taskId, String moduleType, Long resultId) { return findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT); }