diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/CleanupErrorContractTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/CleanupErrorContractTest.java new file mode 100644 index 00000000..a6a18fc4 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/CleanupErrorContractTest.java @@ -0,0 +1,283 @@ +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.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.dao.DuplicateKeyException; +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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * task-136：事务外清理异常不阻断契约（spec 07 §2）。 + * afterCommit 语义的清理抛异常：记录日志、响应不受影响、不进入重试循环、 + * 事务已提交、下次提交仍可再次清理、部分失败不影响后续流程。 + */ +@ExtendWith(MockitoExtension.class) +class CleanupErrorContractTest { + + private static final Long TASK_ID = 5555L; + private static final String PARSED_POINTER = "rustfs:task-parsed/similar-asin/5555/payload.json"; + private static final String CHUNK_POINTER = "rustfs:task-chunk/similar-asin/5555/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 AtomicBoolean transactionActive = new AtomicBoolean(); + private final AtomicInteger cleanupAttempts = new AtomicInteger(); + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.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))) + .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(0L); + lenient().when(taskScopeStateMapper.selectCount(any())).thenReturn(0L); + 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.extractPointer(STORED_CHUNK_POINTER)) + .thenReturn(CHUNK_POINTER); + lenient().doAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + scope.setId(401L); + return 1; + }).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class)); + lenient().doThrow(new DuplicateKeyException("duplicate chunk")) + .when(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class)); + lenient().doAnswer(invocation -> { + cleanupAttempts.incrementAndGet(); + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(anyString()); + } + + @AfterEach + void shutdownExecutors() { + service.shutdownAssembleExecutor(); + } + + private void configureCleanupFailure() { + doThrow(new IllegalStateException("cleanup failed")).when(transientPayloadStorageService) + .deletePayloadIfPresent(anyString()); + } + + @Test + void cleanupErrorDoesNotBlockResponse() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + configureCleanupFailure(); + + service.submitResult(TASK_ID, request()); + + verify(taskCacheService).touchTaskHeartbeat(TASK_ID); + verify(transientPayloadStorageService).deletePayloadIfPresent(STORED_CHUNK_POINTER); + } + + @Test + void cleanupErrorNoRetryLoop() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + configureCleanupFailure(); + + service.submitResult(TASK_ID, request()); + + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent(STORED_CHUNK_POINTER); + } + + @Test + void cleanupRunsAfterTransactionCommitted() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + + service.submitResult(TASK_ID, request()); + + var order = inOrder(transactionManager, transientPayloadStorageService); + order.verify(transactionManager).commit(transactionStatus); + order.verify(transientPayloadStorageService).deletePayloadIfPresent(STORED_CHUNK_POINTER); + } + + @Test + void cleanupObservesTransactionFinished() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + AtomicBoolean cleanupOutsideTx = new AtomicBoolean(); + doAnswer(invocation -> { + cleanupOutsideTx.set(!transactionActive.get()); + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(anyString()); + + service.submitResult(TASK_ID, request()); + + assertTrue(cleanupOutsideTx.get(), "清理必须在事务结束后（等价 afterCommit）"); + } + + @Test + void cleanupFailureKeepsPayloadForNextSubmission() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + configureCleanupFailure(); + + service.submitResult(TASK_ID, request()); + // 恢复后再次提交，同一 payload 再次进入清理（失败可重试，payload 未丢） + service.submitResult(TASK_ID, request()); + + verify(transientPayloadStorageService, times(2)).deletePayloadIfPresent(STORED_CHUNK_POINTER); + } + + @Test + void partialCleanupFailureStillFinishesFlow() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + AtomicInteger failures = new AtomicInteger(); + doAnswer(invocation -> { + if (failures.getAndIncrement() == 0) { + throw new IllegalStateException("first cleanup failed"); + } + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(anyString()); + + service.submitResult(TASK_ID, request()); + + verify(taskCacheService).touchTaskHeartbeat(TASK_ID); + verify(transientPayloadStorageService).deletePayloadIfPresent(STORED_CHUNK_POINTER); + } + + @Test + void cleanupFailureDoesNotAffectCommittedState() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + configureCleanupFailure(); + + service.submitResult(TASK_ID, request()); + + verify(transactionManager).commit(transactionStatus); + verify(transactionManager, org.mockito.Mockito.never()).rollback(transactionStatus); + } + + @Test + void cleanupErrorIntegration() { + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + configureCleanupFailure(); + + // 全链路：提交成功 → 清理失败被吞 → 心跳/后续流程正常 + service.submitResult(TASK_ID, request()); + + verify(transactionManager).commit(transactionStatus); + verify(transientPayloadStorageService).deletePayloadIfPresent(STORED_CHUNK_POINTER); + verify(taskCacheService).touchTaskHeartbeat(TASK_ID); + } + + private SimilarAsinSubmitResultRequest request() { + SimilarAsinSubmitResultRequest request = new SimilarAsinSubmitResultRequest(); + request.setSubmissionId("similar-asin-5555"); + request.setChunkIndex(0); + request.setChunkTotal(1); + request.setDone(false); + return request; + } + + 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; + } +}