feat(任务判死): 心跳正常但 180 分钟无结果上报的二次判死线（13 模块）+ 同期待发改动
判死线（治 28131 型「主线程卡死、心跳线程照发」）： - 判据改看 biz_task_scope_state.last_chunk_at（HTTP 心跳不刷新它）；从未上报跳过不判 - 中央线覆盖 DELETE_BRAND/PRODUCT_RISK_RESOLVE/PRICE_TRACK/SHOP_MATCH/PATROL_DELETE/QUERY_ASIN/WITHDRAW - 自带线接入 COLLECT_DATA/SIMILAR_ASIN/APPEARANCE_PATENT/SHOP_DATA_CRAWL/PUBLISH/BRAND - 客户端心跳带处理位置 progressText，判死文案含最后位置；no-result-upload-timeout-minutes 默认 180（0 关闭） 同期带上另一工作流的待发改动：跟价换 IP 重试、品牌检测重试上限与 LLM 并发下调、 教程包后台管理页与 V126 迁移、admin-vue 教程记录页。
This commit is contained in:
+11
-2
@@ -30,15 +30,24 @@ public class AppearancePatentProperties {
|
||||
private int llmFirstAttemptReadTimeoutMillis = 60000;
|
||||
private int llmBatchSize = 10;
|
||||
/**
|
||||
* 批内行级并发数，默认等于批量大小
|
||||
* 批内行级并发数。批次串行提交，每行串行发 2 个 LLM 请求，
|
||||
* 故该值≈单任务对 LLM 网关的瞬时并发；默认与品牌检测同为 5，避免多任务并行时成倍放大。
|
||||
*/
|
||||
private int llmRowConcurrency = 10;
|
||||
private int llmRowConcurrency = 5;
|
||||
/**
|
||||
* 每行每个 LLM 请求的重试次数（含首次）
|
||||
*/
|
||||
private int llmRetryTimes = 3;
|
||||
private int staleTimeoutMinutes = 30;
|
||||
private String staleFinalizeCron = "0 */2 * * * *";
|
||||
|
||||
/**
|
||||
* 二次判死线（心跳正常但连续 N 分钟无结果分片上报）阈值，分钟；<=0 关闭本线。
|
||||
* 既有判定以 Redis heartbeat stale 为主信号，但心跳随 Python HTTP 心跳每分钟刷新——
|
||||
* 主线程卡死时心跳线程照发，任务永远不 stale（与生产 28131 同型缺口）。
|
||||
* 本线改看 biz_task_scope_state.last_chunk_at（仅分片上传时刷新）。
|
||||
*/
|
||||
private int noResultUploadTimeoutMinutes = 180;
|
||||
/**
|
||||
* 末尾不足一批的数据等待该时长后强制提交检测。
|
||||
* Python 回传超时后会直接封口；这个配置处理 Python 慢回传但仍未超时的零头批次。
|
||||
|
||||
@@ -10,10 +10,19 @@ public class BrandCheckProperties {
|
||||
private String path = "/brand_check";
|
||||
private String token = "";
|
||||
private String defaultStrategy = "Terms";
|
||||
/** 单品牌商标查询的重试次数（含首次），连续失败超过该次数才判定为查询失败。 */
|
||||
private int retryTimes = 3;
|
||||
/** 每次查询失败后到下一次重试前的等待毫秒数。 */
|
||||
/**
|
||||
* 单品牌商标查询的重试次数（含首次），连续失败超过该次数才判定为查询失败。
|
||||
* 原为 3：16890 偶发限流几秒内即恢复，3 次（前两次间隔各 1s）恢复不了就把结论
|
||||
* 写成「查询失败」，对客户是硬伤；2026-09-14 与客户端品牌一致提到 10 次。
|
||||
*/
|
||||
private int retryTimes = 10;
|
||||
/** 每次查询失败后到下一次重试前的等待毫秒数（基准值，按重试轮次递增）。 */
|
||||
private int retryIntervalMillis = 1000;
|
||||
/**
|
||||
* 单次重试等待的上限毫秒数。等待按 retryIntervalMillis × 第几次重试 递增后封顶，
|
||||
* 10 次尝试合计约 45s，既能覆盖限流窗口又不会让单个品牌长时间占住查询线程。
|
||||
*/
|
||||
private int retryMaxIntervalMillis = 10000;
|
||||
private int connectTimeoutMillis = 10000;
|
||||
private int readTimeoutMillis = 60000;
|
||||
}
|
||||
|
||||
@@ -10,4 +10,11 @@ public class BrandProgressProperties {
|
||||
private long failedTtlHours = 2;
|
||||
private long heartbeatTimeoutMinutes = 15;
|
||||
private String staleCheckCron = "0 */2 * * * *";
|
||||
|
||||
/**
|
||||
* 二次判死线（心跳正常但连续 N 分钟无结果上报）阈值，分钟；<=0 关闭本线。
|
||||
* 既有心跳线以 updated_at/last_heartbeat_at 陈旧为判据，而前端心跳会持续刷新它们——
|
||||
* 主线程卡死时心跳线程照发，任务永远命不中。本线改看结果上报时写入的 last_result_at。
|
||||
*/
|
||||
private long noResultUploadTimeoutMinutes = 180;
|
||||
}
|
||||
|
||||
+13
@@ -45,6 +45,19 @@ public class DeleteBrandProgressProperties {
|
||||
*/
|
||||
private long withdrawStaleTimeoutMinutes = 30;
|
||||
|
||||
/**
|
||||
* 「心跳正常但连续 N 分钟无结果分片上报」的二次判死阈值（分钟），默认 3 小时。
|
||||
*
|
||||
* <p>既有各模块心跳线的候选条件是 updated_at 陈旧，而客户端心跳会持续刷新 updated_at——
|
||||
* 主线程卡死时心跳线程照发，任务永远命不中（生产 28131 卡死 12h+ 仍 RUNNING）。
|
||||
* 本线改用 biz_task_scope_state.last_chunk_at（只随结果分片上报刷新）作判据，
|
||||
* 从未上报（无 scope 行或全 NULL）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private long noResultUploadTimeoutMinutes = 180;
|
||||
|
||||
/** 二次判死线开关：false = 整段不扫描（观察期与回滚用，改环境变量即生效）。 */
|
||||
private boolean noResultUploadCheckEnabled = true;
|
||||
|
||||
/**
|
||||
* 组装结果文件时允许缺失的最大分片数：缺失不超过该值时降级组装（缺失行状态写"未回传"），
|
||||
* 保证任务不因少量分片丢失整体白跑；超出时任务失败，但仍会在重试耗尽后尝试
|
||||
|
||||
@@ -23,6 +23,14 @@ public class SimilarAsinProperties {
|
||||
private int staleTimeoutMinutes = 30;
|
||||
private String staleFinalizeCron = "0 */2 * * * *";
|
||||
|
||||
/**
|
||||
* 二次判死线（心跳正常但连续 N 分钟无结果分片上报）阈值，分钟；<=0 关闭本线。
|
||||
* 既有判定以 Redis heartbeat stale 为主信号，但心跳随 Python HTTP 心跳每分钟刷新——
|
||||
* 主线程卡死时心跳线程照发，任务永远不 stale（与生产 28131 同型缺口）。
|
||||
* 本线改看 biz_task_scope_state.last_chunk_at（仅分片上传时刷新）。
|
||||
*/
|
||||
private int noResultUploadTimeoutMinutes = 180;
|
||||
|
||||
/**
|
||||
* 末尾零头 batch 的强制 flush 阈值（分钟）：当不足 llmBatchSize 的零头 row
|
||||
* 长时间挂着（Python 慢回传）时触发提交。
|
||||
|
||||
+61
@@ -40,6 +40,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
||||
@@ -82,6 +83,7 @@ import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
@@ -601,6 +603,7 @@ public class AppearancePatentTaskService {
|
||||
}
|
||||
}
|
||||
}
|
||||
finalizeNoUploadStaleTasks();
|
||||
}
|
||||
|
||||
public void debugFinalizeStaleTask(Long taskId) {
|
||||
@@ -661,6 +664,64 @@ public class AppearancePatentTaskService {
|
||||
return updatedMillis <= thresholdMillis;
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：Python 心跳正常（Redis heartbeat 新鲜）但连续 N 分钟无结果分片上报。
|
||||
*
|
||||
* <p>既有判定以 Redis heartbeat 为 stale 主信号（P1-7），但该心跳随 Python 的 HTTP 心跳
|
||||
* 每分钟刷新——主线程卡死时心跳线程照发，任务永远不 stale（与生产 28131 同型缺口）。
|
||||
* 本线改看 biz_task_scope_state.last_chunk_at（仅在 persistSubmittedChunk 上传分片时刷新），
|
||||
* 命中后走既有 finalizeStaleTask（封口上传 + LLM 收尾，不粗暴杀任务）。
|
||||
* 从未上报（无 scope 行或全 NULL）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void finalizeNoUploadStaleTasks() {
|
||||
long minutes = properties.getNoResultUploadTimeoutMinutes();
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
List<FileTaskEntity> tasks = listStaleFinalizeCandidates();
|
||||
if (tasks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
long heartbeatThresholdMillis = LocalDateTime.now()
|
||||
.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes()))
|
||||
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = new HashMap<>();
|
||||
for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds(
|
||||
tasks.stream().map(FileTaskEntity::getId).toList())) {
|
||||
if (dto.taskId() != null && dto.lastChunkAt() != null) {
|
||||
lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt());
|
||||
}
|
||||
}
|
||||
for (FileTaskEntity task : tasks) {
|
||||
if (isHeartbeatStale(task, heartbeatThresholdMillis)) {
|
||||
// 心跳已 stale：归既有心跳线处理
|
||||
continue;
|
||||
}
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId());
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L);
|
||||
if (lockHandle == null) {
|
||||
continue;
|
||||
}
|
||||
try (lockHandle) {
|
||||
log.warn("[appearance-patent] 心跳正常但 {} 分钟无结果分片上报，按卡死收尾 taskId={} lastResultAt={}",
|
||||
minutes, task.getId(), lastResultAt);
|
||||
String error = "Python heartbeat alive but no result chunk uploaded for " + minutes + " minutes";
|
||||
if (transactionManager != null) {
|
||||
inNewTransaction(() -> {
|
||||
finalizeStaleTask(task.getId(), error);
|
||||
return null;
|
||||
});
|
||||
} else {
|
||||
finalizeStaleTask(task.getId(), error);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* /result 提交的纯计算阶段（无事务注解）：读任务 + 校验归属 + 分组展开 +
|
||||
* 序列化 + payload 存储 + 哈希预计算，全部在事务开始前完成。
|
||||
|
||||
+20
-8
@@ -121,9 +121,10 @@ public class BrandCheckClient {
|
||||
lastFailure = ex;
|
||||
response = null;
|
||||
if (attempt < attempts) {
|
||||
log.warn("[brand-check] 请求异常将重试 brand={} attempt={}/{} err={}",
|
||||
brand, attempt, attempts, ex.getMessage());
|
||||
sleepBeforeRetry();
|
||||
long retryDelayMillis = retryDelayMillis(attempt);
|
||||
log.warn("[brand-check] 请求异常将重试 brand={} attempt={}/{} 等待={}ms err={}",
|
||||
brand, attempt, attempts, retryDelayMillis, ex.getMessage());
|
||||
sleepBeforeRetry(retryDelayMillis);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
@@ -132,9 +133,10 @@ public class BrandCheckClient {
|
||||
return new BrandCheckOutcome(nullToEmpty(response.getFaildData()), List.of());
|
||||
}
|
||||
if (attempt < attempts) {
|
||||
log.warn("[brand-check] 服务端返回查询失败将重试 brand={} attempt={}/{}",
|
||||
brand, attempt, attempts);
|
||||
sleepBeforeRetry();
|
||||
long retryDelayMillis = retryDelayMillis(attempt);
|
||||
log.warn("[brand-check] 服务端返回查询失败将重试 brand={} attempt={}/{} 等待={}ms",
|
||||
brand, attempt, attempts, retryDelayMillis);
|
||||
sleepBeforeRetry(retryDelayMillis);
|
||||
}
|
||||
}
|
||||
log.warn("[brand-check] 重试耗尽判定查询失败 brand={} attempts={} lastErr={}",
|
||||
@@ -144,8 +146,18 @@ public class BrandCheckClient {
|
||||
response == null ? List.of(brand) : nullToEmpty(response.getQueryFaildData()));
|
||||
}
|
||||
|
||||
private void sleepBeforeRetry() {
|
||||
long delayMillis = Math.max(0L, properties.getRetryIntervalMillis());
|
||||
/**
|
||||
* 第 attempt 次重试前的等待毫秒数：按基准间隔随轮次递增后封顶。
|
||||
* 限流窗口通常只有几秒，固定 1s 间隔反复打过去救不回来；递增等待能覆盖窗口，
|
||||
* 封顶则保证单个品牌不会长时间占住查询线程（并发度只有 3）。
|
||||
*/
|
||||
private long retryDelayMillis(int attempt) {
|
||||
long base = Math.max(0L, properties.getRetryIntervalMillis());
|
||||
long cap = Math.max(base, properties.getRetryMaxIntervalMillis());
|
||||
return Math.min(base * Math.max(1, attempt), cap);
|
||||
}
|
||||
|
||||
private void sleepBeforeRetry(long delayMillis) {
|
||||
if (delayMillis <= 0L) {
|
||||
return;
|
||||
}
|
||||
|
||||
+2
@@ -60,6 +60,8 @@ public class BrandTaskProgressCacheService {
|
||||
values.put("finished_files", String.valueOf(Math.max(finishedFiles, 0)));
|
||||
values.put("updated_at", now);
|
||||
values.put("last_heartbeat_at", now);
|
||||
// 结果上报专属信号（二次判死线用）：心跳/touchHeartbeat 不写它，只有结果回传才刷新
|
||||
values.put("last_result_at", now);
|
||||
try {
|
||||
stringRedisTemplate.opsForHash().putAll(key, values);
|
||||
stringRedisTemplate.expire(key, ttl());
|
||||
|
||||
+49
@@ -1537,9 +1537,58 @@ public class BrandTaskService {
|
||||
failStaleRunningTask(task.getId(), "前端长时间无响应，任务已自动失败");
|
||||
}
|
||||
}
|
||||
failNoUploadStaleRunningTasks();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：前端心跳正常但连续 N 分钟无结果上报（治「心跳续命」的假活任务）。
|
||||
*
|
||||
* <p>既有心跳线候选条件是 updated_at/last_heartbeat_at 陈旧，而前端心跳会持续刷新它们——
|
||||
* 主线程卡死时心跳线程照发，任务永远命不中（与生产 28131 同型缺口）。
|
||||
* 本线候选取「心跳新鲜 + 创建超过 N 分钟」，判据用结果上报时写入 progress hash 的
|
||||
* last_result_at（仅 saveProgressFromResult 写）；从未上报（无该字段）跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void failNoUploadStaleRunningTasks() {
|
||||
long minutes = brandProgressProperties.getNoResultUploadTimeoutMinutes();
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
LocalDateTime heartbeatThreshold = LocalDateTime.now()
|
||||
.minusMinutes(brandProgressProperties.getHeartbeatTimeoutMinutes());
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
List<BrandCrawlTaskEntity> runningTasks = brandCrawlTaskMapper.selectList(new LambdaQueryWrapper<BrandCrawlTaskEntity>()
|
||||
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
|
||||
.ge(BrandCrawlTaskEntity::getUpdatedAt, heartbeatThreshold)
|
||||
.lt(BrandCrawlTaskEntity::getCreatedAt, cutoff));
|
||||
for (BrandCrawlTaskEntity task : runningTasks) {
|
||||
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L);
|
||||
if (taskLockHandle == null) {
|
||||
continue;
|
||||
}
|
||||
try (taskLockHandle) {
|
||||
Map<Object, Object> progress = brandTaskProgressCacheService.getProgress(task.getId());
|
||||
long lastResultAt = 0L;
|
||||
try {
|
||||
lastResultAt = Long.parseLong(String.valueOf(progress.getOrDefault("last_result_at", "0")));
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
if (lastResultAt <= 0L) {
|
||||
// 从未上报结果：保守跳过（首批可能较慢）
|
||||
continue;
|
||||
}
|
||||
LocalDateTime lastResult = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastResultAt), ZoneId.systemDefault());
|
||||
if (lastResult.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
log.warn("[brand-stale-check] no-upload failing taskId={} lastResultAt={} timeoutMinutes={}",
|
||||
task.getId(), lastResult, minutes);
|
||||
failStaleRunningTask(task.getId(),
|
||||
"连续 " + minutes + " 分钟无结果回传，疑似客户端任务卡死，任务已自动失败（最后结果上报 " + lastResult + "）");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
|
||||
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis);
|
||||
if (lockHandle == null) {
|
||||
|
||||
+86
@@ -51,6 +51,7 @@ import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||
@@ -179,6 +180,10 @@ public class CollectDataService {
|
||||
@Value("${aiimage.collect-data.stale-timeout-minutes:30}")
|
||||
private long staleTimeoutMinutes;
|
||||
|
||||
/** 二次判死线（心跳正常但连续 N 分钟无结果分片上报）阈值，分钟；<=0 关闭本线。 */
|
||||
@Value("${aiimage.collect-data.no-result-upload-timeout-minutes:180}")
|
||||
private long noResultUploadTimeoutMinutes;
|
||||
|
||||
@Value("${aiimage.collect-data.max-source-file-bytes:0}")
|
||||
private Long maxSourceFileBytes;
|
||||
|
||||
@@ -526,6 +531,7 @@ public class CollectDataService {
|
||||
for (FileTaskEntity task : tasks) {
|
||||
finalizeStaleTask(task.getId(), threshold, timeoutMinutes);
|
||||
}
|
||||
finalizeNoUploadStaleTasks(threshold);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -565,6 +571,86 @@ public class CollectDataService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：心跳正常但连续 N 分钟无结果分片上报（治「心跳续命」的假活任务）。
|
||||
*
|
||||
* <p>既有心跳线候选条件是 updated_at 陈旧，而客户端心跳会持续刷新 updated_at——
|
||||
* 主线程卡死（浏览器自动化等待/异常）时心跳线程照发，任务永远命不中。
|
||||
* 本线候选取「心跳新鲜（既有线放过）+ 创建超过 N 分钟」，判据用
|
||||
* biz_task_scope_state.last_chunk_at（只随结果分片上报刷新）；
|
||||
* 从未上报（无 scope 行或全 NULL）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void finalizeNoUploadStaleTasks(LocalDateTime heartbeatThreshold) {
|
||||
long minutes = noResultUploadTimeoutMinutes;
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
List<FileTaskEntity> candidates = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
|
||||
.ge(FileTaskEntity::getUpdatedAt, heartbeatThreshold)
|
||||
.lt(FileTaskEntity::getCreatedAt, cutoff)
|
||||
.orderByAsc(FileTaskEntity::getCreatedAt)
|
||||
.last("limit 200"));
|
||||
if (candidates.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = new HashMap<>();
|
||||
for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds(
|
||||
candidates.stream().map(FileTaskEntity::getId).toList())) {
|
||||
if (dto.taskId() != null && dto.lastChunkAt() != null) {
|
||||
lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt());
|
||||
}
|
||||
}
|
||||
for (FileTaskEntity task : candidates) {
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId());
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
finalizeNoUploadStaleTask(task.getId(), lastResultAt, minutes);
|
||||
}
|
||||
}
|
||||
|
||||
/** 单个任务的二次判死收尾：锁内复查心跳后，有分片→组装部分工作簿；无分片→标失败。 */
|
||||
private void finalizeNoUploadStaleTask(Long taskId, LocalDateTime lastResultAt, long minutes) {
|
||||
try (TaskDistributedLockService.LockHandle lock =
|
||||
taskDistributedLockService.acquire(MODULE_TYPE, taskId, 0L)) {
|
||||
if (lock == null) {
|
||||
return;
|
||||
}
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
LocalDateTime heartbeatFreshAfter = LocalDateTime.now().minusMinutes(Math.max(1L, staleTimeoutMinutes));
|
||||
if (task == null
|
||||
|| !MODULE_TYPE.equals(task.getModuleType())
|
||||
|| !STATUS_RUNNING.equals(task.getStatus())
|
||||
|| task.getUpdatedAt() == null
|
||||
|| task.getUpdatedAt().isBefore(heartbeatFreshAfter)) {
|
||||
// 已终态、或心跳在排队期间回落到陈旧（交回既有心跳线处理）
|
||||
return;
|
||||
}
|
||||
if (taskFileJobService.countUnfinishedAssembleJobs(taskId, MODULE_TYPE) > 0L) {
|
||||
return;
|
||||
}
|
||||
FileResultEntity result = ensureTaskResult(task);
|
||||
CollectDataStats stats = loadStats(task);
|
||||
if (hasReceivedChunks(taskId)) {
|
||||
enqueueFinalWorkbook(task, result, stats);
|
||||
log.warn("[collect-data] no-upload stale task enqueued partial workbook taskId={} lastResultAt={} timeoutMinutes={} finalRows={}",
|
||||
taskId, lastResultAt, minutes, stats.finalRowCount);
|
||||
return;
|
||||
}
|
||||
markTaskFailed(task, result,
|
||||
"连续 " + minutes + " 分钟无结果回传，疑似客户端任务卡死，任务已自动失败（最后结果上报 " + lastResultAt + "）",
|
||||
stats);
|
||||
log.warn("[collect-data] no-upload stale task failed without chunks taskId={} lastResultAt={} timeoutMinutes={}",
|
||||
taskId, lastResultAt, minutes);
|
||||
} catch (Exception ex) {
|
||||
log.warn("[collect-data] no-upload stale task finalization failed taskId={} msg={}",
|
||||
taskId, ex.getMessage(), ex);
|
||||
}
|
||||
}
|
||||
|
||||
public CollectDataDashboardVo dashboard(Long userId) {
|
||||
CollectDataDashboardVo vo = new CollectDataDashboardVo();
|
||||
vo.setPendingTaskCount(countActiveTasks(userId));
|
||||
|
||||
+137
-1
@@ -21,9 +21,12 @@ import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService;
|
||||
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskCacheService;
|
||||
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
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 lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
@@ -55,6 +58,10 @@ public class DeleteBrandStaleTaskService {
|
||||
private static final String MODULE_TYPE_PATROL_DELETE = "PATROL_DELETE";
|
||||
private static final String MODULE_TYPE_QUERY_ASIN = "QUERY_ASIN";
|
||||
private static final String MODULE_TYPE_WITHDRAW = "WITHDRAW";
|
||||
/** 二次判死线（心跳正常但无结果上报）覆盖的模块：走结果分片上报表、last_chunk_at 有效的模块。 */
|
||||
private static final List<String> NO_RESULT_UPLOAD_CHECKED_MODULES = List.of(
|
||||
MODULE_TYPE_DELETE_BRAND, MODULE_TYPE_PRODUCT_RISK, MODULE_TYPE_PRICE_TRACK,
|
||||
MODULE_TYPE_SHOP_MATCH, MODULE_TYPE_PATROL_DELETE, MODULE_TYPE_QUERY_ASIN, MODULE_TYPE_WITHDRAW);
|
||||
private static final Duration STALE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
|
||||
private static final Duration FINALIZE_CHECK_LOCK_TTL = Duration.ofMinutes(10);
|
||||
private static final Duration TEMP_DIR_CLEANUP_LOCK_TTL = Duration.ofMinutes(10);
|
||||
@@ -83,6 +90,8 @@ public class DeleteBrandStaleTaskService {
|
||||
private final TaskDistributedLockService taskDistributedLockService;
|
||||
private final TaskFileJobService taskFileJobService;
|
||||
private final ShopDataCrawlTaskService shopDataCrawlTaskService;
|
||||
private final TaskScopeStateMapper taskScopeStateMapper;
|
||||
private final TaskHeartbeatPositionService taskHeartbeatPositionService;
|
||||
|
||||
@Value("${aiimage.temp-dir.retention-hours:24}")
|
||||
private long tempDirRetentionHours;
|
||||
@@ -104,19 +113,21 @@ public class DeleteBrandStaleTaskService {
|
||||
ShopMatchStaleCheckStats patrolDeleteStats = failStalePatrolDeleteTasks();
|
||||
ShopMatchStaleCheckStats queryAsinStats = failStaleQueryAsinTasks();
|
||||
ShopMatchStaleCheckStats withdrawStats = failStaleWithdrawTasks();
|
||||
NoResultUploadStaleCheckStats noUploadStats = failNoResultUploadTasks();
|
||||
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={}) 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={}",
|
||||
stats.scannedTaskCount, stats.finalizedTaskCount, stats.failedTaskCount, stats.skippedTaskCount,
|
||||
priceTrackStats.scannedTaskCount, priceTrackStats.finalizedTaskCount, priceTrackStats.failedTaskCount, priceTrackStats.skippedTaskCount,
|
||||
shopMatchStats.scannedTaskCount, shopMatchStats.finalizedTaskCount, shopMatchStats.failedTaskCount, shopMatchStats.skippedTaskCount,
|
||||
patrolDeleteStats.scannedTaskCount, patrolDeleteStats.finalizedTaskCount, patrolDeleteStats.failedTaskCount, patrolDeleteStats.skippedTaskCount,
|
||||
queryAsinStats.scannedTaskCount, queryAsinStats.finalizedTaskCount, queryAsinStats.failedTaskCount, queryAsinStats.skippedTaskCount,
|
||||
withdrawStats.scannedTaskCount, withdrawStats.finalizedTaskCount, withdrawStats.failedTaskCount, withdrawStats.skippedTaskCount,
|
||||
noUploadStats.scannedTaskCount, noUploadStats.failedTaskCount, noUploadStats.skippedTaskCount,
|
||||
System.currentTimeMillis() - startedAt,
|
||||
Thread.currentThread().getName());
|
||||
}
|
||||
@@ -711,6 +722,125 @@ public class DeleteBrandStaleTaskService {
|
||||
return stats;
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：心跳正常但连续 N 分钟无结果分片上报（治「心跳续命」的假活任务）。
|
||||
*
|
||||
* <p>既有各模块心跳线的候选条件都是 updated_at 陈旧，而客户端心跳会持续刷新 updated_at——
|
||||
* 主线程卡死（浏览器自动化等待/异常）时心跳线程照发，任务永远命不中（生产 28131 卡死 12h+ 仍 RUNNING）。
|
||||
* 本线改用 biz_task_scope_state.last_chunk_at（只随结果分片上报刷新）作判据；
|
||||
* 无 scope 行或 last_chunk_at 全 NULL（从未上报）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private NoResultUploadStaleCheckStats failNoResultUploadTasks() {
|
||||
NoResultUploadStaleCheckStats stats = new NoResultUploadStaleCheckStats();
|
||||
if (!deleteBrandProgressProperties.isNoResultUploadCheckEnabled()) {
|
||||
return stats;
|
||||
}
|
||||
long minutes = Math.max(1L, deleteBrandProgressProperties.getNoResultUploadTimeoutMinutes());
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
List<FileTaskEntity> candidates = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.select(FileTaskEntity::getId, FileTaskEntity::getModuleType,
|
||||
FileTaskEntity::getCreatedAt, FileTaskEntity::getUpdatedAt)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.in(FileTaskEntity::getModuleType, NO_RESULT_UPLOAD_CHECKED_MODULES)
|
||||
.lt(FileTaskEntity::getCreatedAt, cutoff)
|
||||
.orderByAsc(FileTaskEntity::getCreatedAt)
|
||||
.last("limit 200"));
|
||||
stats.scannedTaskCount = candidates.size();
|
||||
if (candidates.isEmpty()) {
|
||||
return stats;
|
||||
}
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = taskScopeStateMapper
|
||||
.selectLastChunkAtByTaskIds(candidates.stream().map(FileTaskEntity::getId).toList())
|
||||
.stream()
|
||||
.filter(dto -> dto.taskId() != null && dto.lastChunkAt() != null)
|
||||
.collect(Collectors.toMap(TaskScopeLastChunkDto::taskId, TaskScopeLastChunkDto::lastChunkAt, (a, b) -> a));
|
||||
for (FileTaskEntity task : candidates) {
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId());
|
||||
// 从未上报（无 scope 行或全 NULL）或上报仍新鲜：正常推进，不必处理
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
String moduleType = task.getModuleType();
|
||||
if (hasPendingAssembleJobs(task.getId(), moduleType)) {
|
||||
stats.skippedTaskCount++;
|
||||
log.info("[stale-check] no-upload skip pending-assemble-jobs taskId={} moduleType={}", task.getId(), moduleType);
|
||||
continue;
|
||||
}
|
||||
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(moduleType, task.getId());
|
||||
if (taskLockHandle == null) {
|
||||
stats.skippedTaskCount++;
|
||||
continue;
|
||||
}
|
||||
try (taskLockHandle) {
|
||||
try {
|
||||
tryFinalizeNoResultUploadTask(moduleType, task.getId());
|
||||
} catch (Exception ex) {
|
||||
// finalize 抛异常（组装失败等）不代表任务活着：继续走 CAS 判死
|
||||
log.warn("[stale-check] no-upload finalize threw taskId={} moduleType={} msg={}", task.getId(), moduleType, ex.getMessage());
|
||||
}
|
||||
// finalize 已终结的任务 CAS 自然不命中（状态不再是 RUNNING），无需回读
|
||||
String lastPosition = taskHeartbeatPositionService.describe(task.getId());
|
||||
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getId, task.getId())
|
||||
.eq(FileTaskEntity::getModuleType, moduleType)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.set(FileTaskEntity::getStatus, "FAILED")
|
||||
.set(FileTaskEntity::getErrorMessage, buildNoResultUploadFailReason(minutes, lastResultAt, lastPosition))
|
||||
.set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())
|
||||
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
|
||||
if (updated > 0) {
|
||||
stats.failedTaskCount++;
|
||||
deleteNoResultUploadTaskCache(moduleType, task.getId());
|
||||
log.warn("[stale-check] no-upload failed taskId={} moduleType={} lastResultAt={} taskUpdatedAt={} timeoutMinutes={}",
|
||||
task.getId(), moduleType, lastResultAt, task.getUpdatedAt(), minutes);
|
||||
} else {
|
||||
stats.skippedTaskCount++;
|
||||
log.info("[stale-check] no-upload task already finalized by compensation taskId={} moduleType={}", task.getId(), moduleType);
|
||||
}
|
||||
}
|
||||
}
|
||||
return stats;
|
||||
}
|
||||
|
||||
/** 按模块调用各自的收尾入口（尽力组装部分结果；已终结的任务由后续 CAS 自然放行）。 */
|
||||
private void tryFinalizeNoResultUploadTask(String moduleType, Long taskId) {
|
||||
switch (moduleType) {
|
||||
case MODULE_TYPE_DELETE_BRAND -> deleteBrandRunService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_PRODUCT_RISK -> productRiskTaskService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_PRICE_TRACK -> priceTrackTaskService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_SHOP_MATCH -> shopMatchTaskService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_PATROL_DELETE -> patrolDeleteTaskService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_QUERY_ASIN -> queryAsinTaskService.tryFinalizeTask(taskId, true);
|
||||
case MODULE_TYPE_WITHDRAW -> withdrawTaskService.tryFinalizeTask(taskId, true);
|
||||
default -> { }
|
||||
}
|
||||
}
|
||||
|
||||
/** 按模块删除各自缓存（与既有各段 CAS 翻转后的清理一致）。 */
|
||||
private void deleteNoResultUploadTaskCache(String moduleType, Long taskId) {
|
||||
switch (moduleType) {
|
||||
case MODULE_TYPE_DELETE_BRAND -> deleteBrandTaskCacheService.delete(taskId);
|
||||
case MODULE_TYPE_PRODUCT_RISK -> productRiskTaskCacheService.deleteTaskCache(taskId);
|
||||
case MODULE_TYPE_PRICE_TRACK -> priceTrackTaskCacheService.deleteTaskCache(taskId);
|
||||
case MODULE_TYPE_SHOP_MATCH -> shopMatchTaskCacheService.deleteTaskCache(taskId);
|
||||
case MODULE_TYPE_PATROL_DELETE -> patrolDeleteTaskCacheService.deleteTaskCache(taskId);
|
||||
case MODULE_TYPE_QUERY_ASIN -> queryAsinTaskCacheService.deleteTaskCache(taskId);
|
||||
case MODULE_TYPE_WITHDRAW -> withdrawTaskCacheService.deleteTaskCache(taskId);
|
||||
default -> { }
|
||||
}
|
||||
}
|
||||
|
||||
/** 二次判死文案：写明最后结果上报时间与最后处理位置，便于区分「卡死」与「长间隔」并定位卡点。 */
|
||||
private String buildNoResultUploadFailReason(long minutes, LocalDateTime lastResultAt, String position) {
|
||||
StringBuilder sb = new StringBuilder("连续 ").append(minutes)
|
||||
.append(" 分钟无结果回传，疑似客户端任务卡死，任务已自动失败（最后结果上报 ").append(lastResultAt);
|
||||
if (position != null && !position.isBlank()) {
|
||||
sb.append("，最后处理位置：").append(position);
|
||||
}
|
||||
sb.append("）");
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
private TaskDistributedLockService.LockHandle acquireTaskLock(String moduleType, Long taskId) {
|
||||
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(moduleType, taskId, 0L);
|
||||
if (lockHandle == null) {
|
||||
@@ -849,4 +979,10 @@ public class DeleteBrandStaleTaskService {
|
||||
private int failedTaskCount;
|
||||
private int skippedTaskCount;
|
||||
}
|
||||
|
||||
private static final class NoResultUploadStaleCheckStats {
|
||||
private int scannedTaskCount;
|
||||
private int failedTaskCount;
|
||||
private int skippedTaskCount;
|
||||
}
|
||||
}
|
||||
|
||||
+91
-2
@@ -36,6 +36,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
@@ -70,6 +71,7 @@ import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Predicate;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Collectors;
|
||||
import com.nanri.aiimage.modules.task.model.vo.TaskProgressLightBatchVo;
|
||||
@@ -116,6 +118,10 @@ public class PublishTaskService {
|
||||
@Value("${aiimage.publish.stale-timeout-minutes:30}")
|
||||
private int staleTimeoutMinutes;
|
||||
|
||||
/** 二次判死线（心跳正常但连续 N 分钟无结果分片上报）阈值，分钟；<=0 关闭本线。 */
|
||||
@Value("${aiimage.publish.no-result-upload-timeout-minutes:180}")
|
||||
private int noResultUploadTimeoutMinutes;
|
||||
|
||||
public PublishParseVo parseAndCreateTask(PublishParseRequest request) {
|
||||
validateParseRequest(request);
|
||||
List<PreparedFile> preparedFiles = new ArrayList<>();
|
||||
@@ -459,6 +465,7 @@ public class PublishTaskService {
|
||||
candidate.getId(), ex.getMessage());
|
||||
}
|
||||
}
|
||||
failNoUploadStaleTasks();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -685,10 +692,93 @@ public class PublishTaskService {
|
||||
}
|
||||
|
||||
private void failStaleTaskLocked(Long taskId, LocalDateTime threshold) {
|
||||
failStaleTaskLockedInternal(taskId,
|
||||
task -> task.getUpdatedAt() != null && task.getUpdatedAt().isBefore(threshold),
|
||||
"任务心跳超时");
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：Python 心跳正常但连续 N 分钟无结果分片上报（治「心跳续命」的假活任务）。
|
||||
*
|
||||
* <p>既有心跳线候选条件是 updated_at 陈旧，而客户端心跳会持续刷新 updated_at——
|
||||
* 主线程卡死时心跳线程照发，任务永远命不中（与生产 28131 同型缺口）。
|
||||
* owner 过滤与 A 线保持一致（避免双节点同时处理非本实例任务）；
|
||||
* 判据用 biz_task_scope_state.last_chunk_at（仅分片上传时刷新）；从未上报跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void failNoUploadStaleTasks() {
|
||||
long minutes = noResultUploadTimeoutMinutes;
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
LocalDateTime heartbeatThreshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes));
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
List<FileTaskEntity> candidates = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
|
||||
.and(owner -> owner
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL")
|
||||
.or()
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''")
|
||||
.or()
|
||||
.apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId()))
|
||||
.ge(FileTaskEntity::getUpdatedAt, heartbeatThreshold)
|
||||
.lt(FileTaskEntity::getCreatedAt, cutoff)
|
||||
.orderByAsc(FileTaskEntity::getCreatedAt)
|
||||
.last("limit 100"));
|
||||
if (candidates.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = new LinkedHashMap<>();
|
||||
for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds(
|
||||
candidates.stream().map(FileTaskEntity::getId).toList())) {
|
||||
if (dto.taskId() != null && dto.lastChunkAt() != null) {
|
||||
lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt());
|
||||
}
|
||||
}
|
||||
for (FileTaskEntity candidate : candidates) {
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(candidate.getId());
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) {
|
||||
continue;
|
||||
}
|
||||
try (TaskDistributedLockService.LockHandle lock =
|
||||
taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) {
|
||||
if (lock == null) {
|
||||
continue;
|
||||
}
|
||||
String error = "连续 " + minutes + " 分钟无结果回传，疑似客户端任务卡死，任务已自动失败（最后结果上报 " + lastResultAt + "）";
|
||||
transactionTemplate.executeWithoutResult(status -> failStaleTaskLockedInternal(
|
||||
candidate.getId(),
|
||||
task -> {
|
||||
// 排队期间复查：心跳回落陈旧 → 交回 A 线；出现新结果上报 → 任务已恢复
|
||||
if (task.getUpdatedAt() == null || task.getUpdatedAt().isBefore(heartbeatThreshold)) {
|
||||
return false;
|
||||
}
|
||||
LocalDateTime freshLastChunk = queryLastChunkAt(task.getId());
|
||||
return freshLastChunk != null && !freshLastChunk.isAfter(cutoff);
|
||||
},
|
||||
error));
|
||||
} catch (Exception ex) {
|
||||
log.warn("[publish] no-upload stale task cleanup failed taskId={} msg={}",
|
||||
candidate.getId(), ex.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 单任务最后一次结果分片上报时间；无记录返回 null。 */
|
||||
private LocalDateTime queryLastChunkAt(Long taskId) {
|
||||
List<TaskScopeLastChunkDto> rows = taskScopeStateMapper.selectLastChunkAtByTaskIds(List.of(taskId));
|
||||
return rows.isEmpty() ? null : rows.get(0).lastChunkAt();
|
||||
}
|
||||
|
||||
/** 收尾条件由调用方判定（A 线=心跳陈旧；B 线=心跳新鲜但结果上报陈旧），条件通过后走同一收尾路径。 */
|
||||
private void failStaleTaskLockedInternal(Long taskId, Predicate<FileTaskEntity> staleCheck, String error) {
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType())
|
||||
|| !STATUS_RUNNING.equals(task.getStatus())
|
||||
|| task.getUpdatedAt() == null || !task.getUpdatedAt().isBefore(threshold)) {
|
||||
|| !staleCheck.test(task)) {
|
||||
return;
|
||||
}
|
||||
ensureTaskOwnedByCurrentInstance(task, "cleanup stale publish task");
|
||||
@@ -696,7 +786,6 @@ public class PublishTaskService {
|
||||
return;
|
||||
}
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
String error = "任务心跳超时";
|
||||
List<PublishFileEntity> files = listTaskFiles(taskId);
|
||||
for (PublishFileEntity file : files) {
|
||||
if (!STATUS_PENDING.equals(file.getStatus()) && !STATUS_RUNNING.equals(file.getStatus())) {
|
||||
|
||||
+84
-2
@@ -32,6 +32,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
@@ -126,6 +127,10 @@ public class ShopDataCrawlTaskService {
|
||||
@Value("${aiimage.shop-data-crawl.stale-timeout-minutes:30}")
|
||||
private long staleTimeoutMinutes;
|
||||
|
||||
/** 二次判死线（心跳正常但连续 N 分钟无结果分片上报）阈值，分钟；<=0 关闭本线。 */
|
||||
@Value("${aiimage.shop-data-crawl.no-result-upload-timeout-minutes:180}")
|
||||
private long noResultUploadTimeoutMinutes;
|
||||
|
||||
/**
|
||||
* 陈旧任务判死：扫描本模块「RUNNING 且 Python 心跳超时」的任务并终结。
|
||||
*
|
||||
@@ -175,6 +180,11 @@ public class ShopDataCrawlTaskService {
|
||||
log.warn("[shop-data-crawl] stale task finalization failed taskId={} msg={}", task.getId(), ex.getMessage());
|
||||
}
|
||||
}
|
||||
finalizeNoUploadStaleTasks();
|
||||
}
|
||||
|
||||
private boolean markStaleTaskFailedIfStillRunning(Long taskId) {
|
||||
return markTaskFailedIfStillRunning(taskId, "长时间未收到 Python 结果回传，任务已自动失败");
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -183,13 +193,13 @@ public class ShopDataCrawlTaskService {
|
||||
*
|
||||
* @return true 表示本次确实由 RUNNING 翻转为 FAILED
|
||||
*/
|
||||
private boolean markStaleTaskFailedIfStillRunning(Long taskId) {
|
||||
private boolean markTaskFailedIfStillRunning(Long taskId, String errorMessage) {
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getId, taskId)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.set(FileTaskEntity::getStatus, "FAILED")
|
||||
.set(FileTaskEntity::getErrorMessage, "长时间未收到 Python 结果回传，任务已自动失败")
|
||||
.set(FileTaskEntity::getErrorMessage, errorMessage)
|
||||
.set(FileTaskEntity::getUpdatedAt, now)
|
||||
.set(FileTaskEntity::getFinishedAt, now));
|
||||
if (updated <= 0) {
|
||||
@@ -202,6 +212,78 @@ public class ShopDataCrawlTaskService {
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：Python 心跳正常（Redis heartbeat 新鲜）但连续 N 分钟无结果分片上报。
|
||||
*
|
||||
* <p>既有心跳线候选条件是心跳/updated_at 陈旧，而 Python 的 HTTP 心跳线程在任务主线程
|
||||
* 卡死时照常发送——心跳永远新鲜，任务永远命不中（与生产 28131 同型缺口）。
|
||||
* 本线候选取「心跳新鲜 + 创建超过 N 分钟」，判据用 biz_task_scope_state.last_chunk_at
|
||||
* （仅分片上传时刷新）；从未上报（无 scope 行或全 NULL）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void finalizeNoUploadStaleTasks() {
|
||||
long minutes = noResultUploadTimeoutMinutes;
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
long heartbeatFreshMillis = Duration.ofMinutes(Math.max(1L, staleTimeoutMinutes)).toMillis();
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
long nowMillis = System.currentTimeMillis();
|
||||
List<FileTaskEntity> tasks;
|
||||
try {
|
||||
tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||
.eq(FileTaskEntity::getStatus, "RUNNING")
|
||||
.ge(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(Math.max(1L, staleTimeoutMinutes)))
|
||||
.lt(FileTaskEntity::getCreatedAt, cutoff)
|
||||
.last("limit 200"));
|
||||
} catch (Exception ex) {
|
||||
log.warn("[shop-data-crawl] no-upload stale task scan failed msg={}", ex.getMessage());
|
||||
return;
|
||||
}
|
||||
if (tasks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<Long, Long> heartbeats = taskCacheService.getTaskHeartbeatMillisBatch(
|
||||
tasks.stream().map(FileTaskEntity::getId).toList());
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = new LinkedHashMap<>();
|
||||
for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds(
|
||||
tasks.stream().map(FileTaskEntity::getId).toList())) {
|
||||
if (dto.taskId() != null && dto.lastChunkAt() != null) {
|
||||
lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt());
|
||||
}
|
||||
}
|
||||
for (FileTaskEntity task : tasks) {
|
||||
long heartbeat = heartbeats.getOrDefault(task.getId(), 0L);
|
||||
boolean heartbeatFresh = heartbeat > 0 && nowMillis - heartbeat < heartbeatFreshMillis;
|
||||
if (!heartbeatFresh) {
|
||||
// 心跳已陈旧：归既有心跳线处理（updated_at 同源陈旧，由既有候选捞走）
|
||||
continue;
|
||||
}
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId());
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
TaskDistributedLockService.LockHandle taskLock = acquireTaskLock(task.getId(), 0L);
|
||||
if (taskLock == null) {
|
||||
log.info("[shop-data-crawl] no-upload stale task skipped because task lock is busy taskId={}", task.getId());
|
||||
continue;
|
||||
}
|
||||
try (taskLock) {
|
||||
if (taskFileJobService.countUnfinishedAssembleJobs(task.getId(), MODULE_TYPE) > 0L) continue;
|
||||
if (!tryFinalizeTask(task.getId(), true, true)) {
|
||||
markTaskFailedIfStillRunning(task.getId(),
|
||||
"连续 " + minutes + " 分钟无结果回传，疑似客户端任务卡死，任务已自动失败（最后结果上报 " + lastResultAt + "）");
|
||||
}
|
||||
log.warn("[shop-data-crawl] no-upload stale task finalized taskId={} lastResultAt={} timeoutMinutes={}",
|
||||
task.getId(), lastResultAt, minutes);
|
||||
} catch (TaskOwnerMismatchException ignored) {
|
||||
// The owner may change between the scan and finalization.
|
||||
} catch (Exception ex) {
|
||||
log.warn("[shop-data-crawl] no-upload stale task finalization failed taskId={} msg={}", task.getId(), ex.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private FileTaskEntity loadTaskForExecution(Long taskId) {
|
||||
Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId));
|
||||
FileTaskEntity cached = cachedTasks.get(taskId);
|
||||
|
||||
+54
@@ -63,6 +63,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
||||
@@ -98,6 +99,7 @@ import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
@@ -859,6 +861,7 @@ public class SimilarAsinTaskService implements SimilarAsinPipelineHost {
|
||||
finalizeStaleTask(task.getId(), "Python interrupted before uploading final similar ASIN result");
|
||||
}
|
||||
}
|
||||
finalizeNoUploadStaleTasks();
|
||||
}
|
||||
|
||||
public void debugFinalizeStaleTask(Long taskId) {
|
||||
@@ -906,6 +909,57 @@ public class SimilarAsinTaskService implements SimilarAsinPipelineHost {
|
||||
return updatedMillis <= thresholdMillis;
|
||||
}
|
||||
|
||||
/**
|
||||
* 二次判死：Python 心跳正常（Redis heartbeat 新鲜）但连续 N 分钟无结果分片上报。
|
||||
*
|
||||
* <p>既有判定以 Redis heartbeat 为 stale 主信号（P1-7），但该心跳随 Python 的 HTTP 心跳
|
||||
* 每分钟刷新——主线程卡死时心跳线程照发，任务永远不 stale（与生产 28131 同型缺口）。
|
||||
* 本线改看 biz_task_scope_state.last_chunk_at（仅在 persistSubmittedChunk 上传分片时刷新），
|
||||
* 命中后走既有 finalizeStaleTask（封口上传 + LLM 收尾，不粗暴杀任务）。
|
||||
* 从未上报（无 scope 行或全 NULL）的任务跳过，避免误杀首批较慢的正常任务。
|
||||
*/
|
||||
private void finalizeNoUploadStaleTasks() {
|
||||
long minutes = properties.getNoResultUploadTimeoutMinutes();
|
||||
if (minutes <= 0) {
|
||||
return;
|
||||
}
|
||||
List<FileTaskEntity> tasks = listStaleFinalizeCandidates();
|
||||
if (tasks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes);
|
||||
long heartbeatThresholdMillis = LocalDateTime.now()
|
||||
.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes()))
|
||||
.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
|
||||
Map<Long, LocalDateTime> lastResultAtByTaskId = new HashMap<>();
|
||||
for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds(
|
||||
tasks.stream().map(FileTaskEntity::getId).toList())) {
|
||||
if (dto.taskId() != null && dto.lastChunkAt() != null) {
|
||||
lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt());
|
||||
}
|
||||
}
|
||||
for (FileTaskEntity task : tasks) {
|
||||
if (isHeartbeatStale(task, heartbeatThresholdMillis)) {
|
||||
// 心跳已 stale：归既有心跳线处理
|
||||
continue;
|
||||
}
|
||||
LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId());
|
||||
if (lastResultAt == null || lastResultAt.isAfter(cutoff)) {
|
||||
continue;
|
||||
}
|
||||
TaskDistributedLockService.LockHandle taskLockHandle = ownershipSupport().acquireTaskLock(task.getId(), 0L);
|
||||
if (taskLockHandle == null) {
|
||||
continue;
|
||||
}
|
||||
try (taskLockHandle) {
|
||||
log.warn("[similar-asin] 心跳正常但 {} 分钟无结果分片上报，按卡死收尾 taskId={} lastResultAt={}",
|
||||
minutes, task.getId(), lastResultAt);
|
||||
finalizeStaleTask(task.getId(),
|
||||
"Python heartbeat alive but no result chunk uploaded for " + minutes + " minutes");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void finalizeStaleTask(Long taskId, String error) {
|
||||
FileTaskEntity task = fileTaskMapper.selectById(taskId);
|
||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !STATUS_RUNNING.equals(task.getStatus())) {
|
||||
|
||||
+4
@@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.task.controller;
|
||||
import com.nanri.aiimage.common.api.ApiResponse;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
|
||||
import com.nanri.aiimage.modules.task.model.vo.TaskHeartbeatVo;
|
||||
import com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService;
|
||||
import com.nanri.aiimage.modules.task.service.TaskHeartbeatService;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.Parameter;
|
||||
@@ -22,6 +23,7 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
public class TaskHeartbeatController {
|
||||
|
||||
private final TaskHeartbeatService taskHeartbeatService;
|
||||
private final TaskHeartbeatPositionService taskHeartbeatPositionService;
|
||||
|
||||
@PostMapping("/{taskId}/heartbeat")
|
||||
@Operation(
|
||||
@@ -31,6 +33,8 @@ public class TaskHeartbeatController {
|
||||
@Parameter(description = "任务 ID", required = true, example = "200")
|
||||
@PathVariable Long taskId,
|
||||
@Valid @RequestBody(required = false) TaskHeartbeatRequest request) {
|
||||
// 先记录「当前处理位置」（仅用于判死文案与排查，内部异常降级，不影响心跳本身）
|
||||
taskHeartbeatPositionService.record(taskId, request);
|
||||
return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request));
|
||||
}
|
||||
|
||||
|
||||
+25
@@ -1,9 +1,34 @@
|
||||
package com.nanri.aiimage.modules.task.mapper;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Mapper
|
||||
public interface TaskScopeStateMapper extends BaseMapper<TaskScopeStateEntity> {
|
||||
|
||||
/**
|
||||
* 各任务「最后一次结果分片上报时间」（二次判死线专用）。
|
||||
*
|
||||
* <p>last_chunk_at 只随结果分片上报刷新（TaskScopePayloadStorageService.saveScopePayload），
|
||||
* 不随 HTTP 心跳刷新——是区分「客户端进程活着」与「任务有进展」的可靠信号。
|
||||
* HAVING 过滤掉全 NULL 的组：从未上报的任务不出现在结果里，调用方据此跳过（不判死）。
|
||||
* GROUP BY 走 uk_task_scope 的 task_id 前缀。
|
||||
*/
|
||||
@Select("""
|
||||
<script>
|
||||
SELECT task_id AS taskId, MAX(last_chunk_at) AS lastChunkAt
|
||||
FROM biz_task_scope_state
|
||||
WHERE task_id IN
|
||||
<foreach item="taskId" collection="taskIds" open="(" separator="," close=")">#{taskId}</foreach>
|
||||
GROUP BY task_id
|
||||
HAVING MAX(last_chunk_at) IS NOT NULL
|
||||
</script>
|
||||
""")
|
||||
List<TaskScopeLastChunkDto> selectLastChunkAtByTaskIds(@Param("taskIds") List<Long> taskIds);
|
||||
}
|
||||
|
||||
+4
@@ -38,4 +38,8 @@ public class TaskHeartbeatRequest {
|
||||
|
||||
@Schema(description = "详情页 ASIN 总数", example = "120")
|
||||
private Integer detailTotalAsins;
|
||||
|
||||
@Schema(description = "客户端归一化后的当前处理位置描述（如「店铺 张三(2/10)，国家 德国(1/3)」），"
|
||||
+ "用于判死日志与排查；老客户端不传，服务端不得以位置为空作为判死依据", example = "店铺 魏振峰(2/10)")
|
||||
private String progressText;
|
||||
}
|
||||
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
package com.nanri.aiimage.modules.task.model.dto;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* 任务「最后一次结果分片上报时间」（「心跳正常但无结果上报」二次判死线用）。
|
||||
*
|
||||
* <p>lastChunkAt 取 biz_task_scope_state 的 MAX(last_chunk_at)——该列只随结果分片上报刷新
|
||||
* （TaskScopePayloadStorageService.saveScopePayload），不随 HTTP 心跳刷新。
|
||||
*/
|
||||
public record TaskScopeLastChunkDto(Long taskId, LocalDateTime lastChunkAt) {
|
||||
}
|
||||
+98
@@ -0,0 +1,98 @@
|
||||
package com.nanri.aiimage.modules.task.service;
|
||||
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 任务心跳的「当前处理位置」存储（Redis Hash，24h TTL）。
|
||||
*
|
||||
* <p>客户端主线程（浏览器自动化）卡死时心跳线程照发，仅凭心跳无法区分
|
||||
* 「进程活着」与「任务有进展」。位置文本用于判死日志 / errorMessage 说明卡在哪，
|
||||
* 以及后台人工排查。位置缺失只退化文案，<b>不得作为任何判死依据</b>。
|
||||
* 老客户端不传 progressText、位置全空时不写；读取失败返回 null。
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class TaskHeartbeatPositionService {
|
||||
|
||||
private static final String KEY_PREFIX = "task:heartbeat:position:";
|
||||
private static final Duration POSITION_TTL = Duration.ofHours(24);
|
||||
|
||||
private final StringRedisTemplate stringRedisTemplate;
|
||||
|
||||
public void record(Long taskId, TaskHeartbeatRequest request) {
|
||||
if (taskId == null || taskId <= 0 || request == null) {
|
||||
return;
|
||||
}
|
||||
String progressText = request.getProgressText() == null ? "" : request.getProgressText().trim();
|
||||
boolean hasPosition = !progressText.isEmpty()
|
||||
|| request.getCurrent() != null
|
||||
|| request.getTotal() != null
|
||||
|| (request.getPhase() != null && !request.getPhase().isBlank());
|
||||
if (!hasPosition) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Map<String, String> values = new HashMap<>();
|
||||
values.put("phase", request.getPhase() == null ? "" : request.getPhase());
|
||||
values.put("current", request.getCurrent() == null ? "" : String.valueOf(request.getCurrent()));
|
||||
values.put("total", request.getTotal() == null ? "" : String.valueOf(request.getTotal()));
|
||||
values.put("progressText", progressText);
|
||||
values.put("at", String.valueOf(Instant.now().toEpochMilli()));
|
||||
String key = KEY_PREFIX + taskId;
|
||||
stringRedisTemplate.opsForHash().putAll(key, values);
|
||||
stringRedisTemplate.expire(key, POSITION_TTL);
|
||||
} catch (Exception ex) {
|
||||
log.warn("[heartbeat-position] 记录位置失败 taskId={} msg={}", taskId, ex.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/** 读取「最后处理位置」描述；无位置信息返回 null（调用方自行降级）。 */
|
||||
public String describe(Long taskId) {
|
||||
if (taskId == null || taskId <= 0) {
|
||||
return null;
|
||||
}
|
||||
Map<Object, Object> entries;
|
||||
try {
|
||||
entries = stringRedisTemplate.opsForHash().entries(KEY_PREFIX + taskId);
|
||||
} catch (Exception ex) {
|
||||
log.warn("[heartbeat-position] 读取位置失败 taskId={} msg={}", taskId, ex.getMessage());
|
||||
return null;
|
||||
}
|
||||
if (entries == null || entries.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
String text = value(entries, "progressText");
|
||||
if (!text.isEmpty()) {
|
||||
return text;
|
||||
}
|
||||
String phase = value(entries, "phase");
|
||||
String current = value(entries, "current");
|
||||
String total = value(entries, "total");
|
||||
StringBuilder sb = new StringBuilder();
|
||||
if (!phase.isEmpty()) {
|
||||
sb.append("阶段=").append(phase);
|
||||
}
|
||||
if (!current.isEmpty() || !total.isEmpty()) {
|
||||
if (sb.length() > 0) {
|
||||
sb.append("，");
|
||||
}
|
||||
sb.append(current.isEmpty() ? "0" : current).append("/").append(total.isEmpty() ? "0" : total);
|
||||
}
|
||||
return sb.length() == 0 ? null : sb.toString();
|
||||
}
|
||||
|
||||
private String value(Map<Object, Object> entries, String key) {
|
||||
Object raw = entries.get(key);
|
||||
return raw == null ? "" : String.valueOf(raw).trim();
|
||||
}
|
||||
}
|
||||
+10
-7
@@ -46,10 +46,12 @@ public class TutorialAdminController {
|
||||
@Operation(summary = "签发教程包直传预签名", description = "返回浏览器直传 MinIO client 桶的 PUT 预签名地址；直传完成后调用 /tutorial/confirm 校验落库")
|
||||
public ApiResponse<Map<String, Object>> presignTutorial(
|
||||
HttpServletRequest request,
|
||||
@Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName) {
|
||||
@Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName,
|
||||
@Parameter(description = "教程包版本号（展示与排序用，可空）") @RequestParam(value = "version", required = false) String version) {
|
||||
AdminUserEntity operator = adminAuthSupport.requireAdminOrInternal(request);
|
||||
log.info("[tutorial] 管理端签发教程包直传预签名 operator={} fileName={}", operator.getUsername(), fileName);
|
||||
return ApiResponse.success(tutorialPackageService.presignTutorialPackage(fileName));
|
||||
log.info("[tutorial] 管理端签发教程包直传预签名 operator={} fileName={} version={}",
|
||||
operator.getUsername(), fileName, version);
|
||||
return ApiResponse.success(tutorialPackageService.presignTutorialPackage(fileName, version));
|
||||
}
|
||||
|
||||
@PostMapping("/tutorial/confirm")
|
||||
@@ -57,11 +59,12 @@ public class TutorialAdminController {
|
||||
public ApiResponse<Map<String, Object>> confirmTutorial(
|
||||
HttpServletRequest request,
|
||||
@Parameter(description = "预签名返回的对象 key") @RequestParam(value = "object_key", required = false) String objectKey,
|
||||
@Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName) {
|
||||
@Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName,
|
||||
@Parameter(description = "教程包版本号（展示与排序用，可空）") @RequestParam(value = "version", required = false) String version) {
|
||||
AdminUserEntity operator = adminAuthSupport.requireAdminOrInternal(request);
|
||||
log.info("[tutorial] 管理端确认教程包直传完成 operator={} objectKey={} fileName={}",
|
||||
operator.getUsername(), objectKey, fileName);
|
||||
return ApiResponse.success("上传成功", Map.of("item", tutorialPackageService.confirmTutorialPackage(objectKey, fileName)));
|
||||
log.info("[tutorial] 管理端确认教程包直传完成 operator={} objectKey={} fileName={} version={}",
|
||||
operator.getUsername(), objectKey, fileName, version);
|
||||
return ApiResponse.success("上传成功", Map.of("item", tutorialPackageService.confirmTutorialPackage(objectKey, fileName, version)));
|
||||
}
|
||||
|
||||
@PostMapping("/tutorial/delete")
|
||||
|
||||
+3
@@ -22,6 +22,9 @@ public class TutorialPackageEntity {
|
||||
/** 原始文件名（展示用） */
|
||||
private String fileName;
|
||||
|
||||
/** 教程包版本号（上传时填写；V126 之前的历史行可为空） */
|
||||
private String version;
|
||||
|
||||
/** MinIO 对象 key（client 桶，tutorial/ 前缀） */
|
||||
private String objectKey;
|
||||
|
||||
|
||||
+34
-8
@@ -37,6 +37,9 @@ public class TutorialPackageService {
|
||||
/** 教程包与软件版本包同量级，沿用 512MB 上限 */
|
||||
private static final long MAX_UPLOAD_BYTES = 512L * 1024 * 1024;
|
||||
|
||||
/** 版本号长度上限（与 biz_tutorial_package.version VARCHAR(64) 对齐） */
|
||||
private static final int MAX_VERSION_LENGTH = 64;
|
||||
|
||||
/** 直传预签名有效期：3 分钟足够数百 MB 级压缩包上传 */
|
||||
private static final int PRESIGN_EXPIRY_SECONDS = 180;
|
||||
|
||||
@@ -76,15 +79,17 @@ public class TutorialPackageService {
|
||||
if (entity == null) {
|
||||
log.info("[tutorial] 查询最新教程包：无记录，前端将回退固定直链");
|
||||
result.put("file_name", null);
|
||||
result.put("version", null);
|
||||
result.put("file_url", null);
|
||||
result.put("file_size", null);
|
||||
result.put("created_at", null);
|
||||
return result;
|
||||
}
|
||||
Map<String, Object> item = toItemMap(entity);
|
||||
log.info("[tutorial] 查询最新教程包 id={} fileName={} objectKey={}",
|
||||
entity.getId(), entity.getFileName(), entity.getObjectKey());
|
||||
log.info("[tutorial] 查询最新教程包 id={} fileName={} version={} objectKey={}",
|
||||
entity.getId(), entity.getFileName(), entity.getVersion(), entity.getObjectKey());
|
||||
result.put("file_name", item.get("file_name"));
|
||||
result.put("version", item.get("version"));
|
||||
result.put("file_url", item.get("file_url"));
|
||||
result.put("file_size", item.get("file_size"));
|
||||
result.put("created_at", item.get("created_at"));
|
||||
@@ -94,27 +99,32 @@ public class TutorialPackageService {
|
||||
/**
|
||||
* 签发教程包直传预签名：按 时间戳-原文件名 生成新对象 key（每次上传互不覆盖），
|
||||
* 浏览器直传到 MinIO 后再调 {@link #confirmTutorialPackage} 由服务端核对落库。
|
||||
* <p>版本号为上传时填写的展示信息，一并回传给前端在 confirm 阶段落库。
|
||||
*/
|
||||
public Map<String, Object> presignTutorialPackage(String fileName) {
|
||||
public Map<String, Object> presignTutorialPackage(String fileName, String version) {
|
||||
String normalizedName = requireZipFileName(fileName);
|
||||
String normalizedVersion = normalizeVersion(version);
|
||||
String objectKey = OBJECT_KEY_PREFIX + OBJECT_NAME_FORMATTER.format(LocalDateTime.now())
|
||||
+ "-" + safeFileName(normalizedName);
|
||||
log.info("[tutorial] 签发教程包直传预签名 fileName={} objectKey={} expirySeconds={}",
|
||||
normalizedName, objectKey, PRESIGN_EXPIRY_SECONDS);
|
||||
log.info("[tutorial] 签发教程包直传预签名 fileName={} version={} objectKey={} expirySeconds={}",
|
||||
normalizedName, normalizedVersion, objectKey, PRESIGN_EXPIRY_SECONDS);
|
||||
String uploadUrl = ossStorageService.presignTutorialUpload(objectKey, PRESIGN_EXPIRY_SECONDS);
|
||||
Map<String, Object> result = new LinkedHashMap<>();
|
||||
result.put("object_key", objectKey);
|
||||
result.put("upload_url", uploadUrl);
|
||||
result.put("file_url", ossStorageService.getTutorialDownloadUrl(objectKey));
|
||||
result.put("version", normalizedVersion);
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 直传完成后确认落库：校验对象存在且不超限，新 key 不与其它记录冲突，写入记录并返回新行。 */
|
||||
public Map<String, Object> confirmTutorialPackage(String objectKey, String fileName) {
|
||||
public Map<String, Object> confirmTutorialPackage(String objectKey, String fileName, String version) {
|
||||
if (objectKey == null || !objectKey.startsWith(OBJECT_KEY_PREFIX)) {
|
||||
log.warn("[tutorial] 直传确认失败：非法对象路径 objectKey={}", objectKey);
|
||||
throw new BusinessException("非法的对象路径，请重新上传");
|
||||
}
|
||||
String normalizedName = requireZipFileName(fileName);
|
||||
String normalizedVersion = normalizeVersion(version);
|
||||
long size = ossStorageService.tutorialObjectSize(objectKey);
|
||||
if (size < 0) {
|
||||
log.warn("[tutorial] 直传确认失败：对象不存在 objectKey={}", objectKey);
|
||||
@@ -134,11 +144,12 @@ public class TutorialPackageService {
|
||||
|
||||
TutorialPackageEntity entity = new TutorialPackageEntity();
|
||||
entity.setFileName(normalizedName);
|
||||
entity.setVersion(normalizedVersion);
|
||||
entity.setObjectKey(objectKey);
|
||||
entity.setFileSize(size);
|
||||
tutorialPackageMapper.insert(entity);
|
||||
log.info("[tutorial] 教程包登记成功 id={} fileName={} objectKey={} bytes={}",
|
||||
entity.getId(), normalizedName, objectKey, size);
|
||||
log.info("[tutorial] 教程包登记成功 id={} fileName={} version={} objectKey={} bytes={}",
|
||||
entity.getId(), normalizedName, normalizedVersion, objectKey, size);
|
||||
TutorialPackageEntity saved = tutorialPackageMapper.selectById(entity.getId());
|
||||
return toItemMap(saved == null ? entity : saved);
|
||||
}
|
||||
@@ -201,6 +212,20 @@ public class TutorialPackageService {
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/**
|
||||
* 版本号归一：去空白后作为展示字段入库，允许为空（老客户端/历史行兼容）。
|
||||
* <p>与软件版本号不同，教程包版本号只是展示与排序用的标签，不参与对象 key，
|
||||
* 因此超长时截断而不是直接拒绝上传。
|
||||
*/
|
||||
private String normalizeVersion(String version) {
|
||||
String normalized = version == null ? "" : version.trim();
|
||||
if (normalized.length() > MAX_VERSION_LENGTH) {
|
||||
log.warn("[tutorial] 版本号超长已截断 length={} limit={}", normalized.length(), MAX_VERSION_LENGTH);
|
||||
return normalized.substring(0, MAX_VERSION_LENGTH);
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
/** 文件名 → 安全对象名片段：去掉路径分隔，只保留中文/字母/数字/._-，空则兜底 tutorial.zip。 */
|
||||
private String safeFileName(String fileName) {
|
||||
String base = fileName.replace('\\', '/');
|
||||
@@ -216,6 +241,7 @@ public class TutorialPackageService {
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("id", entity.getId());
|
||||
item.put("file_name", entity.getFileName() == null ? "" : entity.getFileName());
|
||||
item.put("version", entity.getVersion() == null ? "" : entity.getVersion());
|
||||
item.put("object_key", entity.getObjectKey() == null ? "" : entity.getObjectKey());
|
||||
item.put("file_size", entity.getFileSize() == null ? 0L : entity.getFileSize());
|
||||
item.put("file_url", entity.getObjectKey() == null || entity.getObjectKey().isBlank()
|
||||
|
||||
@@ -66,7 +66,7 @@ AIIMAGE_APPEARANCE_PATENT_TITLE_MODEL=gemini-3.8-flash
|
||||
AIIMAGE_APPEARANCE_PATENT_APPEARANCE_MODEL=gemini-3.8-flash
|
||||
AIIMAGE_APPEARANCE_PATENT_LLM_MAX_TOKENS=64000
|
||||
AIIMAGE_APPEARANCE_PATENT_LLM_BATCH_SIZE=10
|
||||
AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY=10
|
||||
AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY=5
|
||||
AIIMAGE_APPEARANCE_PATENT_LLM_RETRY_TIMES=3
|
||||
AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES=30
|
||||
|
||||
|
||||
@@ -186,6 +186,7 @@ aiimage:
|
||||
failed-ttl-hours: ${AIIMAGE_BRAND_PROGRESS_FAILED_TTL_HOURS:2}
|
||||
heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:30}
|
||||
stale-check-cron: ${AIIMAGE_BRAND_PROGRESS_STALE_CHECK_CRON:0 */2 * * * *}
|
||||
no-result-upload-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180}
|
||||
delete-brand-progress:
|
||||
heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:30}
|
||||
stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:*/30 * * * * *}
|
||||
@@ -196,6 +197,8 @@ aiimage:
|
||||
patrol-delete-stale-timeout-minutes: ${AIIMAGE_PATROL_DELETE_STALE_TIMEOUT_MINUTES:30}
|
||||
query-asin-stale-timeout-minutes: ${AIIMAGE_QUERY_ASIN_STALE_TIMEOUT_MINUTES:30}
|
||||
withdraw-stale-timeout-minutes: ${AIIMAGE_WITHDRAW_STALE_TIMEOUT_MINUTES:30}
|
||||
no-result-upload-timeout-minutes: ${AIIMAGE_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180}
|
||||
no-result-upload-check-enabled: ${AIIMAGE_NO_RESULT_UPLOAD_CHECK_ENABLED:true}
|
||||
module-cleanup:
|
||||
enabled: ${AIIMAGE_MODULE_CLEANUP_ENABLED:true}
|
||||
cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *}
|
||||
@@ -249,11 +252,13 @@ aiimage:
|
||||
path: ${AIIMAGE_BRAND_CHECK_PATH:/brand_check}
|
||||
token: ${AIIMAGE_BRAND_CHECK_TOKEN:}
|
||||
default-strategy: ${AIIMAGE_BRAND_CHECK_DEFAULT_STRATEGY:Terms}
|
||||
retry-times: ${AIIMAGE_BRAND_CHECK_RETRY_TIMES:3}
|
||||
retry-times: ${AIIMAGE_BRAND_CHECK_RETRY_TIMES:10}
|
||||
retry-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_INTERVAL_MILLIS:1000}
|
||||
retry-max-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_MAX_INTERVAL_MILLIS:10000}
|
||||
connect-timeout-millis: ${AIIMAGE_BRAND_CHECK_CONNECT_TIMEOUT_MILLIS:10000}
|
||||
read-timeout-millis: ${AIIMAGE_BRAND_CHECK_READ_TIMEOUT_MILLIS:60000}
|
||||
appearance-patent:
|
||||
no-result-upload-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180}
|
||||
llm-host: ${AIIMAGE_APPEARANCE_PATENT_LLM_HOST:https://ai.t8star.org}
|
||||
title-model: ${AIIMAGE_APPEARANCE_PATENT_TITLE_MODEL:gemini-3.8-flash}
|
||||
appearance-model: ${AIIMAGE_APPEARANCE_PATENT_APPEARANCE_MODEL:gemini-3.8-flash}
|
||||
@@ -262,7 +267,10 @@ aiimage:
|
||||
llm-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_LLM_READ_TIMEOUT_MILLIS:180000}
|
||||
llm-first-attempt-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_LLM_FIRST_ATTEMPT_READ_TIMEOUT_MILLIS:60000}
|
||||
llm-batch-size: ${AIIMAGE_APPEARANCE_PATENT_LLM_BATCH_SIZE:10}
|
||||
llm-row-concurrency: ${AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY:10}
|
||||
# 批内行并发：每行会串行发 2 个 LLM 请求（标题提取 + 外观检测），且批次串行提交，
|
||||
# 故该值≈单任务对 LLM 网关的瞬时并发。原 10 与品牌检测的 8 同一量级，
|
||||
# 多任务并行时会成倍放大、打满用户自己的网关密钥配额（2026-09-14 随品牌一起下调）。
|
||||
llm-row-concurrency: ${AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY:5}
|
||||
max-parse-rows: ${AIIMAGE_APPEARANCE_PATENT_MAX_PARSE_ROWS:50000}
|
||||
llm-retry-times: ${AIIMAGE_APPEARANCE_PATENT_LLM_RETRY_TIMES:3}
|
||||
flush-pending-minutes: ${AIIMAGE_APPEARANCE_PATENT_FLUSH_PENDING_MINUTES:${AIIMAGE_APPEARANCE_PATENT_COZE_FLUSH_PENDING_MINUTES:1}}
|
||||
@@ -274,6 +282,7 @@ aiimage:
|
||||
db-task-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_TASK_TOUCH_INTERVAL_MILLIS:120000}
|
||||
db-job-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_JOB_TOUCH_INTERVAL_MILLIS:60000}
|
||||
llm-flush-pending-minutes: ${AIIMAGE_SIMILAR_ASIN_LLM_FLUSH_PENDING_MINUTES:1}
|
||||
no-result-upload-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180}
|
||||
image-download-pool-size: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE:2}
|
||||
image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:5}
|
||||
image-prefetch-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_PREFETCH_TIMEOUT_SECONDS:1800}
|
||||
@@ -299,6 +308,7 @@ aiimage:
|
||||
llm-image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_LLM_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:10}
|
||||
collect-data:
|
||||
stale-timeout-minutes: ${AIIMAGE_COLLECT_DATA_STALE_TIMEOUT_MINUTES:30}
|
||||
no-result-upload-timeout-minutes: ${AIIMAGE_COLLECT_DATA_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180}
|
||||
stale-check-cron: ${AIIMAGE_COLLECT_DATA_STALE_CHECK_CRON:*/30 * * * * *}
|
||||
max-source-file-bytes: ${AIIMAGE_COLLECT_DATA_MAX_SOURCE_FILE_BYTES:0}
|
||||
max-parse-rows: ${AIIMAGE_COLLECT_DATA_MAX_PARSE_ROWS:0}
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
-- V126: biz_tutorial_package 增加 version 列（后台「教程管理」展示与上传时填写的版本号）
|
||||
--
|
||||
-- 背景：教程包后台列表原先只有文件名/大小/上传时间，无法区分同名的多轮教程包；
|
||||
-- 上传时补填版本号（与软件版本管理页同交互），列表新增「版本号」列并支持排序。
|
||||
--
|
||||
-- 历史行无版本号：列可空，Java 侧读取为 null 时展示为空（历史种子行不受影响）。
|
||||
-- 幂等：ADD COLUMN 前先查 information_schema，重复执行安全。
|
||||
-- 回滚：ALTER TABLE biz_tutorial_package DROP COLUMN version;
|
||||
|
||||
SET @db_name = DATABASE();
|
||||
|
||||
SET @col_exists := (
|
||||
SELECT COUNT(*) FROM information_schema.COLUMNS
|
||||
WHERE TABLE_SCHEMA = @db_name AND TABLE_NAME = 'biz_tutorial_package' AND COLUMN_NAME = 'version'
|
||||
);
|
||||
SET @sql := IF(@col_exists = 0,
|
||||
'ALTER TABLE biz_tutorial_package ADD COLUMN version VARCHAR(64) NULL COMMENT ''教程包版本号（历史行可为空）'' AFTER file_name',
|
||||
'SELECT 1'
|
||||
);
|
||||
PREPARE stmt FROM @sql;
|
||||
EXECUTE stmt;
|
||||
DEALLOCATE PREPARE stmt;
|
||||
+143
@@ -0,0 +1,143 @@
|
||||
package com.nanri.aiimage.modules.collectdata.service;
|
||||
|
||||
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
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.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.Spy;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
|
||||
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.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* collect-data 二次判死线（心跳正常但连续 N 分钟无结果分片上报）。
|
||||
*
|
||||
* <p>既有心跳线候选条件是 updated_at 陈旧（心跳线），而 Python 心跳线程在主线程卡死时照发——
|
||||
* 任务永远命不中。本线候选取「心跳新鲜 + 创建超过 N 分钟」，判据用
|
||||
* biz_task_scope_state.last_chunk_at（仅分片上传时刷新）；从未上报（无 scope 行）跳过。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class CollectDataNoUploadStaleTest {
|
||||
|
||||
@Mock private FileTaskMapper fileTaskMapper;
|
||||
@Mock private FileResultMapper fileResultMapper;
|
||||
@Mock private TaskChunkMapper taskChunkMapper;
|
||||
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||
@Mock private TaskFileJobService taskFileJobService;
|
||||
@Mock private TaskDistributedLockService.LockHandle lockHandle;
|
||||
|
||||
@Spy
|
||||
private com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper();
|
||||
|
||||
@InjectMocks
|
||||
private CollectDataService service;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L);
|
||||
ReflectionTestUtils.setField(service, "noResultUploadTimeoutMinutes", 180L);
|
||||
ReflectionTestUtils.setField(service, "progressThrottleMillis", 0L);
|
||||
ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 0L);
|
||||
}
|
||||
|
||||
private FileTaskEntity runningTask(long id) {
|
||||
FileTaskEntity task = new FileTaskEntity();
|
||||
task.setId(id);
|
||||
task.setModuleType("COLLECT_DATA");
|
||||
task.setStatus("RUNNING");
|
||||
task.setCreatedAt(LocalDateTime.now().minusDays(1));
|
||||
task.setUpdatedAt(LocalDateTime.now().minusMinutes(1));
|
||||
return task;
|
||||
}
|
||||
|
||||
private void invokeNoUpload() {
|
||||
ReflectionTestUtils.invokeMethod(service, "finalizeNoUploadStaleTasks",
|
||||
LocalDateTime.now().minusMinutes(30));
|
||||
}
|
||||
|
||||
@Test
|
||||
void 上报仍新鲜时不判死() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(1L)));
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(1L, LocalDateTime.now().minusMinutes(5))));
|
||||
|
||||
invokeNoUpload();
|
||||
|
||||
verify(taskDistributedLockService, never()).acquire(any(), anyLong(), anyLong());
|
||||
verify(fileTaskMapper, never()).updateById(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void 从未上报时跳过() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(2L)));
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())).thenReturn(List.of());
|
||||
|
||||
invokeNoUpload();
|
||||
|
||||
verify(taskDistributedLockService, never()).acquire(any(), anyLong(), anyLong());
|
||||
verify(fileTaskMapper, never()).updateById(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void 开关为0时不扫描() {
|
||||
ReflectionTestUtils.setField(service, "noResultUploadTimeoutMinutes", 0L);
|
||||
|
||||
invokeNoUpload();
|
||||
|
||||
verify(fileTaskMapper, never()).selectList(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void 任务锁忙时跳过() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(3L)));
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(3L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskDistributedLockService.acquire(any(), anyLong(), anyLong())).thenReturn(null);
|
||||
|
||||
invokeNoUpload();
|
||||
|
||||
verify(fileTaskMapper, never()).updateById(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void 心跳正常但上报超时触发判死() {
|
||||
FileTaskEntity task = runningTask(5L);
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(task));
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(5L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskDistributedLockService.acquire(any(), anyLong(), anyLong())).thenReturn(lockHandle);
|
||||
when(fileTaskMapper.selectById(5L)).thenReturn(task);
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(5L, "COLLECT_DATA")).thenReturn(0L);
|
||||
when(taskChunkMapper.selectCount(any())).thenReturn(0L);
|
||||
when(fileResultMapper.selectOne(any())).thenReturn(null);
|
||||
|
||||
invokeNoUpload();
|
||||
|
||||
ArgumentCaptor<FileTaskEntity> taskCaptor = ArgumentCaptor.forClass(FileTaskEntity.class);
|
||||
verify(fileTaskMapper).updateById(taskCaptor.capture());
|
||||
assertEquals("FAILED", taskCaptor.getValue().getStatus());
|
||||
assertTrue(taskCaptor.getValue().getErrorMessage().contains("无结果回传"),
|
||||
taskCaptor.getValue().getErrorMessage());
|
||||
}
|
||||
}
|
||||
+184
-2
@@ -5,10 +5,15 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
|
||||
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||
import com.nanri.aiimage.config.DeleteBrandProgressProperties;
|
||||
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskCacheService;
|
||||
import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService;
|
||||
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto;
|
||||
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 org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -50,6 +55,10 @@ class DeleteBrandStaleTaskServiceTest {
|
||||
@Mock private DeleteBrandProgressProperties deleteBrandProgressProperties;
|
||||
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||
@Mock private TaskFileJobService taskFileJobService;
|
||||
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||
@Mock private ProductRiskTaskService productRiskTaskService;
|
||||
@Mock private ProductRiskTaskCacheService productRiskTaskCacheService;
|
||||
@Mock private TaskHeartbeatPositionService taskHeartbeatPositionService;
|
||||
|
||||
@BeforeAll
|
||||
static void initializeTableInfo() {
|
||||
@@ -212,11 +221,184 @@ class DeleteBrandStaleTaskServiceTest {
|
||||
assertTrue(errorMessage.contains("缺失分片"), errorMessage);
|
||||
}
|
||||
|
||||
// ---------- 二次判死线：心跳正常但连续 N 分钟无结果分片上报 ----------
|
||||
|
||||
@Test
|
||||
void noUploadStaleTaskIsFailedWithCasAndCacheEvicted() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(901L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(901L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(901L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L);
|
||||
noUploadTaskLockAvailable();
|
||||
when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(productRiskTaskService).tryFinalizeTask(901L, true);
|
||||
verify(productRiskTaskCacheService).deleteTaskCache(901L);
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
ArgumentCaptor<LambdaUpdateWrapper> update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
|
||||
verify(fileTaskMapper).update(isNull(), update.capture());
|
||||
List<String> values = update.getValue().getParamNameValuePairs().values().stream()
|
||||
.map(String::valueOf)
|
||||
.toList();
|
||||
assertTrue(values.stream().anyMatch(v -> v.contains("无结果回传")), values.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadFreshResultIsSkipped() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(902L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(902L, LocalDateTime.now().minusMinutes(5))));
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(taskDistributedLockService, never()).acquire(any(), any(), anyLong());
|
||||
verify(fileTaskMapper, never()).update(any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadNeverUploadedIsSkipped() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(903L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())).thenReturn(List.of());
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(taskDistributedLockService, never()).acquire(any(), any(), anyLong());
|
||||
verify(fileTaskMapper, never()).update(any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadAlreadyFinalizedTaskKeepsCache() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(904L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(904L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(904L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L);
|
||||
noUploadTaskLockAvailable();
|
||||
// CAS 未命中：finalize 已把任务终结（状态不再是 RUNNING）
|
||||
when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(0);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(productRiskTaskCacheService, never()).deleteTaskCache(anyLong());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadBusyTaskLockIsSkipped() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(905L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(905L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(905L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L);
|
||||
when(taskDistributedLockService.acquire(any(), any(), anyLong())).thenReturn(null);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(productRiskTaskService, never()).tryFinalizeTask(anyLong(), anyBoolean());
|
||||
verify(fileTaskMapper, never()).update(any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadPendingAssembleJobsIsSkipped() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(906L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(906L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(906L, "PRODUCT_RISK_RESOLVE")).thenReturn(1L);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(productRiskTaskService, never()).tryFinalizeTask(anyLong(), anyBoolean());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadDisabledByConfigSkipsScan() {
|
||||
when(deleteBrandProgressProperties.isNoResultUploadCheckEnabled()).thenReturn(false);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(fileTaskMapper, never()).selectList(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadDeleteBrandDispatchesToRunService() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(907L, "DELETE_BRAND")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(907L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(907L, "DELETE_BRAND")).thenReturn(0L);
|
||||
noUploadTaskLockAvailable();
|
||||
when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
verify(deleteBrandRunService).tryFinalizeTask(907L, true);
|
||||
verify(deleteBrandTaskCacheService).delete(907L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void noUploadFailReasonIncludesLastPosition() {
|
||||
when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(908L, "PRODUCT_RISK_RESOLVE")));
|
||||
noUploadEnabled();
|
||||
when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any()))
|
||||
.thenReturn(List.of(new TaskScopeLastChunkDto(908L, LocalDateTime.now().minusHours(4))));
|
||||
when(taskFileJobService.countUnfinishedAssembleJobs(908L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L);
|
||||
noUploadTaskLockAvailable();
|
||||
when(taskHeartbeatPositionService.describe(908L)).thenReturn("店铺 魏振峰(2/10)，国家 德国(1/3)");
|
||||
when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1);
|
||||
|
||||
DeleteBrandStaleTaskService service = service();
|
||||
ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks");
|
||||
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
ArgumentCaptor<LambdaUpdateWrapper> update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
|
||||
verify(fileTaskMapper).update(isNull(), update.capture());
|
||||
List<String> values = update.getValue().getParamNameValuePairs().values().stream()
|
||||
.map(String::valueOf)
|
||||
.toList();
|
||||
assertTrue(values.stream().anyMatch(v -> v.contains("最后处理位置：店铺 魏振峰(2/10)")), values.toString());
|
||||
}
|
||||
|
||||
private void noUploadEnabled() {
|
||||
when(deleteBrandProgressProperties.isNoResultUploadCheckEnabled()).thenReturn(true);
|
||||
when(deleteBrandProgressProperties.getNoResultUploadTimeoutMinutes()).thenReturn(180L);
|
||||
}
|
||||
|
||||
private void noUploadTaskLockAvailable() {
|
||||
when(taskDistributedLockService.acquire(any(), any(), anyLong()))
|
||||
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||
}
|
||||
|
||||
private static FileTaskEntity noUploadTask(Long id, String moduleType) {
|
||||
FileTaskEntity task = new FileTaskEntity();
|
||||
task.setId(id);
|
||||
task.setModuleType(moduleType);
|
||||
task.setStatus("RUNNING");
|
||||
task.setCreatedAt(LocalDateTime.now().minusDays(1));
|
||||
task.setUpdatedAt(LocalDateTime.now().minusMinutes(1));
|
||||
return task;
|
||||
}
|
||||
|
||||
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, taskFileJobService, null);
|
||||
productRiskTaskService, productRiskTaskCacheService,
|
||||
null, null, null, null, null, null, null, null, null, null,
|
||||
null, null, null,
|
||||
deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService, null,
|
||||
taskScopeStateMapper, taskHeartbeatPositionService);
|
||||
}
|
||||
|
||||
private void lockAvailable() {
|
||||
|
||||
+2
-1
@@ -81,7 +81,8 @@ class TaskModuleCoverageTest {
|
||||
mock(AppearancePatentTaskService.class),
|
||||
mock(SimilarAsinTaskService.class),
|
||||
null, null, null, null,
|
||||
mock(ShopDataCrawlTaskService.class));
|
||||
mock(ShopDataCrawlTaskService.class),
|
||||
null, null);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
package com.nanri.aiimage.modules.task.service;
|
||||
|
||||
import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.data.redis.core.HashOperations;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* 心跳「当前处理位置」存储：写读往返、空位置不写、Redis 异常只降级不抛出。
|
||||
* 位置仅用于判死文案与排查，任何异常都不得影响心跳本身。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class TaskHeartbeatPositionServiceTest {
|
||||
|
||||
@Mock private StringRedisTemplate stringRedisTemplate;
|
||||
@Mock private HashOperations<String, Object, Object> hashOperations;
|
||||
|
||||
private TaskHeartbeatPositionService service() {
|
||||
return new TaskHeartbeatPositionService(stringRedisTemplate);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordWritesHashWithProgressText() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
TaskHeartbeatRequest request = new TaskHeartbeatRequest();
|
||||
request.setPhase("crawling");
|
||||
request.setCurrent(2);
|
||||
request.setTotal(10);
|
||||
request.setProgressText("店铺 魏振峰(2/10)");
|
||||
|
||||
service().record(123L, request);
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
ArgumentCaptor<Map<String, String>> values = ArgumentCaptor.forClass(Map.class);
|
||||
verify(hashOperations).putAll(eq("task:heartbeat:position:123"), values.capture());
|
||||
assertEquals("店铺 魏振峰(2/10)", values.getValue().get("progressText"));
|
||||
assertEquals("crawling", values.getValue().get("phase"));
|
||||
assertEquals("2", values.getValue().get("current"));
|
||||
assertEquals("10", values.getValue().get("total"));
|
||||
verify(stringRedisTemplate).expire(eq("task:heartbeat:position:123"), any(Duration.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordSkipsWhenNoPosition() {
|
||||
TaskHeartbeatRequest request = new TaskHeartbeatRequest();
|
||||
|
||||
service().record(123L, request);
|
||||
|
||||
verify(stringRedisTemplate, never()).opsForHash();
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordSwallowsRedisError() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
doThrow(new RuntimeException("redis down")).when(hashOperations).putAll(any(), any());
|
||||
TaskHeartbeatRequest request = new TaskHeartbeatRequest();
|
||||
request.setProgressText("店铺 A(1/2)");
|
||||
|
||||
service().record(123L, request);
|
||||
|
||||
verify(stringRedisTemplate, never()).expire(any(), any(Duration.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void describePrefersProgressText() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
when(hashOperations.entries("task:heartbeat:position:123"))
|
||||
.thenReturn(Map.of("progressText", "店铺 魏振峰(2/10)", "current", "2", "total", "10"));
|
||||
|
||||
assertEquals("店铺 魏振峰(2/10)", service().describe(123L));
|
||||
}
|
||||
|
||||
@Test
|
||||
void describeBuildsFromPhaseAndCountersWhenNoText() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
when(hashOperations.entries("task:heartbeat:position:123"))
|
||||
.thenReturn(Map.of("phase", "crawling", "current", "2", "total", "10"));
|
||||
|
||||
assertEquals("阶段=crawling，2/10", service().describe(123L));
|
||||
}
|
||||
|
||||
@Test
|
||||
void describeReturnsNullWhenMissing() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
when(hashOperations.entries("task:heartbeat:position:123")).thenReturn(Map.of());
|
||||
|
||||
assertNull(service().describe(123L));
|
||||
}
|
||||
|
||||
@Test
|
||||
void describeReturnsNullOnRedisError() {
|
||||
when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations);
|
||||
when(hashOperations.entries("task:heartbeat:position:123"))
|
||||
.thenThrow(new RuntimeException("redis down"));
|
||||
|
||||
assertNull(service().describe(123L));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user