task-125: similarasin /result 纯计算移出事务（payload 哈希在 prepare 阶段预计算，落库直接引用）+ 9 条事务边界守门测试

2026-09-02 03:34:15 +08:00
parent 56afee0386
commit 79b3efd365
2 changed files with 325 additions and 4 deletions
@@ -1030,15 +1030,17 @@ public class SimilarAsinTaskService {
        if (existing != null) {
            return new PreparedSubmittedChunk(
                    taskId, scopeKey, scopeHash, chunkIndex, chunkTotal, done, request.getError(),
-                    null, null, false, taskMetadata);
+                    null, null, false, taskMetadata, null);
        }
        String payloadJson = writeJson(flattenSubmittedRows(request), "结果序列化失败");
        String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(
                MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
        boolean localFallback = transientPayloadStorageService.wasLastStoreLocalFallback();
+        // 纯计算在事务外完成：payload 哈希预计算，persist 落库时直接引用
        return new PreparedSubmittedChunk(
                taskId, scopeKey, scopeHash, chunkIndex, chunkTotal, done, request.getError(),
-                payloadJson, storedPayload, localFallback, taskMetadata);
+                payloadJson, storedPayload, localFallback, taskMetadata,
+                DigestUtil.sha256Hex(payloadJson));
    }

    private SubmittedTaskMetadata readSubmittedTaskMetadata(FileTaskEntity task) {
@@ -1146,7 +1148,7 @@ public class SimilarAsinTaskService {
            chunk.setChunkIndex(prepared.chunkIndex());
            chunk.setChunkTotal(prepared.chunkTotal());
            chunk.setPayloadJson(prepared.storedPayload());
-            chunk.setPayloadHash(DigestUtil.sha256Hex(prepared.payloadJson()));
+            chunk.setPayloadHash(prepared.payloadHash());
            chunk.setCreatedAt(LocalDateTime.now());
            chunk.setUpdatedAt(LocalDateTime.now());
            try {
@@ -5750,7 +5752,8 @@ public class SimilarAsinTaskService {
                                          String payloadJson,
                                          String storedPayload,
                                          boolean localFallback,
-                                          SubmittedTaskMetadata taskMetadata) {
+                                          SubmittedTaskMetadata taskMetadata,
+                                          String payloadHash) {
    }

    private record PersistSubmittedChunkResult(SubmitContext context,
@@ -0,0 +1,318 @@
+package com.nanri.aiimage.modules.similarasin.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.common.service.DistributedJobLockService;
+import com.nanri.aiimage.config.InstanceMetadata;
+import com.nanri.aiimage.config.SimilarAsinProperties;
+import com.nanri.aiimage.config.StorageProperties;
+import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
+import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
+import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper;
+import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinSubmitResultRequest;
+import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder;
+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.FileTaskEntity;
+import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
+import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
+import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
+import com.nanri.aiimage.modules.task.service.TaskFileJobService;
+import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.junit.jupiter.api.AfterEach;
+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.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
+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-125：similarasin /result 提交事务边界守门。
+ * 纯计算（分组/校验/payload 哈希）在 prepareSubmittedChunk 中、事务开始前完成；
+ * 落库（chunk insert / scope upsert）仍在事务内；回滚语义不变。
+ */
+@ExtendWith(MockitoExtension.class)
+class SimilarAsinTaskServiceTxBoundaryTest {
+
+    private static final Long TASK_ID = 21879L;
+    private static final String PARSED_POINTER = "rustfs:task-parsed/similar-asin/21879/payload.json";
+    private static final String CHUNK_POINTER = "rustfs:task-chunk/similar-asin/21879/chunk.json";
+    private static final String STORED_CHUNK_POINTER = "\"" + CHUNK_POINTER + "\"";
+
+    @Mock private LocalFileStorageService localFileStorageService;
+    @Mock private OssStorageService ossStorageService;
+    @Mock private StorageProperties storageProperties;
+    @Mock private FileTaskMapper fileTaskMapper;
+    @Mock private FileResultMapper fileResultMapper;
+    @Mock private TaskScopeStateMapper taskScopeStateMapper;
+    @Mock private TaskChunkMapper taskChunkMapper;
+    @Mock private SimilarAsinFilterConditionMapper filterConditionMapper;
+    @Spy private ObjectMapper objectMapper = new ObjectMapper();
+    @Mock private SimilarAsinTaskCacheService taskCacheService;
+    @Mock private SimilarAsinProperties properties;
+    @Mock private TaskFileJobService taskFileJobService;
+    @Mock private TaskDistributedLockService taskDistributedLockService;
+    @Mock private TaskProgressSnapshotService taskProgressSnapshotService;
+    @Mock private TransientPayloadStorageService transientPayloadStorageService;
+    @Mock private PlatformTransactionManager transactionManager;
+    @Mock private DistributedJobLockService distributedJobLockService;
+    @Mock private InstanceMetadata instanceMetadata;
+    @Mock private SimilarAsinImageEmbedder imageEmbedder;
+    @Mock private SimilarAsinImagePrefetchService imagePrefetchService;
+    @Mock private TransactionStatus transactionStatus;
+
+    @InjectMocks private SimilarAsinTaskService service;
+
+    private final AtomicBoolean transactionActive = new AtomicBoolean();
+    private final AtomicReference<TaskChunkEntity> insertedChunk = new AtomicReference<>();
+    private final AtomicReference<String> storedPayloadJson = new AtomicReference<>();
+    private final AtomicBoolean storageOutsideTx = new AtomicBoolean(true);
+
+    @BeforeAll
+    static void initializeMybatisMetadata() {
+        MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
+        TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
+    }
+
+    @BeforeEach
+    void setUpTransactionAndLock() {
+        lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
+        lenient().when(taskDistributedLockService.acquire(
+                        eq(SimilarAsinTaskService.MODULE_TYPE), anyLong(), any(Duration.class), eq(10_000L)))
+                .thenReturn(mock(TaskDistributedLockService.LockHandle.class));
+        lenient().when(transactionManager.getTransaction(any(TransactionDefinition.class)))
+                .thenAnswer(invocation -> {
+                    transactionActive.set(true);
+                    return transactionStatus;
+                });
+        lenient().doAnswer(invocation -> {
+            transactionActive.set(false);
+            return null;
+        }).when(transactionManager).commit(transactionStatus);
+        lenient().doAnswer(invocation -> {
+            transactionActive.set(false);
+            return null;
+        }).when(transactionManager).rollback(transactionStatus);
+        lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of());
+        lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of());
+        lenient().when(taskChunkMapper.selectCount(any())).thenReturn(1L);
+        lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null);
+        lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
+                        eq(SimilarAsinTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString()))
+                .thenAnswer(invocation -> {
+                    storedPayloadJson.set(invocation.getArgument(4));
+                    storageOutsideTx.set(!transactionActive.get());
+                    return STORED_CHUNK_POINTER;
+                });
+        lenient().when(transientPayloadStorageService.wasLastStoreLocalFallback()).thenReturn(false);
+        lenient().doAnswer(invocation -> {
+            TaskChunkEntity chunk = invocation.getArgument(0);
+            assertTrue(transactionActive.get(), "chunk 落库必须在事务内");
+            chunk.setId(301L);
+            insertedChunk.set(chunk);
+            return 1;
+        }).when(taskChunkMapper).insert(any(TaskChunkEntity.class));
+        lenient().doAnswer(invocation -> {
+            TaskScopeStateEntity scope = invocation.getArgument(0);
+            scope.setId(401L);
+            return 1;
+        }).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
+    }
+
+    @AfterEach
+    void shutdownExecutors() {
+        service.shutdownAssembleExecutor();
+    }
+
+    @Test
+    void payloadHashIsPrecomputedInPrepareBeforeTransaction() throws Exception {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        Object prepared = ReflectionTestUtils.invokeMethod(
+                service, "prepareSubmittedChunk", TASK_ID, request(false));
+        assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()),
+                ReflectionTestUtils.getField(prepared, "payloadHash"),
+                "prepare 阶段必须产出预计算的 payload 哈希（事务开始前可用）");
+        // prepare 是纯计算：不触发任何落库
+        verify(taskChunkMapper, never()).insert(any(TaskChunkEntity.class));
+        verify(taskScopeStateMapper, never()).insert(any(TaskScopeStateEntity.class));
+    }
+
+    @Test
+    void pureComputeStaysOutsideTransactionAndPersistenceInside() throws Exception {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+
+        TaskChunkEntity chunk = insertedChunk.get();
+        assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()), chunk.getPayloadHash(),
+                "payload 哈希必须在 prepare 阶段预计算，落库行直接使用");
+        assertTrue(storageOutsideTx.get(), "prepare 阶段（哈希计算所在）不得处于事务内");
+        // 事务必须晚于 prepare 开始，chunk 落库在事务内
+        var order = inOrder(transientPayloadStorageService, transactionManager, taskChunkMapper);
+        order.verify(transientPayloadStorageService).storeChunkPayloadVersioned(
+                eq(SimilarAsinTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString());
+        order.verify(transactionManager).getTransaction(any(TransactionDefinition.class));
+        order.verify(taskChunkMapper).insert(any(TaskChunkEntity.class));
+    }
+
+    @Test
+    void precomputedHashMatchesComputedFromSamePayloadJson() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+
+        String payloadJson = storedPayloadJson.get();
+        assertEquals(DigestUtil.sha256Hex(payloadJson), insertedChunk.get().getPayloadHash());
+    }
+
+    @Test
+    void persistenceStaysInsideTransaction() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+
+        verify(taskChunkMapper).insert(any(TaskChunkEntity.class));
+        verify(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
+        verify(transactionManager).commit(transactionStatus);
+        verify(transactionManager, never()).rollback(transactionStatus);
+    }
+
+    @Test
+    void persistFailureRollsBackAndComputeHasNoSideEffect() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+        org.mockito.Mockito.doThrow(new IllegalStateException("scope upsert failed"))
+                .when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
+
+        assertThrows(IllegalStateException.class, () -> service.submitResult(TASK_ID, request(false)));
+
+        verify(transactionManager).rollback(transactionStatus);
+        verify(transactionManager, never()).commit(transactionStatus);
+        verify(taskCacheService, never()).touchTaskHeartbeat(TASK_ID);
+        assertTrue(storageOutsideTx.get(), "计算阶段先于事务执行，回滚不影响已完成的 prepare");
+    }
+
+    @Test
+    void repeatedSubmissionComputesIdenticalHash() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+        String first = insertedChunk.get().getPayloadHash();
+        service.submitResult(TASK_ID, request(false));
+
+        assertEquals(first, insertedChunk.get().getPayloadHash(), "相同入参重复提交哈希必须一致");
+        verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class));
+    }
+
+    @Test
+    void repeatedNonFinalSubmissionStaysSafe() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+        service.submitResult(TASK_ID, request(false));
+
+        verify(taskCacheService, times(2)).touchTaskHeartbeat(TASK_ID);
+        assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()), insertedChunk.get().getPayloadHash());
+    }
+
+    @Test
+    void movedComputationMethodsCarryNoTransactionAnnotation() throws Exception {
+        assertNull(SimilarAsinTaskService.class
+                        .getDeclaredMethod("prepareSubmittedChunk", Long.class, SimilarAsinSubmitResultRequest.class)
+                        .getAnnotation(Transactional.class),
+                "prepareSubmittedChunk 不得带 @Transactional");
+        boolean persistUnannotated = java.util.Arrays.stream(SimilarAsinTaskService.class.getDeclaredMethods())
+                .filter(method -> method.getName().equals("persistSubmittedChunk"))
+                .allMatch(method -> method.getAnnotation(Transactional.class) == null);
+        assertTrue(persistUnannotated, "persistSubmittedChunk 保持无注解（事务由调用方 inNewTransaction 控制）");
+    }
+
+    @Test
+    void insertedChunkSnapshotUnchanged() {
+        FileTaskEntity task = runningTask("instance-a");
+        when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
+
+        service.submitResult(TASK_ID, request(false));
+
+        TaskChunkEntity chunk = insertedChunk.get();
+        assertEquals(TASK_ID, chunk.getTaskId());
+        assertEquals(SimilarAsinTaskService.MODULE_TYPE, chunk.getModuleType());
+        assertEquals("similar-asin-21879", chunk.getScopeKey());
+        assertEquals(DigestUtil.sha256Hex("similar-asin-21879"), chunk.getScopeHash());
+        assertEquals(0, chunk.getChunkIndex());
+        assertEquals(1, chunk.getChunkTotal());
+        assertEquals(STORED_CHUNK_POINTER, chunk.getPayloadJson(), "payloadJson 存存储指针（现状不变）");
+        assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()), chunk.getPayloadHash());
+        org.junit.jupiter.api.Assertions.assertNotNull(chunk.getCreatedAt());
+        org.junit.jupiter.api.Assertions.assertNotNull(chunk.getUpdatedAt());
+    }
+
+    private SimilarAsinSubmitResultRequest request(boolean done) {
+        SimilarAsinSubmitResultRequest request = new SimilarAsinSubmitResultRequest();
+        request.setSubmissionId("similar-asin-21879");
+        request.setChunkIndex(0);
+        request.setChunkTotal(1);
+        request.setDone(done);
+        return request;
+    }
+
+    private FileTaskEntity runningTask(String owner) {
+        FileTaskEntity task = new FileTaskEntity();
+        task.setId(TASK_ID);
+        task.setModuleType(SimilarAsinTaskService.MODULE_TYPE);
+        task.setStatus("RUNNING");
+        task.setUserId(7L);
+        String resultJson = "{\"parsedPayloadRef\":\"" + PARSED_POINTER + "\"";
+        if (owner != null) {
+            resultJson += ",\"ownerInstanceId\":\"" + owner + "\"";
+        }
+        task.setResultJson(resultJson + "}");
+        return task;
+    }
+}