From 5e71e1bb3d45ef90e2cbeed37e8c18f4c97a9d84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Wed, 2 Sep 2026 04:15:44 +0800 Subject: [PATCH] =?UTF-8?q?task-129:=20shopdatacrawl=20=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E6=94=B6=E7=BC=A9=EF=BC=88/result=20=E6=97=A0=E9=95=BF?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1=E5=A5=91=E7=BA=A6=E5=9B=BA=E5=8C=96=20+=20de?= =?UTF-8?q?leteTask=20=E7=BA=AF=E8=AE=A1=E7=AE=97=E6=8A=BD=E9=9D=99?= =?UTF-8?q?=E6=80=81=E7=BA=AF=E5=87=BD=E6=95=B0=EF=BC=89+=208=20=E6=9D=A1?= =?UTF-8?q?=E8=BE=B9=E7=95=8C=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/ShopDataCrawlTaskService.java | 36 +- ...hopDataCrawlTaskServiceTxBoundaryTest.java | 312 ++++++++++++++++++ 2 files changed, 340 insertions(+), 8 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceTxBoundaryTest.java 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 c991d6fc..3262d51b 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 @@ -686,19 +686,13 @@ public class ShopDataCrawlTaskService { List taskRows = listTaskRows(taskId); try (DailyLockSet dailyLocks = acquireDailyLocks(taskRows)) { ensureDailySyncCompletedBeforeDelete(taskRows); - Set removedResultIds = taskRows.stream() - .map(FileResultEntity::getId) - .filter(id -> id != null && id > 0) - .collect(java.util.stream.Collectors.toSet()); + Set removedResultIds = collectResultIds(taskRows); // A task deletion is only a frontend task-record cleanup. The daily // workbook is an independent backend aggregate and must not roll // back when its source task is removed. DailyDeletionResult dailyResult = preserveDailyForTaskDeletion(removedResultIds); registerUploadedObjectRollback(dailyResult.uploadedObjectKeys()); - List resultFileUrls = new ArrayList<>(dailyResult.obsoleteObjectKeys()); - resultFileUrls.addAll(taskRows.stream() - .map(FileResultEntity::getResultFileUrl).filter(url -> !blank(url)).distinct().toList()); - resultFileUrls = resultFileUrls.stream().filter(url -> !blank(url)).distinct().toList(); + List resultFileUrls = collectResultFileUrls(taskRows, dailyResult.obsoleteObjectKeys()); fileResultMapper.delete(new LambdaQueryWrapper() .eq(FileResultEntity::getTaskId, taskId) .eq(FileResultEntity::getModuleType, MODULE_TYPE)); @@ -713,6 +707,32 @@ public class ShopDataCrawlTaskService { } } + /** 删除任务的纯计算：收集有效结果行 id 集合（过滤 null/非正数），无副作用。 */ + static Set collectResultIds(List taskRows) { + if (taskRows == null || taskRows.isEmpty()) { + return Set.of(); + } + return taskRows.stream() + .filter(Objects::nonNull) + .map(FileResultEntity::getId) + .filter(id -> id != null && id > 0) + .collect(java.util.stream.Collectors.toSet()); + } + + /** 删除任务的纯计算：合并待清理对象键，过滤空白并去重，无副作用。 */ + static List collectResultFileUrls(List taskRows, List obsoleteObjectKeys) { + List urls = new ArrayList<>(obsoleteObjectKeys == null ? List.of() : obsoleteObjectKeys); + if (taskRows != null) { + for (FileResultEntity row : taskRows) { + if (row == null || row.getResultFileUrl() == null || row.getResultFileUrl().isBlank()) { + continue; + } + urls.add(row.getResultFileUrl()); + } + } + return urls.stream().filter(url -> url != null && !url.isBlank()).distinct().toList(); + } + private void ensureDailySyncCompletedBeforeDelete(List taskRows) { if (taskRows == null || taskRows.isEmpty()) { throw new BusinessException("后台店铺数据尚未完成同步，暂不能删除任务"); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceTxBoundaryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceTxBoundaryTest.java new file mode 100644 index 00000000..a5897b06 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceTxBoundaryTest.java @@ -0,0 +1,312 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import cn.hutool.crypto.digest.DigestUtil; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +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.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler; +import com.nanri.aiimage.modules.task.service.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.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.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.Transactional; + +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +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.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-129：shopdatacrawl 事务收缩。 + * 审计结论：/result 提交路径（submitResult → persistResultChunk/mergeShopPayload/ + * persistProgressOrSnapshot）无 @Transactional、无长事务——写为单条原子写或 + * executeShortTransaction 短事务，事务边界已窄；deleteTask 事务方法内的纯计算 + * 段抽为无事务静态纯函数。本测试固化上述契约。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlTaskServiceTxBoundaryTest { + + private static final Long TASK_ID = 4242L; + private static final Long USER_ID = 7L; + private static final String SHOP_KEY = "amazon.de"; + + @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; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @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; + @Mock private PlatformTransactionManager transactionManager; + @Mock private TaskProgressLightAssembler taskProgressLightAssembler; + + @InjectMocks private ShopDataCrawlTaskService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(taskPressureProperties.getDbSelectBatchSize()).thenReturn(500); + } + + @Test + void collectResultIdsFiltersBrokenRows() { + FileResultEntity ok = resultRow(1L); + FileResultEntity nullId = new FileResultEntity(); + FileResultEntity zeroId = new FileResultEntity(); + zeroId.setId(0L); + + assertEquals(java.util.Set.of(1L), + service.collectResultIds(java.util.Arrays.asList(ok, nullId, zeroId, null))); + } + + @Test + void collectResultFileUrlsDeduplicatesAndSkipsBlank() { + FileResultEntity a = resultRow(1L); + a.setResultFileUrl("url-a"); + FileResultEntity b = resultRow(2L); + b.setResultFileUrl("url-b"); + FileResultEntity dup = resultRow(3L); + dup.setResultFileUrl("url-a"); + FileResultEntity blank = resultRow(4L); + blank.setResultFileUrl(" "); + + assertEquals(List.of("old-key", "url-a", "url-b"), + service.collectResultFileUrls(List.of(a, b, dup, blank), List.of("old-key"))); + } + + @Test + void pureHelpersAreStaticAndCarryNoTransactionAnnotation() throws Exception { + for (String name : List.of("collectResultIds", "collectResultFileUrls")) { + Method method = java.util.Arrays.stream(ShopDataCrawlTaskService.class.getDeclaredMethods()) + .filter(candidate -> candidate.getName().equals(name)) + .findFirst() + .orElseThrow(); + assertTrue(Modifier.isStatic(method.getModifiers()), name + " 应为静态纯函数"); + assertNull(method.getAnnotation(Transactional.class), name + " 不得带 @Transactional"); + } + } + + @Test + void submitResultCarriesNoTransactionAnnotation() throws Exception { + Method submit = ShopDataCrawlTaskService.class.getMethod( + "submitResult", Long.class, ShopDataCrawlSubmitResultRequest.class); + assertNull(submit.getAnnotation(Transactional.class), + "submitResult 无 @Transactional（/result 提交路径无长事务）"); + } + + @Test + void submitResultPathUsesNoLongTransaction() throws Exception { + FileTaskEntity task = runningTask(); + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task); + when(fileResultMapper.selectList(any())).thenReturn(List.of(resultRow(1L))); + when(taskDistributedLockService.acquire(any(), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class), null); + + service.submitResult(TASK_ID, submitRequest(false)); + + verify(transactionManager, never()).getTransaction(any()); + verify(taskCacheService).touchTaskHeartbeat(TASK_ID); + verify(fileTaskMapper, times(1)).updateById(any(FileTaskEntity.class)); + } + + @Test + void chunkPathPersistsWithSingleInsertWithoutTransaction() throws Exception { + FileTaskEntity task = runningTask(); + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task); + FileResultEntity row = resultRow(1L); + row.setSourceFilename(SHOP_KEY); + when(fileResultMapper.selectList(any())).thenReturn(List.of(row)); + when(taskDistributedLockService.acquire(any(), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class), null); + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + when(transientPayloadStorageService.storeChunkPayloadVersioned( + eq("SHOP_DATA_CRAWL"), eq(TASK_ID), any(), eq(1), any())) + .thenReturn("\"rustfs:chunk\""); + lenient().when(transientPayloadStorageService.extractPointer("\"rustfs:chunk\"")) + .thenReturn("rustfs:chunk"); + lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenReturn("[{\"country\":\"DE\",\"items\":[{\"asin\":\"B0TEST1234\"}]}]"); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null); + lenient().when(taskChunkMapper.selectCount(any())).thenReturn(1L); + lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(completedChunk())); + + service.submitResult(TASK_ID, chunkRequest()); + + verify(transactionManager, never()).getTransaction(any()); + verify(taskChunkMapper).insert(any(TaskChunkEntity.class)); + verify(fileResultMapper).updateById(any(FileResultEntity.class)); + } + + @Test + void deleteTaskKeepsTransactionAnnotationAndPureCompute() throws Exception { + Method delete = ShopDataCrawlTaskService.class.getMethod("deleteTask", Long.class, Long.class); + assertTrue(delete.getAnnotation(Transactional.class) != null, + "deleteTask 落库删除必须保留 @Transactional（锁内删除语义不变）"); + // 删除路径使用抽出的纯函数（行为等价由既有 ShopDataCrawlCleanupTest 守门） + assertTrue(ShopDataCrawlTaskService.collectResultIds(List.of()).isEmpty()); + } + + @Test + void duplicateChunkPathIsIdempotentWithoutLongTransaction() throws Exception { + FileTaskEntity task = runningTask(); + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task); + FileResultEntity row = resultRow(1L); + row.setSourceFilename(SHOP_KEY); + when(fileResultMapper.selectList(any())).thenReturn(List.of(row)); + when(taskDistributedLockService.acquire(any(), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class), null); + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + java.util.concurrent.atomic.AtomicReference chunkPayload = + new java.util.concurrent.atomic.AtomicReference<>(); + when(transientPayloadStorageService.storeChunkPayloadVersioned( + eq("SHOP_DATA_CRAWL"), eq(TASK_ID), any(), eq(1), any())) + .thenAnswer(invocation -> { + chunkPayload.set(invocation.getArgument(4)); + return "\"rustfs:chunk\""; + }); + lenient().when(transientPayloadStorageService.extractPointer("\"rustfs:chunk\"")) + .thenReturn("rustfs:chunk"); + lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenReturn("[{\"country\":\"DE\",\"items\":[{\"asin\":\"B0TEST1234\"}]}]"); + lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> { + TaskChunkEntity winner = completedChunk(); + winner.setPayloadHash(DigestUtil.sha256Hex(chunkPayload.get())); + return winner; + }); + lenient().when(taskChunkMapper.selectCount(any())).thenReturn(1L); + lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> { + // 重复提交：scope 计数器已持久化（received=1），幂等判定 completed + com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity scope = + new com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity(); + scope.setId(901L); + scope.setTaskId(TASK_ID); + scope.setReceivedChunkCount(1); + scope.setChunkTotal(1); + return scope; + }); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(completedChunk())); + org.mockito.Mockito.doThrow(new org.springframework.dao.DuplicateKeyException("dup")) + .when(taskChunkMapper).insert(any(TaskChunkEntity.class)); + + service.submitResult(TASK_ID, chunkRequest()); + + verify(transactionManager, never()).getTransaction(any()); + verify(transientPayloadStorageService).deletePayloadIfPresent("\"rustfs:chunk\""); + } + + private FileTaskEntity runningTask() { + FileTaskEntity task = new FileTaskEntity(); + task.setId(TASK_ID); + task.setModuleType("SHOP_DATA_CRAWL"); + task.setStatus("RUNNING"); + task.setUserId(USER_ID); + task.setResultJson("{\"ownerInstanceId\":\"instance-a\"}"); + return task; + } + + private FileResultEntity resultRow(Long id) { + FileResultEntity row = new FileResultEntity(); + row.setId(id); + row.setTaskId(TASK_ID); + row.setSourceFilename(SHOP_KEY); + row.setModuleType("SHOP_DATA_CRAWL"); + return row; + } + + private TaskChunkEntity completedChunk() { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(801L); + chunk.setTaskId(TASK_ID); + chunk.setModuleType("SHOP_DATA_CRAWL"); + chunk.setScopeKey("result-chunks:" + SHOP_KEY); + chunk.setScopeHash("result-chunks-hash"); + chunk.setChunkIndex(1); + chunk.setChunkTotal(1); + chunk.setPayloadJson("\"rustfs:chunk\""); + return chunk; + } + + private ShopDataCrawlSubmitResultRequest submitRequest(boolean done) { + ShopDataCrawlShopPayloadDto shop = new ShopDataCrawlShopPayloadDto(); + shop.setShopName(SHOP_KEY); + shop.setShopDone(done); + ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest(); + request.setShops(List.of(shop)); + return request; + } + + private ShopDataCrawlSubmitResultRequest chunkRequest() { + ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); + row.setAsin("B0TEST1234"); + ShopDataCrawlCountryResultDto country = new ShopDataCrawlCountryResultDto(); + country.setCountry("DE"); + country.setItems(List.of(row)); + ShopDataCrawlShopPayloadDto shop = new ShopDataCrawlShopPayloadDto(); + shop.setShopName(SHOP_KEY); + shop.setChunkIndex(1); + shop.setChunkTotal(1); + shop.setCountryResults(List.of(country)); + ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest(); + request.setShops(List.of(shop)); + return request; + } +}