task-133: 回滚语义硬约束契约（完整回滚/状态不改写/payload 保留/重试安全/计算幂等/双模块一致）+ 8 条守门测试
This commit is contained in:
+324
@@ -0,0 +1,324 @@
|
|||||||
|
package com.nanri.aiimage.modules.task.contract;
|
||||||
|
|
||||||
|
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.appearancepatent.client.AppearancePatentLlmClient;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskCacheService;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskService;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentSubmitResultRequest;
|
||||||
|
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.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.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.TaskProgressLightAssembler;
|
||||||
|
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.test.util.ReflectionTestUtils;
|
||||||
|
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.assertThrows;
|
||||||
|
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.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-133：回滚语义硬约束契约（spec 07 §2）。
|
||||||
|
* 落库异常 → 完整回滚：任务状态/结果不被改写、payload 保留、异常抛出；
|
||||||
|
* 纯计算（prepare）无副作用且幂等；重试安全。similarasin + appearancepatent
|
||||||
|
* 双模块统一守门。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class RollbackSemanticsContractTest {
|
||||||
|
|
||||||
|
private static final Long TASK_ID = 9999L;
|
||||||
|
private static final String PARSED_POINTER = "rustfs:task-parsed/similar-asin/9999/payload.json";
|
||||||
|
private static final String CHUNK_POINTER = "rustfs:task-chunk/similar-asin/9999/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 AtomicReference<String> storedPayloadJson = new AtomicReference<>();
|
||||||
|
|
||||||
|
@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(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));
|
||||||
|
return STORED_CHUNK_POINTER;
|
||||||
|
});
|
||||||
|
lenient().when(transientPayloadStorageService.wasLastStoreLocalFallback()).thenReturn(false);
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
|
||||||
|
@AfterEach
|
||||||
|
void shutdownExecutors() {
|
||||||
|
service.shutdownAssembleExecutor();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void rollbackIsFullAndExceptionPropagates() {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
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()));
|
||||||
|
|
||||||
|
verify(transactionManager).rollback(transactionStatus);
|
||||||
|
verify(transactionManager, never()).commit(transactionStatus);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void rollbackLeavesTaskRowUntouched() {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
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()));
|
||||||
|
|
||||||
|
verify(fileTaskMapper, never()).updateById(any(FileTaskEntity.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void rollbackKeepsPayloadNotDeleted() {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
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()));
|
||||||
|
|
||||||
|
verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void retryAfterRollbackSucceeds() {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
AtomicInteger failures = new AtomicInteger();
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
if (failures.getAndIncrement() == 0) {
|
||||||
|
throw new IllegalStateException("first attempt failed");
|
||||||
|
}
|
||||||
|
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||||
|
scope.setId(401L);
|
||||||
|
return 1;
|
||||||
|
}).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class, () -> service.submitResult(TASK_ID, request()));
|
||||||
|
service.submitResult(TASK_ID, request());
|
||||||
|
|
||||||
|
verify(transactionManager, times(1)).rollback(transactionStatus);
|
||||||
|
verify(transactionManager).commit(transactionStatus);
|
||||||
|
verify(taskCacheService).touchTaskHeartbeat(TASK_ID);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void computePhaseIsSideEffectFreeAndIdempotent() throws Exception {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
|
||||||
|
Object first = ReflectionTestUtils.invokeMethod(service, "prepareSubmittedChunk", TASK_ID, request());
|
||||||
|
Object second = ReflectionTestUtils.invokeMethod(service, "prepareSubmittedChunk", TASK_ID, request());
|
||||||
|
|
||||||
|
assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()),
|
||||||
|
ReflectionTestUtils.getField(first, "payloadHash"));
|
||||||
|
assertEquals(ReflectionTestUtils.getField(first, "payloadHash"),
|
||||||
|
ReflectionTestUtils.getField(second, "payloadHash"), "prepare 幂等：重复调用哈希一致");
|
||||||
|
verify(taskChunkMapper, never()).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class));
|
||||||
|
verify(taskScopeStateMapper, never()).insert(any(TaskScopeStateEntity.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void partialWriteDoesNotContinueAfterFailure() {
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
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()));
|
||||||
|
|
||||||
|
// chunk insert 已发生（部分写入），但后续 updateById/心跳/schedule 全部不执行
|
||||||
|
verify(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class));
|
||||||
|
verify(fileTaskMapper, never()).updateById(any(FileTaskEntity.class));
|
||||||
|
verify(taskCacheService, never()).touchTaskHeartbeat(TASK_ID);
|
||||||
|
verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), anyString(), anyLong(), anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void appearancePatentRollbackSkipsSecondTransaction() {
|
||||||
|
AppearancePatentTaskService apService = appearancePatentService();
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(TASK_ID);
|
||||||
|
task.setModuleType("APPEARANCE_PATENT");
|
||||||
|
task.setStatus("RUNNING");
|
||||||
|
task.setUserId(7L);
|
||||||
|
task.setResultJson("{\"parsedPayloadRef\":\"" + PARSED_POINTER + "\",\"ownerInstanceId\":\"instance-a\"}");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||||
|
when(transientPayloadStorageService.storeChunkPayload(
|
||||||
|
eq("APPEARANCE_PATENT"), eq(TASK_ID), anyString(), any(), anyString()))
|
||||||
|
.thenReturn(STORED_CHUNK_POINTER);
|
||||||
|
lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null);
|
||||||
|
when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenThrow(
|
||||||
|
new IllegalStateException("persist tx failed"));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class, () -> apService.submitResult(TASK_ID, new AppearancePatentSubmitResultRequest()));
|
||||||
|
|
||||||
|
verify(transactionManager, times(1)).getTransaction(any(TransactionDefinition.class));
|
||||||
|
verify(transactionManager).rollback(transactionStatus);
|
||||||
|
verify(transactionManager, never()).commit(transactionStatus);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void rollbackConsistencyAcrossModules() {
|
||||||
|
// 双模块回滚观测一致：rollback 调用、无 commit、异常传播（同类断言已覆盖）
|
||||||
|
FileTaskEntity task = runningTask();
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
org.mockito.Mockito.doThrow(new IllegalStateException("boom"))
|
||||||
|
.when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class, () -> service.submitResult(TASK_ID, request()));
|
||||||
|
|
||||||
|
verify(transactionManager, never()).commit(transactionStatus);
|
||||||
|
}
|
||||||
|
|
||||||
|
private AppearancePatentTaskService appearancePatentService() {
|
||||||
|
return new AppearancePatentTaskService(
|
||||||
|
localFileStorageService, ossStorageService, storageProperties, fileTaskMapper, fileResultMapper,
|
||||||
|
taskScopeStateMapper, taskChunkMapper, objectMapper, mock(AppearancePatentLlmClient.class),
|
||||||
|
mock(AppearancePatentTaskCacheService.class), mock(com.nanri.aiimage.config.AppearancePatentProperties.class),
|
||||||
|
taskFileJobService, taskProgressSnapshotService, transientPayloadStorageService, transactionManager,
|
||||||
|
distributedJobLockService, taskDistributedLockService, instanceMetadata,
|
||||||
|
mock(TaskProgressLightAssembler.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
private SimilarAsinSubmitResultRequest request() {
|
||||||
|
SimilarAsinSubmitResultRequest request = new SimilarAsinSubmitResultRequest();
|
||||||
|
request.setSubmissionId("similar-asin-9999");
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user