From 79b3efd365319306ae4bcbe27e113faafe100035 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 03:34:15 +0800 Subject: [PATCH] =?UTF-8?q?task-125:=20similarasin=20/result=20=E7=BA=AF?= =?UTF-8?q?=E8=AE=A1=E7=AE=97=E7=A7=BB=E5=87=BA=E4=BA=8B=E5=8A=A1=EF=BC=88?= =?UTF-8?q?payload=20=E5=93=88=E5=B8=8C=E5=9C=A8=20prepare=20=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=E9=A2=84=E8=AE=A1=E7=AE=97=EF=BC=8C=E8=90=BD=E5=BA=93?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=EF=BC=89+=209=20=E6=9D=A1?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1=E8=BE=B9=E7=95=8C=E5=AE=88=E9=97=A8=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/SimilarAsinTaskService.java | 11 +- .../SimilarAsinTaskServiceTxBoundaryTest.java | 318 ++++++++++++++++++ 2 files changed, 325 insertions(+), 4 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTxBoundaryTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 80f0aae0..62fd8fb7 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -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, diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTxBoundaryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTxBoundaryTest.java new file mode 100644 index 00000000..cd3e03e0 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTxBoundaryTest.java @@ -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 insertedChunk = new AtomicReference<>(); + private final AtomicReference 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; + } +}