feat(任务恢复): 客户端中断的任务自动续跑(保留失败记录 + 重新排队)
2026-09-18 任务 28587(跟价,uid 977 店铺「张美莺」)在客户端 09:51/09:57 被更新脚本 重启后中断,任务一直挂 RUNNING 到 stale 兜底;跟价循环还有第二重问题——子任务失败 即整个循环判 FAILED,一次客户端更新就让无限循环彻底停下。 契约:中断任务**保留失败记录**(用户看得到「因客户端重启中断」),同时由服务端自动 重新排队续跑,长任务不再因为一次更新整个白跑。 - V129:biz_file_task 加 resume_of_task_id / resume_attempt, biz_price_track_loop_run 加 resume_attempt(代数封顶用); - 新增 TaskResumeService:扫描最近 30 分钟内因客户端中断而失败、未续跑过、代数未超限的 任务,复制请求参数重新排队成 PENDING,交给已有兜底拉取通道(客户端每分钟 pull-pending) 领走执行——不新造第二套派发机制。只对注册了 ClientTaskPullSpi 的模块生效 (相似ASIN/采集/外观专利),上架/改价等写操作模块刻意排除,避免盲目重跑; - 幂等:按 resume_of_task_id 反查,同一原任务不会重复排队; - 挂在 stale-check 巡检线(每 2 分钟、有分布式锁),并用隔离壳包住异常: 续跑失败绝不拖垮判死主流程(判死优先级更高); - PriceTrackLoopRunService:子任务因「客户端异常中断」失败时不终止循环, 清 active_task_id 后保持 RUNNING,客户端下次 dispatchNext 拿到同一店铺/同一轮, 即原地续跑(页内已处理 ASIN 由服务端 skip_asins 去重,不会重复改价); 连续重派超过 3 次才真正判失败,避免会话持续不可用时无限重开浏览器 (28587 就是这样白烧了 5 小时)。子任务成功一轮后计数归零; - application.yml:aiimage.task-resume.enabled 默认跟随 client-task-pull 开关 (兜底拉取关着时续跑任务无人领取,只会积压被告死)。 测试:TaskResumeServiceTest 6 例(排队/幂等/开关/无实现/插入失败/无归属用户)、 PriceTrackLoopRunServiceTest 新增 4 例(中断续跑/达上限/非中断仍终止/成功归零), 连带更新 DeleteBrandStaleTaskServiceTest 与 TaskModuleCoverageTest 的构造参数。
This commit is contained in:
+22
-1
@@ -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<String, Runnable> 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 的模块都必须在这里登记,
|
||||
|
||||
+2
@@ -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;
|
||||
|
||||
+42
@@ -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<PriceTrackMatchShopsVo.PriceTrackShopQueueItem> 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;
|
||||
|
||||
+4
@@ -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;
|
||||
|
||||
+164
@@ -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 续跑任务**。
|
||||
*
|
||||
* <p>场景:客户端被更新脚本/任务管理器杀掉时,{@code TaskHeartbeatService.markInterrupted} 会把在跑的任务
|
||||
* 标 FAILED(原因「客户端异常中断: …」),用户能看到发生过什么;但长任务(采集/相似ASIN/外观专利/单次跟价)
|
||||
* 一遇客户端重启就整个白跑,需要有人把它重新放回队列。本服务负责这件事。
|
||||
*
|
||||
* <p>续跑方式刻意复用已有链路:新建的任务落 PENDING,由 {@link TaskClientPullService} 的兜底拉取
|
||||
* (客户端每分钟 {@code GET /api/tasks/pull-pending})领走执行——不再新造第二套派发机制。
|
||||
* 只有注册了 {@link ClientTaskPullSpi} 的模块才会被续跑(这些模块的载荷能由服务端自行组装);
|
||||
* 上架/改价等写操作模块不在其中,避免盲目重跑。
|
||||
*
|
||||
* <p>封顶 {@code max-attempt}:会话持续不可用(账号被风控、紫鸟未就绪)时,不封顶会无限重排队。
|
||||
* 计数写在 {@code biz_file_task.resume_attempt}(V129),随续跑任务代代递增。
|
||||
*
|
||||
* <p>幂等:同一原任务已存在续跑任务({@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<String, ClientTaskPullSpi> 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<ClientTaskPullSpi> pullSpiHandlers) {
|
||||
this.fileTaskMapper = fileTaskMapper;
|
||||
Map<String, ClientTaskPullSpi> 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<FileTaskEntity> candidates = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.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<FileTaskEntity>()
|
||||
.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;
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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;
|
||||
+6
-1
@@ -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() {
|
||||
|
||||
+1
-1
@@ -82,7 +82,7 @@ class TaskModuleCoverageTest {
|
||||
mock(SimilarAsinTaskService.class),
|
||||
null, null, null, null,
|
||||
mock(ShopDataCrawlTaskService.class),
|
||||
null, null);
|
||||
null, null, null);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
+157
@@ -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);
|
||||
|
||||
+155
@@ -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 续跑任务。
|
||||
*
|
||||
* <p>背景: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<String, Object> buildQueuePayload(FileTaskEntity task) {
|
||||
return Map.of("type", "x-run", "data", Map.of("taskId", task.getId()));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static TaskResumeService service(FileTaskMapper mapper, List<ClientTaskPullSpi> 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<FileTaskEntity> 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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user