diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskPullSpiImpl.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskPullSpiImpl.java new file mode 100644 index 00000000..5a022f02 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskPullSpiImpl.java @@ -0,0 +1,60 @@ +package com.nanri.aiimage.modules.appearancepatent.service; + +import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentParsedPayloadDto; +import com.nanri.aiimage.modules.appearancepatent.model.vo.AppearancePatentParsedGroupVo; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * 外观专利的客户端兜底拉取实现。 + * + *

Python 消费端需要 groups（解析分组，页面上是现拉 /queue-payload 再入队）， + * 这里直接复用同一个 service 方法；payload 与页面保持一致（含 prompt / api_key）。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class AppearancePatentTaskPullSpiImpl implements ClientTaskPullSpi { + + private static final String QUEUE_TYPE = "appearance-patent-run"; + + private final AppearancePatentTaskService taskService; + private final AppearancePatentTaskCacheService taskCacheService; + + @Override + public String moduleType() { + return AppearancePatentTaskService.MODULE_TYPE; + } + + @Override + public Map buildQueuePayload(FileTaskEntity task) { + AppearancePatentParsedPayloadDto payload = taskService.queuePayload(task.getId(), task.getUserId()); + List groups = payload.getGroups() == null ? List.of() : payload.getGroups(); + if (groups.isEmpty()) { + // 空 groups 在 Python 侧会被静默跳过（"groups/rows is empty, skip"），宁可在服务端直接判失败 + log.warn("[appearance-patent] 兜底拉取失败：解析分组为空 taskId={}", task.getId()); + return null; + } + Map data = new LinkedHashMap<>(); + data.put("taskId", task.getId()); + data.put("user_id", task.getUserId()); + data.put("prompt", payload.getAiPrompt()); + data.put("api_key", payload.getApiKey()); + data.put("groups", groups); + log.info("[appearance-patent] 兜底载荷已组装 taskId={} groups={}", task.getId(), groups.size()); + return Map.of("type", QUEUE_TYPE, "data", data); + } + + @Override + public void onClaimed(FileTaskEntity task) { + // 对齐 activate：刷新模块缓存心跳，让页面立刻看到 RUNNING + taskCacheService.touchTaskHeartbeat(task.getId()); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index 2b42b65d..7675e268 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -280,12 +280,20 @@ public class AppearancePatentTaskService { throw new BusinessException("任务不存在"); } ensureTaskOwnedByCurrentInstance(task, "activate"); - if (STATUS_SUCCESS.equals(task.getStatus()) || STATUS_FAILED.equals(task.getStatus())) { + // 只允许 PENDING→RUNNING（条件更新）：与客户端「兜底拉取」的原子认领互斥， + // 谁先翻转谁执行，避免页面与客户端重复执行同一任务 + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, taskId) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) + .set(FileTaskEntity::getStatus, STATUS_RUNNING) + .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())); + if (updated == 0) { + FileTaskEntity latest = fileTaskMapper.selectById(taskId); + if (latest != null && STATUS_RUNNING.equals(latest.getStatus())) { + throw new BusinessException("任务已在执行中（可能已由客户端自动接管），无需重复启动"); + } throw new BusinessException("任务已结束"); } - task.setStatus(STATUS_RUNNING); - task.setUpdatedAt(LocalDateTime.now()); - fileTaskMapper.updateById(task); taskCacheService.touchTaskHeartbeat(taskId); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 49e1e70c..c79e3e30 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -381,14 +381,18 @@ public class CollectDataService { @Transactional public void activateTask(Long taskId, Long userId) { FileTaskEntity task = requireTask(taskId, userId); - // 条件更新：上面的「已结束」判断与写入之间存在窗口（TOCTOU），期间 /fail 可能已把任务 - // 标为 FAILED —— 整行 updateById 会把它复活成 RUNNING（前端显示"执行中"但无人推进） + // 只允许 PENDING→RUNNING（条件更新）：既堵住 TOCTOU（/fail 抢先标 FAILED 后被整行 + // updateById 复活成 RUNNING），又与客户端「兜底拉取」的原子认领互斥，谁先翻转谁执行 int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() .eq(FileTaskEntity::getId, task.getId()) - .notIn(FileTaskEntity::getStatus, STATUS_SUCCESS, STATUS_FAILED) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) .set(FileTaskEntity::getStatus, STATUS_RUNNING) .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())); if (updated == 0) { + FileTaskEntity latest = fileTaskMapper.selectById(task.getId()); + if (latest != null && STATUS_RUNNING.equals(latest.getStatus())) { + throw new BusinessException("任务已在执行中（可能已由客户端自动接管），无需重复启动"); + } throw new BusinessException("任务已结束"); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImpl.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImpl.java new file mode 100644 index 00000000..4c92f432 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImpl.java @@ -0,0 +1,71 @@ +package com.nanri.aiimage.modules.collectdata.service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * 集采（collect-data）的客户端兜底拉取实现。 + * + *

Python 消费端需要 taskId / totalRows / pageSize / filters（明细行自行按 /items 分页拉取）。 + * filters 取自任务行 request_json 里 parse 时落库的那份（CollectDataService#persistParsedTask）， + * 序列化后就是 Python 读取的 camelCase 键（countryCode/minAmount/...）。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class CollectDataTaskPullSpiImpl implements ClientTaskPullSpi { + + private static final String QUEUE_TYPE = "collect-data-run"; + private static final String TASK_TYPE = "collect-data"; + + private final ObjectMapper objectMapper; + + @Override + public String moduleType() { + return CollectDataService.MODULE_TYPE; + } + + @Override + public Map buildQueuePayload(FileTaskEntity task) { + JsonNode request = parseJson(task.getRequestJson()); + if (request == null) { + log.warn("[collect-data] 兜底拉取失败：任务请求参数缺失或不可解析 taskId={}", task.getId()); + return null; + } + JsonNode filtersNode = request.get("filters"); + Map filters = filtersNode == null || filtersNode.isNull() + ? Map.of() + : objectMapper.convertValue(filtersNode, Map.class); + JsonNode stats = parseJson(task.getResultJson()); + int totalRows = stats == null ? 0 : stats.path("totalRows").asInt(0); + Map data = new LinkedHashMap<>(); + data.put("taskId", task.getId()); + data.put("taskNo", task.getTaskNo()); + data.put("taskType", TASK_TYPE); + data.put("totalRows", totalRows); + data.put("pageSize", CollectDataService.DEFAULT_PAGE_SIZE); + data.put("filters", filters); + log.info("[collect-data] 兜底载荷已组装 taskId={} totalRows={} filters={}", task.getId(), totalRows, filters); + return Map.of("type", QUEUE_TYPE, "data", data); + } + + private JsonNode parseJson(String json) { + if (json == null || json.isBlank()) { + return null; + } + try { + return objectMapper.readTree(json); + } catch (Exception ex) { + log.warn("[collect-data] 兜底拉取解析任务 JSON 失败 err={}", ex.getMessage()); + return null; + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskPullSpiImpl.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskPullSpiImpl.java new file mode 100644 index 00000000..a80ed058 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskPullSpiImpl.java @@ -0,0 +1,47 @@ +package com.nanri.aiimage.modules.similarasin.service; + +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * 相似 ASIN 的客户端兜底拉取实现。 + * + *

Python 消费端只需 taskId 即可自行回拉解析载荷（parsed-payload）与本地 aliprice 账号/代理， + * 因此这里只补 user_id（价格回调需要）—— 与页面 payload 相比刻意不带 counts/api_key， + * 避免与 Python 侧已有的拉取、服务端密钥兜底逻辑重复。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class SimilarAsinTaskPullSpiImpl implements ClientTaskPullSpi { + + private static final String QUEUE_TYPE = "similar-asin-run"; + + private final SimilarAsinTaskCacheService taskCacheService; + + @Override + public String moduleType() { + return SimilarAsinTaskService.MODULE_TYPE; + } + + @Override + public Map buildQueuePayload(FileTaskEntity task) { + Map data = new LinkedHashMap<>(); + data.put("taskId", task.getId()); + data.put("user_id", task.getUserId()); + log.info("[similar-asin] 兜底载荷已组装 taskId={} userId={}", task.getId(), task.getUserId()); + return Map.of("type", QUEUE_TYPE, "data", data); + } + + @Override + public void onClaimed(FileTaskEntity task) { + // 对齐 activate：刷新模块缓存心跳，让页面立刻看到 RUNNING + taskCacheService.touchTaskHeartbeat(task.getId()); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 6d646c61..6a0312ca 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -532,12 +532,20 @@ public class SimilarAsinTaskService implements SimilarAsinPipelineHost { throw new BusinessException("任务不存在"); } ownershipSupport().ensureTaskOwnedByCurrentInstance(task, "activate"); - if (STATUS_SUCCESS.equals(task.getStatus()) || STATUS_FAILED.equals(task.getStatus())) { + // 只允许 PENDING→RUNNING（条件更新）：与客户端「兜底拉取」的原子认领互斥， + // 谁先翻转谁执行，避免页面与客户端重复执行同一任务 + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, taskId) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) + .set(FileTaskEntity::getStatus, STATUS_RUNNING) + .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now())); + if (updated == 0) { + FileTaskEntity latest = fileTaskMapper.selectById(taskId); + if (latest != null && STATUS_RUNNING.equals(latest.getStatus())) { + throw new BusinessException("任务已在执行中（可能已由客户端自动接管），无需重复启动"); + } throw new BusinessException("任务已结束"); } - task.setStatus(STATUS_RUNNING); - task.setUpdatedAt(LocalDateTime.now()); - fileTaskMapper.updateById(task); taskCacheService.touchTaskHeartbeat(taskId); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskClientPullController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskClientPullController.java new file mode 100644 index 00000000..a6f1bf2b --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskClientPullController.java @@ -0,0 +1,42 @@ +package com.nanri.aiimage.modules.task.controller; + +import com.nanri.aiimage.common.api.ApiResponse; +import com.nanri.aiimage.common.model.entity.AdminUserEntity; +import com.nanri.aiimage.common.security.AdminAuthSupport; +import com.nanri.aiimage.modules.task.model.vo.TaskClientPullVo; +import com.nanri.aiimage.modules.task.service.TaskClientPullService; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.List; + +/** + * 客户端兜底拉取：桌面客户端定时调用，把"页面没推送成功、长期停在 PENDING"的任务领走执行。 + * + *

用户身份一律从 JWT 解析（客户端 4.0.14+ 的请求钩子自动带 Bearer），未登录直接 401 —— + * 不允许匿名按 user_id 参数领别人的任务。 + */ +@RestController +@RequiredArgsConstructor +@RequestMapping("/api/tasks") +@Tag(name = "任务兜底拉取", description = "供桌面客户端领取页面未推送成功的待执行任务") +public class TaskClientPullController { + + private final TaskClientPullService taskClientPullService; + private final AdminAuthSupport adminAuthSupport; + + @GetMapping("/pull-pending") + @Operation( + summary = "拉取并认领长期未推送的任务", + description = "只返回当前登录用户、创建超过 N 分钟仍为 PENDING 的任务；服务端原子认领（PENDING→RUNNING），" + + "与页面推送互斥，不会重复执行。开关关闭时返回空列表。") + public ApiResponse> pullPending(HttpServletRequest request) { + AdminUserEntity me = adminAuthSupport.requireUser(request); + return ApiResponse.success(taskClientPullService.pullPendingTasks(me.getId())); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskClientPullVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskClientPullVo.java new file mode 100644 index 00000000..91dce26d --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskClientPullVo.java @@ -0,0 +1,27 @@ +package com.nanri.aiimage.modules.task.model.vo; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.Map; + +/** 客户端兜底拉取返回的任务项：任务元信息 + 客户端可直接入队的载荷。 */ +@Data +@Schema(description = "客户端兜底拉取的任务项") +public class TaskClientPullVo { + + @Schema(description = "任务 ID", example = "7004") + private Long taskId; + + @Schema(description = "模块类型", example = "SIMILAR_ASIN") + private String moduleType; + + @Schema(description = "任务编号", example = "SIMILAR_ASIN-2063533925058785280") + private String taskNo; + + @Schema(description = "任务创建时间", example = "2026-09-14 10:00:00") + private String createdAt; + + @Schema(description = "客户端可直接入队的载荷：{type, data}") + private Map queuePayload; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairService.java index 0da71dce..3fdae083 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairService.java @@ -55,6 +55,7 @@ public class StaleTaskRepairService { } try (lock) { repairFileTaskStaleIdle(); + repairFileTaskStaleScheduled(); repairFileTaskStaleRunning(); repairBrandStale(); } catch (Exception ex) { @@ -62,12 +63,12 @@ public class StaleTaskRepairService { } } - /** 中间态 PENDING/SCHEDULED 超时未接单 → FAILED。 */ + /** PENDING 长时间未被领取 → FAILED（页面没推进客户端队列，客户端兜底拉取也没领到）。 */ private void repairFileTaskStaleIdle() { LocalDateTime cutoff = LocalDateTime.now().minusMinutes(STALE_IDLE_MINUTES); List stale = fileTaskMapper.selectList(new LambdaQueryWrapper() .select(FileTaskEntity::getId, FileTaskEntity::getModuleType) - .in(FileTaskEntity::getStatus, STATUS_PENDING, STATUS_SCHEDULED) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) .lt(FileTaskEntity::getUpdatedAt, cutoff) .last("limit 500")); if (stale.isEmpty()) { @@ -77,7 +78,7 @@ public class StaleTaskRepairService { LocalDateTime now = LocalDateTime.now(); int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() .in(FileTaskEntity::getId, stale.stream().map(FileTaskEntity::getId).toList()) - .in(FileTaskEntity::getStatus, STATUS_PENDING, STATUS_SCHEDULED) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) .set(FileTaskEntity::getStatus, STATUS_FAILED) .set(FileTaskEntity::getErrorMessage, reason) .set(FileTaskEntity::getFinishedAt, now)); @@ -87,6 +88,42 @@ public class StaleTaskRepairService { } } + /** + * SCHEDULED 到时后长时间未启动 → FAILED。 + * + *

只按 updated_at 判死会误杀：定时任务（店铺匹配）到点前本来就不会有人动它， + * 排期在 STALE_IDLE_MINUTES 之后的任务必被判死，而 activate 又被「未到定时执行时间」挡住。 + * 因此有 scheduled_at 时以「到点时间」为判死基准（到点后再宽限 STALE_IDLE_MINUTES）， + * scheduled_at 为空时退回 updated_at 逻辑。completeStage 会把 scheduled_at 刷成下一轮时间， + * 多轮定时任务不会被上一轮的时间误判。 + */ + private void repairFileTaskStaleScheduled() { + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(STALE_IDLE_MINUTES); + List stale = fileTaskMapper.selectList(new LambdaQueryWrapper() + .select(FileTaskEntity::getId) + .eq(FileTaskEntity::getStatus, STATUS_SCHEDULED) + .and(w -> w.isNull(FileTaskEntity::getScheduledAt) + .lt(FileTaskEntity::getUpdatedAt, cutoff) + .or() + .lt(FileTaskEntity::getScheduledAt, cutoff)) + .last("limit 500")); + if (stale.isEmpty()) { + return; + } + String reason = "定时任务到点后长时间未启动，已自动失败"; + LocalDateTime now = LocalDateTime.now(); + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .in(FileTaskEntity::getId, stale.stream().map(FileTaskEntity::getId).toList()) + .eq(FileTaskEntity::getStatus, STATUS_SCHEDULED) + .set(FileTaskEntity::getStatus, STATUS_FAILED) + .set(FileTaskEntity::getErrorMessage, reason) + .set(FileTaskEntity::getFinishedAt, now)); + if (updated > 0) { + log.warn("[stale-task-repair] 陈旧定时任务已标失败 count={} ids={}", + updated, stale.stream().map(FileTaskEntity::getId).toList()); + } + } + /** RUNNING 但心跳超过 2h 未刷新 → FAILED（心跳残留续命即僵尸）。 */ private void repairFileTaskStaleRunning() { LocalDateTime cutoff = LocalDateTime.now().minusMinutes(STALE_IDLE_MINUTES); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskClientPullService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskClientPullService.java new file mode 100644 index 00000000..a388af29 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskClientPullService.java @@ -0,0 +1,202 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.vo.TaskClientPullVo; +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.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +/** + * 客户端兜底拉取：把"页面没推送成功、长期停在 PENDING"的任务交给在线客户端执行。 + * + *

PENDING 表示任务已解析落库、等页面把它推进本机 Python 队列；页面这一环缺失时任务会一直 + * 停在这里，2 小时后被 {@code StaleTaskRepairService} 标失败。本服务让客户端主动来领： + * 只挑创建超过 N 分钟仍是 PENDING 的本用户任务，逐条用条件更新认领（PENDING→RUNNING）， + * 只有把状态翻过来的调用方算领取成功 —— 与页面 activate 天然互斥，不会重复执行。 + * + *

开关 {@code aiimage.client-task-pull.enabled} 默认关闭；模块白名单默认只放三个 + * "执行参数全在服务端"的模块，其余模块实现 {@link ClientTaskPullSpi} 后可逐步加入。 + */ +@Slf4j +@Service +public class TaskClientPullService { + + private static final String STATUS_PENDING = "PENDING"; + private static final String STATUS_RUNNING = "RUNNING"; + private static final String STATUS_FAILED = "FAILED"; + private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + /** 单次拉取上限，防止配置写大后一次认领过多把客户端队列压满。 */ + private static final int MAX_LIMIT = 20; + + private final FileTaskMapper fileTaskMapper; + private final InstanceMetadata instanceMetadata; + /** moduleType → 模块兜底实现（启动时建索引并校验重复注册）。 */ + private final Map pullHandlers; + + @Value("${aiimage.client-task-pull.enabled:false}") + private boolean enabled; + + @Value("${aiimage.client-task-pull.module-types:SIMILAR_ASIN,COLLECT_DATA,APPEARANCE_PATENT}") + private String moduleTypes; + + @Value("${aiimage.client-task-pull.min-pending-minutes:5}") + private long minPendingMinutes; + + @Value("${aiimage.client-task-pull.limit:5}") + private int limit; + + public TaskClientPullService(FileTaskMapper fileTaskMapper, + InstanceMetadata instanceMetadata, + List handlers) { + this.fileTaskMapper = fileTaskMapper; + this.instanceMetadata = instanceMetadata; + Map index = new LinkedHashMap<>(); + for (ClientTaskPullSpi handler : handlers == null ? List.of() : handlers) { + String moduleType = handler.moduleType(); + if (moduleType == null || moduleType.isBlank()) { + throw new IllegalStateException("ClientTaskPullSpi 未声明 moduleType: " + + handler.getClass().getName()); + } + ClientTaskPullSpi exists = index.put(moduleType.trim().toUpperCase(Locale.ROOT), handler); + if (exists != null) { + throw new IllegalStateException("moduleType=" + moduleType + " 注册了多个兜底拉取实现: " + + exists.getClass().getName() + " / " + handler.getClass().getName()); + } + } + this.pullHandlers = Map.copyOf(index); + log.info("[client-task-pull] 模块兜底实现注册完成 count={} modules={}", index.size(), index.keySet()); + } + + /** 拉取并认领该用户"长期未被领取"的任务，按创建时间升序返回，最多 limit 条。 */ + public List pullPendingTasks(Long userId) { + if (!enabled) { + return List.of(); + } + if (userId == null || userId <= 0) { + return List.of(); + } + List types = resolveModuleTypes(); + if (types.isEmpty()) { + log.warn("[client-task-pull] 模块白名单为空，本次不拉取 userId={}", userId); + return List.of(); + } + long startedAt = System.currentTimeMillis(); + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(Math.max(1L, minPendingMinutes)); + int safeLimit = Math.max(1, Math.min(limit, MAX_LIMIT)); + List candidates = fileTaskMapper.selectList(new LambdaQueryWrapper() + .select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getCreatedAt) + .eq(FileTaskEntity::getUserId, userId) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) + .in(FileTaskEntity::getModuleType, types) + .lt(FileTaskEntity::getCreatedAt, cutoff) + .orderByAsc(FileTaskEntity::getCreatedAt) + .last("limit " + safeLimit)); + if (candidates.isEmpty()) { + return List.of(); + } + List claimed = new ArrayList<>(); + for (FileTaskEntity candidate : candidates) { + ClientTaskPullSpi handler = pullHandlers.get(candidate.getModuleType()); + if (handler == null) { + // 白名单配了未实现 SPI 的模块：只告警、不动状态（页面仍可正常推送） + log.warn("[client-task-pull] 模块未实现兜底拉取，跳过 taskId={} moduleType={}", + candidate.getId(), candidate.getModuleType()); + continue; + } + if (!claim(candidate.getId())) { + log.info("[client-task-pull] 任务已被页面领取或状态已变化，跳过 taskId={}", candidate.getId()); + continue; + } + FileTaskEntity task = fileTaskMapper.selectById(candidate.getId()); + if (task == null) { + continue; + } + Map queuePayload = null; + try { + queuePayload = handler.buildQueuePayload(task); + } catch (Exception ex) { + log.error("[client-task-pull] 组装兜底载荷异常 taskId={} moduleType={} err={}", + task.getId(), task.getModuleType(), ex.getMessage(), ex); + } + if (queuePayload == null) { + failClaimedTask(task, "任务数据不完整，无法自动执行，请重新提交"); + continue; + } + try { + handler.onClaimed(task); + } catch (Exception ex) { + // 缓存/心跳刷新失败不影响执行：客户端已拿到载荷，心跳会由执行侧补上 + log.warn("[client-task-pull] 认领后刷新模块缓存失败 taskId={} err={}", task.getId(), ex.getMessage()); + } + TaskClientPullVo vo = new TaskClientPullVo(); + vo.setTaskId(task.getId()); + vo.setModuleType(task.getModuleType()); + vo.setTaskNo(task.getTaskNo()); + vo.setCreatedAt(task.getCreatedAt() == null ? null : TIME_FORMATTER.format(task.getCreatedAt())); + vo.setQueuePayload(queuePayload); + claimed.add(vo); + log.warn("[client-task-pull] 兜底认领成功 taskId={} moduleType={} userId={} taskNo={} 创建于={}", + task.getId(), task.getModuleType(), task.getUserId(), task.getTaskNo(), vo.getCreatedAt()); + } + log.info("[client-task-pull] 拉取完成 userId={} 候选={} 认领={} 耗时={}ms", + userId, candidates.size(), claimed.size(), System.currentTimeMillis() - startedAt); + return claimed; + } + + /** + * 条件更新认领：只有把 PENDING 翻成 RUNNING 的一方算领取成功。 + * PENDING 无活跃 owner（页面尚未执行），认领即接管，因此不做归属转发。 + */ + private boolean claim(Long taskId) { + return fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, taskId) + .eq(FileTaskEntity::getStatus, STATUS_PENDING) + .set(FileTaskEntity::getStatus, STATUS_RUNNING) + .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) + .set(FileTaskEntity::getOwnerInstanceId, currentInstanceId())) > 0; + } + + /** 认领后组装不出载荷：直接标失败，避免留成 RUNNING 孤儿等 2 小时心跳线。 */ + private void failClaimedTask(FileTaskEntity task, String reason) { + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, task.getId()) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .set(FileTaskEntity::getStatus, STATUS_FAILED) + .set(FileTaskEntity::getErrorMessage, reason) + .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); + log.warn("[client-task-pull] 认领后组装载荷失败，已标失败 taskId={} moduleType={} 命中={} 原因={}", + task.getId(), task.getModuleType(), updated, reason); + } + + private List resolveModuleTypes() { + if (moduleTypes == null || moduleTypes.isBlank()) { + return List.of(); + } + List types = new ArrayList<>(); + for (String raw : moduleTypes.split(",")) { + String value = raw == null ? "" : raw.trim().toUpperCase(Locale.ROOT); + if (!value.isEmpty() && !types.contains(value)) { + types.add(value); + } + } + return types; + } + + private String currentInstanceId() { + String instanceId = instanceMetadata == null ? null : instanceMetadata.getInstanceId(); + return instanceId == null || instanceId.isBlank() ? "unknown-instance" : instanceId; + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/spi/ClientTaskPullSpi.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/spi/ClientTaskPullSpi.java new file mode 100644 index 00000000..a1831d7e --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/spi/ClientTaskPullSpi.java @@ -0,0 +1,36 @@ +package com.nanri.aiimage.modules.task.spi; + +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; + +import java.util.Map; + +/** + * 客户端兜底拉取的模块侧扩展点。 + * + *

背景：任务由页面"解析落库 PENDING → activate → 经 pywebview 桥推给本机 Python 队列"派发， + * 页面这一段缺失（只解析没点启动 / 推送前关页面 / 在纯浏览器打开）时任务永远停在 PENDING， + * 2 小时后被 StaleTaskRepairService 标失败（「任务长期未被领取，已自动失败」）。 + * 本接口让业务模块自行组装"客户端可直接入队"的 payload，由 {@code TaskClientPullService} + * 认领（PENDING→RUNNING 条件更新）后下发给客户端执行 —— 等于替页面补上"推送"这一步。 + * + *

与 {@link TaskModuleHeartbeatSpi} 同理（G5）：task 侧不 import 业务模块， + * 各模块实现本接口，Spring 注入 List 后由 task 侧建索引。 + */ +public interface ClientTaskPullSpi { + + /** 本实现负责的 moduleType（与 biz_file_task.module_type 一致）。 */ + String moduleType(); + + /** + * 组装客户端可直接入队的载荷：{@code {"type": "...", "data": {...}}}。 + * 返回 null 表示该任务当前不具备兜底执行条件（调用方会把任务标失败并回写原因）。 + */ + Map buildQueuePayload(FileTaskEntity task); + + /** + * 认领成功后的模块侧动作：对齐各自的 activate（多数模块实现为 + * {@code cacheService.touchTaskHeartbeat(taskId)}，让页面立刻看到 RUNNING）。 + */ + default void onClaimed(FileTaskEntity task) { + } +} diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 043bbb4f..08e5b1f4 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -234,6 +234,14 @@ aiimage: stuck-timeout-minutes: ${AIIMAGE_RESULT_FILE_JOB_STUCK_TIMEOUT_MINUTES:30} heartbeat-interval-ms: ${AIIMAGE_RESULT_FILE_JOB_HEARTBEAT_INTERVAL_MS:60000} batch-size: ${AIIMAGE_RESULT_FILE_JOB_BATCH_SIZE:20} + # 客户端兜底拉取：页面没推送成功（只解析没点启动 / 推送前关页面 / 纯浏览器打开）、 + # 长期停在 PENDING 的任务由在线客户端领走执行，替代 2 小时后的"任务长期未被领取"标失败。 + # 默认关闭；打开前需先发布带兜底拉取线程的客户端版本，旧客户端不受影响。 + client-task-pull: + enabled: ${AIIMAGE_CLIENT_TASK_PULL_ENABLED:false} + 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} coze-task: max-concurrent: ${AIIMAGE_COZE_TASK_MAX_CONCURRENT:12} brand-check: diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataActivateGuardTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataActivateGuardTest.java new file mode 100644 index 00000000..a2b3d74a --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataActivateGuardTest.java @@ -0,0 +1,163 @@ +package com.nanri.aiimage.modules.collectdata.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.modules.collectdata.util.CollectDataBatchQuery; +import com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter; +import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailReader; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper; +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.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +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.TaskProgressLightAssembler; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * activate 收紧契约（2026-09-14）：只允许 PENDING→RUNNING。 + * + *

任务被客户端「兜底拉取」领走后已是 RUNNING，页面再点启动必须被拒且不改状态 —— + * 否则同一任务会被执行两次（页面入队一次、客户端一次）。同理，已结束的任务不能被复活。 + */ +class CollectDataActivateGuardTest { + + private static final long USER_ID = 1095L; + private static final long TASK_ID = 11L; + + private final FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + private CollectDataService service; + + @BeforeEach + void setUp() { + // LambdaWrapper 需要实体列缓存，单测无 Spring/MyBatis 上下文，手动初始化 + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), FileTaskEntity.class); + service = buildService(fileTaskMapper); + } + + @Test + void 已被客户端领走时拒绝重复启动() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task("RUNNING")); + + BusinessException error = assertThrows(BusinessException.class, + () -> service.activateTask(TASK_ID, USER_ID)); + + assertTrue(error.getMessage().contains("无需重复启动"), "错误信息应说明任务已在执行中：" + error.getMessage()); + // 只有一次条件更新，且谓词限定 PENDING —— 已被客户端认领（RUNNING）的行不可能被再翻一次 + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper).update(isNull(), captor.capture()); + assertTrue(paramsOf(captor.getValue()).containsValue("PENDING"), + "条件更新必须限定 PENDING，否则已认领的任务会被重复执行"); + } + + @Test + void 已结束的任务不能被复活() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task("SUCCESS")); + + BusinessException error = assertThrows(BusinessException.class, + () -> service.activateTask(TASK_ID, USER_ID)); + + assertEquals("任务已结束", error.getMessage()); + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper).update(isNull(), captor.capture()); + assertTrue(paramsOf(captor.getValue()).containsValue("PENDING"), + "已结束的任务不能命中条件更新"); + } + + @Test + void 非本用户的任务被拒() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task("PENDING")); + + assertThrows(BusinessException.class, () -> service.activateTask(TASK_ID, USER_ID + 1)); + verify(fileTaskMapper, never()).update(isNull(), any()); + } + + @Test + void PENDING_任务激活成功且条件更新限定_PENDING() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task("PENDING")); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + + service.activateTask(TASK_ID, USER_ID); + + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper).update(isNull(), captor.capture()); + Map params = paramsOf(captor.getValue()); + assertTrue(params.containsValue("PENDING"), "条件更新的谓词必须限定 PENDING"); + assertTrue(params.containsValue("RUNNING"), "条件更新必须写入 RUNNING"); + } + + private static FileTaskEntity task(String status) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(TASK_ID); + task.setModuleType("COLLECT_DATA"); + task.setTaskNo("COLLECT_DATA-11"); + task.setUserId(USER_ID); + task.setStatus(status); + return task; + } + + /** 复用既有测试的服务装配方式：只关心 activateTask 用到的 fileTaskMapper，其余协作者为 mock。 */ + private static CollectDataService buildService(FileTaskMapper fileTaskMapper) { + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + return new CollectDataService(null, fileTaskMapper, mock(FileResultMapper.class), + mock(com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper.class), + mock(com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper.class), + mock(InvalidAsinDataMapper.class), + mock(TaskChunkMapper.class), mock(TaskScopeStateMapper.class), mock(TaskResultItemMapper.class), + mock(TaskDistributedLockService.class), mock(TaskFileJobService.class), + mock(TransientPayloadStorageService.class), + mock(CollectDataExcelAssemblyService.class), mock(OssStorageService.class), + new ObjectMapper(), new TransactionTemplate(transactionManager), + mock(DistributedJobLockService.class), + mock(CollectDataBatchQuery.class), mock(CollectDataBrandBatchFilter.class), + mock(CollectDataInvalidAsinBatchWriter.class), mock(CollectDataResultDetailCodec.class), + mock(CollectDataResultItemBatchWriter.class), mock(CollectDataResultDetailReader.class), + mock(TaskProgressLightAssembler.class)); + } + + @SuppressWarnings("unchecked") + private static ArgumentCaptor> wrapperCaptor() { + return ArgumentCaptor.forClass(Wrapper.class); + } + + /** 读取 wrapper 参数容器：MyBatis-Plus 在生成 SQL 片段时才填充，故先触发一次。 */ + @SuppressWarnings("unchecked") + private static Map paramsOf(Wrapper wrapper) { + wrapper.getSqlSegment(); + ((com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper) wrapper).getSqlSet(); + Map params = (Map) ReflectionTestUtils.getField(wrapper, "paramNameValuePairs"); + assertNotNull(params, "wrapper 参数容器字段名可能已变化（paramNameValuePairs）"); + return params; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImplTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImplTest.java new file mode 100644 index 00000000..c61ec566 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataTaskPullSpiImplTest.java @@ -0,0 +1,100 @@ +package com.nanri.aiimage.modules.collectdata.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * 集采兜底载荷契约（与 Python SearchTask 的读取口径对齐）： + * filters 原样透传（Python 读 countryCode/minAmount/maxAmount/rank/fba/fbm 这些 camelCase 键）、 + * totalRows 取任务 result_json 的初始统计、pageSize 固定 50； + * request_json 缺失/不可解析时返回 null —— 由调用方标失败，绝不用默认国家"凑"一次采集。 + */ +class CollectDataTaskPullSpiImplTest { + + private final CollectDataTaskPullSpiImpl spi = new CollectDataTaskPullSpiImpl(new ObjectMapper()); + + @Test + void 载荷包含明细拉取所需的全部字段() { + FileTaskEntity task = task(""" + {"userId":1095,"taskType":"collect-data", + "filters":{"countryCode":"DE","minAmount":10.5,"maxAmount":99.9,"rank":true,"fba":false,"fbm":true}, + "files":[]}""", """ + {"totalRows":81}"""); + + Map payload = spi.buildQueuePayload(task); + + assertEquals("collect-data-run", payload.get("type")); + @SuppressWarnings("unchecked") + Map data = (Map) payload.get("data"); + assertEquals(11L, data.get("taskId")); + assertEquals("COLLECT_DATA-11", data.get("taskNo")); + assertEquals("collect-data", data.get("taskType")); + assertEquals(81, data.get("totalRows")); + assertEquals(CollectDataService.DEFAULT_PAGE_SIZE, data.get("pageSize")); + + @SuppressWarnings("unchecked") + Map filters = (Map) data.get("filters"); + assertEquals("DE", filters.get("countryCode")); + assertEquals(10.5, filters.get("minAmount")); + assertEquals(99.9, filters.get("maxAmount")); + assertEquals(true, filters.get("rank")); + assertEquals(false, filters.get("fba")); + assertEquals(true, filters.get("fbm")); + } + + @Test + void 模块类型与入队类型固定() { + assertEquals("COLLECT_DATA", spi.moduleType()); + assertEquals("collect-data-run", spi.buildQueuePayload(task("{\"filters\":{}}", "{}")).get("type")); + } + + @Test + void 结果统计缺失时_totalRows_为_0() { + Map payload = spi.buildQueuePayload(task("{\"filters\":{}}", null)); + + @SuppressWarnings("unchecked") + Map data = (Map) payload.get("data"); + assertEquals(0, data.get("totalRows")); + } + + @Test + void filters_为空时透传空对象由_python_默认值兜底() { + Map payload = spi.buildQueuePayload(task("{\"filters\":null}", "{}")); + + @SuppressWarnings("unchecked") + Map data = (Map) payload.get("data"); + assertEquals(Map.of(), data.get("filters")); + } + + @Test + void 请求参数缺失或不可解析时返回_null_由调用方标失败() { + assertNull(spi.buildQueuePayload(task(null, "{}"))); + assertNull(spi.buildQueuePayload(task("not-a-json", "{}"))); + } + + private static FileTaskEntity task(String requestJson, String resultJson) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(11L); + task.setTaskNo("COLLECT_DATA-11"); + task.setModuleType("COLLECT_DATA"); + task.setUserId(1095L); + task.setStatus("PENDING"); + task.setRequestJson(requestJson); + task.setResultJson(resultJson); + return task; + } + + /** 防回归：whitelist 里的模块类型必须与业务模块常量一致（配置默认值写的是字符串）。 */ + @Test + void 模块类型常量未被改名() { + assertTrue(List.of("SIMILAR_ASIN", "COLLECT_DATA", "APPEARANCE_PATENT").contains(spi.moduleType())); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairServiceTest.java new file mode 100644 index 00000000..5ff57606 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/StaleTaskRepairServiceTest.java @@ -0,0 +1,149 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.spi.BrandTaskStaleRepairSpi; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeEach; +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +/** + * 陈旧任务判死契约（2026-09-14 修订）： + * PENDING 仍按 updated_at 判死；SCHEDULED（店铺匹配定时任务）改以 scheduled_at 为基准 —— + * 只按 updated_at 会误杀排期在 2 小时之后的定时任务，而 activate 又被「未到定时执行时间」挡住。 + */ +class StaleTaskRepairServiceTest { + + private final FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + private final BrandTaskStaleRepairSpi brandTaskStaleRepairSpi = mock(BrandTaskStaleRepairSpi.class); + private final DistributedJobLockService distributedJobLockService = mock(DistributedJobLockService.class); + private final StaleTaskRepairService service = new StaleTaskRepairService( + fileTaskMapper, brandTaskStaleRepairSpi, distributedJobLockService); + + @BeforeEach + void initTableInfo() { + // LambdaWrapper 需要实体列缓存，单测无 Spring/MyBatis 上下文，手动初始化 + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), FileTaskEntity.class); + } + + @Test + void PENDING_判死只命中_PENDING_不再波及_SCHEDULED() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(task(11L))); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + + ReflectionTestUtils.invokeMethod(service, "repairFileTaskStaleIdle"); + + Map selectParams = selectParams(); + assertTrue(selectParams.containsValue("PENDING")); + assertFalse(selectParams.containsValue("SCHEDULED"), "PENDING 分支命中 SCHEDULED 是定时任务被误杀的原因"); + + Map updateParams = updateParams(1); + assertTrue(updateParams.containsValue("FAILED")); + assertTrue(containsMessage(updateParams, "任务长期未被领取")); + } + + @Test + void SCHEDULED_判死以_scheduled_at_为基准并保留_updated_at_兜底() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(task(21L))); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + + ReflectionTestUtils.invokeMethod(service, "repairFileTaskStaleScheduled"); + + ArgumentCaptor> selectCaptor = wrapperCaptor(); + verify(fileTaskMapper).selectList(selectCaptor.capture()); + String segment = selectCaptor.getValue().getSqlSegment(); + assertTrue(segment.contains("scheduled_at"), "SCHEDULED 判死必须用到 scheduled_at"); + assertTrue(segment.contains("updated_at"), "scheduled_at 为空时要退回 updated_at 逻辑"); + Map selectParams = paramsOf(selectCaptor.getValue()); + assertTrue(selectParams.containsValue("SCHEDULED"), "只处理 SCHEDULED 状态"); + long cutoffs = selectParams.values().stream().filter(LocalDateTime.class::isInstance).count(); + assertEquals(2L, cutoffs, "updated_at 与 scheduled_at 两个分支各有一个判死时间点"); + + Map updateParams = updateParams(1); + assertTrue(updateParams.containsValue("FAILED")); + assertTrue(updateParams.containsValue("SCHEDULED")); + assertTrue(containsMessage(updateParams, "定时任务到点后长时间未启动")); + } + + @Test + void 没有候选任务时不写库() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of()); + + ReflectionTestUtils.invokeMethod(service, "repairFileTaskStaleIdle"); + ReflectionTestUtils.invokeMethod(service, "repairFileTaskStaleScheduled"); + + verify(fileTaskMapper, never()).update(any(), any()); + verifyNoMoreInteractions(brandTaskStaleRepairSpi); + } + + private Map selectParams() { + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper).selectList(captor.capture()); + return paramsOf(captor.getValue()); + } + + private Map updateParams(int times) { + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper, org.mockito.Mockito.times(times)).update(isNull(), captor.capture()); + return paramsOf(captor.getValue()); + } + + @SuppressWarnings("unchecked") + private static ArgumentCaptor> wrapperCaptor() { + return ArgumentCaptor.forClass(Wrapper.class); + } + + /** + * 读取 wrapper 参数容器：用来断言判死谓词与写入值。 + * 注意 MyBatis-Plus 是在生成 SQL 片段时才把参数写进 paramNameValuePairs，故先触发一次。 + */ + @SuppressWarnings("unchecked") + private static Map paramsOf(Wrapper wrapper) { + wrapper.getSqlSegment(); + if (wrapper instanceof LambdaUpdateWrapper) { + ((LambdaUpdateWrapper) wrapper).getSqlSet(); + } + Map params = (Map) ReflectionTestUtils.getField(wrapper, "paramNameValuePairs"); + assertNotNull(params, "wrapper 参数容器字段名可能已变化（paramNameValuePairs）"); + return params; + } + + private static boolean containsMessage(Map params, String keyword) { + return params.values().stream() + .anyMatch(value -> value instanceof String text && text.contains(keyword)); + } + + private static FileTaskEntity task(Long id) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType("SHOP_MATCH"); + task.setStatus("SCHEDULED"); + task.setCreatedAt(LocalDateTime.now().minusHours(5)); + task.setUpdatedAt(LocalDateTime.now().minusHours(5)); + task.setScheduledAt(LocalDateTime.now().plusHours(3)); + return task; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskClientPullServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskClientPullServiceTest.java new file mode 100644 index 00000000..7c8592d1 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskClientPullServiceTest.java @@ -0,0 +1,260 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.vo.TaskClientPullVo; +import com.nanri.aiimage.modules.task.spi.ClientTaskPullSpi; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.LocalDateTime; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * 客户端兜底拉取契约： + * 开关关闭不查库；候选限定「本用户 + PENDING + 白名单模块 + 创建超过 N 分钟」； + * 认领走条件更新（限定 PENDING，写 RUNNING 与 owner）——与页面 activate 互斥，认领失败即返回空； + * 认领后组装不出载荷时把任务标失败（不留 RUNNING 孤儿）；白名单未实现 SPI 的模块跳过；同模块重复注册直接报错。 + */ +class TaskClientPullServiceTest { + + private static final long USER_ID = 1095L; + + private final FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class); + + @BeforeEach + void initTableInfo() { + // LambdaWrapper 需要实体列缓存，单测无 Spring/MyBatis 上下文，手动初始化 + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), FileTaskEntity.class); + } + + @Test + void 开关关闭时不查库也不返回任务() { + TaskClientPullService service = service(new StubHandler("SIMILAR_ASIN", payload("similar-asin-run"))); + ReflectionTestUtils.setField(service, "enabled", false); + + assertTrue(service.pullPendingTasks(USER_ID).isEmpty()); + verifyNoInteractions(fileTaskMapper); + } + + @Test + void 认领成功后返回载荷并写_RUNNING_与_owner() { + FileTaskEntity candidate = task(11L, "SIMILAR_ASIN"); + when(fileTaskMapper.selectList(any())).thenReturn(List.of(candidate)); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + when(fileTaskMapper.selectById(11L)).thenReturn(candidate); + StubHandler handler = new StubHandler("SIMILAR_ASIN", payload("similar-asin-run")); + + List result = service(handler).pullPendingTasks(USER_ID); + + assertEquals(1, result.size(), "应返回一条已认领任务"); + assertEquals(11L, result.get(0).getTaskId()); + assertEquals("similar-asin-run", result.get(0).getQueuePayload().get("type")); + assertEquals("SIMILAR_ASIN", result.get(0).getModuleType()); + assertEquals(1, handler.claimedCalls, "认领成功后必须执行模块侧后置处理"); + + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper).update(isNull(), captor.capture()); + Map params = paramsOf(captor.getValue()); + assertTrue(params.containsValue("PENDING"), "认领条件必须限定 PENDING"); + assertTrue(params.containsValue("RUNNING"), "认领必须写入 RUNNING"); + assertTrue(sqlSetOf(captor.getValue()).contains("owner_instance_id"), "认领必须接管归属实例"); + + ArgumentCaptor> selectCaptor = wrapperCaptor(); + verify(fileTaskMapper).selectList(selectCaptor.capture()); + Map selectParams = paramsOf(selectCaptor.getValue()); + assertTrue(selectParams.containsValue("PENDING"), "候选只挑 PENDING"); + assertTrue(selectParams.containsValue(USER_ID), "候选限定任务所属用户"); + assertTrue(selectCaptor.getValue().getSqlSegment().contains("created_at"), "候选要有创建时间窗口"); + } + + @Test + void 页面抢先认领后不再返回任务() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(task(11L, "SIMILAR_ASIN"))); + when(fileTaskMapper.update(isNull(), any())).thenReturn(0); + StubHandler handler = new StubHandler("SIMILAR_ASIN", payload("similar-asin-run")); + + List result = service(handler).pullPendingTasks(USER_ID); + + assertTrue(result.isEmpty(), "条件更新没命中说明页面已接管，不能再返回给客户端"); + assertEquals(0, handler.buildCalls, "未认领成功时不应组装载荷"); + verify(fileTaskMapper, never()).selectById(anyLong()); + } + + @Test + void 组装不出载荷时任务标失败() { + FileTaskEntity candidate = task(11L, "SIMILAR_ASIN"); + when(fileTaskMapper.selectList(any())).thenReturn(List.of(candidate)); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + when(fileTaskMapper.selectById(11L)).thenReturn(candidate); + + List result = service(new StubHandler("SIMILAR_ASIN", null)).pullPendingTasks(USER_ID); + + assertTrue(result.isEmpty()); + Map failParams = secondUpdateParams(); + assertTrue(failParams.containsValue("FAILED"), "组装失败必须标 FAILED，不留 RUNNING 孤儿"); + assertTrue(containsMessage(failParams, "无法自动执行")); + } + + @Test + void 组装抛异常时任务标失败且不抛出() { + FileTaskEntity candidate = task(11L, "SIMILAR_ASIN"); + when(fileTaskMapper.selectList(any())).thenReturn(List.of(candidate)); + when(fileTaskMapper.update(isNull(), any())).thenReturn(1); + when(fileTaskMapper.selectById(11L)).thenReturn(candidate); + StubHandler broken = new StubHandler("SIMILAR_ASIN", null, new IllegalStateException("boom")); + + List result = service(broken).pullPendingTasks(USER_ID); + + assertTrue(result.isEmpty()); + Map failParams = secondUpdateParams(); + assertTrue(failParams.containsValue("FAILED")); + assertTrue(containsMessage(failParams, "无法自动执行")); + } + + @Test + void 白名单模块未实现兜底时跳过且不改状态() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(task(21L, "COLLECT_DATA"))); + + List result = service(new StubHandler("SIMILAR_ASIN", payload("similar-asin-run"))) + .pullPendingTasks(USER_ID); + + assertTrue(result.isEmpty()); + verify(fileTaskMapper, never()).update(isNull(), any()); + verify(fileTaskMapper, never()).selectById(anyLong()); + } + + @Test + void 同一模块重复注册直接报错() { + assertThrows(IllegalStateException.class, () -> service( + new StubHandler("SIMILAR_ASIN", payload("similar-asin-run")), + new StubHandler("SIMILAR_ASIN", payload("similar-asin-run")))); + } + + private TaskClientPullService service(ClientTaskPullSpi... handlers) { + TaskClientPullService service = new TaskClientPullService( + fileTaskMapper, new InstanceMetadata("server-test"), List.of(handlers)); + ReflectionTestUtils.setField(service, "enabled", true); + ReflectionTestUtils.setField(service, "moduleTypes", "SIMILAR_ASIN,COLLECT_DATA"); + ReflectionTestUtils.setField(service, "minPendingMinutes", 5L); + ReflectionTestUtils.setField(service, "limit", 5); + return service; + } + + private Map secondUpdateParams() { + ArgumentCaptor> captor = wrapperCaptor(); + verify(fileTaskMapper, times(2)).update(isNull(), captor.capture()); + return paramsOf(captor.getAllValues().get(1)); + } + + @SuppressWarnings("unchecked") + private static ArgumentCaptor> wrapperCaptor() { + return ArgumentCaptor.forClass(Wrapper.class); + } + + /** + * 读取 MyBatis-Plus wrapper 里的参数容器：用来断言条件更新的谓词与写入值。 + * 注意 MyBatis-Plus 是在生成 SQL 片段时才把参数写进 paramNameValuePairs，故先触发一次。 + */ + @SuppressWarnings("unchecked") + private static Map paramsOf(Wrapper wrapper) { + wrapper.getSqlSegment(); + if (wrapper instanceof LambdaUpdateWrapper) { + ((LambdaUpdateWrapper) wrapper).getSqlSet(); + } + Map params = (Map) ReflectionTestUtils.getField(wrapper, "paramNameValuePairs"); + assertNotNull(params, "wrapper 参数容器字段名可能已变化（paramNameValuePairs）"); + return params; + } + + @SuppressWarnings("unchecked") + private static String sqlSetOf(Wrapper wrapper) { + return ((LambdaUpdateWrapper) wrapper).getSqlSet(); + } + + private static boolean containsMessage(Map params, String keyword) { + return params.values().stream() + .anyMatch(value -> value instanceof String text && text.contains(keyword)); + } + + private static Map payload(String type) { + Map data = new LinkedHashMap<>(); + data.put("taskId", 11L); + return Map.of("type", type, "data", data); + } + + private static FileTaskEntity task(Long id, String moduleType) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType(moduleType); + task.setTaskNo(moduleType + "-" + id); + task.setUserId(USER_ID); + task.setStatus("PENDING"); + task.setCreatedAt(LocalDateTime.now().minusMinutes(10)); + task.setUpdatedAt(LocalDateTime.now().minusMinutes(10)); + return task; + } + + private static final class StubHandler implements ClientTaskPullSpi { + + private final String moduleType; + private final Map queuePayload; + private final RuntimeException failure; + private int buildCalls; + private int claimedCalls; + + StubHandler(String moduleType, Map queuePayload) { + this(moduleType, queuePayload, null); + } + + StubHandler(String moduleType, Map queuePayload, RuntimeException failure) { + this.moduleType = moduleType; + this.queuePayload = queuePayload; + this.failure = failure; + } + + @Override + public String moduleType() { + return moduleType; + } + + @Override + public Map buildQueuePayload(FileTaskEntity task) { + buildCalls++; + if (failure != null) { + throw failure; + } + return queuePayload; + } + + @Override + public void onClaimed(FileTaskEntity task) { + claimedCalls++; + } + } +} diff --git a/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue b/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue index 948def1c..a7b35737 100644 --- a/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandAppearancePatentTab.vue @@ -456,7 +456,15 @@ async function pushToPythonQueue() { nonEmptyArrayKeys: ['groups'], }) if (!(await passGuard(guard))) return - await activateAppearancePatentTask(taskId) + // 任务可能已被客户端「兜底拉取」领走（页面长时间没点启动）：服务端会拒绝 activate， + // 此时必须直接返回，不能再入队，否则同一任务会被执行两次 + try { + await activateAppearancePatentTask(taskId) + } catch (activateError) { + ElMessage.warning(activateError instanceof Error ? activateError.message : '任务已在执行中，无需重复启动') + await Promise.all([loadDashboard(), loadHistory({ force: true })]) + return + } const result = await api.enqueue_json(payload) if (!result?.success) { ElMessage.error(result?.error || '启动失败') diff --git a/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue b/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue index 5c4ae8ab..42276ff4 100644 --- a/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue +++ b/frontend-vue/src/pages/brand/components/BrandSimilarAsinTab.vue @@ -533,7 +533,15 @@ async function pushToPythonQueue() { requiredDataKeys: ['taskId', 'aliprice_usename', 'aliprice_pwd'], }) if (!(await passGuard(guard))) return - await activateSimilarAsinTask(taskId) + // 任务可能已被客户端「兜底拉取」领走（页面长时间没点启动）：服务端会拒绝 activate， + // 此时必须直接返回，不能再入队，否则同一任务会被执行两次 + try { + await activateSimilarAsinTask(taskId) + } catch (activateError) { + ElMessage.warning(activateError instanceof Error ? activateError.message : '任务已在执行中，无需重复启动') + await Promise.all([loadDashboard(), loadHistory({ force: true })]) + return + } const result = await api.enqueue_json(payload) if (!result?.success) { ElMessage.error(result?.error || '启动失败')