diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DoneFinalizationContractTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DoneFinalizationContractTest.java new file mode 100644 index 00000000..3ba8f539 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DoneFinalizationContractTest.java @@ -0,0 +1,282 @@ +package com.nanri.aiimage.modules.task.contract; + +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.service.SimilarAsinImagePrefetchService; +import com.nanri.aiimage.modules.similarasin.service.SimilarAsinLlmService; +import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskCacheService; +import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService; +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.FileResultEntity; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +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.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +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-134：done=true 封口时机硬约束契约（spec 07 §2）。 + * done=true 的最后一批：分片先落库 → 强制 finalize（处理剩余 + 生成结果文件）； + * 重复 done 安全不重复生成；done=false 只触心跳不封口。similarasin 路径统一守门。 + */ +@ExtendWith(MockitoExtension.class) +class DoneFinalizationContractTest { + + private static final Long TASK_ID = 7777L; + private static final String PARSED_POINTER = "rustfs:task-parsed/similar-asin/7777/payload.json"; + private static final String CHUNK_POINTER = "rustfs:task-chunk/similar-asin/7777/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 SimilarAsinLlmService similarAsinLlmService; + @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 AtomicReference taskState = new AtomicReference<>(); + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + 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))) + .thenReturn(transactionStatus); + lenient().doAnswer(invocation -> null).when(transactionManager).commit(transactionStatus); + lenient().doAnswer(invocation -> 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())) + .thenReturn(STORED_CHUNK_POINTER); + lenient().when(transientPayloadStorageService.wasLastStoreLocalFallback()).thenReturn(false); + lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenReturn("{\"allItems\":[{},{}],\"items\":[{},{}],\"sourceFiles\":[]}"); + lenient().doAnswer(invocation -> { + com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity chunk = invocation.getArgument(0); + chunk.setId(301L); + return 1; + }).when(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class)); + lenient().doAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + scope.setId(401L); + return 1; + }).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class)); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.insert(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity result = invocation.getArgument(0); + result.setId(601L); + return 1; + }); + lenient().when(taskFileJobService.enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString())) + .thenReturn(null); + } + + @AfterEach + void shutdownExecutors() { + service.shutdownAssembleExecutor(); + } + + private FileTaskEntity runningTask() { + FileTaskEntity task = new FileTaskEntity(); + task.setId(TASK_ID); + task.setModuleType(SimilarAsinTaskService.MODULE_TYPE); + task.setStatus("RUNNING"); + task.setUserId(7L); + task.setResultJson("{\"parsedPayloadRef\":\"" + PARSED_POINTER + "\",\"ownerInstanceId\":\"instance-a\"}"); + return task; + } + + private SimilarAsinSubmitResultRequest request(boolean done) { + SimilarAsinSubmitResultRequest request = new SimilarAsinSubmitResultRequest(); + request.setSubmissionId("similar-asin-7777"); + request.setChunkIndex(0); + request.setChunkTotal(1); + request.setDone(done); + return request; + } + + @Test + void doneTriggersFinalizeInsteadOfHeartbeat() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + verify(taskCacheService, never()).touchTaskHeartbeat(TASK_ID); + verify(taskFileJobService).enqueueAssembleResult(eq(TASK_ID), eq(SimilarAsinTaskService.MODULE_TYPE), + anyLong(), anyString()); + verify(fileTaskMapper).updateById(any(FileTaskEntity.class)); + } + + @Test + void chunkPersistsBeforeFinalizeOnDone() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + var order = inOrder(taskChunkMapper, taskFileJobService); + order.verify(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class)); + order.verify(taskFileJobService).enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString()); + } + + @Test + void doneProcessesRemainderWithFullRowCount() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + // resolvePayload 提供 2 行 → finalize 的 result 行数 = 全量 2（处理剩余） + org.mockito.ArgumentCaptor resultCaptor = + org.mockito.ArgumentCaptor.forClass(FileResultEntity.class); + verify(fileResultMapper).insert(resultCaptor.capture()); + assertEquals(2, resultCaptor.getValue().getRowCount(), "封口时结果行数必须覆盖剩余全量"); + } + + @Test + void doneGeneratesResultFileJob() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + verify(taskFileJobService).enqueueAssembleResult(eq(TASK_ID), eq(SimilarAsinTaskService.MODULE_TYPE), + eq(601L), anyString()); + } + + @Test + void repeatedDoneDoesNotDoubleFinalize() { + FileTaskEntity task = runningTask(); + taskState.set(task); + when(fileTaskMapper.selectById(TASK_ID)).thenAnswer(invocation -> taskState.get()); + lenient().doAnswer(invocation -> { + FileTaskEntity updated = invocation.getArgument(0); + taskState.set(updated); + return 1; + }).when(fileTaskMapper).updateById(any(FileTaskEntity.class)); + AtomicReference resultState = new AtomicReference<>(); + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> + resultState.get() == null ? List.of() : List.of(resultState.get())); + lenient().doAnswer(invocation -> { + FileResultEntity result = invocation.getArgument(0); + result.setId(601L); + resultState.set(result); + return 1; + }).when(fileResultMapper).insert(any(FileResultEntity.class)); + + service.submitResult(TASK_ID, request(true)); + // 第二次 done：result 行已存在（复用，不重复创建）；封口不重复生成 + service.submitResult(TASK_ID, request(true)); + + verify(fileResultMapper, times(1)).insert(any(FileResultEntity.class)); + } + + @Test + void doneTimingChunkFirstThenTaskUpdate() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + var order = inOrder(taskChunkMapper, fileTaskMapper); + order.verify(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class)); + order.verify(fileTaskMapper).updateById(any(FileTaskEntity.class)); + } + + @Test + void nonDoneOnlyTouchesHeartbeatNoFinalize() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(false)); + + verify(taskCacheService).touchTaskHeartbeat(TASK_ID); + verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString()); + } + + @Test + void doneContractFreezesFinalizeEffect() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request(true)); + + // 封口效应固定：result 行落库 + assemble job 入队 + 任务行更新 + verify(fileResultMapper).insert(any(FileResultEntity.class)); + verify(taskFileJobService).enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString()); + verify(fileTaskMapper).updateById(any(FileTaskEntity.class)); + verify(taskCacheService, never()).touchTaskHeartbeat(TASK_ID); + } +}