super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-34: 将 ownerInstanceId 从 JSON 查询迁移到显式列并补充索引

Browse Source
This commit is contained in:
黄自达
2026-08-29 20:44:21 +08:00
parent 790267560a
commit 11e19ef4f1
4 changed files with 413 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+24 -2
View File
@@ -116,12 +116,18 @@ public class ShopDataCrawlTaskService {
public void finalizeOwnedStaleTasks() {
long minutes = Math.max(1L, staleTimeoutMinutes);
long nowMillis = System.currentTimeMillis();
List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
List<FileTaskEntity> tasks;
try {
tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
.eq(FileTaskEntity::getStatus, "RUNNING")
.apply("JSON_UNQUOTE(JSON_EXTRACT(request_json, '$.ownerInstanceId')) = {0}", currentInstanceId())
.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<Long, Long> 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("");
backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/FileTaskEntity.java
+1
View File
@@ -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;
backend-java/src/main/resources/db/V92__file_task_owner_instance_column.sql
+3
View File
@@ -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`);
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlOwnerColumnTest.java
+381
View File
@@ -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<FileTaskEntity> dbTasks = new ArrayList<>();
/** 最近一次 stale 扫描返回的结果。 */
private final List<FileTaskEntity> 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<FileTaskEntity> 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<FileTaskEntity> applyTaskScanFilter(Wrapper<FileTaskEntity> wrapper, List<FileTaskEntity> candidates) {
if (!(wrapper instanceof LambdaQueryWrapper<?> query)) {
return candidates;
}
String sql = query.getSqlSegment();
Map<String, Object> 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();
}
}