fix(任务收尾): 打破「数据永久缺失 → 卡死恢复无限重建 job」的死循环

线上任务 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 重建全部归零。
This commit is contained in:
2026-09-17 01:19:14 +08:00
parent d5952945dd
commit 188aedec84
3 changed files with 59 additions and 0 deletions
@@ -1065,6 +1065,14 @@ public class AppearancePatentTaskService {
long activeAssembleJobs = taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE); long activeAssembleJobs = taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE);
log.info("[appearance-patent] stale recovery probe taskId={} uploadComplete={} activeAssembleJobs={} persistedRows={}", log.info("[appearance-patent] stale recovery probe taskId={} uploadComplete={} activeAssembleJobs={} persistedRows={}",
taskId, uploadComplete, activeAssembleJobs, hasPersistedResultRows(taskId)); 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)) { if (!hasPersistedResultRows(taskId)) {
log.info("[appearance-patent] stale recovery aborted because no persisted rows taskId={}", taskId); log.info("[appearance-patent] stale recovery aborted because no persisted rows taskId={}", taskId);
return false; return false;
@@ -2915,12 +2923,35 @@ public class AppearancePatentTaskService {
} catch (Exception ex) { } catch (Exception ex) {
log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}", log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}",
chunk.getTaskId(), chunk.getChunkIndex(), ex.getMessage()); 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=" throw new BusinessException("appearance patent chunk payload read failed chunk="
+ chunk.getChunkIndex() + ": " + ex.getMessage(), ex); + chunk.getChunkIndex() + ": " + ex.getMessage(), ex);
} }
return rows; return rows;
} }
/** payload 对象已不存在(RustFS 返回 NoSuchKeymessage 为 "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) { private String rowKey(AppearancePatentParsedRowVo row) {
if (row == null) { if (row == null) {
return ""; return "";
@@ -87,6 +87,15 @@ public class SimilarAsinChunkPayloadSupport {
recordChunkReadFailure(chunk, false, msg); recordChunkReadFailure(chunk, false, msg);
return rows; 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={}", log.warn("[similar-asin] read chunk payload failed taskId={} chunk={} crossInstance={} err={}",
chunk.getTaskId(), chunk.getChunkIndex(), crossInstance, msg); chunk.getTaskId(), chunk.getChunkIndex(), crossInstance, msg);
recordChunkReadFailure(chunk, crossInstance, msg); recordChunkReadFailure(chunk, crossInstance, msg);
@@ -643,6 +643,25 @@ public class TaskFileJobService {
.set(TaskFileJobEntity::getTerminalCallbackAt, LocalDateTime.now())); .set(TaskFileJobEntity::getTerminalCallbackAt, LocalDateTime.now()));
} }
/**
* 该任务是否已有「重试耗尽且已走完终态回调」的组装 job。
*
* <p>用于卡死恢复(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<TaskFileJobEntity>()
.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) { public TaskFileJobEntity findAssembleJob(Long taskId, String moduleType, Long resultId) {
return findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT); return findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT);
} }