diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index 641fdd6e..82f8b4c5 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -116,12 +116,18 @@ public class ShopDataCrawlTaskService { public void finalizeOwnedStaleTasks() { long minutes = Math.max(1L, staleTimeoutMinutes); long nowMillis = System.currentTimeMillis(); - List tasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .eq(FileTaskEntity::getModuleType, MODULE_TYPE) - .eq(FileTaskEntity::getStatus, "RUNNING") - .apply("JSON_UNQUOTE(JSON_EXTRACT(request_json, '$.ownerInstanceId')) = {0}", currentInstanceId()) - .lt(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(minutes)) - .last("limit 200")); + List tasks; + try { + tasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, "RUNNING") + .eq(FileTaskEntity::getOwnerInstanceId, currentInstanceId()) + .lt(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(minutes)) + .last("limit 200")); + } catch (Exception ex) { + log.warn("[shop-data-crawl] stale task scan failed msg={}", ex.getMessage()); + return; + } Map heartbeats = taskCacheService.getTaskHeartbeatMillisBatch( tasks.stream().map(FileTaskEntity::getId).toList()); for (FileTaskEntity task : tasks) { @@ -1137,6 +1143,7 @@ public class ShopDataCrawlTaskService { requestSnapshot.put("items", requestItems); requestSnapshot.put("countryCodes", countryCodes); task.setRequestJson(objectMapper.writeValueAsString(requestSnapshot)); + task.setOwnerInstanceId(currentInstanceId()); task.setResultJson(objectMapper.writeValueAsString(snapshots)); syncSnapshotTables(task, snapshots); fileTaskMapper.updateById(task); @@ -2555,7 +2562,22 @@ public class ShopDataCrawlTaskService { throw new TaskOwnerMismatchException(task == null ? null : task.getId(), operation, owner, currentInstanceId()); } + /** + * 任务归属实例 id：优先读显式列 owner_instance_id（Task 34 迁移）， + * 旧任务列缺失时兼容回退到 request_json.ownerInstanceId。 + */ + public String ownerInstanceIdOf(FileTaskEntity task) { + if (task != null && !blank(task.getOwnerInstanceId())) { + return task.getOwnerInstanceId(); + } + return ownerFromJson(task); + } + private String ownerFromTask(FileTaskEntity task) { + return ownerInstanceIdOf(task); + } + + private String ownerFromJson(FileTaskEntity task) { if (task == null || blank(task.getRequestJson())) return null; try { String owner = objectMapper.readTree(task.getRequestJson()).path("ownerInstanceId").asText(""); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java index f83cad32..e33e6372 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java @@ -25,6 +25,7 @@ public class FileTaskEntity { private String errorMessage; private String createdBy; private Long userId; + private String ownerInstanceId; private LocalDateTime createdAt; private LocalDateTime updatedAt; private LocalDateTime finishedAt; diff --git a/backend-java/src/main/resources/db/V92__file_task_owner_instance_column.sql b/backend-java/src/main/resources/db/V92__file_task_owner_instance_column.sql new file mode 100644 index 00000000..c743cbf6 --- /dev/null +++ b/backend-java/src/main/resources/db/V92__file_task_owner_instance_column.sql @@ -0,0 +1,3 @@ +ALTER TABLE `biz_file_task` + ADD COLUMN `owner_instance_id` VARCHAR(128) NULL DEFAULT NULL COMMENT '任务归属实例 id（原存于 request_json.ownerInstanceId，迁移为显式列）' AFTER `user_id`, + ADD KEY `idx_biz_file_task_owner_status_updated` (`owner_instance_id`, `status`, `updated_at`); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlOwnerColumnTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlOwnerColumnTest.java new file mode 100644 index 00000000..2869cd1d --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlOwnerColumnTest.java @@ -0,0 +1,381 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +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.exception.TaskOwnerMismatchException; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.TaskPressureProperties; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCreateTaskRequest; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlTaskItemDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlCreateTaskVo; +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.entity.FileResultEntity; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TaskResultItemService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo; +import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +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.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 34：将 ownerInstanceId 从 JSON 查询迁移到显式列并补充索引。 + * 原 ownerInstanceId 只存在于 request_json 的 JSON 字段里，stale 扫描用 + * JSON_UNQUOTE(JSON_EXTRACT(...)) 过滤、任务归属校验用 Jackson 解析整棵 JSON 树； + * 迁移后写入/读取/查询都走 biz_file_task.owner_instance_id 显式列 + * （V92 迁移新增列并补 (owner_instance_id, status, updated_at) 索引）， + * JSON 解析不再参与 owner 判定，stale 扫描直接按列过滤。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlOwnerColumnTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "Demo Shop"; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private ShopDataCrawlResolveService shopDataCrawlResolveService; + @Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService; + @Mock private ShopDataCrawlTaskCacheService taskCacheService; + @Mock private OssStorageService ossStorageService; + @Mock private ZiniaoShopSwitchService ziniaoShopSwitchService; + @Mock private TaskPressureProperties taskPressureProperties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskResultItemService taskResultItemService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private ShopDataCrawlDailyFileService dailyFileService; + @Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules(); + + private ShopDataCrawlTaskService service; + /** 内存中的任务表（createTask 写入 / stale 扫描过滤的源数据）。 */ + private final List dbTasks = new ArrayList<>(); + /** 最近一次 stale 扫描返回的结果。 */ + private final List lastScan = new ArrayList<>(); + + @BeforeEach + void configureStorage() { + service = new ShopDataCrawlTaskService( + fileTaskMapper, + fileResultMapper, + shopDataCrawlResolveService, + excelAssemblyService, + taskCacheService, + ossStorageService, + ziniaoShopSwitchService, + objectMapper, + taskPressureProperties, + taskFileJobService, + taskResultItemService, + taskProgressSnapshotService, + taskDistributedLockService, + taskChunkMapper, + taskScopeStateMapper, + transientPayloadStorageService, + instanceMetadata, + dailyFileService, + null); + + dbTasks.clear(); + lastScan.clear(); + + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); + lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); + lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of()); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L); + lenient().when(excelAssemblyService.countRows(any())).thenReturn(1); + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileTaskMapper.selectById(anyLong())).thenReturn(null); + lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + lastScan.clear(); + lastScan.addAll(applyTaskScanFilter(wrapper, dbTasks)); + return new ArrayList<>(lastScan); + }); + lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1); + lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); + } + + @Test + void test_task_034_owner_normal_default_path() { + // 正常路径：createTask 把 ownerInstanceId 写入显式列，持久化后 owner 路由可读。 + ShopDataCrawlCreateTaskVo vo = createTask(3401L); + assertNotNull(vo); + assertEquals("instance-a", dbTask(3401L).getOwnerInstanceId(), "显式列写入当前实例 id"); + assertTrue(dbTask(3401L).getRequestJson().contains("ownerInstanceId"), "兼容字段仍保留在快照 JSON"); + + // stale 扫描按显式列过滤当前实例 RUNNING 任务，命中 owner 路由。 + runStaleScan(); + assertEquals(1, lastScan.size(), "stale 扫描按 owner_instance_id 列过滤命中"); + assertEquals(3401L, lastScan.get(0).getId()); + } + + @Test + void test_task_034_owner_normal_multiple_items() { + // 批量场景：多个任务各自带 owner 列，扫描只返回当前实例的任务，其他实例不命中。 + dbTasks.add(task(3402L, "instance-a", "RUNNING")); + dbTasks.add(task(3403L, "instance-a", "RUNNING")); + dbTasks.add(task(3404L, "instance-b", "RUNNING")); + dbTasks.add(task(3405L, "instance-a", "SUCCESS")); + + runStaleScan(); + + assertEquals(2, lastScan.size(), "只返回当前实例的 RUNNING 任务"); + assertEquals(3402L, lastScan.get(0).getId()); + assertEquals(3403L, lastScan.get(1).getId()); + } + + @Test + void test_task_034_owner_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入重复提交创建的两个任务 owner 列值稳定一致； + // 重复 stale 扫描不重复处理已终结任务（终态任务不再命中）。 + ShopDataCrawlCreateTaskVo first = createTask(3406L); + ShopDataCrawlCreateTaskVo second = createTask(3407L); + + assertNotNull(first); + assertNotNull(second); + assertEquals("instance-a", dbTask(3406L).getOwnerInstanceId(), "重复提交 owner 列值稳定"); + assertEquals("instance-a", dbTask(3407L).getOwnerInstanceId(), "重复提交 owner 列值一致"); + + runStaleScan(); + assertEquals(2, lastScan.size(), "两个任务都被当前实例接管"); + + // 首次扫描已把 stale 任务终结（FAILED），再次扫描不再命中，不重复处理。 + runStaleScan(); + assertEquals(0, lastScan.size(), "重复扫描不重复处理已终结任务"); + assertEquals("FAILED", dbTask(3406L).getStatus(), "任务已被终结"); + } + + @Test + void test_task_034_owner_boundary_empty_input() { + // 空输入：无 RUNNING 任务时扫描返回空，不抛异常、不触碰无关资源。 + runStaleScan(); + assertEquals(0, lastScan.size(), "空任务列表安全跳过"); + verify(fileTaskMapper, never()).updateById(any(FileTaskEntity.class)); + } + + @Test + void test_task_034_owner_boundary_single_item() { + // 单元素：仅一个当前实例 RUNNING 任务被扫描命中，owner 列解析不依赖批量路径。 + FileTaskEntity single = task(3407L, "instance-a", "RUNNING"); + single.setUpdatedAt(LocalDateTime.now().minusMinutes(45)); + dbTasks.add(single); + + runStaleScan(); + + assertEquals(1, lastScan.size()); + assertEquals(3407L, lastScan.get(0).getId()); + assertEquals("instance-a", dbTask(3407L).getOwnerInstanceId()); + } + + @Test + void test_task_034_owner_boundary_limit_and_overflow() { + // 上限/超限：owner 列值接近上限长度（110 字符，可稳定写入）仍完整读写； + // 大量任务扫描不放大结果。 + String longOwner = "instance-" + "x".repeat(110); + lenient().when(instanceMetadata.getInstanceId()).thenReturn(longOwner); + ShopDataCrawlCreateTaskVo vo = createTask(3408L); + assertNotNull(vo); + assertEquals(longOwner, dbTask(3408L).getOwnerInstanceId(), "长 owner 值完整保留"); + assertTrue(dbTask(3408L).getOwnerInstanceId().length() <= 128, "不超列长度上限"); + + // 恢复当前实例 id 后扫描 40 个 instance-a 任务。 + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + for (int i = 0; i < 40; i++) { + dbTasks.add(task(10000L + i, "instance-a", "RUNNING")); + } + runStaleScan(); + assertEquals(40, lastScan.size(), "大量任务逐一命中，无重复无丢失"); + } + + @Test + void test_task_034_owner_invalid_input_rejected() { + // 非法参数：无 owner 列的 RUNNING 任务不属于任何实例，扫描不命中（区别于旧 JSON 兼容分支）。 + dbTasks.add(task(3409L, null, "RUNNING")); + + runStaleScan(); + + assertEquals(0, lastScan.size(), "owner 列缺失的任务不属于当前实例"); + // 归属性判定对缺 owner 列的任务放行兼容读取，但不归属任何实例。 + FileTaskEntity legacy = task(3410L, null, "RUNNING"); + legacy.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + service.ensureTaskOwnedByCurrentInstance(legacy, "legacy callback"); + assertEquals("instance-a", service.ownerInstanceIdOf(legacy), "旧 JSON 兜底仍可读取 owner"); + } + + @Test + void test_task_034_owner_dependency_failure_releases_resources() { + // 依赖失败：扫描 DB 异常时安全降级不抛错、不泄漏任务；他人实例任务被拒且无副作用。 + FileTaskEntity foreign = task(3411L, "instance-b", "RUNNING"); + foreign.setRequestJson("{\"ownerInstanceId\":\"instance-b\"}"); + assertThrows(TaskOwnerMismatchException.class, () -> + service.ensureTaskOwnedByCurrentInstance(foreign, "submit shop data crawl result")); + + doThrow(new RuntimeException("db down")).when(fileTaskMapper).selectList(any()); + service.finalizeOwnedStaleTasks(); + assertTrue(lastScan.isEmpty(), "DB 异常时扫描降级为空"); + } + + // ---- helpers ---- + + private ShopDataCrawlCreateTaskVo createTask(long taskId) { + ShopDataCrawlCreateTaskRequest request = new ShopDataCrawlCreateTaskRequest(); + request.setUserId(7L); + request.setCountryCodes(List.of("DE")); + ShopDataCrawlTaskItemDto item = new ShopDataCrawlTaskItemDto(); + item.setShopName(SHOP_NAME); + request.setItems(List.of(item)); + + ZiniaoShopMatchResultVo matched = new ZiniaoShopMatchResultVo(); + matched.setMatched(true); + matched.setShopId("shop-1"); + matched.setPlatform("AMAZON"); + matched.setCompanyName("Demo Co"); + matched.setMatchStatus("MATCHED"); + matched.setMatchMessage("ok"); + lenient().when(shopDataCrawlResolveService.validateCountryCodes(any())).thenReturn(List.of("DE")); + lenient().when(shopDataCrawlResolveService.requireMatchedShop(nullable(String.class))).thenReturn(matched); + lenient().doNothing().when(taskResultItemService).replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any()); + lenient().doNothing().when(taskProgressSnapshotService) + .save(anyLong(), any(), any(), anyInt(), anyInt(), anyInt(), any(), any(), any()); + lenient().when(fileTaskMapper.insert(any(FileTaskEntity.class))).thenAnswer(invocation -> { + FileTaskEntity inserted = invocation.getArgument(0); + boolean duplicate = dbTasks.stream() + .anyMatch(t -> Objects.equals(t.getId(), inserted.getId()) + || Objects.equals(t.getTaskNo(), inserted.getTaskNo())); + if (duplicate) { + throw new org.springframework.dao.DuplicateKeyException("duplicate task"); + } + inserted.setId(taskId); + dbTasks.add(inserted); + return 1; + }); + lenient().when(fileResultMapper.insert(any(FileResultEntity.class))).thenReturn(1); + + return service.createTask(request); + } + + private FileTaskEntity dbTask(long taskId) { + for (FileTaskEntity t : dbTasks) { + if (Objects.equals(t.getId(), taskId)) { + return t; + } + } + return null; + } + + private FileTaskEntity task(long id, String ownerInstanceId, String status) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType(MODULE_TYPE); + task.setStatus(status); + task.setOwnerInstanceId(ownerInstanceId); + task.setRequestJson(ownerInstanceId == null ? "{}" : "{\"ownerInstanceId\":\"" + ownerInstanceId + "\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + return task; + } + + private void runStaleScan() { + service.finalizeOwnedStaleTasks(); + } + + /** + * 按 SQL 片段中出现的列名与参数占位符顺序提取 owner_instance_id / status 的查询值， + * 模拟 MySQL 按显式列过滤（与生产查询的语义一致，仅用于筛选 dbTasks）。 + */ + @SuppressWarnings("unchecked") + private List applyTaskScanFilter(Wrapper wrapper, List candidates) { + if (!(wrapper instanceof LambdaQueryWrapper query)) { + return candidates; + } + String sql = query.getSqlSegment(); + Map params = query.getParamNameValuePairs(); + final String[] ownerFilter = {null}; + final String[] statusFilter = {null}; + if (sql != null && sql.contains("ownerInstanceId")) { + Matcher matcher = Pattern.compile("([a-zA-Z_]+)\\s*=\\s*#\\{ew\\.paramNameValuePairs\\.([A-Za-z0-9]+)\\}") + .matcher(sql); + while (matcher.find()) { + String column = matcher.group(1); + Object value = params.get(matcher.group(2)); + if ("ownerInstanceId".equals(column) && value instanceof String s) { + ownerFilter[0] = s; + } else if ("status".equals(column) && value instanceof String s) { + statusFilter[0] = s; + } + } + } + return candidates.stream() + .filter(t -> ownerFilter[0] == null || Objects.equals(ownerFilter[0], t.getOwnerInstanceId())) + .filter(t -> statusFilter[0] == null || Objects.equals(statusFilter[0], t.getStatus())) + .toList(); + } +}