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 47cdb5b8..d51c93fe 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 @@ -27,6 +27,7 @@ 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 com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService; +import com.nanri.aiimage.modules.task.service.TaskResumeService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -92,6 +93,8 @@ public class DeleteBrandStaleTaskService { private final ShopDataCrawlTaskService shopDataCrawlTaskService; private final TaskScopeStateMapper taskScopeStateMapper; private final TaskHeartbeatPositionService taskHeartbeatPositionService; + /** 客户端中断任务的自动续跑（保留失败记录 + 重排队续跑任务，V129）。 */ + private final TaskResumeService taskResumeService; @Value("${aiimage.temp-dir.retention-hours:24}") private long tempDirRetentionHours; @@ -114,13 +117,16 @@ public class DeleteBrandStaleTaskService { ShopMatchStaleCheckStats queryAsinStats = failStaleQueryAsinTasks(); ShopMatchStaleCheckStats withdrawStats = failStaleWithdrawTasks(); NoResultUploadStaleCheckStats noUploadStats = failNoResultUploadTasks(); + // 客户端重启中断的任务：除了标失败（客户端上报，用户能看到原因），还要重新排队 + // 一条 PENDING 续跑任务交给客户端兜底拉取执行，否则长任务一遇客户端更新就整个白跑 + TaskResumeService.ResumeStats resumeStats = resumeInterruptedSafely(); for (Map.Entry delegated : delegatedStaleChecks().entrySet()) { runModuleStaleCheck(delegated.getKey(), delegated.getValue()); } // 周期每 2 分钟一轮，各模块 summary 合并为单行，避免定期刷屏 // 注意：每段占位符数量必须与实参一致——此前每段 5 个占位符只传 4 个参数， // 导致 withdraw 之后的取值整体错位、末尾 elapsedMs/thread 打成字面量 - log.info("[stale-check] summary product-risk(s={} f={} x={} p={}) price-track(s={} f={} x={} p={}) shop-match(s={} f={} x={} p={}) patrol-delete(s={} f={} x={} p={}) query-asin(s={} f={} x={} p={}) withdraw(s={} f={} x={} p={}) no-upload(c={} f={} x={}) elapsedMs={} thread={}", + log.info("[stale-check] summary product-risk(s={} f={} x={} p={}) price-track(s={} f={} x={} p={}) shop-match(s={} f={} x={} p={}) patrol-delete(s={} f={} x={} p={}) query-asin(s={} f={} x={} p={}) withdraw(s={} f={} x={} p={}) no-upload(c={} f={} x={}) resume(s={} r={} k={}) elapsedMs={} thread={}", stats.scannedTaskCount, stats.finalizedTaskCount, stats.failedTaskCount, stats.skippedTaskCount, priceTrackStats.scannedTaskCount, priceTrackStats.finalizedTaskCount, priceTrackStats.failedTaskCount, priceTrackStats.skippedTaskCount, shopMatchStats.scannedTaskCount, shopMatchStats.finalizedTaskCount, shopMatchStats.failedTaskCount, shopMatchStats.skippedTaskCount, @@ -128,11 +134,26 @@ public class DeleteBrandStaleTaskService { queryAsinStats.scannedTaskCount, queryAsinStats.finalizedTaskCount, queryAsinStats.failedTaskCount, queryAsinStats.skippedTaskCount, withdrawStats.scannedTaskCount, withdrawStats.finalizedTaskCount, withdrawStats.failedTaskCount, withdrawStats.skippedTaskCount, noUploadStats.scannedTaskCount, noUploadStats.failedTaskCount, noUploadStats.skippedTaskCount, + resumeStats.scannedTaskCount, resumeStats.resumedTaskCount, resumeStats.skippedTaskCount, System.currentTimeMillis() - startedAt, Thread.currentThread().getName()); } } + /** + * 自动续跑巡检的隔离壳：续跑失败绝不能拖垮判死主流程。 + * 判死是任务状态的兜底（不做会留下永远 RUNNING 的孤儿），优先级高于续跑； + * 这里失败只记日志并返回空统计，2 分钟后的下一轮自然重试。 + */ + private TaskResumeService.ResumeStats resumeInterruptedSafely() { + try { + return taskResumeService.resumeInterruptedTasks(); + } catch (Exception ex) { + log.warn("[task-resume] 自动续跑巡检失败（不影响本轮判死）: {}", ex.getMessage(), ex); + return new TaskResumeService.ResumeStats(); + } + } + /** * 委派式陈旧判死：moduleType → 处理动作。 * {@code TaskModuleRegistry} 中 delegatedStaleCheck=true 的模块都必须在这里登记， diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/model/entity/PriceTrackLoopRunEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/model/entity/PriceTrackLoopRunEntity.java index f25352ec..32ab3811 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/model/entity/PriceTrackLoopRunEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/model/entity/PriceTrackLoopRunEntity.java @@ -29,6 +29,8 @@ public class PriceTrackLoopRunEntity { @TableField(value = "active_task_id", updateStrategy = FieldStrategy.ALWAYS) private Long activeTaskId; private Boolean stopRequested; + /** 因客户端中断自动重派当前轮的次数：封顶用，避免会话持续不可用时无限重派（V129）。 */ + private Integer resumeAttempt; private String errorMessage; private LocalDateTime createdAt; private LocalDateTime updatedAt; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunService.java index 947d5b0e..ac5149f6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunService.java @@ -39,6 +39,14 @@ public class PriceTrackLoopRunService { private static final String STATUS_STOPPED = "STOPPED"; private static final String EXECUTION_MODE_FINITE = "FINITE"; private static final String EXECUTION_MODE_INFINITE = "INFINITE"; + /** 客户端异常中断的任务 errorMessage 前缀（TaskHeartbeatService.markInterrupted 写入）。 */ + private static final String CLIENT_INTERRUPT_ERROR_PREFIX = "客户端异常中断"; + /** + * 中断后自动重派当前轮的次数上限。 + * 会话持续不可用（账号被风控、紫鸟未就绪）时，不封顶会无限重派、无限重开浏览器—— + * 2026-09-18 任务 28587 就是这样白烧了 5 小时。 + */ + private static final int MAX_AUTO_RESUME_ATTEMPT = 3; private final PriceTrackLoopRunMapper loopRunMapper; private final FileTaskMapper fileTaskMapper; @@ -259,6 +267,28 @@ public class PriceTrackLoopRunService { entity.setActiveTaskId(null); entity.setUpdatedAt(LocalDateTime.now()); if (STATUS_FAILED.equals(task.getStatus())) { + // 客户端重启中断（markInterrupted 写入的前缀）不终止循环：清空 active_task_id 后保持 + // RUNNING，客户端下次 dispatchNext 会拿到**同一店铺、同一轮次**的 childTaskRequest， + // 等于原地续跑——页内已处理的 ASIN 由服务端 skip_asins 去重，不会重复改价。 + if (isClientInterrupt(task) && currentResumeAttempt(entity) < MAX_AUTO_RESUME_ATTEMPT) { + int attempt = currentResumeAttempt(entity) + 1; + entity.setStatus(STATUS_RUNNING); + entity.setErrorMessage(null); + entity.setFinishedAt(null); + entity.setResumeAttempt(attempt); + loopRunMapper.updateById(entity); + log.warn("[price-track-loop] 子任务因客户端中断失败，自动重派当前轮 loopRunId={} childTaskId={} " + + "round={} shopIndex={} resumeAttempt={}/{} error={}", + entity.getId(), childTaskId, entity.getCurrentRound(), entity.getCurrentShopIndex(), + attempt, MAX_AUTO_RESUME_ATTEMPT, task.getErrorMessage()); + return; + } + if (isClientInterrupt(task)) { + log.warn("[price-track-loop] 客户端中断续跑已达上限 {} 次，终止循环 loopRunId={} childTaskId={} " + + "round={} shopIndex={}", + MAX_AUTO_RESUME_ATTEMPT, entity.getId(), childTaskId, + entity.getCurrentRound(), entity.getCurrentShopIndex()); + } entity.setStatus(STATUS_FAILED); entity.setErrorMessage(task.getErrorMessage() == null || task.getErrorMessage().isBlank() ? "子任务执行失败" @@ -269,6 +299,8 @@ public class PriceTrackLoopRunService { entity.getId(), childTaskId, entity.getCurrentRound(), entity.getCurrentShopIndex(), entity.getErrorMessage()); return; } + // 子任务成功 → 中断续跑计数归零（否则历史上的中断会一直占用封顶额度） + entity.setResumeAttempt(0); List items = parseShops(entity); if (items.isEmpty()) { entity.setStatus(STATUS_FAILED); @@ -300,6 +332,16 @@ public class PriceTrackLoopRunService { entity.getId(), childTaskId, entity.getStatus(), entity.getCurrentRound(), entity.getCurrentShopIndex(), entity.getActiveTaskId()); } + /** 子任务失败原因是否为「客户端异常中断」（客户端重启上报，可自动重派续跑）。 */ + private static boolean isClientInterrupt(FileTaskEntity task) { + String message = task == null ? null : task.getErrorMessage(); + return message != null && message.startsWith(CLIENT_INTERRUPT_ERROR_PREFIX); + } + + private static int currentResumeAttempt(PriceTrackLoopRunEntity entity) { + return entity.getResumeAttempt() == null ? 0 : entity.getResumeAttempt(); + } + private void reconcileWithTerminalChild(PriceTrackLoopRunEntity entity) { if (entity == null || entity.getActiveTaskId() == null || isTerminal(entity.getStatus())) { return; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java index e33e6372..c9489432 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java @@ -26,6 +26,10 @@ public class FileTaskEntity { private String createdBy; private Long userId; private String ownerInstanceId; + /** 续跑来源任务 ID：本行是客户端中断后由服务端自动重排队的续跑任务时非空（V129）。 */ + private Long resumeOfTaskId; + /** 续跑代数：0=原始任务，N=第 N 次自动续跑（封顶用，V129）。 */ + private Integer resumeAttempt; private LocalDateTime createdAt; private LocalDateTime updatedAt; private LocalDateTime finishedAt; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResumeService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResumeService.java new file mode 100644 index 00000000..a881bc65 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResumeService.java @@ -0,0 +1,164 @@ +package com.nanri.aiimage.modules.task.service; + +import cn.hutool.core.util.IdUtil; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +/** + * 客户端中断任务的自动续跑：**保留失败记录 + 重新排队一条 PENDING 续跑任务**。 + * + *

场景：客户端被更新脚本/任务管理器杀掉时，{@code TaskHeartbeatService.markInterrupted} 会把在跑的任务 + * 标 FAILED（原因「客户端异常中断: …」），用户能看到发生过什么；但长任务（采集/相似ASIN/外观专利/单次跟价） + * 一遇客户端重启就整个白跑，需要有人把它重新放回队列。本服务负责这件事。 + * + *

续跑方式刻意复用已有链路：新建的任务落 PENDING，由 {@link TaskClientPullService} 的兜底拉取 + * （客户端每分钟 {@code GET /api/tasks/pull-pending}）领走执行——不再新造第二套派发机制。 + * 只有注册了 {@link ClientTaskPullSpi} 的模块才会被续跑（这些模块的载荷能由服务端自行组装）； + * 上架/改价等写操作模块不在其中，避免盲目重跑。 + * + *

封顶 {@code max-attempt}：会话持续不可用（账号被风控、紫鸟未就绪）时，不封顶会无限重排队。 + * 计数写在 {@code biz_file_task.resume_attempt}（V129），随续跑任务代代递增。 + * + *

幂等：同一原任务已存在续跑任务（{@code resume_of_task_id} 反查）则跳过，重复扫描安全。 + */ +@Slf4j +@Service +public class TaskResumeService { + + private static final String STATUS_FAILED = "FAILED"; + private static final String STATUS_PENDING = "PENDING"; + /** 客户端重启中断的失败原因前缀，由 TaskHeartbeatService.markInterrupted 写入。 */ + private static final String CLIENT_INTERRUPT_PREFIX = "客户端异常中断"; + + private final FileTaskMapper fileTaskMapper; + /** moduleType → 模块兜底载荷实现；只有注册了 SPI 的模块才能被自动续跑。 */ + private final Map resumeHandlers; + + @Value("${aiimage.task-resume.enabled:true}") + private boolean enabled; + + /** 续跑代数上限：达到后不再重排队，任务保持 FAILED 等人工介入。 */ + @Value("${aiimage.task-resume.max-attempt:3}") + private int maxAttempt; + + /** 只处理最近这段时间内中断的任务，避免开机扫描历史积压。 */ + @Value("${aiimage.task-resume.window-minutes:30}") + private long windowMinutes; + + @Value("${aiimage.task-resume.limit:20}") + private int limit; + + public TaskResumeService(FileTaskMapper fileTaskMapper, List pullSpiHandlers) { + this.fileTaskMapper = fileTaskMapper; + Map index = new LinkedHashMap<>(); + if (pullSpiHandlers != null) { + for (ClientTaskPullSpi handler : pullSpiHandlers) { + String moduleType = handler.moduleType(); + if (moduleType == null || moduleType.isBlank()) { + continue; + } + index.put(moduleType.trim().toUpperCase(Locale.ROOT), handler); + } + } + this.resumeHandlers = Map.copyOf(index); + log.info("[task-resume] 可自动续跑模块注册完成 count={} modules={}", index.size(), index.keySet()); + } + + /** 扫描一轮：把最近因客户端中断而失败、且未续跑过的任务重新排队。 */ + public ResumeStats resumeInterruptedTasks() { + ResumeStats stats = new ResumeStats(); + if (!enabled) { + return stats; + } + if (resumeHandlers.isEmpty()) { + log.warn("[task-resume] 没有注册任何 ClientTaskPullSpi，跳过本轮续跑"); + return stats; + } + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(Math.max(1L, windowMinutes)); + int safeLimit = Math.max(1, limit); + List candidates = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getStatus, STATUS_FAILED) + .likeRight(FileTaskEntity::getErrorMessage, CLIENT_INTERRUPT_PREFIX) + .in(FileTaskEntity::getModuleType, resumeHandlers.keySet()) + .lt(FileTaskEntity::getResumeAttempt, maxAttempt) + .ge(FileTaskEntity::getFinishedAt, cutoff) + .orderByAsc(FileTaskEntity::getId) + .last("limit " + safeLimit)); + if (candidates.isEmpty()) { + return stats; + } + stats.scannedTaskCount = candidates.size(); + for (FileTaskEntity original : candidates) { + if (original.getUserId() == null || original.getUserId() <= 0) { + log.warn("[task-resume] 原任务没有归属用户，跳过 taskId={}", original.getId()); + stats.skippedTaskCount++; + continue; + } + // 幂等：同一原任务已经排过一次续跑就跳过（重复扫描 / 双实例竞态下不会重复建单） + if (hasResumeChild(original.getId())) { + stats.skippedTaskCount++; + continue; + } + FileTaskEntity resume = buildResumeTask(original); + try { + fileTaskMapper.insert(resume); + stats.resumedTaskCount++; + log.warn("[task-resume] 已自动重新排队 taskId={} moduleType={} userId={} 续跑任务={} 代数={}/{} 中断原因={}", + original.getId(), original.getModuleType(), original.getUserId(), + resume.getId(), resume.getResumeAttempt(), maxAttempt, original.getErrorMessage()); + } catch (Exception ex) { + stats.skippedTaskCount++; + log.error("[task-resume] 重新排队失败 taskId={} moduleType={} err={}", + original.getId(), original.getModuleType(), ex.getMessage(), ex); + } + } + return stats; + } + + /** 该原任务是否已经有续跑任务（反查 resume_of_task_id）。 */ + private boolean hasResumeChild(Long originalTaskId) { + Long count = fileTaskMapper.selectCount(new LambdaQueryWrapper() + .eq(FileTaskEntity::getResumeOfTaskId, originalTaskId)); + return count != null && count > 0; + } + + private FileTaskEntity buildResumeTask(FileTaskEntity original) { + LocalDateTime now = LocalDateTime.now(); + FileTaskEntity resume = new FileTaskEntity(); + resume.setTaskNo(original.getModuleType() + "-" + IdUtil.getSnowflakeNextIdStr()); + resume.setModuleType(original.getModuleType()); + resume.setTaskMode(original.getTaskMode()); + resume.setStatus(STATUS_PENDING); + resume.setSourceFileCount(original.getSourceFileCount()); + resume.setSuccessFileCount(0); + resume.setFailedFileCount(0); + // 原任务的请求参数整体带走：各模块的 ClientTaskPullSpi 从 request_json / 关联表还原执行参数 + resume.setRequestJson(original.getRequestJson()); + resume.setCreatedBy(original.getCreatedBy()); + resume.setUserId(original.getUserId()); + resume.setResumeOfTaskId(original.getId()); + int attempt = original.getResumeAttempt() == null ? 0 : original.getResumeAttempt(); + resume.setResumeAttempt(attempt + 1); + resume.setCreatedAt(now); + resume.setUpdatedAt(now); + return resume; + } + + /** 单轮统计（合并进 stale-check 的 summary 日志，避免定期刷屏）。 */ + public static final class ResumeStats { + public int scannedTaskCount; + public int resumedTaskCount; + public int skippedTaskCount; + } +} diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 07776513..e6e0315b 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -279,6 +279,17 @@ aiimage: module-types: ${AIIMAGE_CLIENT_TASK_PULL_MODULE_TYPES:SIMILAR_ASIN,COLLECT_DATA,APPEARANCE_PATENT} min-pending-minutes: ${AIIMAGE_CLIENT_TASK_PULL_MIN_PENDING_MINUTES:5} limit: ${AIIMAGE_CLIENT_TASK_PULL_LIMIT:5} + # 客户端中断任务的自动续跑（V129）：客户端被更新脚本/任务管理器杀掉时，在跑的任务会被 + # 客户端启动时上报中断并标 FAILED（用户能看到真实原因），这里把这些任务重新排队成 PENDING， + # 由上面的兜底拉取通道交给在线客户端继续跑——长任务不再因为一次客户端更新就整个白跑。 + # 只有注册了 ClientTaskPullSpi 的模块会被续跑（相似ASIN/采集/外观专利）， + # 上架/改价等写操作模块刻意不在内，避免盲目重跑。 + # enabled 默认跟随兜底拉取开关：拉了没人领的话，续跑任务只会积压并被 stale 判死。 + task-resume: + enabled: ${AIIMAGE_TASK_RESUME_ENABLED:${AIIMAGE_CLIENT_TASK_PULL_ENABLED:false}} + max-attempt: ${AIIMAGE_TASK_RESUME_MAX_ATTEMPT:3} + window-minutes: ${AIIMAGE_TASK_RESUME_WINDOW_MINUTES:30} + limit: ${AIIMAGE_TASK_RESUME_LIMIT:20} coze-task: max-concurrent: ${AIIMAGE_COZE_TASK_MAX_CONCURRENT:12} brand-check: diff --git a/backend-java/src/main/resources/db/V129__task_auto_resume.sql b/backend-java/src/main/resources/db/V129__task_auto_resume.sql new file mode 100644 index 00000000..e1670bb0 --- /dev/null +++ b/backend-java/src/main/resources/db/V129__task_auto_resume.sql @@ -0,0 +1,55 @@ +-- V129: 客户端中断后的自动续跑（保留失败记录 + 服务端重新排队） +-- +-- 背景：2026-09-18 任务 28587（跟价，uid 977 店铺「张美莺」）在客户端 09:51/09:57 被重启后中断， +-- 旧行为是任务一直挂 RUNNING 到 stale 兜底判死（最长 2 小时），长任务一旦撞上客户端更新就白跑。 +-- 新契约：中断任务**保留失败记录**（用户能看到「因客户端重启中断」），同时由服务端自动重新排队续跑。 +-- +-- 两条续跑路径各需要一个计数： +-- 1) biz_file_task.resume_of_task_id + resume_attempt —— 通用模块（相似ASIN/采集/外观专利/跟价单次任务） +-- 由 TaskResumeService 复制出一条 PENDING 续跑任务，交给客户端兜底拉取；代数用于封顶。 +-- 2) biz_price_track_loop_run.resume_attempt —— 跟价循环由 loop_run 驱动，中断时不让循环终止， +-- 而是清掉 active_task_id 重新派发当前轮（客户端下次 dispatch 即拿到同一店铺/轮次），同样封顶。 +-- +-- 封顶的意义：会话持续不可用（如账号被风控）时，不封顶会无限重排队、无限重开浏览器。 +-- +-- 风险：ADD COLUMN 走 INSTANT，两张表均为小表，秒级完成；建议低峰执行。 +-- 回滚：ALTER TABLE biz_file_task DROP COLUMN resume_of_task_id, DROP COLUMN resume_attempt; +-- ALTER TABLE biz_price_track_loop_run DROP COLUMN resume_attempt; + +SET @db_name = DATABASE(); + +SET @col_exists := ( + SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = @db_name AND TABLE_NAME = 'biz_file_task' AND COLUMN_NAME = 'resume_of_task_id' +); +SET @sql := IF(@col_exists = 0, + 'ALTER TABLE biz_file_task ADD COLUMN resume_of_task_id BIGINT NULL COMMENT ''续跑来源任务ID（客户端中断后自动重排队）'' AFTER owner_instance_id', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @col_exists := ( + SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = @db_name AND TABLE_NAME = 'biz_file_task' AND COLUMN_NAME = 'resume_attempt' +); +SET @sql := IF(@col_exists = 0, + 'ALTER TABLE biz_file_task ADD COLUMN resume_attempt INT NOT NULL DEFAULT 0 COMMENT ''续跑代数：0=原始任务，N=第N次自动续跑'' AFTER resume_of_task_id', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @col_exists := ( + SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = @db_name AND TABLE_NAME = 'biz_price_track_loop_run' AND COLUMN_NAME = 'resume_attempt' +); +SET @sql := IF(@col_exists = 0, + 'ALTER TABLE biz_price_track_loop_run ADD COLUMN resume_attempt INT NOT NULL DEFAULT 0 COMMENT ''中断后自动重派当前轮的次数（用于封顶）'' AFTER stop_requested', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; 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 1e9f66d4..fa456fcd 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 @@ -14,6 +14,7 @@ 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 com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService; +import com.nanri.aiimage.modules.task.service.TaskResumeService; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -59,6 +60,7 @@ class DeleteBrandStaleTaskServiceTest { @Mock private ProductRiskTaskService productRiskTaskService; @Mock private ProductRiskTaskCacheService productRiskTaskCacheService; @Mock private TaskHeartbeatPositionService taskHeartbeatPositionService; + @Mock private TaskResumeService taskResumeService; @BeforeAll static void initializeTableInfo() { @@ -374,6 +376,9 @@ class DeleteBrandStaleTaskServiceTest { private void noUploadEnabled() { when(deleteBrandProgressProperties.isNoResultUploadCheckEnabled()).thenReturn(true); when(deleteBrandProgressProperties.getNoResultUploadTimeoutMinutes()).thenReturn(180L); + // 续跑巡检在判死主流程里被调用：默认返回空统计（无中断任务可续跑）； + // 只有跑到 failStaleRunningTasks 的用例会用到，故 lenient + lenient().when(taskResumeService.resumeInterruptedTasks()).thenReturn(new TaskResumeService.ResumeStats()); } private void noUploadTaskLockAvailable() { @@ -398,7 +403,7 @@ class DeleteBrandStaleTaskServiceTest { null, null, null, null, null, null, null, null, null, null, null, null, null, deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService, null, - taskScopeStateMapper, taskHeartbeatPositionService); + taskScopeStateMapper, taskHeartbeatPositionService, taskResumeService); } private void lockAvailable() { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java index ce4068dc..05e76fe7 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java @@ -82,7 +82,7 @@ class TaskModuleCoverageTest { mock(SimilarAsinTaskService.class), null, null, null, null, mock(ShopDataCrawlTaskService.class), - null, null); + null, null, null); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunServiceTest.java index 7ef9d324..33159439 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunServiceTest.java @@ -14,6 +14,7 @@ import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -120,6 +121,162 @@ class PriceTrackLoopRunServiceTest { verify(loopRunMapper, never()).updateById(loop); } + @Test + void 子任务因客户端中断失败时循环保持运行并重派当前轮() throws Exception { + ObjectMapper objectMapper = new ObjectMapper(); + PriceTrackLoopRunMapper loopRunMapper = mock(PriceTrackLoopRunMapper.class); + FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + PriceTrackLoopRunService service = new PriceTrackLoopRunService( + loopRunMapper, fileTaskMapper, objectMapper, mock(ZiniaoShopSwitchService.class)); + + PriceTrackLoopRunEntity loop = new PriceTrackLoopRunEntity(); + loop.setId(2699L); + loop.setUserId(977L); + loop.setStatus("RUNNING"); + loop.setExecutionMode("INFINITE"); + loop.setCurrentRound(1); + loop.setCurrentShopIndex(0); + loop.setActiveTaskId(28587L); + loop.setResumeAttempt(0); + loop.setStopRequested(false); + loop.setShopsJson(objectMapper.writeValueAsString(List.of(shop("张美莺")))); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(28587L); + task.setUserId(977L); + task.setModuleType("PRICE_TRACK"); + task.setStatus("FAILED"); + task.setErrorMessage("客户端异常中断: 客户端重启恢复上报（上次任务类型: price-track-run）"); + task.setRequestJson(objectMapper.writeValueAsString(Map.of( + "loopRunId", 2699, "roundIndex", 1, "shopIndex", 0))); + + when(loopRunMapper.selectList(any())).thenReturn(List.of(loop)); + when(fileTaskMapper.selectById(28587L)).thenReturn(task); + + service.syncLoopRunAfterChildTerminal(28587L); + + assertEquals("RUNNING", loop.getStatus(), "中断不该终止循环"); + assertNull(loop.getActiveTaskId(), "必须清空 active_task_id 才能重派当前轮"); + assertNull(loop.getFinishedAt()); + assertNull(loop.getErrorMessage()); + assertEquals(1, loop.getCurrentRound(), "轮次不变：原地续跑"); + assertEquals(0, loop.getCurrentShopIndex(), "店铺不变：原地续跑"); + assertEquals(1, loop.getResumeAttempt(), "中断计数递增"); + } + + @Test + void 中断重派达上限后循环判失败() throws Exception { + ObjectMapper objectMapper = new ObjectMapper(); + PriceTrackLoopRunMapper loopRunMapper = mock(PriceTrackLoopRunMapper.class); + FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + PriceTrackLoopRunService service = new PriceTrackLoopRunService( + loopRunMapper, fileTaskMapper, objectMapper, mock(ZiniaoShopSwitchService.class)); + + PriceTrackLoopRunEntity loop = new PriceTrackLoopRunEntity(); + loop.setId(2699L); + loop.setUserId(977L); + loop.setStatus("RUNNING"); + loop.setExecutionMode("INFINITE"); + loop.setCurrentRound(1); + loop.setCurrentShopIndex(0); + loop.setActiveTaskId(28587L); + loop.setResumeAttempt(3); + loop.setStopRequested(false); + loop.setShopsJson(objectMapper.writeValueAsString(List.of(shop("张美莺")))); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(28587L); + task.setUserId(977L); + task.setModuleType("PRICE_TRACK"); + task.setStatus("FAILED"); + task.setErrorMessage("客户端异常中断: 客户端重启恢复上报"); + task.setRequestJson(objectMapper.writeValueAsString(Map.of( + "loopRunId", 2699, "roundIndex", 1, "shopIndex", 0))); + + when(loopRunMapper.selectList(any())).thenReturn(List.of(loop)); + when(fileTaskMapper.selectById(28587L)).thenReturn(task); + + service.syncLoopRunAfterChildTerminal(28587L); + + assertEquals("FAILED", loop.getStatus(), "达上限后不再重派，交人工介入"); + assertNotNull(loop.getErrorMessage()); + } + + @Test + void 非中断原因的子任务失败仍终止循环() throws Exception { + ObjectMapper objectMapper = new ObjectMapper(); + PriceTrackLoopRunMapper loopRunMapper = mock(PriceTrackLoopRunMapper.class); + FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + PriceTrackLoopRunService service = new PriceTrackLoopRunService( + loopRunMapper, fileTaskMapper, objectMapper, mock(ZiniaoShopSwitchService.class)); + + PriceTrackLoopRunEntity loop = new PriceTrackLoopRunEntity(); + loop.setId(2699L); + loop.setUserId(977L); + loop.setStatus("RUNNING"); + loop.setExecutionMode("INFINITE"); + loop.setCurrentRound(1); + loop.setCurrentShopIndex(0); + loop.setActiveTaskId(28587L); + loop.setResumeAttempt(0); + loop.setStopRequested(false); + loop.setShopsJson(objectMapper.writeValueAsString(List.of(shop("张美莺")))); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(28587L); + task.setUserId(977L); + task.setModuleType("PRICE_TRACK"); + task.setStatus("FAILED"); + task.setErrorMessage("张美莺: 国家 德国 未跑完：重试后仍未获取到页码信息"); + task.setRequestJson(objectMapper.writeValueAsString(Map.of( + "loopRunId", 2699, "roundIndex", 1, "shopIndex", 0))); + + when(loopRunMapper.selectList(any())).thenReturn(List.of(loop)); + when(fileTaskMapper.selectById(28587L)).thenReturn(task); + + service.syncLoopRunAfterChildTerminal(28587L); + + assertEquals("FAILED", loop.getStatus(), "业务失败照旧终止循环，只有中断才续跑"); + assertEquals(0, loop.getResumeAttempt()); + } + + @Test + void 子任务成功后中断计数归零() throws Exception { + ObjectMapper objectMapper = new ObjectMapper(); + PriceTrackLoopRunMapper loopRunMapper = mock(PriceTrackLoopRunMapper.class); + FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + PriceTrackLoopRunService service = new PriceTrackLoopRunService( + loopRunMapper, fileTaskMapper, objectMapper, mock(ZiniaoShopSwitchService.class)); + + PriceTrackLoopRunEntity loop = new PriceTrackLoopRunEntity(); + loop.setId(2699L); + loop.setUserId(977L); + loop.setStatus("RUNNING"); + loop.setExecutionMode("FINITE"); + loop.setTargetRounds(1); + loop.setCurrentRound(1); + loop.setCurrentShopIndex(0); + loop.setActiveTaskId(28587L); + loop.setResumeAttempt(2); + loop.setStopRequested(false); + loop.setShopsJson(objectMapper.writeValueAsString(List.of(shop("张美莺")))); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(28587L); + task.setUserId(977L); + task.setModuleType("PRICE_TRACK"); + task.setStatus("SUCCESS"); + task.setRequestJson(objectMapper.writeValueAsString(Map.of( + "loopRunId", 2699, "roundIndex", 1, "shopIndex", 0))); + + when(loopRunMapper.selectList(any())).thenReturn(List.of(loop)); + when(fileTaskMapper.selectById(28587L)).thenReturn(task); + + service.syncLoopRunAfterChildTerminal(28587L); + + assertEquals(0, loop.getResumeAttempt(), "成功一轮后中断计数归零，避免历史中断占用额度"); + } + private PriceTrackMatchShopsVo.PriceTrackShopQueueItem shop(String name) { PriceTrackMatchShopsVo.PriceTrackShopQueueItem item = new PriceTrackMatchShopsVo.PriceTrackShopQueueItem(); item.setShopName(name); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResumeServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResumeServiceTest.java new file mode 100644 index 00000000..458ef86f --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskResumeServiceTest.java @@ -0,0 +1,155 @@ +package com.nanri.aiimage.modules.task.service; + +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 客户端中断任务的自动续跑：保留失败记录 + 重新排队 PENDING 续跑任务。 + * + *

背景：2026-09-18 任务 28587 因客户端被更新脚本重启而中断后一直挂 RUNNING； + * 中断上报把任务标 FAILED 之后，还需要有人把长任务放回队列，否则一遇客户端更新就整个白跑。 + */ +class TaskResumeServiceTest { + + private static ClientTaskPullSpi spi(String moduleType) { + return new ClientTaskPullSpi() { + @Override + public String moduleType() { + return moduleType; + } + + @Override + public Map buildQueuePayload(FileTaskEntity task) { + return Map.of("type", "x-run", "data", Map.of("taskId", task.getId())); + } + }; + } + + private static TaskResumeService service(FileTaskMapper mapper, List handlers, + boolean enabled, int maxAttempt) { + TaskResumeService service = new TaskResumeService(mapper, handlers); + ReflectionTestUtils.setField(service, "enabled", enabled); + ReflectionTestUtils.setField(service, "maxAttempt", maxAttempt); + ReflectionTestUtils.setField(service, "windowMinutes", 30L); + ReflectionTestUtils.setField(service, "limit", 20); + return service; + } + + private static FileTaskEntity interruptedTask(Long id, String moduleType) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType(moduleType); + task.setUserId(977L); + task.setTaskMode("PYTHON_QUEUE"); + task.setStatus("FAILED"); + task.setErrorMessage("客户端异常中断: 客户端重启恢复上报（上次任务类型: similar-asin-run）"); + task.setRequestJson("{\"items\":[]}"); + task.setResumeAttempt(0); + task.setFinishedAt(LocalDateTime.now()); + return task; + } + + @Test + void 中断任务会被重新排队为待执行的续跑任务() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(interruptedTask(28587L, "SIMILAR_ASIN"))); + when(mapper.selectCount(any())).thenReturn(0L); + TaskResumeService service = service(mapper, List.of(spi("SIMILAR_ASIN")), true, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(1, stats.scannedTaskCount); + assertEquals(1, stats.resumedTaskCount); + ArgumentCaptor captor = ArgumentCaptor.forClass(FileTaskEntity.class); + verify(mapper).insert(captor.capture()); + FileTaskEntity resumed = captor.getValue(); + assertEquals("PENDING", resumed.getStatus(), "续跑任务必须是待领取状态，交给客户端兜底拉取"); + assertEquals(28587L, resumed.getResumeOfTaskId(), "必须回指原任务，用于追溯与幂等"); + assertEquals(1, resumed.getResumeAttempt()); + assertEquals(977L, resumed.getUserId()); + assertEquals("SIMILAR_ASIN", resumed.getModuleType()); + assertEquals("{\"items\":[]}", resumed.getRequestJson(), "原任务参数必须整体带走"); + assertTrue(resumed.getTaskNo().startsWith("SIMILAR_ASIN-"), "任务号沿用模块前缀"); + } + + @Test + void 已有续跑任务时跳过避免重复排队() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(interruptedTask(28587L, "SIMILAR_ASIN"))); + when(mapper.selectCount(any())).thenReturn(1L, 1L); + TaskResumeService service = service(mapper, List.of(spi("SIMILAR_ASIN")), true, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(1, stats.skippedTaskCount); + assertEquals(0, stats.resumedTaskCount); + verify(mapper, never()).insert(any(FileTaskEntity.class)); + } + + @Test + void 开关关闭时不扫描() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + TaskResumeService service = service(mapper, List.of(spi("SIMILAR_ASIN")), false, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(0, stats.scannedTaskCount); + verify(mapper, never()).selectList(any()); + } + + @Test + void 没有注册模块兜底实现时不扫描() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + TaskResumeService service = service(mapper, List.of(), true, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(0, stats.scannedTaskCount); + verify(mapper, never()).selectList(any()); + } + + @Test + void 插入失败只计入跳过不抛出() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(interruptedTask(28587L, "SIMILAR_ASIN"))); + when(mapper.selectCount(any())).thenReturn(0L); + doThrow(new RuntimeException("库挂了")).when(mapper).insert(any(FileTaskEntity.class)); + TaskResumeService service = service(mapper, List.of(spi("SIMILAR_ASIN")), true, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(1, stats.skippedTaskCount); + assertEquals(0, stats.resumedTaskCount); + } + + @Test + void 没有归属用户的中断任务跳过() { + FileTaskMapper mapper = mock(FileTaskMapper.class); + FileTaskEntity orphan = interruptedTask(28587L, "SIMILAR_ASIN"); + orphan.setUserId(null); + when(mapper.selectList(any())).thenReturn(List.of(orphan)); + TaskResumeService service = service(mapper, List.of(spi("SIMILAR_ASIN")), true, 3); + + TaskResumeService.ResumeStats stats = service.resumeInterruptedTasks(); + + assertEquals(1, stats.skippedTaskCount); + verify(mapper, never()).insert(any(FileTaskEntity.class)); + } +}