task-129: shopdatacrawl 事务收缩（/result 无长事务契约固化 + deleteTask 纯计算抽静态纯函数）+ 8 条边界测试

2026-09-02 04:15:44 +08:00
parent ecc62f095b
commit 5e71e1bb3d
2 changed files with 340 additions and 8 deletions
@@ -686,19 +686,13 @@ public class ShopDataCrawlTaskService {
            List<FileResultEntity> taskRows = listTaskRows(taskId);
            try (DailyLockSet dailyLocks = acquireDailyLocks(taskRows)) {
                ensureDailySyncCompletedBeforeDelete(taskRows);
-                Set<Long> removedResultIds = taskRows.stream()
-                        .map(FileResultEntity::getId)
-                        .filter(id -> id != null && id > 0)
-                        .collect(java.util.stream.Collectors.toSet());
+                Set<Long> 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<String> 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<String> resultFileUrls = collectResultFileUrls(taskRows, dailyResult.obsoleteObjectKeys());
                fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
                        .eq(FileResultEntity::getTaskId, taskId)
                        .eq(FileResultEntity::getModuleType, MODULE_TYPE));
@@ -713,6 +707,32 @@ public class ShopDataCrawlTaskService {
        }
    }

+    /** 删除任务的纯计算：收集有效结果行 id 集合（过滤 null/非正数），无副作用。 */
+    static Set<Long> collectResultIds(List<FileResultEntity> 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<String> collectResultFileUrls(List<FileResultEntity> taskRows, List<String> obsoleteObjectKeys) {
+        List<String> 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<FileResultEntity> taskRows) {
        if (taskRows == null || taskRows.isEmpty()) {
            throw new BusinessException("后台店铺数据尚未完成同步，暂不能删除任务");
@@ -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<String> 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;
+    }
+}