diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java index 6ec284d8..80622f56 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java @@ -44,4 +44,11 @@ public class DeleteBrandProgressProperties { * Withdraw RUNNING timeout auto-finalize/fail threshold in minutes. */ private long withdrawStaleTimeoutMinutes = 30; + + /** + * 组装结果文件时允许缺失的最大分片数：缺失不超过该值时降级组装（缺失行状态写"未回传"）， + * 保证任务不因少量分片丢失整体白跑；超出时任务失败，但仍会在重试耗尽后尝试 + * fallbackAssembleOnFailure 生成部分结果文件。 + */ + private int missingChunkFallbackLimit = 20; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java index 81a41104..d2465e02 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/controller/DeleteBrandRunController.java @@ -129,6 +129,13 @@ public class DeleteBrandRunController { } } + @GetMapping("/tasks/{taskId}/missing-chunks") + @Operation(summary = "查询删除品牌任务缺失的分片号", description = "供插件端在组装失败后补传缺失分片：携带缺失 chunkIndex 重新 POST 同一 /tasks/{taskId}/result 即可触发恢复组装。") + public ApiResponse missingChunks( + @PathVariable Long taskId) { + return ApiResponse.success(deleteBrandRunService.listMissingChunks(taskId)); + } + @GetMapping("/tasks/{taskId}/download") @Operation(summary = "下载删除品牌任务结果", description = "返回任务结果文件流，便于桌面端通过 save_file_from_url_new 直接保存。") public void download(@PathVariable Long taskId, jakarta.servlet.http.HttpServletResponse response) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandMissingChunksVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandMissingChunksVo.java new file mode 100644 index 00000000..dee511d7 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/vo/DeleteBrandMissingChunksVo.java @@ -0,0 +1,37 @@ +package com.nanri.aiimage.modules.deletebrand.model.vo; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.ArrayList; +import java.util.List; + +/** + * 删除品牌任务缺失分片查询结果（供插件端补传）。 + */ +@Data +@Schema(description = "删除品牌任务缺失分片") +public class DeleteBrandMissingChunksVo { + + @Schema(description = "任务ID") + private Long taskId; + + @Schema(description = "按文件维度的缺失分片列表") + private List files = new ArrayList<>(); + + @Data + @Schema(description = "单个文件的缺失分片") + public static class MissingFileItem { + @Schema(description = "文件标识（hash）") + private String fileIdentity; + + @Schema(description = "文件总分片数") + private Integer chunkTotal; + + @Schema(description = "已回传分片数") + private Integer received; + + @Schema(description = "缺失的分片号（1 起）") + private List missing = new ArrayList<>(); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandResultFileJobHandler.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandResultFileJobHandler.java index eb9a8ae0..2a6b3c57 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandResultFileJobHandler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandResultFileJobHandler.java @@ -30,4 +30,10 @@ public class DeleteBrandResultFileJobHandler implements ResultFileJobHandler { public void cleanup(TaskFileJobEntity job) { deleteBrandRunService.cleanupResultFileJob(job); } + + @Override + public boolean fallbackAssembleOnFailure(TaskFileJobEntity job, String message) { + // 分片缺失等导致组装 job 重试耗尽时：忽略缺失阈值，尽力用已回传数据生成部分结果文件供下载。 + return deleteBrandRunService.fallbackAssembleResultFile(job); + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java index 713d55dc..ddb45d43 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java @@ -21,6 +21,7 @@ import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryGroupVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandHistoryVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandLineProgressInfoVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandLineProgressVo; +import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandMissingChunksVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandPreviewRowVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandResultItemVo; import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandRunVo; @@ -78,6 +79,9 @@ public class DeleteBrandRunService { private static final String MODULE_TYPE = "DELETE_BRAND"; + /** 降级组装时缺失分片对应的结果状态（客户端未回传该行）。 */ + static final String MISSING_CHUNK_STATUS = "未回传"; + public TaskProgressLightBatchVo progressLight(List taskIds) { return taskProgressLightAssembler.assemble(MODULE_TYPE, null, taskIds); } @@ -1342,6 +1346,8 @@ public class DeleteBrandRunService { progress.put("last_heartbeat_at", String.valueOf(System.currentTimeMillis())); deleteBrandTaskCacheService.saveProgress(taskId, progress, true); + maybeRecoverAssembleOnCompletedChunks(taskId, fileIdentity, parsedFile); + if (isFileCompleted(fileDto)) { log.info("[DeleteBrand] File chunk transmission fully reached via chunkTotal for taskId: {}, file: {}", taskId, fileDto.getSourceFilename()); final String identityForTx = fileIdentity; @@ -1357,6 +1363,39 @@ public class DeleteBrandRunService { } } + /** + * 补传恢复：某文件的分片此前因缺失导致组装 job 终态失败（重试耗尽）， + * 客户端补传缺口后这里把失败的组装 job 重置为 PENDING 重新派发。 + * 常态（无终态失败 job）下只查一次 job 状态即返回，不触发分片扫描。 + */ + private void maybeRecoverAssembleOnCompletedChunks(Long taskId, String fileIdentity, DeleteBrandParsedFileCacheDto parsedFile) { + if (taskId == null || fileIdentity == null || fileIdentity.isBlank()) { + return; + } + FileResultEntity resultEntity = findResultEntity(taskId, fileIdentity, parsedFile); + if (resultEntity == null) { + return; + } + if (taskFileJobService.hasSuccessfulAssembleJob(taskId, MODULE_TYPE, resultEntity.getId())) { + return; + } + if (!taskFileJobService.isTerminalFailedAssembleJob(taskId, MODULE_TYPE, resultEntity.getId())) { + return; + } + List chunks = deleteBrandTaskStorageService.loadMergedChunks(taskId).getOrDefault(fileIdentity, List.of()); + if (!isMergedFileCompleted(chunks)) { + return; + } + FileTaskEntity task = fileTaskMapper.selectById(taskId); + if (task == null) { + return; + } + log.info("[DeleteBrand] missing chunks patched, recovering terminal failed assemble -> taskId={} file={} resultId={}", + taskId, fileIdentity, resultEntity.getId()); + taskFileJobService.resetTerminalFailedForRecovery(taskId, MODULE_TYPE, resultEntity.getId()); + enqueueCompletedFileAssembleJob(task, fileIdentity, parsedFile); + } + private void enqueueCompletedFileAssembleJob(FileTaskEntity task, String fileIdentity, DeleteBrandParsedFileCacheDto parsedFile) { @@ -1558,6 +1597,30 @@ public class DeleteBrandRunService { return true; } + /** 缺失的分片号（1..chunkTotal 中未回传的序号），升序返回；chunkTotal 未知时返回空表。 */ + private List missingChunkIndexes(List chunks) { + if (chunks == null || chunks.isEmpty()) { + return List.of(); + } + Integer chunkTotal = chunks.get(0).getChunkTotal(); + if (chunkTotal == null || chunkTotal <= 0) { + return List.of(); + } + Set indexes = new LinkedHashSet<>(); + for (DeleteBrandResultFileDto chunk : chunks) { + if (chunk.getChunkIndex() != null && chunkTotal.equals(chunk.getChunkTotal())) { + indexes.add(chunk.getChunkIndex()); + } + } + List missing = new ArrayList<>(); + for (int i = 1; i <= chunkTotal; i++) { + if (!indexes.contains(i)) { + missing.add(i); + } + } + return missing; + } + private void finalizeTask(FileTaskEntity task, Map parsedPayload, Map> mergedByFile) { @@ -1591,10 +1654,19 @@ public class DeleteBrandRunService { String fileIdentity = entry.getKey(); DeleteBrandParsedFileCacheDto parsedFile = entry.getValue(); List chunks = mergedByFile.get(fileIdentity); - if (!isMergedFileCompleted(chunks)) { - throw new BusinessException("存在未完成的结果分片: " + parsedFile.getSourceFilename()); + boolean chunksComplete = isMergedFileCompleted(chunks); + if (!chunksComplete) { + List missing = missingChunkIndexes(chunks); + int limit = Math.max(1, deleteBrandProgressProperties.getMissingChunkFallbackLimit()); + if (missing.size() > limit) { + throw new BusinessException("存在未完成的结果分片: " + parsedFile.getSourceFilename() + "，缺失分片: " + missing + + "（已收 " + (chunks == null ? 0 : chunks.size()) + "/" + + (chunks == null || chunks.isEmpty() || chunks.get(0).getChunkTotal() == null ? 0 : chunks.get(0).getChunkTotal()) + "）"); + } + log.warn("[DeleteBrand] finalize assemble with missing chunks -> taskId={} file={} missing={} fallbackLimit={}", + task.getId(), parsedFile.getSourceFilename(), missing, limit); } - mergeChunks(parsedFile, chunks); + mergeChunks(parsedFile, chunks, !chunksComplete); FileResultEntity resultEntity = resultEntityByIdentity.get(fileIdentity); if (resultEntity == null) { @@ -1697,6 +1769,15 @@ public class DeleteBrandRunService { } public void processResultFileJob(TaskFileJobEntity job) { + processResultFileJob(job, false); + } + + /** + * 处理结果文件组装 job。 + * + * @param forceFallback 失败兜底模式：忽略缺失阈值，缺失分片一律降级为"未回传"并生成部分结果文件 + */ + public void processResultFileJob(TaskFileJobEntity job, boolean forceFallback) { if (job == null || job.getTaskId() == null || job.getResultId() == null) { throw new BusinessException("结果文件任务参数不完整"); } @@ -1722,10 +1803,20 @@ public class DeleteBrandRunService { throw new BusinessException("任务原始数据不存在: " + fileIdentity); } List chunks = mergedByFile.get(fileIdentity); - if (!isMergedFileCompleted(chunks)) { - throw new BusinessException("结果分片未完整: " + fileIdentity); + boolean chunksComplete = isMergedFileCompleted(chunks); + if (!chunksComplete) { + List missing = missingChunkIndexes(chunks); + int limit = Math.max(1, deleteBrandProgressProperties.getMissingChunkFallbackLimit()); + if (!forceFallback && missing.size() > limit) { + int chunkTotal = chunks == null || chunks.isEmpty() || chunks.get(0).getChunkTotal() == null + ? 0 : chunks.get(0).getChunkTotal(); + throw new BusinessException("结果分片未完整: " + fileIdentity + "，缺失分片: " + missing + + "（已收 " + (chunks == null ? 0 : chunks.size()) + "/" + chunkTotal + "）"); + } + log.warn("[DeleteBrand] result chunks incomplete, fallback assemble -> taskId={} file={} missing={} received={} forceFallback={} fallbackLimit={}", + job.getTaskId(), fileIdentity, missing, chunks == null ? 0 : chunks.size(), forceFallback, limit); } - MergedDeleteBrandFile mergedFile = mergeChunks(parsedFile, chunks); + MergedDeleteBrandFile mergedFile = mergeChunks(parsedFile, chunks, !chunksComplete); File outputFile = buildResultWorkbookPreserveLayout(job.getTaskId(), parsedFile, mergedFile); try { deleteBrandTaskCacheService.saveProgress(job.getTaskId(), Map.of( @@ -1758,6 +1849,55 @@ public class DeleteBrandRunService { cleanupCompletedTaskDataIfNoPendingFileJobs(job.getTaskId()); } + /** + * 失败兜底：job 重试耗尽后调用，忽略缺失阈值、尽力用已回传分片生成部分结果文件。 + * 失败只记日志（job 已终态失败，不改变其状态），返回是否有可下载结果。 + */ + public boolean fallbackAssembleResultFile(TaskFileJobEntity job) { + if (job == null || job.getTaskId() == null) { + return false; + } + try { + processResultFileJob(job, true); + log.warn("[DeleteBrand] fallback assemble produced partial result -> taskId={} resultId={} file={}", + job.getTaskId(), job.getResultId(), job.getScopeKey()); + return true; + } catch (Exception ex) { + log.warn("[DeleteBrand] fallback assemble also failed taskId={} resultId={} msg={}", + job.getTaskId(), job.getResultId(), ex.getMessage()); + return false; + } + } + + /** + * 查询任务每个文件缺失的分片号，供插件端补传（POST 同一 /tasks/{taskId}/result 携带对应 chunkIndex）。 + */ + public DeleteBrandMissingChunksVo listMissingChunks(Long taskId) { + DeleteBrandMissingChunksVo vo = new DeleteBrandMissingChunksVo(); + vo.setTaskId(taskId); + if (taskId == null || taskId <= 0) { + return vo; + } + Map> mergedByFile = deleteBrandTaskStorageService.loadMergedChunks(taskId); + List files = new ArrayList<>(); + for (Map.Entry> entry : mergedByFile.entrySet()) { + List missing = missingChunkIndexes(entry.getValue()); + if (missing.isEmpty()) { + continue; + } + DeleteBrandMissingChunksVo.MissingFileItem item = new DeleteBrandMissingChunksVo.MissingFileItem(); + item.setFileIdentity(entry.getKey()); + item.setChunkTotal(entry.getValue().isEmpty() || entry.getValue().get(0).getChunkTotal() == null + ? 0 : entry.getValue().get(0).getChunkTotal()); + item.setReceived(entry.getValue().size()); + item.setMissing(missing); + files.add(item); + } + vo.setFiles(files); + log.info("[DeleteBrand] missing chunks queried taskId={} files={}", taskId, files.size()); + return vo; + } + private void cleanupCompletedTaskDataIfNoPendingFileJobs(Long taskId) { if (taskId == null || taskId <= 0) { return; @@ -1780,7 +1920,31 @@ public class DeleteBrandRunService { return filename; } - private MergedDeleteBrandFile mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List chunks) { + /** 没有任何已回传分片时的降级合并：所有国家/ASIN 行标记"未回传"，布局与全量结果一致。 */ + private MergedDeleteBrandFile mergeChunksAsMissingAll(DeleteBrandParsedFileCacheDto parsedFile) { + List mergedCountries = new ArrayList<>(); + for (DeleteBrandCountryGroupVo parsedCountry : parsedFile.getCountries()) { + DeleteBrandProcessedCountryDto mergedCountry = new DeleteBrandProcessedCountryDto(); + mergedCountry.setCountry(parsedCountry.getCountry()); + for (DeleteBrandCountryAsinVo parsedAsin : parsedCountry.getItems()) { + DeleteBrandCountryResultItemDto row = new DeleteBrandCountryResultItemDto(); + row.setAsin(parsedAsin.getAsin()); + row.setStatus(MISSING_CHUNK_STATUS); + mergedCountry.getItems().add(row); + } + mergedCountries.add(mergedCountry); + } + return new MergedDeleteBrandFile("", "", 0, mergedCountries); + } + + private MergedDeleteBrandFile mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List chunks, boolean allowMissing) { + if (chunks == null || chunks.isEmpty()) { + // 没有任何已回传分片：降级时全部行标记"未回传"，严格模式抛业务异常（而非索引越界） + if (!allowMissing) { + throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename()); + } + return mergeChunksAsMissingAll(parsedFile); + } Integer chunkTotal = chunks.get(0).getChunkTotal(); for (DeleteBrandResultFileDto chunk : chunks) { if (!chunkTotal.equals(chunk.getChunkTotal())) { @@ -1804,18 +1968,26 @@ public class DeleteBrandRunService { for (DeleteBrandCountryGroupVo parsedCountry : parsedFile.getCountries()) { Map processedItems = processedByCountry.get(parsedCountry.getCountry()); if (processedItems == null) { - throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry()); + // 允许缺失时整国家降级：全部行标记"未回传"，否则文件缺口会让用户误以为已删除。 + if (!allowMissing) { + throw new BusinessException("缺少国家结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry()); + } + processedItems = Map.of(); } DeleteBrandProcessedCountryDto mergedCountry = new DeleteBrandProcessedCountryDto(); mergedCountry.setCountry(parsedCountry.getCountry()); for (DeleteBrandCountryAsinVo parsedAsin : parsedCountry.getItems()) { DeleteBrandCountryResultItemDto processed = processedItems.get(normalizeAsinKey(parsedAsin.getAsin())); - if (processed == null) { - throw new BusinessException("缺少 ASIN 结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry() + " / " + parsedAsin.getAsin()); - } DeleteBrandCountryResultItemDto row = new DeleteBrandCountryResultItemDto(); row.setAsin(parsedAsin.getAsin()); - row.setStatus(processed.getStatus()); + if (processed == null) { + if (!allowMissing) { + throw new BusinessException("缺少 ASIN 结果: " + parsedFile.getSourceFilename() + " / " + parsedCountry.getCountry() + " / " + parsedAsin.getAsin()); + } + row.setStatus(MISSING_CHUNK_STATUS); + } else { + row.setStatus(processed.getStatus()); + } mergedCountry.getItems().add(row); } mergedCountries.add(mergedCountry); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java index 3781cc73..e72a64e5 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -257,7 +257,7 @@ public class DeleteBrandStaleTaskService { .eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND) .eq(FileTaskEntity::getStatus, "RUNNING") .set(FileTaskEntity::getStatus, "FAILED") - .set(FileTaskEntity::getErrorMessage, "结果回传长时间无响应，任务已自动失败") + .set(FileTaskEntity::getErrorMessage, buildStaleFailReason(task, entry, minutes, nowMillis)) .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); @@ -272,6 +272,27 @@ public class DeleteBrandStaleTaskService { boolean hasStartedProgress, boolean finalizeThrew) { } + /** + * 组装超时失败文案：带最后心跳时间与最近一次组装失败的真实原因， + * 避免"分片缺失导致组装失败"被误报成"客户端长时间无响应"。 + */ + private String buildStaleFailReason(FileTaskEntity task, StaleFinalizeEntry entry, long minutes, long nowMillis) { + StringBuilder sb = new StringBuilder("结果回传中断超过 ").append(minutes).append(" 分钟"); + if (entry.lastHeartbeatAt > 0L) { + long elapsedMinutes = Math.max(1L, (nowMillis - entry.lastHeartbeatAt) / 60000L); + sb.append("（最后心跳 ").append(Instant.ofEpochMilli(entry.lastHeartbeatAt)) + .append("，距今 ").append(elapsedMinutes).append(" 分钟）"); + } else { + sb.append("（从未收到心跳或结果分片）"); + } + String assembleError = taskFileJobService.latestFailedAssembleError(task.getId(), MODULE_TYPE_DELETE_BRAND); + if (assembleError != null && !assembleError.isBlank()) { + sb.append("；最近一次组装失败：").append(assembleError); + } + sb.append("，任务已自动失败（可下载的部分结果见任务结果文件，或重新提交任务）"); + return sb.toString(); + } + private ProductRiskStaleCheckStats failStaleProductRiskResolveTasks() { ProductRiskStaleCheckStats stats = new ProductRiskStaleCheckStats(); long minutes = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutMinutes()); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ResultFileJobHandler.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ResultFileJobHandler.java index e5f42bd7..6bda251d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ResultFileJobHandler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ResultFileJobHandler.java @@ -27,6 +27,16 @@ public interface ResultFileJobHandler { default void onFailure(TaskFileJobEntity job, String message) { } + /** + * 失败兜底：job 重试耗尽（终态 FAILED）时，用已回传的部分数据尝试生成部分结果文件， + * 让用户仍能下载已做好的部分（缺失行由业务侧降级填充"未回传"等占位状态）。 + * 实现应自吞异常（兜底失败只记日志，不影响 job 已终态失败的事实）； + * 返回 true 表示已生成可下载的结果文件。 + */ + default boolean fallbackAssembleOnFailure(TaskFileJobEntity job, String message) { + return false; + } + default boolean supportsAsyncOffload() { return false; } 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 86d18129..1df2cdc0 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 @@ -6,6 +6,7 @@ import com.nanri.aiimage.modules.task.mapper.TaskFileJobMapper; import com.nanri.aiimage.modules.task.model.dto.TaskFileJobDispatchEvent; import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationEventPublisher; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; @@ -21,6 +22,7 @@ import java.util.stream.Collectors; @Service @RequiredArgsConstructor +@Slf4j public class TaskFileJobService { public static final String JOB_TYPE_ASSEMBLE_RESULT = "ASSEMBLE_RESULT"; @@ -88,6 +90,66 @@ public class TaskFileJobService { return refreshed; } + /** + * 该 result 是否已有重试耗尽的终态失败组装 job（分片缺失类失败后用于补传恢复判断）。 + */ + public boolean isTerminalFailedAssembleJob(Long taskId, String moduleType, Long resultId) { + TaskFileJobEntity existing = findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT); + if (existing == null || !"FAILED".equals(existing.getStatus())) { + return false; + } + int retry = existing.getRetryCount() == null ? 0 : existing.getRetryCount(); + return retry >= MAX_RETRY_COUNT; + } + + /** + * 补传完整后恢复终态失败的组装 job：重置为 PENDING（retryCount 清零）重新派发。 + * 幂等：job 不存在、非 FAILED 或未达重试上限时不动作，返回当前 job。 + */ + public TaskFileJobEntity resetTerminalFailedForRecovery(Long taskId, String moduleType, Long resultId) { + TaskFileJobEntity existing = findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT); + if (existing == null || !"FAILED".equals(existing.getStatus())) { + return existing; + } + int retry = existing.getRetryCount() == null ? 0 : existing.getRetryCount(); + if (retry < MAX_RETRY_COUNT) { + return existing; + } + LocalDateTime now = LocalDateTime.now(); + taskFileJobMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskFileJobEntity::getId, existing.getId()) + .set(TaskFileJobEntity::getStatus, "PENDING") + .set(TaskFileJobEntity::getRetryCount, 0) + .set(TaskFileJobEntity::getErrorMessage, null) + .set(TaskFileJobEntity::getFinishedAt, null) + .set(TaskFileJobEntity::getTerminalCallbackAt, null) + .set(TaskFileJobEntity::getUpdatedAt, now)); + log.info("[task-file-job] terminal failed assemble job reset for recovery jobId={} taskId={} moduleType={} resultId={}", + existing.getId(), taskId, moduleType, resultId); + TaskFileJobEntity refreshed = taskFileJobMapper.selectById(existing.getId()); + publishDispatchEvent(refreshed); + return refreshed; + } + + /** + * 最近一次失败组装 job 的错误信息（用于任务超时失败文案里拼接真实原因）。 + */ + public String latestFailedAssembleError(Long taskId, String moduleType) { + if (taskId == null || moduleType == null || moduleType.isBlank()) { + return null; + } + TaskFileJobEntity job = taskFileJobMapper.selectOne(new LambdaQueryWrapper() + .eq(TaskFileJobEntity::getTaskId, taskId) + .eq(TaskFileJobEntity::getModuleType, moduleType) + .eq(TaskFileJobEntity::getJobType, JOB_TYPE_ASSEMBLE_RESULT) + .eq(TaskFileJobEntity::getStatus, "FAILED") + .orderByDesc(TaskFileJobEntity::getId) + .last("limit 1")); + return job == null || job.getErrorMessage() == null || job.getErrorMessage().isBlank() + ? null + : job.getErrorMessage(); + } + public List listRunnableJobs(int limit) { return taskFileJobMapper.selectList(new LambdaQueryWrapper() .in(TaskFileJobEntity::getStatus, List.of("PENDING", "FAILED")) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java index d947dcf9..454afd23 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java @@ -276,6 +276,16 @@ public class TaskResultFileJobWorker { ResultFileJobHandler handler = handlerRegistry.asMap().get(job.getModuleType()); if (handler != null) { handler.onFailure(job, message); + try { + if (handler.fallbackAssembleOnFailure(job, message)) { + log.info("[task-file-job] retry-exhausted fallback assemble produced downloadable result jobId={} taskId={} moduleType={} resultId={} stage=FALLBACK", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId()); + } + } catch (Exception ex) { + // 兜底失败不影响 job 终态；下次 stuck/exhausted 扫描会重入（幂等），持续失败仅记日志。 + log.warn("[task-file-job] fallback assemble threw jobId={} taskId={} moduleType={} resultId={} msg={} stage=FALLBACK", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), ex.getMessage()); + } } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java b/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java index dcaa8b0b..1bef16fa 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/config/TaskResultFileJobWorkerLogAuditTest.java @@ -100,7 +100,7 @@ class TaskResultFileJobWorkerLogAuditTest { } // 阶段字面量来自允许集合（防新增阶段拼写漂移） List allowed = List.of("SKIP_OWNER", "REQUEUE", "WAIT_LLM", "DEFER", "SUCCESS", - "ORPHAN", "FAILED", "FINALIZE", "HEARTBEAT"); + "ORPHAN", "FAILED", "FINALIZE", "HEARTBEAT", "FALLBACK"); for (String l : perJob) { if (l.contains("stage=")) { int idx = l.indexOf("stage="); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunServiceFallbackTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunServiceFallbackTest.java new file mode 100644 index 00000000..eb1ddacb --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunServiceFallbackTest.java @@ -0,0 +1,188 @@ +package com.nanri.aiimage.modules.deletebrand.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.config.DeleteBrandProgressProperties; +import com.nanri.aiimage.config.TaskPressureProperties; +import com.nanri.aiimage.modules.deletebrand.model.cache.DeleteBrandParsedFileCacheDto; +import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandCountryResultItemDto; +import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandProcessedCountryDto; +import com.nanri.aiimage.modules.deletebrand.model.dto.DeleteBrandResultFileDto; +import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryAsinVo; +import com.nanri.aiimage.modules.deletebrand.model.vo.DeleteBrandCountryGroupVo; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler; +import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +/** + * 删除品牌分片缺失降级组装（修复：缺分片不整任务白跑）纯逻辑单测。 + */ +class DeleteBrandRunServiceFallbackTest { + + private final DeleteBrandRunService service = new DeleteBrandRunService( + mock(FileTaskMapper.class), + mock(FileResultMapper.class), + mock(LocalFileStorageService.class), + mock(DeleteBrandTaskCacheService.class), + mock(DeleteBrandTaskStorageService.class), + mock(ZiniaoShopSwitchService.class), + mock(OssStorageService.class), + mock(ObjectMapper.class), + mock(DeleteBrandProgressProperties.class), + mock(TaskPressureProperties.class), + mock(TaskDistributedLockService.class), + mock(TaskFileJobService.class), + mock(PlatformTransactionManager.class), + mock(TaskProgressLightAssembler.class)); + + private DeleteBrandResultFileDto chunk(int index, int total, String asin, String status) { + DeleteBrandResultFileDto dto = new DeleteBrandResultFileDto(); + dto.setFileKey("scope-key"); + dto.setSourceFilename("魏振峰.xlsx"); + dto.setChunkIndex(index); + dto.setChunkTotal(total); + dto.setProcessedRows(index); + dto.setTotalRows(total); + dto.setCurrentCountry("US"); + dto.setCurrentAsin(asin); + DeleteBrandProcessedCountryDto country = new DeleteBrandProcessedCountryDto(); + country.setCountry("US"); + DeleteBrandCountryResultItemDto item = new DeleteBrandCountryResultItemDto(); + item.setAsin(asin); + item.setStatus(status); + country.setItems(List.of(item)); + dto.setCountries(List.of(country)); + return dto; + } + + private DeleteBrandParsedFileCacheDto parsedFile(String... asins) { + DeleteBrandParsedFileCacheDto parsed = new DeleteBrandParsedFileCacheDto(); + parsed.setFileKey("scope-key"); + parsed.setSourceFilename("魏振峰.xlsx"); + parsed.setTotalRows(asins.length); + DeleteBrandCountryGroupVo country = new DeleteBrandCountryGroupVo(); + country.setCountry("US"); + for (String asin : asins) { + DeleteBrandCountryAsinVo asinVo = new DeleteBrandCountryAsinVo(); + asinVo.setAsin(asin); + country.getItems().add(asinVo); + } + parsed.setCountries(List.of(country)); + return parsed; + } + + @SuppressWarnings("unchecked") + private List missingChunkIndexes(List chunks) { + return (List) ReflectionTestUtils.invokeMethod(service, "missingChunkIndexes", chunks); + } + + @SuppressWarnings("unchecked") + private Object mergeChunks(DeleteBrandParsedFileCacheDto parsedFile, List chunks, boolean allowMissing) { + return ReflectionTestUtils.invokeMethod(service, "mergeChunks", parsedFile, chunks, allowMissing); + } + + @Test + void missingChunkIndexesFindsGap() { + assertEquals(List.of(3), missingChunkIndexes(List.of(chunk(1, 4, "A", "成功"), chunk(2, 4, "B", "成功"), chunk(4, 4, "D", "成功")))); + } + + @Test + void missingChunkIndexesMiddleGapOf27171Scenario() { + // 事故现场：4284 个分片、只缺 247 一个 + List chunks = new java.util.ArrayList<>(); + for (int i = 1; i <= 4284; i++) { + if (i == 247) { + continue; + } + chunks.add(chunk(i, 4284, "B0" + i, "成功")); + } + assertEquals(List.of(247), missingChunkIndexes(chunks)); + } + + @Test + void missingChunkIndexesEmptyAndUnknownTotal() { + assertEquals(List.of(), missingChunkIndexes(List.of())); + DeleteBrandResultFileDto noTotal = new DeleteBrandResultFileDto(); + noTotal.setChunkIndex(1); + assertEquals(List.of(), missingChunkIndexes(List.of(noTotal))); + } + + @Test + void mergeChunksWithAllowMissingFillsPlaceholderStatus() { + // parsed 有 2 个 ASIN，但只回传了 1 个（缺 B0DELETED） + DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED"); + List chunks = List.of(chunk(1, 2, "B0FINE", "成功")); + Object merged = mergeChunks(parsed, chunks, true); + assertEquals(2, mergedAsins(merged).size()); + assertEquals("成功", mergedAsins(merged).get("B0FINE")); + assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0DELETED")); + } + + @Test + void mergeChunksStrictStillThrowsOnMissingAsin() { + DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED"); + List chunks = List.of(chunk(1, 2, "B0FINE", "成功")); + BusinessException ex = assertThrows(BusinessException.class, + () -> mergeChunks(parsed, chunks, false)); + assertTrue(ex.getMessage().contains("缺少 ASIN 结果")); + assertTrue(ex.getMessage().contains("B0DELETED")); + } + + @Test + void mergeChunksAllowMissingWholeCountryBecomesPlaceholder() { + DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE", "B0DELETED"); + List chunks = List.of(); + Object merged = mergeChunks(parsed, chunks, true); + assertEquals(2, mergedAsins(merged).size()); + assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0FINE")); + assertEquals(DeleteBrandRunService.MISSING_CHUNK_STATUS, mergedAsins(merged).get("B0DELETED")); + } + + @Test + void mergeChunksStrictThrowsOnMissingCountry() { + DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE"); + BusinessException ex = assertThrows(BusinessException.class, + () -> mergeChunks(parsed, List.of(), false)); + assertTrue(ex.getMessage().contains("缺少国家结果")); + } + + @Test + void mergeChunksInconsistentChunkTotalStillThrows() { + DeleteBrandParsedFileCacheDto parsed = parsedFile("B0FINE"); + BusinessException ex = assertThrows(BusinessException.class, + () -> mergeChunks(parsed, List.of(chunk(1, 3, "B0FINE", "成功"), chunk(2, 4, "X", "成功")), true)); + assertTrue(ex.getMessage().contains("chunkTotal 不一致")); + } + + /** + * 把 mergeChunks 返回的 MergedDeleteBrandFile（record）转成 asin→status 映射便于断言。 + */ + private Map mergedAsins(Object merged) { + LinkedHashMap map = new LinkedHashMap<>(); + List countries = (List) ReflectionTestUtils.getField(merged, "countries"); + for (Object country : countries) { + List items = (List) ReflectionTestUtils.getField(country, "items"); + for (Object item : items) { + map.put((String) ReflectionTestUtils.getField(item, "asin"), (String) ReflectionTestUtils.getField(item, "status")); + } + } + return map; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java index 838c3b76..a5d2d8c0 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java @@ -8,6 +8,7 @@ import com.nanri.aiimage.config.DeleteBrandProgressProperties; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -26,7 +27,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -46,6 +49,7 @@ class DeleteBrandStaleTaskServiceTest { @Mock private DeleteBrandRunService deleteBrandRunService; @Mock private DeleteBrandProgressProperties deleteBrandProgressProperties; @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskFileJobService taskFileJobService; @BeforeAll static void initializeTableInfo() { @@ -178,17 +182,48 @@ class DeleteBrandStaleTaskServiceTest { verify(deleteBrandRunService, never()).tryFinalizeTask(anyLong(), anyBoolean()); } + @Test + void staleFailReasonIncludesHeartbeatAndAssembleError() { + FileTaskEntity t1 = runningTask(801L); + when(fileTaskMapper.selectList(any())).thenReturn(List.of(t1)); + when(fileTaskMapper.selectBatchIds(any())).thenReturn(List.of(runningTask(801L))); + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); + when(deleteBrandTaskCacheService.getProgress(801L)) + .thenReturn(Map.of("last_heartbeat_at", System.currentTimeMillis() - 35 * 60_000L)); + lockAvailable(); + // 必须放在 lockAvailable() 之后：后设置的 stub 覆盖前面的 lenient 默认值 + when(taskFileJobService.latestFailedAssembleError(801L, "DELETE_BRAND")) + .thenReturn("结果分片未完整: 240d70221b104a3c8140555efed4fd21，缺失分片: [247]（已收 4283/4284）"); + DeleteBrandStaleTaskService service = service(); + + failStaleDeleteBrandTasks(service); + + @SuppressWarnings({"rawtypes", "unchecked"}) + ArgumentCaptor update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class); + verify(fileTaskMapper).update(isNull(), update.capture()); + // 组装后的 SQL 只含占位符，真实文案在 paramNameValuePairs 中 + List values = update.getValue().getParamNameValuePairs().values().stream() + .map(String::valueOf) + .toList(); + String errorMessage = values.stream().filter(v -> v.contains("最后心跳")).findFirst().orElse(""); + // 文案必须包含真实原因（最后心跳 + 最近一次组装失败），不能只写"回传长时间无响应" + assertTrue(errorMessage.contains("最后心跳"), errorMessage); + assertTrue(errorMessage.contains("最近一次组装失败"), errorMessage); + assertTrue(errorMessage.contains("缺失分片"), errorMessage); + } + private DeleteBrandStaleTaskService service() { return new DeleteBrandStaleTaskService( fileTaskMapper, deleteBrandTaskCacheService, deleteBrandTaskStorageService, deleteBrandRunService, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, - deleteBrandProgressProperties, null, taskDistributedLockService, null); + deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService); } private void lockAvailable() { when(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes()).thenReturn(15L); when(taskDistributedLockService.acquire(any(), any(), anyLong())) .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(taskFileJobService.latestFailedAssembleError(any(), anyString())).thenReturn(null); } private static void failStaleDeleteBrandTasks(DeleteBrandStaleTaskService service) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/DeleteBrandResultFileJobHandlerTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/DeleteBrandResultFileJobHandlerTest.java index 523b87ac..8e47b76d 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/DeleteBrandResultFileJobHandlerTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/DeleteBrandResultFileJobHandlerTest.java @@ -10,6 +10,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * DeleteBrandResultFileJobHandler 测试（任务 70）。 @@ -64,6 +65,16 @@ class DeleteBrandResultFileJobHandlerTest { assertFalse(handler.isOwnerScoped()); } + @Test + void fallbackAssembleDelegates() { + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setTaskId(27171L); + job.setResultId(30195L); + when(deleteBrandRunService.fallbackAssembleResultFile(job)).thenReturn(true); + assertEquals(true, handler.fallbackAssembleOnFailure(job, "结果分片未完整")); + verify(deleteBrandRunService).fallbackAssembleResultFile(job); + } + @Test void nullJobGuarded() { assertTrue(handler.process(null));