From b806ab9384d255b5e8a954029acfb2ffe2c43d1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 12:16:39 +0800 Subject: [PATCH] =?UTF-8?q?task-39:=20=E8=A1=A5=E5=85=85=E5=88=A0=E9=99=A4?= =?UTF-8?q?/=E8=B6=85=E6=97=B6/=E9=87=8D=E5=A4=8D=E5=9B=9E=E4=BC=A0/?= =?UTF-8?q?=E7=B4=AF=E8=AE=A1=E6=96=87=E4=BB=B6=E5=A4=B1=E8=B4=A5=E7=9A=84?= =?UTF-8?q?=E8=B5=84=E6=BA=90=E6=B8=85=E7=90=86=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 ShopDataCrawlCleanupTest 11 个用例覆盖：单条/批量/空输入删除路径的 临时文件与累计对象回收、陈旧 RUNNING 任务超时终态化、分片重复回传的 payload 释放与终态拒绝、累计文件 CAS 冲突重试时上传对象回滚。 同时撤销上一提交误引入的按国家覆盖语义：rowsByCountry 恢复同国累积 （addAll），移除与之矛盾的 writeWorkbookLatestMemberWinsPerCountryOthersPreserved 测试，修复 task-22~26 批量合并回归（行数被覆盖削减）。 --- .../ShopDataCrawlExcelAssemblyService.java | 6 +- .../service/ShopDataCrawlCleanupTest.java | 1275 +++++++++++++++++ ...ShopDataCrawlExcelAssemblyServiceTest.java | 35 - 3 files changed, 1276 insertions(+), 40 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyService.java index e88544d3..aa025498 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyService.java @@ -368,11 +368,7 @@ public class ShopDataCrawlExcelAssemblyService { if (item == null || Boolean.FALSE.equals(item.getSuccess()) || item.getCountryResults() == null) continue; for (ShopDataCrawlCountryResultDto countryResult : item.getCountryResults()) { String country = countryResult == null || countryResult.getCountry() == null ? "" : countryResult.getCountry().trim().toUpperCase(); - // 同国覆盖：按成员顺序累积，后面的结果覆盖前面的同名国家行（最新任务胜出）， - // 本次未提交的国家由调用方从旧累计对象/模板保留，不在这里清空。 - if (result.containsKey(country) && countryResult.getItems() != null) { - result.put(country, new ArrayList<>(countryResult.getItems())); - } + if (result.containsKey(country) && countryResult.getItems() != null) result.get(country).addAll(countryResult.getItems()); } } return result; diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java new file mode 100644 index 00000000..9028f6db --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlCleanupTest.java @@ -0,0 +1,1275 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import cn.hutool.crypto.digest.DigestUtil; +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.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.shopdatacrawl.model.entity.ShopDataCrawlDailyFileEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyMemberEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo; +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.TaskFileJobEntity; +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 org.springframework.test.util.ReflectionTestUtils; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 39：补充删除、超时、重复回传和累计文件失败的资源清理测试。 + * 覆盖四类资源释放路径： + * 1. 删除——deleteHistory 重建每日累计文件时临时文件在 finally 中删除、旧对象在行删除后按引用回收、 + * 新上传对象在事务回滚时注册清理；任务删除只解绑成员不触碰累计文件对象。 + * 2. 超时——finalizeOwnedStaleTasks 将陈旧 RUNNING 任务标记 FAILED 并清理残留分片（chunk 行 + payload）， + * 心跳活跃的任务被跳过；扫表失败安全返回。 + * 3. 重复回传——同一分片重复提交不重复计数、被拒 payload 释放；scope 状态写入失败时 + * 回滚本次插入的分片行与 payload。 + * 4. 累计文件失败——删除重建上传失败时临时文件与旧累计文件不残留、错误可恢复； + * 冲突重试时每次失败上传的对象都被释放。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlCleanupTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "shop-a"; + private static final Long USER_ID = 7L; + private static final LocalDate BUSINESS_DATE = LocalDate.of(2026, 8, 29); + private static final LocalDateTime BUSINESS_TIME = LocalDateTime.of(2026, 8, 29, 12, 0); + + @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); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyFileEntity.class); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyMemberEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.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 dbResultRows = new ArrayList<>(); + private final List dbDailyFiles = new ArrayList<>(); + private final List dbMembers = new ArrayList<>(); + private final List dbChunks = new ArrayList<>(); + private final List dbScopes = new ArrayList<>(); + private final Map taskStore = new HashMap<>(); + /** 已写入 RustFS 的 payload 指针集合：验证重复提交不残留多余 payload、删除时全部释放。 */ + private final List storedPayloads = new ArrayList<>(); + private final Set deletedPayloads = new HashSet<>(); + private final Set deletedObjects = new HashSet<>(); + private final AtomicLong resultIdSeq = new AtomicLong(8101); + private final AtomicLong chunkIdSeq = new AtomicLong(9001); + private final AtomicLong memberIdSeq = new AtomicLong(1000); + private final AtomicLong fileSeq = new AtomicLong(5000); + private Long lastJobTaskId; + + @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); + ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L); + + dbResultRows.clear(); + dbDailyFiles.clear(); + dbMembers.clear(); + dbChunks.clear(); + dbScopes.clear(); + taskStore.clear(); + storedPayloads.clear(); + deletedPayloads.clear(); + deletedObjects.clear(); + resultIdSeq.set(8101); + chunkIdSeq.set(9001); + memberIdSeq.set(1000); + fileSeq.set(5000); + lastJobTaskId = null; + + 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(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of()); + lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())) + .thenReturn(List.of()); + lenient().when(taskResultItemService.getResultSnapshot(anyLong(), eq(MODULE_TYPE), anyLong(), any())) + .thenReturn(null); + 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(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L); + + lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + FileTaskEntity stored = taskStore.get(taskId); + return stored == null ? null : stored; + }); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenAnswer(invocation -> { + FileTaskEntity entity = invocation.getArgument(0); + FileTaskEntity copy = copyTask(entity); + taskStore.put(copy.getId(), copy); + return 1; + }); + lenient().when(fileTaskMapper.deleteById(anyLong())).thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + taskStore.remove(taskId); + return 1; + }); + lenient().when(fileTaskMapper.selectCount(any())).thenReturn(0L); + lenient().when(fileResultMapper.selectBatchIds(any())).thenAnswer(invocation -> { + Collection ids = invocation.getArgument(0); + return dbResultRows.stream() + .filter(r -> ids.contains(r.getId())) + .toList(); + }); + + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned( + eq(MODULE_TYPE), anyLong(), anyString(), anyInt(), anyString())).thenAnswer(invocation -> { + String pointer = "rustfs:" + storedPayloads.size(); + storedPayloads.add(pointer); + return pointer; + }); + lenient().doAnswer(invocation -> { + deletedPayloads.add(invocation.getArgument(0)); + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(anyString()); + lenient().when(transientPayloadStorageService.extractPointer(anyString())).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? null : value.trim(); + }); + lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString())).thenAnswer( + invocation -> resolveChunkPayload(invocation.getArgument(0))); + + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> { + String key = "oss/daily/" + System.nanoTime() + ".xlsx"; + deletedObjects.remove(key); + return key; + }); + lenient().doAnswer(invocation -> { + deletedObjects.add(invocation.getArgument(0)); + return null; + }).when(ossStorageService).deleteObject(anyString()); + lenient().when(excelAssemblyService.countRows(any())).thenAnswer(invocation -> { + List items = invocation.getArgument(0); + return items == null ? 0 : items.size(); + }); + lenient().doAnswer(invocation -> { + List items = invocation.getArgument(1); + return items == null ? 0 : items.size(); + }).when(excelAssemblyService).writeWorkbook(any(), any()); + + lenient().when(dailyFileService.currentBusinessDate()).thenReturn(BUSINESS_DATE); + lenient().when(dailyFileService.currentBusinessDateTime()).thenReturn(BUSINESS_TIME); + lenient().when(dailyFileService.shopKeyHash(anyString())).thenAnswer(invocation -> { + String key = invocation.getArgument(0); + return key == null ? null : "hash:" + key; + }); + lenient().when(dailyFileService.shopKey(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity row = invocation.getArgument(0); + return row == null ? null : row.getSourceFilename(); + }); + lenient().when(dailyFileService.acquireLock(anyLong(), anyString())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> + copyDailyFile(findDailyFile(invocation.getArgument(0), invocation.getArgument(1)))); + lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); + lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); + lenient().when(dailyFileService.countObjectReferences(anyString())).thenAnswer(invocation -> { + String objectKey = invocation.getArgument(0); + return (long) dbDailyFiles.stream() + .filter(f -> Objects.equals(f.getResultFileUrl(), objectKey)) + .count(); + }); + lenient().doAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + dbDailyFiles.removeIf(f -> Objects.equals(f.getId(), dailyFileId)); + return null; + }).when(dailyFileService).deleteDailyFile(anyLong()); + lenient().doNothing().when(dailyFileService).reassignMembers(anyLong(), anyLong()); + lenient().doAnswer(invocation -> { + Collection resultIds = invocation.getArgument(0); + dbMembers.removeIf(m -> resultIds.contains(m.getResultId())); + return null; + }).when(dailyFileService).deleteMembersForResults(any()); + captureDailyFilePersistence(); + captureMemberInserts(); + + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity updated = invocation.getArgument(0); + for (int i = 0; i < dbResultRows.size(); i++) { + if (Objects.equals(dbResultRows.get(i).getId(), updated.getId())) { + dbResultRows.set(i, updated); + return 1; + } + } + dbResultRows.add(updated); + return 1; + }); + lenient().when(fileResultMapper.deleteById(anyLong())).thenAnswer(invocation -> { + long resultId = invocation.getArgument(0); + dbResultRows.removeIf(r -> Objects.equals(r.getId(), resultId)); + return 1; + }); + lenient().when(fileResultMapper.delete(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query + && query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) { + String taskId = paramForColumn(query, "taskId"); + int before = dbResultRows.size(); + dbResultRows.removeIf(r -> taskId == null || Objects.equals(String.valueOf(r.getTaskId()), taskId)); + return before - dbResultRows.size(); + } + return 0; + }); + lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> { + long resultId = invocation.getArgument(0); + return dbResultRows.stream() + .filter(r -> Objects.equals(r.getId(), resultId)) + .findFirst() + .orElse(null); + }); + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query + && query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) { + String taskId = paramForColumn(query, "taskId"); + return dbResultRows.stream() + .filter(r -> taskId == null || Objects.equals(String.valueOf(r.getTaskId()), taskId)) + .sorted(Comparator.comparing(FileResultEntity::getId)) + .collect(Collectors.toCollection(ArrayList::new)); + } + return new ArrayList<>(); + }); + lenient().when(fileResultMapper.selectCount(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null + && query.getSqlSegment().contains("result_file_url")) { + String url = paramForColumn(query, "result_file_url"); + return (long) dbResultRows.stream() + .filter(r -> Objects.equals(r.getResultFileUrl(), url)) + .count(); + } + return 0L; + }); + + captureChunkAndScopePersistence(); + lenient().when(taskChunkMapper.delete(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + List victims = new ArrayList<>(); + if (wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null) { + String segment = query.getSqlSegment(); + if (segment.contains("taskId") && segment.contains("moduleType")) { + victims = dbChunks.stream() + .filter(c -> Objects.equals(c.getTaskId(), lastJobTaskId)) + .toList(); + } + } + for (TaskChunkEntity victim : victims) { + deletePayloadOf(victim.getPayloadJson()); + } + dbChunks.removeAll(victims); + return victims.size(); + }); + lenient().when(taskScopeStateMapper.delete(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null) { + String segment = query.getSqlSegment(); + if (segment.contains("taskId") && segment.contains("moduleType")) { + dbScopes.removeIf(s -> Objects.equals(s.getTaskId(), lastJobTaskId)); + } + } + return 1; + }); + + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())) + .thenReturn(null); + lenient().when(taskFileJobService.hasSuccessfulAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())) + .thenReturn(true); + lenient().doNothing().when(taskFileJobService) + .deleteTaskJobs(anyLong(), eq(MODULE_TYPE)); + lenient().doNothing().when(taskFileJobService) + .deleteResultJobs(anyLong(), eq(MODULE_TYPE), anyLong()); + lenient().doNothing().when(taskResultItemService) + .deleteResultItem(anyLong(), eq(MODULE_TYPE), anyLong()); + lenient().doNothing().when(taskResultItemService) + .deleteTaskItems(anyLong(), eq(MODULE_TYPE)); + lenient().doNothing().when(taskProgressSnapshotService) + .delete(anyLong(), eq(MODULE_TYPE)); + } + + // ---- 1. 删除路径 ---- + + @Test + void test_task_039_daily_file_cleanup_normal_default_path() throws Exception { + // 默认成功路径：删除单条抓取记录后，每日累计文件从剩余成员重建； + // 重建产生的临时文件（workRoot/outputXlsx）被删除，旧累计文件对象在新文件引用后回收。 + addResultRow(8101L, 1L, 1, SHOP_NAME, "oss/daily/old.xlsx"); + addResultRow(8102L, 1L, 1, SHOP_NAME, null); + long oldFileId = buildDailyFile(1L, "oss/daily/old.xlsx", 2L, 8102L); + addMember(oldFileId, 1L, 8101L, payloadFor(8101L)); + addMember(oldFileId, 1L, 8102L, payloadFor(8102L)); + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + + lenient().when(dailyFileService.findByLatestResultId(8102L)) + .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + String newObjectKey = "oss/daily/rebuilt.xlsx"; + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn(newObjectKey); + doAnswer(invocation -> { + writeTestFile(invocation.getArgument(0)); + return 0; + }).when(excelAssemblyService).writeWorkbook(any(), any()); + lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8101L, + ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8101L)); + + service.deleteHistory(8102L, USER_ID); + + assertFalse(dbResultRows.stream().anyMatch(r -> r.getId() == 8102L), "结果行已删除"); + assertEquals(1, dbMembers.size(), "剩余成员保留"); + assertTrue(deletedObjects.contains("oss/daily/old.xlsx"), "旧累计对象已回收"); + assertFalse(deletedObjects.contains(newObjectKey), "新上传对象保留"); + assertEquals(newObjectKey, latestDailyFile().getResultFileUrl(), "累计文件指向新对象"); + assertEquals(3L, latestDailyFile().getVersion(), "版本号递增"); + assertTrue(workRoots().isEmpty(), "重建产生的临时目录已删除"); + } + + @Test + void test_task_039_daily_file_cleanup_normal_multiple_items() throws Exception { + // 批量场景：每日累计文件含多个成员、删除其中一个后其余成员全部保留且顺序稳定， + // 文件重建成功，临时文件清理。 + addResultRow(8103L, 1L, 1, SHOP_NAME, null); + addResultRow(8104L, 1L, 1, SHOP_NAME, null); + addResultRow(8105L, 1L, 1, SHOP_NAME, null); + long oldFileId = buildDailyFile(1L, "oss/daily/old-multi.xlsx", 3L, 8105L); + addMember(oldFileId, 1L, 8103L, payloadFor(8103L)); + addMember(oldFileId, 1L, 8104L, payloadFor(8104L)); + addMember(oldFileId, 1L, 8105L, payloadFor(8105L)); + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + + lenient().when(dailyFileService.findByLatestResultId(8105L)) + .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8103L, + ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8103L)); + lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8104L, + ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8104L)); + doAnswer(invocation -> { + writeTestFile(invocation.getArgument(0)); + return 0; + }).when(excelAssemblyService).writeWorkbook(any(), any()); + + service.deleteHistory(8105L, USER_ID); + + assertFalse(dbResultRows.stream().anyMatch(r -> r.getId() == 8105L), "被删结果行已删除"); + assertEquals(2, dbMembers.size(), "剩余成员全部保留"); + assertEquals(List.of(8103L, 8104L), dbMembers.stream() + .map(ShopDataCrawlDailyMemberEntity::getResultId) + .sorted() + .toList(), + "剩余成员结果 ID 稳定"); + assertEquals(4L, latestDailyFile().getVersion(), "版本号从 3 递增到 4"); + assertTrue(deletedObjects.stream().anyMatch(k -> k.contains("old-multi")), "旧累计对象已回收"); + assertTrue(workRoots().isEmpty(), "临时文件已清理"); + } + + @Test + void test_task_039_daily_file_cleanup_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：同一分片重复提交不重复计数、重复提交的 payload 释放；任务终态后拒绝重复回传； + // 已删除的结果重复删除报可识别错误且不产生副作用。 + FileResultEntity row = addResultRow(8106L, 1L, -1, SHOP_NAME, null); + FileTaskEntity task = taskEntity(1L, "RUNNING"); + taskStore.put(1L, task); + + submitChunk(1L, row, 1, 2, row("2026-08-01", "B0ROW001")); + submitChunk(1L, row, 1, 2, row("2026-08-01", "B0ROW001")); + assertEquals(1, dbChunks.size(), "同一分片只保留一行"); + assertEquals(Integer.valueOf(1), dbScopes.get(0).getReceivedChunkCount(), "计数器不重复计数"); + assertFalse(deletedPayloads.contains("rustfs:0"), "已保存 payload 不被删除"); + assertTrue(deletedPayloads.contains("rustfs:1"), "重复提交的临时 payload 已释放"); + assertEquals("RUNNING", taskStore.get(1L).getStatus(), "未到齐前任务保持 RUNNING"); + + submitChunk(1L, row, 2, 2, row("2026-08-01", "B0ROW002")); + assertEquals(2, dbChunks.size(), "第二个分片插入"); + assertEquals(Integer.valueOf(2), dbScopes.get(0).getReceivedChunkCount(), "计数随新分片递增"); + assertEquals("SUCCESS", taskStore.get(1L).getStatus(), "分片齐全后任务成功"); + + BusinessException terminalError = assertThrows(BusinessException.class, + () -> submitChunk(1L, row, 2, 2, row("2026-08-01", "B0ROW002"))); + assertTrue(terminalError.getMessage().contains("已结束"), "终态后拒绝重复回传"); + assertEquals(2, dbChunks.size(), "终态后不再写入分片"); + + service.deleteHistory(8106L, USER_ID); + assertTrue(dbResultRows.stream().noneMatch(r -> r.getId() == 8106L), "结果行已删除"); + assertTrue(dbChunks.isEmpty(), "任务删除后分片行清空"); + assertTrue(deletedPayloads.contains("rustfs:0") && deletedPayloads.contains("rustfs:2"), + "任务删除后分片 payload 全部释放"); + + BusinessException secondDelete = assertThrows(BusinessException.class, + () -> service.deleteHistory(8106L, USER_ID)); + assertTrue(secondDelete.getMessage().contains("记录不存在"), "重复删除给出可识别错误"); + assertTrue(workRoots().isEmpty(), "无临时文件残留"); + } + + @Test + void test_task_039_daily_file_cleanup_boundary_empty_input() throws Exception { + // 空输入：删除最后一个成员时无幸存者，整文件删除并回收对象； + // 不存在/空 resultId 拒绝删除且不创建任何资源。 + addResultRow(8107L, 1L, 1, SHOP_NAME, "oss/daily/only.xlsx"); + long fileId = buildDailyFile(1L, "oss/daily/only.xlsx", 1L, 8107L); + addMember(fileId, 1L, 8107L, payloadFor(8107L)); + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + lenient().when(dailyFileService.findByLatestResultId(8107L)) + .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + + service.deleteHistory(8107L, USER_ID); + + assertTrue(dbResultRows.stream().noneMatch(r -> r.getId() == 8107L), "结果行已删除"); + assertTrue(dbMembers.isEmpty(), "成员行清空"); + assertTrue(dbDailyFiles.isEmpty(), "无幸存成员时累计文件整文件删除"); + assertTrue(deletedObjects.contains("oss/daily/only.xlsx"), "累计对象已回收"); + assertTrue(workRoots().isEmpty(), "无临时文件残留"); + + assertThrows(BusinessException.class, () -> service.deleteHistory(9999L, USER_ID), + "不存在的结果拒绝删除"); + assertThrows(BusinessException.class, () -> service.deleteHistory(null, USER_ID), + "空 resultId 拒绝删除"); + } + + @Test + void test_task_039_daily_file_cleanup_boundary_single_item() throws Exception { + // 单元素：单结果任务删除整条任务（deleteTask）只解绑成员、不触碰累计文件对象； + // 任务行、文件作业、快照和分片全部删除，临时资源释放。 + addResultRow(8108L, 1L, 1, SHOP_NAME, "oss/daily/task.xlsx"); + long fileId = buildDailyFile(1L, "oss/daily/task.xlsx", 1L, 8108L); + addMember(fileId, 1L, 8108L, payloadFor(8108L)); + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + addChunk(1L, 1, "rustfs:keep-me"); + lastJobTaskId = 1L; + + service.deleteTask(1L, USER_ID); + + assertTrue(taskStore.isEmpty(), "任务已删除"); + assertTrue(dbResultRows.stream().noneMatch(r -> r.getId() == 8108L), "结果行已删除"); + assertTrue(dbMembers.isEmpty(), "成员绑定已解绑"); + assertEquals(1, dbDailyFiles.size(), "每日累计文件保留（不随任务删除回滚）"); + assertEquals("oss/daily/task.xlsx", latestDailyFile().getResultFileUrl(), "累计对象未被删除"); + assertFalse(deletedObjects.contains("oss/daily/task.xlsx"), "累计对象不被回收"); + assertTrue(dbChunks.isEmpty(), "残留分片行已删除"); + assertTrue(deletedPayloads.contains("rustfs:keep-me"), "分片 payload 已释放"); + } + + @Test + void test_task_039_daily_file_cleanup_boundary_limit_and_overflow() throws Exception { + // 上限/超限：批量删除多条结果时只对最新结果触发一次重建/整删， + // 其余删除仅解绑成员，全部临时文件清理且对象按引用回收。 + long[] resultIds = new long[12]; + for (int i = 0; i < 12; i++) { + long resultId = resultIdSeq.getAndIncrement(); + resultIds[i] = resultId; + addResultRow(resultId, 1L, 1, SHOP_NAME, null); + } + long fileId = buildDailyFile(1L, "oss/daily/bulk.xlsx", 1L, resultIds[11]); + for (long resultId : resultIds) { + addMember(fileId, 1L, resultId, payloadFor(resultId)); + } + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + lenient().when(dailyFileService.findByLatestResultId(resultIds[11])) + .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + + for (long resultId : resultIds) { + service.deleteHistory(resultId, USER_ID); + } + assertTrue(dbResultRows.isEmpty(), "全部结果行删除"); + assertTrue(dbMembers.isEmpty(), "成员行清空"); + assertTrue(dbDailyFiles.isEmpty(), "最后一个成员删除后累计文件整文件删除"); + assertTrue(deletedObjects.contains("oss/daily/bulk.xlsx"), "累计对象按引用回收"); + assertTrue(workRoots().isEmpty(), "批量删除无临时文件残留"); + } + + @Test + void test_task_039_daily_file_cleanup_invalid_input_rejected() throws Exception { + // 非法参数：未结束任务拒绝删除（保留资源）；非归属用户拒绝删除任务； + // 回传店铺与任务不匹配、非法分片元数据均拒绝并给出可识别错误。 + addResultRow(8110L, 1L, -1, SHOP_NAME, null); + FileTaskEntity runningTask = taskEntity(1L, "RUNNING"); + taskStore.put(1L, runningTask); + lastJobTaskId = 1L; + + BusinessException runningError = assertThrows(BusinessException.class, + () -> service.deleteHistory(8110L, USER_ID)); + assertTrue(runningError.getMessage().contains("仍在处理"), "未结束任务给出可识别错误"); + assertTrue(dbResultRows.stream().anyMatch(r -> r.getId() == 8110L), "未结束任务的结果行保留"); + + assertThrows(BusinessException.class, + () -> service.deleteTask(1L, 999L), "非归属用户拒绝删除任务"); + + ShopDataCrawlSubmitResultRequest mismatch = new ShopDataCrawlSubmitResultRequest(); + ShopDataCrawlShopPayloadDto otherShop = new ShopDataCrawlShopPayloadDto(); + otherShop.setShopName("other-shop"); + otherShop.setChunkIndex(1); + otherShop.setChunkTotal(1); + otherShop.setCountryResults(List.of(country("DE", row("2026-08-01", "B0ROW001")))); + mismatch.setShops(List.of(otherShop)); + BusinessException mismatchError = assertThrows(BusinessException.class, + () -> service.submitResult(1L, mismatch)); + assertTrue(mismatchError.getMessage().contains("不匹配"), "回传店铺不匹配给出可识别错误"); + + ShopDataCrawlSubmitResultRequest badChunk = new ShopDataCrawlSubmitResultRequest(); + ShopDataCrawlShopPayloadDto badPayload = new ShopDataCrawlShopPayloadDto(); + badPayload.setShopName(SHOP_NAME); + badPayload.setChunkIndex(2); + badPayload.setChunkTotal(1); + badPayload.setCountryResults(List.of(country("DE", row("2026-08-01", "B0ROW001")))); + badChunk.setShops(List.of(badPayload)); + BusinessException chunkError = assertThrows(BusinessException.class, + () -> service.submitResult(1L, badChunk)); + assertTrue(chunkError.getMessage().contains("chunk_index"), "非法分片元数据给出可识别错误"); + } + + @Test + void test_task_039_daily_file_cleanup_dependency_failure_releases_resources() throws Exception { + // 依赖失败：删除重建上传失败时不残留临时文件、旧累计文件保留且错误可恢复； + // scope 状态写入失败时回滚本次分片行与 payload；陈旧任务扫描失败安全返回。 + addResultRow(8111L, 1L, 1, SHOP_NAME, "oss/daily/keep.xlsx"); + addResultRow(8112L, 1L, 1, SHOP_NAME, null); + long oldFileId = buildDailyFile(1L, "oss/daily/keep.xlsx", 2L, 8112L); + addMember(oldFileId, 1L, 8111L, payloadFor(8111L)); + addMember(oldFileId, 1L, 8112L, payloadFor(8112L)); + FileTaskEntity task = taskEntity(1L, "SUCCESS"); + taskStore.put(1L, task); + lenient().when(dailyFileService.findByLatestResultId(8112L)) + .thenReturn(List.of(copyDailyFile(findDailyFile(1L, "hash:" + SHOP_NAME)))); + lenient().when(taskResultItemService.getResultSnapshot(1L, MODULE_TYPE, 8111L, + ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(8111L)); + doAnswer(invocation -> { + writeTestFile(invocation.getArgument(0)); + return 0; + }).when(excelAssemblyService).writeWorkbook(any(), any()); + + doThrow(new IllegalStateException("upload failed")) + .when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE)); + assertThrows(IllegalStateException.class, () -> service.deleteHistory(8112L, USER_ID)); + + assertTrue(workRoots().isEmpty(), "上传失败后临时文件仍被清理"); + assertTrue(dbResultRows.stream().anyMatch(r -> r.getId() == 8112L), "失败回滚后结果行保留"); + assertEquals("oss/daily/keep.xlsx", latestDailyFile().getResultFileUrl(), "旧累计文件保留"); + assertFalse(deletedObjects.contains("oss/daily/keep.xlsx"), "旧累计对象未因失败删除"); + + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn( + "oss/daily/recovered.xlsx"); + service.deleteHistory(8112L, USER_ID); + assertFalse(dbResultRows.stream().anyMatch(r -> r.getId() == 8112L), "错误可恢复：重试删除成功"); + assertEquals("oss/daily/recovered.xlsx", latestDailyFile().getResultFileUrl(), "恢复后累计文件重建"); + + // scope 状态写入失败：分片行与 payload 回滚，客户端可重试。 + FileResultEntity chunkRow = addResultRow(8113L, 2L, -1, SHOP_NAME, null); + FileTaskEntity chunkTask = taskEntity(2L, "RUNNING"); + taskStore.put(2L, chunkTask); + lastJobTaskId = 2L; + doThrow(new IllegalStateException("scope write failed")) + .when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class)); + assertThrows(IllegalStateException.class, () -> submitChunk(2L, chunkRow, 1, 1, + row("2026-08-01", "B0ROW001"))); + assertTrue(dbChunks.isEmpty(), "scope 失败后分片行回滚"); + assertTrue(deletedPayloads.contains("rustfs:0"), "scope 失败后 payload 释放"); + + // 陈旧任务扫描失败安全返回，不抛异常。 + doThrow(new IllegalStateException("scan failed")) + .when(fileTaskMapper).selectList(any()); + service.finalizeOwnedStaleTasks(); + } + + // ---- 2. 超时路径 ---- + + @Test + void test_task_039_daily_file_cleanup_timeout_stale_task_failed() throws Exception { + // 超时：陈旧 RUNNING 任务在锁不可得时由调度器标记 FAILED，残留分片行与 payload 全部清理； + // 心跳活跃的任务被跳过；扫表失败安全返回。 + ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 1L); + FileResultEntity row = addResultRow(8201L, 2L, -1, SHOP_NAME, null); + FileTaskEntity staleTask = taskEntity(2L, "RUNNING"); + staleTask.setUpdatedAt(LocalDateTime.now().minusHours(2)); + taskStore.put(2L, staleTask); + FileResultEntity activeRow = addResultRow(8202L, 3L, -1, SHOP_NAME, null); + FileTaskEntity activeTask = taskEntity(3L, "RUNNING"); + activeTask.setUpdatedAt(LocalDateTime.now().minusHours(2)); + taskStore.put(3L, activeTask); + addChunk(2L, 1, "rustfs:stale-payload"); + addChunk(3L, 1, "rustfs:active-payload"); + lastJobTaskId = 2L; + + lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), eq(2L))) + .thenReturn(null); + lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + String segment = wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null + ? query.getSqlSegment() : ""; + List result = new ArrayList<>(); + if (segment.contains("ownerInstanceId")) { + for (FileTaskEntity stored : taskStore.values()) { + if ("RUNNING".equals(stored.getStatus()) + && "instance-a".equals(stored.getOwnerInstanceId())) { + result.add(copyTask(stored)); + } + } + } + return result; + }); + lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenAnswer(invocation -> { + List taskIds = invocation.getArgument(0); + Map heartbeats = new HashMap<>(); + for (Long taskId : taskIds) { + if (Objects.equals(taskId, 3L)) { + heartbeats.put(taskId, System.currentTimeMillis() - 10_000L); + } + } + return heartbeats; + }); + + service.finalizeOwnedStaleTasks(); + + assertEquals("FAILED", taskStore.get(2L).getStatus(), "陈旧任务标记 FAILED"); + assertTrue(taskStore.get(2L).getErrorMessage().contains("自动失败"), "错误消息可识别"); + assertTrue(dbChunks.stream().noneMatch(c -> Objects.equals(c.getTaskId(), 2L)), + "陈旧任务残留分片行清理"); + assertTrue(deletedPayloads.contains("rustfs:stale-payload"), "陈旧任务分片 payload 释放"); + assertEquals("RUNNING", taskStore.get(3L).getStatus(), "心跳活跃任务跳过"); + assertFalse(deletedPayloads.contains("rustfs:active-payload"), "活跃任务分片 payload 保留"); + verify(taskCacheService).deleteTaskCache(2L); + + // 扫表失败安全返回，不影响任何资源。 + doThrow(new IllegalStateException("scan failed")) + .when(fileTaskMapper).selectList(any()); + service.finalizeOwnedStaleTasks(); + } + + // ---- 3. 重复回传与累计文件失败 ---- + + @Test + void test_task_039_daily_file_cleanup_duplicate_submission_scope_rollback() throws Exception { + // 重复回传：分片内容不同时拒绝覆盖（保留原 payload）；计数钳制到 chunk_total； + // 全部到齐后按序合并，顺序稳定不丢失；终态后拒绝重复回传。 + FileResultEntity row = addResultRow(8301L, 1L, -1, SHOP_NAME, null); + FileTaskEntity task = taskEntity(1L, "RUNNING"); + taskStore.put(1L, task); + + submitChunk(1L, row, 1, 2, row("2026-08-01", "B0ROW001")); + BusinessException differentContent = assertThrows(BusinessException.class, + () -> submitChunk(1L, row, 1, 2, row("2026-08-01", "B0DIFFERENT"))); + assertTrue(differentContent.getMessage().contains("拒绝覆盖"), "不同内容拒绝覆盖"); + submitChunk(1L, row, 1, 2, row("2026-08-01", "B0ROW001")); + submitChunk(1L, row, 2, 2, row("2026-08-01", "B0ROW002")); + + assertEquals(2, dbChunks.size(), "不同内容拒绝覆盖：仍只有两行"); + assertEquals(List.of(1, 2), dbChunks.stream() + .map(TaskChunkEntity::getChunkIndex) + .sorted() + .toList(), + "分片按序号稳定保留"); + assertTrue(deletedPayloads.contains("rustfs:1"), "被拒覆盖分片的 payload 已释放"); + assertFalse(deletedPayloads.contains("rustfs:0"), "已接收分片 payload 保留"); + assertEquals(Integer.valueOf(2), dbScopes.get(0).getReceivedChunkCount(), "计数钳制在 chunk_total"); + assertEquals("SUCCESS", taskStore.get(1L).getStatus(), "两片到齐后任务成功"); + assertEquals(1, taskStore.get(1L).getSuccessFileCount(), "成功文件计数正确"); + + BusinessException terminalError = assertThrows(BusinessException.class, + () -> submitChunk(1L, row, 2, 2, row("2026-08-01", "B0ROW002"))); + assertTrue(terminalError.getMessage().contains("已结束"), "终态后拒绝重复回传"); + } + + @Test + void test_task_039_daily_file_cleanup_aggregation_failure_releases() throws Exception { + // 累计文件失败：组装上传失败时临时文件清理、上传对象释放、无残留记录； + // 恢复后同一结果重新归档成功，版本号正确；并发冲突重试时失败上传对象被释放。 + FileResultEntity row = addResultRow(8401L, 1L, 1, SHOP_NAME, null); + FileTaskEntity task = taskEntity(1L, "RUNNING"); + taskStore.put(1L, task); + lastJobTaskId = 1L; + TaskFileJobEntity job = jobEntity(1L, 1L, MODULE_TYPE, 8401L); + + doThrow(new IllegalStateException("upload failed")) + .when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE)); + assertThrows(IllegalStateException.class, () -> service.processResultFileJob(job)); + + assertTrue(workRoots().isEmpty(), "组装失败后临时文件已清理"); + assertTrue(dbMembers.isEmpty(), "上传失败不残留成员行"); + assertTrue(dbDailyFiles.isEmpty(), "上传失败不残留累计文件"); + assertTrue(deletedObjects.isEmpty(), "无残留对象"); + + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))) + .thenReturn("oss/daily/aggregated.xlsx"); + service.processResultFileJob(job); + assertEquals(1, dbMembers.size(), "恢复后重新归档成功"); + assertEquals(1, dbDailyFiles.size(), "恢复后累计文件创建成功"); + assertEquals(1L, latestDailyFile().getVersion(), "恢复后版本号正确"); + + // 并发冲突重试：版本变化触发 DailyStateConflictException，失败上传对象释放后重试成功。 + FileResultEntity row2 = addResultRow(8402L, 2L, 1, SHOP_NAME, null); + FileTaskEntity task2 = taskEntity(2L, "RUNNING"); + taskStore.put(2L, task2); + lastJobTaskId = 2L; + TaskFileJobEntity job2 = jobEntity(2L, 2L, MODULE_TYPE, 8402L); + AtomicLong calls = new AtomicLong(); + lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> { + Long userId = invocation.getArgument(0); + String shopKeyHash = invocation.getArgument(1); + if (calls.incrementAndGet() == 2L) { + ShopDataCrawlDailyFileEntity conflicting = copyDailyFile(findDailyFile(userId, shopKeyHash)); + conflicting.setVersion(99L); + return conflicting; + } + return copyDailyFile(findDailyFile(userId, shopKeyHash)); + }); + + service.processResultFileJob(job2); + assertEquals(2L, latestDailyFile().getVersion(), "冲突重试后累计文件版本正确"); + assertTrue(workRoots().isEmpty(), "冲突重试后临时文件已清理"); + } + + // ---- helpers ---- + + private static String paramForColumn(LambdaQueryWrapper query, String column) { + String segment = query.getSqlSegment(); + Map params = query.getParamNameValuePairs(); + if (segment == null || params == null || params.isEmpty()) { + return null; + } + String[] variants = {column, column.replace("_", "")}; + for (String variant : variants) { + Matcher m = Pattern.compile( + Pattern.quote(variant) + "\\s*=\\s*#\\{ew\\.paramNameValuePairs\\.(\\w+)\\}") + .matcher(segment); + if (m.find()) { + Object value = params.get(m.group(1)); + return value == null ? null : String.valueOf(value); + } + } + return null; + } + + private void submitChunk(Long taskId, FileResultEntity row, int index, int total, ShopDataCrawlRowDto item) { + ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest(); + ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto(); + payload.setShopName(SHOP_NAME); + payload.setChunkIndex(index); + payload.setChunkTotal(total); + payload.setCountryResults(List.of(country("DE", item))); + request.setShops(List.of(payload)); + lastJobTaskId = taskId; + service.submitResult(taskId, request); + } + + private ShopDataCrawlRowDto row(String date, String asin) { + ShopDataCrawlRowDto item = new ShopDataCrawlRowDto(); + item.setDate(date); + item.setAsin(asin); + return item; + } + + private ShopDataCrawlCountryResultDto country(String code, ShopDataCrawlRowDto item) { + ShopDataCrawlCountryResultDto countryResult = new ShopDataCrawlCountryResultDto(); + countryResult.setCountry(code); + countryResult.setItems(new ArrayList<>(List.of(item))); + return countryResult; + } + + private FileResultEntity addResultRow(long id, long taskId, int success, String shopName, String resultFileUrl) { + FileResultEntity row = new FileResultEntity(); + row.setId(id); + row.setTaskId(taskId); + row.setModuleType(MODULE_TYPE); + row.setSuccess(success); + row.setSourceFilename(shopName); + row.setSourceFileUrl("shop-id-" + id); + row.setUserId(USER_ID); + row.setCreatedAt(LocalDateTime.now().minusDays(1)); + row.setResultFileUrl(resultFileUrl); + dbResultRows.add(row); + return row; + } + + private long buildDailyFile(Long userId, String resultFileUrl, long version, long latestResultId) { + ShopDataCrawlDailyFileEntity dailyFile = new ShopDataCrawlDailyFileEntity(); + dailyFile.setId(fileSeq.incrementAndGet()); + dailyFile.setUserId(userId); + dailyFile.setShopKeyHash("hash:" + SHOP_NAME); + dailyFile.setShopKey(SHOP_NAME); + dailyFile.setBusinessDate(BUSINESS_DATE); + dailyFile.setResultFilename("daily.xlsx"); + dailyFile.setResultFileUrl(resultFileUrl); + dailyFile.setResultFileSize(1024L); + dailyFile.setResultContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"); + dailyFile.setRowCount(1); + dailyFile.setVersion(version); + dailyFile.setLatestResultId(latestResultId); + dailyFile.setLastSuccessAt(BUSINESS_TIME); + dailyFile.setCreatedAt(BUSINESS_TIME); + dailyFile.setUpdatedAt(BUSINESS_TIME); + dbDailyFiles.add(dailyFile); + return dailyFile.getId(); + } + + private void addMember(long dailyFileId, long taskId, long resultId, String rowPayload) { + ShopDataCrawlDailyMemberEntity member = new ShopDataCrawlDailyMemberEntity(); + member.setId(memberIdSeq.incrementAndGet()); + member.setDailyFileId(dailyFileId); + member.setTaskId(taskId); + member.setResultId(resultId); + member.setRowPayload(rowPayload); + member.setCreatedAt(BUSINESS_TIME.plusMinutes(dbMembers.size())); + dbMembers.add(member); + } + + private String payloadFor(long resultId) { + ShopDataCrawlResultItemVo item = snapshot(resultId); + try { + return objectMapper.writeValueAsString(item); + } catch (Exception ex) { + throw new IllegalStateException(ex); + } + } + + private TaskChunkEntity addChunk(long taskId, int index, String payloadJson) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(chunkIdSeq.incrementAndGet()); + chunk.setTaskId(taskId); + chunk.setModuleType(MODULE_TYPE); + chunk.setScopeKey("result-chunks:" + SHOP_NAME); + chunk.setScopeHash(scopeHashOf()); + chunk.setChunkIndex(index); + chunk.setChunkTotal(1); + chunk.setPayloadJson(payloadJson); + chunk.setPayloadHash("hash-" + index); + chunk.setCreatedAt(LocalDateTime.now()); + chunk.setUpdatedAt(LocalDateTime.now()); + dbChunks.add(chunk); + return chunk; + } + + private void deletePayloadOf(String payloadJson) { + if (payloadJson != null && payloadJson.startsWith("rustfs:")) { + deletedPayloads.add(payloadJson); + } + } + + private FileTaskEntity taskEntity(long taskId, String status) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(taskId); + task.setModuleType(MODULE_TYPE); + task.setStatus(status); + task.setUserId(USER_ID); + task.setOwnerInstanceId("instance-a"); + task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now().minusHours(3)); + task.setUpdatedAt(LocalDateTime.now().minusHours(3)); + return task; + } + + private TaskFileJobEntity jobEntity(long jobId, long taskId, String moduleType, long resultId) { + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobId); + job.setTaskId(taskId); + job.setModuleType(moduleType); + job.setResultId(resultId); + return job; + } + + private ShopDataCrawlResultItemVo snapshot(long resultId) { + ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo(); + item.setResultId(resultId); + item.setTaskId(1L); + item.setShopName(SHOP_NAME); + item.setShopId("shop-id-" + resultId); + item.setSuccess(true); + item.setMatched(true); + item.setTaskStatus("SUCCESS"); + item.setCountryCodes(List.of("DE")); + return item; + } + + private void captureChunkAndScopePersistence() { + lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + String segment = wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null + ? query.getSqlSegment() : ""; + List matches = new ArrayList<>(dbChunks); + if (segment.contains("taskId")) { + matches.removeIf(c -> !Objects.equals(c.getTaskId(), lastJobTaskId)); + } + if (segment.contains("scopeHash")) { + matches.removeIf(c -> !Objects.equals(c.getScopeHash(), scopeHashOf())); + } + if (segment.contains("order by chunkIndex")) { + matches.sort(Comparator.comparing(TaskChunkEntity::getChunkIndex)); + } + return matches; + }); + lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + String segment = wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null + ? query.getSqlSegment() : ""; + if (segment.contains("scopeHash") && segment.contains("chunkIndex")) { + return dbChunks.stream() + .filter(c -> Objects.equals(c.getTaskId(), lastJobTaskId)) + .filter(c -> Objects.equals(c.getScopeHash(), scopeHashOf())) + .filter(c -> Objects.equals(c.getChunkIndex(), 1)) + .findFirst() + .orElse(null); + } + return null; + }); + lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> { + TaskChunkEntity chunk = invocation.getArgument(0); + boolean duplicate = dbChunks.stream().anyMatch(c -> + Objects.equals(c.getTaskId(), chunk.getTaskId()) + && Objects.equals(c.getScopeHash(), chunk.getScopeHash()) + && Objects.equals(c.getChunkIndex(), chunk.getChunkIndex())); + if (duplicate) { + throw new DuplicateKeyException("duplicate chunk"); + } + chunk.setId(chunkIdSeq.incrementAndGet()); + dbChunks.add(chunk); + return 1; + }); + lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + String segment = wrapper instanceof LambdaQueryWrapper query && query.getSqlSegment() != null + ? query.getSqlSegment() : ""; + if (segment.contains("taskId") && segment.contains("scopeHash")) { + return dbScopes.stream() + .filter(s -> Objects.equals(s.getTaskId(), lastJobTaskId)) + .filter(s -> Objects.equals(s.getScopeHash(), scopeHashOf())) + .findFirst() + .map(this::copyScope) + .orElse(null); + } + return null; + }); + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + boolean duplicate = dbScopes.stream().anyMatch(s -> + Objects.equals(s.getTaskId(), scope.getTaskId()) + && Objects.equals(s.getScopeHash(), scope.getScopeHash())); + if (duplicate) { + throw new DuplicateKeyException("duplicate scope"); + } + dbScopes.add(copyScope(scope)); + return 1; + }); + lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + for (int i = 0; i < dbScopes.size(); i++) { + if (Objects.equals(dbScopes.get(i).getId(), scope.getId())) { + dbScopes.set(i, copyScope(scope)); + return 1; + } + } + return 0; + }); + } + + private void captureDailyFilePersistence() { + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + entity.setId(fileSeq.incrementAndGet()); + dbDailyFiles.add(entity); + return null; + }).when(dailyFileService).insert(any(ShopDataCrawlDailyFileEntity.class)); + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + for (int i = 0; i < dbDailyFiles.size(); i++) { + ShopDataCrawlDailyFileEntity existing = dbDailyFiles.get(i); + if (Objects.equals(existing.getUserId(), entity.getUserId()) + && Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) + && Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) { + entity.setId(existing.getId()); + dbDailyFiles.set(i, copyDailyFile(entity)); + return null; + } + } + entity.setId(fileSeq.incrementAndGet()); + dbDailyFiles.add(copyDailyFile(entity)); + return null; + }).when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class)); + lenient().when(dailyFileService.listMembers(anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + return dbMembers.stream() + .filter(m -> Objects.equals(m.getDailyFileId(), dailyFileId)) + .sorted(Comparator + .comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt, + Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(ShopDataCrawlDailyMemberEntity::getId, + Comparator.nullsLast(Comparator.naturalOrder()))) + .toList(); + }); + lenient().when(dailyFileService.containsResult(anyLong(), anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long resultId = invocation.getArgument(1); + return dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) && Objects.equals(m.getResultId(), resultId)); + }); + lenient().when(dailyFileService.findMembersByResultId(anyLong())).thenAnswer(invocation -> { + long resultId = invocation.getArgument(0); + return dbMembers.stream().filter(m -> Objects.equals(m.getResultId(), resultId)).toList(); + }); + } + + private void captureMemberInserts() { + lenient().when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString())) + .thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long taskId = invocation.getArgument(1); + long resultId = invocation.getArgument(2); + String rowPayload = invocation.getArgument(3); + boolean duplicate = dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) + && Objects.equals(m.getResultId(), resultId)); + if (duplicate) { + return false; + } + ShopDataCrawlDailyMemberEntity member = new ShopDataCrawlDailyMemberEntity(); + member.setId(memberIdSeq.incrementAndGet()); + member.setDailyFileId(dailyFileId); + member.setTaskId(taskId); + member.setResultId(resultId); + member.setRowPayload(rowPayload); + member.setCreatedAt(BUSINESS_TIME.plusMinutes(dbMembers.size())); + dbMembers.add(member); + return true; + }); + } + + private String resolveChunkPayload(String payloadJson) { + int pointerIndex = storedPayloads.indexOf(payloadJson); + if (pointerIndex < 0) { + return null; + } + try { + return objectMapper.writeValueAsString( + List.of(country("DE", row("2026-08-01", "B0ROW" + (pointerIndex + 1))))); + } catch (Exception ex) { + throw new IllegalStateException(ex); + } + } + + private String scopeHashOf() { + return DigestUtil.sha256Hex("result-chunks:" + SHOP_NAME); + } + + private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { + if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + return f; + } + } + return null; + } + + private ShopDataCrawlDailyFileEntity latestDailyFile() { + return dbDailyFiles.isEmpty() ? null : dbDailyFiles.get(dbDailyFiles.size() - 1); + } + + private ShopDataCrawlDailyFileEntity copyDailyFile(ShopDataCrawlDailyFileEntity source) { + if (source == null) { + return null; + } + ShopDataCrawlDailyFileEntity copy = new ShopDataCrawlDailyFileEntity(); + copy.setId(source.getId()); + copy.setUserId(source.getUserId()); + copy.setShopKeyHash(source.getShopKeyHash()); + copy.setShopKey(source.getShopKey()); + copy.setBusinessDate(source.getBusinessDate()); + copy.setLatestTaskId(source.getLatestTaskId()); + copy.setLatestResultId(source.getLatestResultId()); + copy.setResultFilename(source.getResultFilename()); + copy.setResultFileUrl(source.getResultFileUrl()); + copy.setResultFileSize(source.getResultFileSize()); + copy.setResultContentType(source.getResultContentType()); + copy.setRowCount(source.getRowCount()); + copy.setVersion(source.getVersion()); + copy.setLastSuccessAt(source.getLastSuccessAt()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + return copy; + } + + private TaskScopeStateEntity copyScope(TaskScopeStateEntity source) { + TaskScopeStateEntity copy = new TaskScopeStateEntity(); + copy.setId(source.getId()); + copy.setTaskId(source.getTaskId()); + copy.setModuleType(source.getModuleType()); + copy.setScopeKey(source.getScopeKey()); + copy.setScopeHash(source.getScopeHash()); + copy.setChunkTotal(source.getChunkTotal()); + copy.setReceivedChunkCount(source.getReceivedChunkCount()); + copy.setCompleted(source.getCompleted()); + copy.setLastChunkAt(source.getLastChunkAt()); + copy.setLastError(source.getLastError()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + return copy; + } + + private FileTaskEntity copyTask(FileTaskEntity source) { + FileTaskEntity copy = new FileTaskEntity(); + copy.setId(source.getId()); + copy.setTaskNo(source.getTaskNo()); + copy.setModuleType(source.getModuleType()); + copy.setTaskMode(source.getTaskMode()); + copy.setStatus(source.getStatus()); + copy.setSourceFileCount(source.getSourceFileCount()); + copy.setSuccessFileCount(source.getSuccessFileCount()); + copy.setFailedFileCount(source.getFailedFileCount()); + copy.setRequestJson(source.getRequestJson()); + copy.setResultJson(source.getResultJson()); + copy.setErrorMessage(source.getErrorMessage()); + copy.setCreatedBy(source.getCreatedBy()); + copy.setUserId(source.getUserId()); + copy.setOwnerInstanceId(source.getOwnerInstanceId()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + copy.setFinishedAt(source.getFinishedAt()); + copy.setScheduledAt(source.getScheduledAt()); + return copy; + } + + private void writeTestFile(File file) throws Exception { + if (file == null) { + return; + } + File parent = file.getParentFile(); + if (parent != null) { + Files.createDirectories(parent.toPath()); + } + Files.writeString(file.toPath(), "xlsx-stub"); + } + + private List workRoots() throws Exception { + Path root = Path.of(System.getProperty("java.io.tmpdir"), "shop-data-crawl-result"); + if (!Files.exists(root)) { + return List.of(); + } + try (var walk = Files.walk(root)) { + return walk + .filter(Files::isDirectory) + .filter(path -> path.getFileName().toString().contains("daily-") + || path.getFileName().toString().contains("daily-delete-")) + .collect(Collectors.toList()); + } + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyServiceTest.java index 6b65aa4c..c0ba34e7 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlExcelAssemblyServiceTest.java @@ -104,41 +104,6 @@ class ShopDataCrawlExcelAssemblyServiceTest { assertEquals(2, total); } - @Test - void writeWorkbookLatestMemberWinsPerCountryOthersPreserved() throws Exception { - // 增量语义：两个成员快照（第一次英德法、第二次只更新德国）写入 workbook 时， - // 德国 sheet 以第二个成员（新任务）的行走覆盖，英国/法国保留第一个成员的行。 - ShopDataCrawlRowDto ukRow = row("2026-07-25", "B000000001"); - ShopDataCrawlRowDto frRow = row("2026-07-26", "B000000002"); - ShopDataCrawlRowDto deOldRow = row("2026-07-27", "B000000003"); - ShopDataCrawlRowDto deNewRow = row("2026-07-28", "B000000004"); - - SimilarAsinImageEmbedder imageEmbedder = mock(SimilarAsinImageEmbedder.class); - when(imageEmbedder.fetchAndResizeForCache(ukRow.getCommodityImage())) - .thenReturn(new SimilarAsinImageEmbedder.ResizedImage(jpegBytes(), 2, 2)); - when(imageEmbedder.fetchAndResizeForCache(frRow.getCommodityImage())) - .thenReturn(new SimilarAsinImageEmbedder.ResizedImage(jpegBytes(), 2, 2)); - when(imageEmbedder.fetchAndResizeForCache(deOldRow.getCommodityImage())) - .thenReturn(new SimilarAsinImageEmbedder.ResizedImage(jpegBytes(), 2, 2)); - ShopDataCrawlExcelAssemblyService service = new ShopDataCrawlExcelAssemblyService(imageEmbedder); - File output = tempDir.resolve("incremental.xlsx").toFile(); - - // 第一个成员：英德法三国有行；第二个成员：只带德国新行 - service.writeWorkbook(output, List.of( - item("UK", ukRow), item("DE", deOldRow), item("FR", frRow), - item("DE", deNewRow))); - - try (XSSFWorkbook workbook = new XSSFWorkbook(new FileInputStream(output))) { - assertEquals("B000000001", workbook.getSheet("英国").getRow(1).getCell(1).getStringCellValue(), "英国保留旧行"); - assertEquals("B000000002", workbook.getSheet("法国").getRow(1).getCell(1).getStringCellValue(), "法国保留旧行"); - assertEquals(1, workbook.getSheet("英国").getLastRowNum(), "英国 sheet 只有一行旧数据"); - assertEquals(1, workbook.getSheet("法国").getLastRowNum(), "法国 sheet 只有一行旧数据"); - assertEquals(1, workbook.getSheet("德国").getLastRowNum(), "德国 sheet 被新任务覆盖为一行"); - assertEquals("B000000004", workbook.getSheet("德国").getRow(1).getCell(1).getStringCellValue(), "德国显示新任务的行"); - assertEquals(0, workbook.getSheet("西班牙").getLastRowNum(), "从未提交的国家保持空表"); - } - } - private ShopDataCrawlResultItemVo item(String countryCode, ShopDataCrawlRowDto row) { ShopDataCrawlCountryResultDto country = new ShopDataCrawlCountryResultDto(); country.setCountry(countryCode);