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 ef2cfe28..9c17d889 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 @@ -1257,15 +1257,11 @@ public class ShopDataCrawlTaskService { validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal); List countryResults = copyCountryResults(incoming.getCountryResults()); + if (!hasProcessableChunkData(incoming.getCountryResults())) { + throw new BusinessException("店铺数据抓取结果分片内容为空，拒绝接收"); + } String payloadJson = writeJson(countryResults, "序列化店铺数据抓取结果分片失败"); String payloadHash = DigestUtil.sha256Hex(payloadJson); - TaskChunkEntity existing = findResultChunk(taskId, scopeHash, chunkIndex); - if (existing != null) { - validateExistingChunk(existing, chunkTotal, payloadHash); - int receivedChunkCount = countResultChunks(taskId, scopeHash); - persistResultScope(taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount); - return new ResultChunkReceipt(scopeHash, chunkTotal, receivedChunkCount >= chunkTotal); - } ensureRustfsPayloadStorageEnabled(); String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned( @@ -2349,6 +2345,30 @@ public class ShopDataCrawlTaskService { return copy; } + /** + * 分片是否含可处理数据：与 copyCountryResults 的拷贝语义一致 —— + * 任一国家含至少一个非空行（国家名非空白且行内容非空）即视为有数据。 + */ + private boolean hasProcessableChunkData(List results) { + if (results == null || results.isEmpty()) { + return false; + } + for (ShopDataCrawlCountryResultDto source : results) { + if (source == null || blank(source.getCountry())) { + continue; + } + if (source.getItems() == null) { + continue; + } + for (ShopDataCrawlRowDto sourceRow : source.getItems()) { + if (sourceRow != null && !rowEmpty(sourceRow)) { + return true; + } + } + } + return false; + } + private ShopDataCrawlRowDto copyRow(ShopDataCrawlRowDto source) { ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); row.setDate(trim(source.getDate())); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlChunkUpsertTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlChunkUpsertTest.java new file mode 100644 index 00000000..e56a428e --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlChunkUpsertTest.java @@ -0,0 +1,603 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +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.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.ShopDataCrawlCountryResultDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest; +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.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 org.springframework.dao.DuplicateKeyException; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.lenient; +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.when; + +/** + * Task 27：将 chunk 接收改为原子插入/幂等 upsert，减少先查后插。 + * persistResultChunk 去掉首包提交前的 findResultChunk 预查，改为直接 + * store payload + insert；唯一索引 uk_task_scope_chunk 兜底幂等 —— + * 重复提交（同内容）命中 DuplicateKeyException 后重查 winner 校验， + * 并清理本次重存的 payload；空分片（无可处理数据）在落库前被拒绝， + * 不创建 RustFS payload 与分片行。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlChunkUpsertTest { + + 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; + + private final List storedChunks = new ArrayList<>(); + private final List storedScopes = new ArrayList<>(); + private final Map rustfsPayloads = new LinkedHashMap<>(); + private FileTaskEntity task; + private FileResultEntity result; + private int nextPayloadId; + /** 模拟依赖故障开关：RustFS 存储失败 / 数据库插入失败 / 唯一键竞态失败。 */ + private boolean storeFails; + private boolean insertFails; + private boolean insertDuplicate; + + @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); + + storedChunks.clear(); + storedScopes.clear(); + rustfsPayloads.clear(); + nextPayloadId = 0; + storeFails = false; + insertFails = false; + insertDuplicate = false; + + 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.getAllShopMergedPayload(anyLong())).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(excelAssemblyService.countRows(any())).thenReturn(1); + + lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> { + Long taskId = invocation.getArgument(0); + return task != null && Objects.equals(taskId, task.getId()) ? task : null; + }); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> { + Long resultId = invocation.getArgument(0); + return result != null && Objects.equals(resultId, result.getId()) ? result : null; + }); + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of(); + }); + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1); + + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + + configureTransientPayloadStorage(); + configureChunkMapper(); + configureScopeMapper(); + } + + @Test + void test_task_027_chunk_normal_default_path() { + // 正常输入：首个分片直接原子插入，不再先查后插—— + // taskChunkMapper 不出现预查 selectOne，分片行与 RustFS payload 各落一份。 + givenRunningTask(1001L, 2001L); + + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))); + + assertEquals(1, storedChunks.size()); + assertEquals("rustfs:payload-1", storedChunks.get(0).getPayloadJson()); + assertEquals(1, rustfsPayloads.size()); + assertEquals(-1, result.getSuccess(), "1/2 未齐，任务继续运行"); + assertEquals("RUNNING", task.getStatus()); + verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class)); + verify(taskChunkMapper, never()).selectOne(any()); + verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString()); + } + + @Test + void test_task_027_chunk_normal_multiple_items() { + // 批量场景：乱序分片（2/2 先到、1/2 后到）全部原子插入，合并结果按 chunk_index 稳定， + // 全程无预查 selectOne；齐集时触发组装任务。 + givenRunningTask(1002L, 2002L); + + service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002")))); + + assertEquals(-1, result.getSuccess()); + assertEquals(1, storedChunks.size()); + verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString()); + + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))); + + assertEquals(1, result.getSuccess()); + assertEquals(2, storedChunks.size()); + assertEquals(2, rustfsPayloads.size()); + assertTrue(task.getResultJson().indexOf("B001") < task.getResultJson().indexOf("B002"), + "合并结果按 chunk_index 顺序稳定"); + verify(taskChunkMapper, never()).selectOne(any()); + verify(taskFileJobService).enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), + "task:" + task.getId() + ":owner:instance-a"); + verify(excelAssemblyService, never()).writeWorkbook(any(), any()); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE)); + } + + @Test + void test_task_027_chunk_normal_repeated_operation_is_idempotent() { + // 重复执行：同一分片重试靠唯一索引 uk_task_scope_chunk 兜底幂等—— + // 不产生重复分片行、重试重存的 payload 立即清理，insert 共两次（首次成功+重试被唯一键拒绝）。 + givenRunningTask(1003L, 2003L); + ShopDataCrawlSubmitResultRequest request = request(chunk(1, 2, "DE", row("2026-07-25", "B001"))); + + service.submitResult(task.getId(), request); + service.submitResult(task.getId(), request); + + assertEquals(1, storedChunks.size(), "重复提交不产生重复分片行"); + assertEquals(1, rustfsPayloads.size(), "重试重存的 payload 被清理"); + verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class)); + verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2"); + assertEquals(-1, result.getSuccess()); + } + + @Test + void test_task_027_chunk_boundary_empty_input() { + // 空输入：无可处理数据的分片（空 items / 空白国家）在落库前拒绝，不创建 payload 与分片行。 + givenRunningTask(1004L, 2004L); + + ShopDataCrawlShopPayloadDto emptyItems = legacyChunk(false, "DE", null); + emptyItems.setChunkIndex(1); + emptyItems.setChunkTotal(1); + emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of()))); + + BusinessException error = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(emptyItems))); + assertTrue(error.getMessage().contains("内容为空")); + + ShopDataCrawlShopPayloadDto blankCountry = chunk(1, 1, "DE", row("2026-07-25", "B001")); + blankCountry.setCountryResults(List.of(countryWithItems(" ", List.of(row("2026-07-25", "B001"))))); + + assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(blankCountry))); + + assertEquals(0, storedChunks.size(), "空分片不落库"); + assertEquals(0, rustfsPayloads.size(), "空分片不写 RustFS"); + verify(taskChunkMapper, never()).insert(any(TaskChunkEntity.class)); + } + + @Test + void test_task_027_chunk_boundary_single_item() { + // 单元素：单分片 1/1 原子插入即齐集，直接合并成功，无预查 selectOne。 + givenRunningTask(1005L, 2005L); + + service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001")))); + + assertEquals(1, result.getSuccess()); + assertEquals(1, storedChunks.size()); + assertEquals(1, rustfsPayloads.size()); + assertEquals(1, storedScopes.size()); + assertEquals(1, storedScopes.get(0).getReceivedChunkCount()); + assertEquals(1, storedScopes.get(0).getCompleted()); + assertTrue(task.getResultJson().contains("B001")); + verify(taskChunkMapper, never()).selectOne(any()); + } + + @Test + void test_task_027_chunk_boundary_limit_and_overflow() { + // 上限/超限：chunk_index 超 chunk_total 拒绝；5/5 分片全量到达时计数一致、顺序稳定、任务完成， + // 期间全部为原子插入（无预查），无重复对象累积。 + givenRunningTask(1006L, 2006L); + + BusinessException overflow = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(2, 1, "DE", row("2026-07-25", "B001"))))); + assertTrue(overflow.getMessage().contains("chunk_index")); + + String[] countries = {"UK", "DE", "FR", "ES", "IT"}; + String[] dates = {"2026-07-25", "2026-07-26", "2026-07-27", "2026-07-28", "2026-07-29"}; + for (int i = 1; i <= 5; i++) { + service.submitResult(task.getId(), request(chunk(i, 5, countries[i - 1], row(dates[i - 1], "B00" + i)))); + if (i < 5) { + assertEquals(-1, result.getSuccess(), i + "/5 未齐"); + } + } + + assertEquals(1, result.getSuccess(), "5/5 齐集后任务完成"); + assertEquals(5, storedChunks.size()); + assertEquals(5, storedScopes.get(0).getReceivedChunkCount()); + assertEquals(1, storedScopes.get(0).getCompleted()); + String json = task.getResultJson(); + for (int i = 1; i < 5; i++) { + assertTrue(json.indexOf("B00" + i) < json.indexOf("B00" + (i + 1)), "合并顺序按 chunk_index 稳定"); + } + verify(taskChunkMapper, times(5)).insert(any(TaskChunkEntity.class)); + verify(taskChunkMapper, never()).selectOne(any()); + } + + @Test + void test_task_027_chunk_invalid_input_rejected() { + // 非法参数：非正 chunk_index/chunk_total、跨分片改 chunk_total、同 index 不同内容 → + // 项目约定异常及可识别消息；重存 payload 不残留。 + givenRunningTask(1007L, 2007L); + + BusinessException zeroIndex = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001"))))); + assertTrue(zeroIndex.getMessage().contains("chunk_index")); + + BusinessException zeroTotal = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001"))))); + assertTrue(zeroTotal.getMessage().contains("chunk_total")); + + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))); + + BusinessException changedTotal = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(2, 3, "UK", row("2026-07-26", "B002"))))); + assertTrue(changedTotal.getMessage().contains("chunk_total")); + + BusinessException differentContent = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B099"))))); + assertTrue(differentContent.getMessage().contains("不同内容")); + + assertEquals(1, storedChunks.size(), "非法输入不产生额外分片行"); + assertEquals(1, rustfsPayloads.size(), "同 index 不同内容的重存 payload 被清理"); + verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class)); + verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2"); + } + + @Test + void test_task_027_chunk_dependency_failure_releases_resources() { + // 依赖失败：RustFS 存储异常/插入异常/唯一键竞态 winner 丢失 —— 临时 payload 清理、 + // 锁释放（故障后可重试成功）、不残留分片行。 + givenRunningTask(1008L, 2008L); + + storeFails = true; + assertThrows(RuntimeException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))))); + assertEquals(0, storedChunks.size()); + assertEquals(0, rustfsPayloads.size()); + storeFails = false; + + insertFails = true; + assertThrows(RuntimeException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))))); + assertEquals(0, storedChunks.size(), "插入失败不残留分片行"); + assertEquals(0, rustfsPayloads.size(), "插入失败清理已存 payload"); + verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-1"); + insertFails = false; + + service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))); + assertEquals(1, storedChunks.size(), "故障恢复后重试成功"); + assertEquals(1, rustfsPayloads.size()); + assertEquals(-1, result.getSuccess()); + + insertDuplicate = true; + assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))))); + insertDuplicate = false; + + service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002")))); + + assertEquals(1, result.getSuccess(), "竞态失败后可重试成功"); + assertEquals(2, storedChunks.size()); + assertEquals(2, rustfsPayloads.size()); + } + + private void configureTransientPayloadStorage() { + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned( + anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> { + if (storeFails) { + throw new RuntimeException("rustfs store down"); + } + String pointer = "rustfs:payload-" + (++nextPayloadId); + rustfsPayloads.put(pointer, invocation.getArgument(4)); + return pointer; + }); + lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + if (value == null) { + return null; + } + return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:") + ? value : null; + }); + lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> { + String pointer = invocation.getArgument(0); + String payload = rustfsPayloads.get(pointer); + if (payload == null) { + throw new IllegalStateException("missing test RustFS payload: " + pointer); + } + return payload; + }); + lenient().doAnswer(invocation -> { + rustfsPayloads.remove(invocation.getArgument(0)); + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(any()); + } + + private void configureChunkMapper() { + lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> { + if (insertDuplicate) { + throw new DuplicateKeyException("duplicate chunk key (race)"); + } + if (insertFails) { + throw new RuntimeException("db down"); + } + TaskChunkEntity chunk = invocation.getArgument(0); + boolean duplicate = storedChunks.stream().anyMatch(existing -> + Objects.equals(existing.getTaskId(), chunk.getTaskId()) + && Objects.equals(existing.getScopeHash(), chunk.getScopeHash()) + && Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex())); + if (duplicate) { + throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex()); + } + chunk.setId((long) storedChunks.size() + 1L); + storedChunks.add(chunk); + return 1; + }); + lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + Integer chunkIndex = queryInteger(query); + return storedChunks.stream() + .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex)) + .findFirst() + .orElse(null); + }); + lenient().when(taskChunkMapper.selectCount(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + return storedChunks.stream() + .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null)) + .count(); + }); + lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + return storedChunks.stream() + .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null)) + .sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex)) + .toList(); + }); + } + + private void configureScopeMapper() { + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + scope.setId((long) storedScopes.size() + 1L); + storedScopes.add(scope); + return 1; + }); + lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + return storedScopes.stream() + .filter(scope -> Objects.equals(taskId, scope.getTaskId())) + .filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash())) + .findFirst() + .orElse(null); + }); + lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenReturn(1); + lenient().when(taskScopeStateMapper.delete(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + int before = storedScopes.size(); + storedScopes.removeIf(scope -> taskId == null || Objects.equals(taskId, scope.getTaskId())); + return before - storedScopes.size(); + }); + } + + private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) { + return (taskId == null || Objects.equals(taskId, chunk.getTaskId())) + && (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash())) + && (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex())); + } + + private Long queryLong(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(Long.class::isInstance) + .map(Long.class::cast) + .findFirst() + .orElse(null); + } + + private Integer queryInteger(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(Integer.class::isInstance) + .map(Integer.class::cast) + .findFirst() + .orElse(null); + } + + private String queryScopeHash(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(String.class::isInstance) + .map(String.class::cast) + .filter(value -> value.length() == 64) + .findFirst() + .orElse(null); + } + + private void givenRunningTask(long taskId, long resultId) { + task = new FileTaskEntity(); + task.setId(taskId); + task.setUserId(7L); + task.setModuleType(MODULE_TYPE); + task.setStatus("RUNNING"); + task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + + result = new FileResultEntity(); + result.setId(resultId); + result.setTaskId(taskId); + result.setUserId(7L); + result.setModuleType(MODULE_TYPE); + result.setSourceFilename(SHOP_NAME); + result.setSourceFileUrl("shop-1"); + result.setSuccess(-1); + result.setCreatedAt(LocalDateTime.now()); + } + + private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) { + ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest(); + request.setShops(List.of(payload)); + return request; + } + + private ShopDataCrawlShopPayloadDto chunk(int chunkIndex, + int chunkTotal, + String country, + ShopDataCrawlRowDto row) { + ShopDataCrawlShopPayloadDto payload = legacyChunk(false, country, row); + payload.setChunkIndex(chunkIndex); + payload.setChunkTotal(chunkTotal); + return payload; + } + + private ShopDataCrawlShopPayloadDto legacyChunk(boolean shopDone, + String country, + ShopDataCrawlRowDto row) { + ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto(); + payload.setShopName(SHOP_NAME); + if (row != null) { + payload.setCountryResults(List.of(country(country, row))); + } + payload.setShopDone(shopDone); + return payload; + } + + private ShopDataCrawlCountryResultDto country(String country, ShopDataCrawlRowDto row) { + return countryWithItems(country, List.of(row)); + } + + private ShopDataCrawlCountryResultDto countryWithItems(String country, List items) { + ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto(); + result.setCountry(country); + result.setItems(items); + return result; + } + + private ShopDataCrawlRowDto row(String date, String asin) { + ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); + row.setDate(date); + row.setAsin(asin); + row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg"); + row.setInventorySales("10"); + row.setSalesRank("20"); + row.setPageViews("30"); + row.setUnitsSold("40"); + row.setPrice("50"); + row.setRecommendedOffer("60"); + return row; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java index 24ef3246..eac4ac5f 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java @@ -35,6 +35,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.Spy; import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.dao.DuplicateKeyException; import java.time.LocalDateTime; import java.util.ArrayList; @@ -213,7 +214,10 @@ class ShopDataCrawlTaskServiceChunkTest { assertEquals(1, storedChunks.size()); assertEquals(1, rustfsPayloads.size()); - verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class)); + // 原子插入路径：首次 insert 成功，重试命中唯一索引 uk_task_scope_chunk 抛键冲突被吸收； + // 重试重存的 payload 立即清理，不产生重复分片。 + verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class)); + verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2"); assertEquals(-1, result.getSuccess()); } @@ -388,6 +392,15 @@ class ShopDataCrawlTaskServiceChunkTest { private void configureChunkMapper() { lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> { TaskChunkEntity chunk = invocation.getArgument(0); + // 与 V30 的 uk_task_scope_chunk 唯一索引一致：同 (task_id, scope_hash, chunk_index) 重复插入抛键冲突。 + boolean duplicate = storedChunks.stream().anyMatch(existing -> + Objects.equals(existing.getTaskId(), chunk.getTaskId()) + && Objects.equals(existing.getScopeHash(), chunk.getScopeHash()) + && Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex())); + if (duplicate) { + throw new DuplicateKeyException("duplicate chunk key: " + + chunk.getScopeHash() + "/" + chunk.getChunkIndex()); + } chunk.setId((long) storedChunks.size() + 1L); storedChunks.add(chunk); return 1;