task-137: 重复提交幂等契约（同 id 幂等接受/不重复落库/终态拒绝/并发撞键不覆盖/多次安全）+ 8 条守门测试
This commit is contained in:
+283
@@ -0,0 +1,283 @@
|
||||
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.exception.BusinessException;
|
||||
import com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper;
|
||||
import com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper;
|
||||
import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataSubmitResultRequest;
|
||||
import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataSubmitRowDto;
|
||||
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
|
||||
import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataSubmitResultVo;
|
||||
import com.nanri.aiimage.modules.collectdata.service.CollectDataExcelAssemblyService;
|
||||
import com.nanri.aiimage.modules.collectdata.service.CollectDataService;
|
||||
import com.nanri.aiimage.modules.collectdata.util.CollectDataBatchQuery;
|
||||
import com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter;
|
||||
import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec;
|
||||
import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter;
|
||||
import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter;
|
||||
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||
import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
|
||||
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.TaskResultItemMapper;
|
||||
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.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.TransientPayloadStorageService;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
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.support.TransactionTemplate;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
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.anyInt;
|
||||
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-137：重复提交幂等契约（spec 07 §2）。
|
||||
* 同 submissionId 重复提交 /result：幂等接受（成功返回、不重复落库）；
|
||||
* 同 chunk_index 内容不同拒绝；任务终态后重复提交拒绝；多次重复安全。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class DuplicateSubmissionContractTest {
|
||||
|
||||
private static final Long TASK_ID = 4444L;
|
||||
private static final Long USER_ID = 7L;
|
||||
private static final String SUBMISSION_ID = "collect-data-4444";
|
||||
|
||||
@Mock private LocalFileStorageService localFileStorageService;
|
||||
@Mock private FileTaskMapper fileTaskMapper;
|
||||
@Mock private FileResultMapper fileResultMapper;
|
||||
@Mock private CollectDataItemMapper collectDataItemMapper;
|
||||
@Mock private CollectDataCountryPrefMapper collectDataCountryPrefMapper;
|
||||
@Mock private InvalidAsinDataMapper invalidAsinDataMapper;
|
||||
@Mock private TaskChunkMapper taskChunkMapper;
|
||||
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||
@Mock private TaskResultItemMapper taskResultItemMapper;
|
||||
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||
@Mock private TaskFileJobService taskFileJobService;
|
||||
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||
@Mock private CollectDataExcelAssemblyService excelAssemblyService;
|
||||
@Mock private OssStorageService ossStorageService;
|
||||
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||
@Mock private TransactionTemplate transactionTemplate;
|
||||
@Mock private CollectDataBatchQuery collectDataBatchQuery;
|
||||
@Mock private CollectDataBrandBatchFilter brandBatchFilter;
|
||||
@Mock private CollectDataInvalidAsinBatchWriter invalidAsinBatchWriter;
|
||||
@Mock private CollectDataResultItemBatchWriter resultItemBatchWriter;
|
||||
@Mock private CollectDataResultDetailCodec resultDetailCodec;
|
||||
|
||||
@InjectMocks private CollectDataService service;
|
||||
|
||||
@BeforeAll
|
||||
static void initializeMybatisMetadata() {
|
||||
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||
}
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
lenient().when(taskDistributedLockService.acquire(eq("COLLECT_DATA"), anyLong(), anyLong()))
|
||||
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||
lenient().when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask());
|
||||
lenient().when(fileResultMapper.selectOne(any())).thenReturn(null);
|
||||
lenient().doAnswer(invocation -> {
|
||||
FileResultEntity result = invocation.getArgument(0);
|
||||
result.setId(3001L);
|
||||
return 1;
|
||||
}).when(fileResultMapper).insert(any(FileResultEntity.class));
|
||||
lenient().when(taskChunkMapper.selectCount(any())).thenReturn(0L);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null);
|
||||
lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null);
|
||||
lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenReturn(1);
|
||||
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||
lenient().when(transientPayloadStorageService.extractPointer(anyString())).thenReturn("rustfs:detail");
|
||||
lenient().when(transientPayloadStorageService.storeResultPayload(
|
||||
eq("COLLECT_DATA"), eq(TASK_ID), anyString(), anyString(), anyString()))
|
||||
.thenReturn("\"rustfs:detail\"");
|
||||
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||
eq("COLLECT_DATA"), eq(TASK_ID), anyString(), anyInt(), anyString()))
|
||||
.thenReturn("\"rustfs:chunk\"");
|
||||
lenient().doAnswer(invocation -> {
|
||||
TaskChunkEntity chunk = invocation.getArgument(0);
|
||||
chunk.setId(701L);
|
||||
return 1;
|
||||
}).when(taskChunkMapper).insert(any(TaskChunkEntity.class));
|
||||
CollectDataResultRowVo row = new CollectDataResultRowVo();
|
||||
row.setAsin("B0DUP1234");
|
||||
lenient().when(collectDataBatchQuery.filter(any())).thenReturn(
|
||||
new CollectDataBatchQuery.FilterResult(List.of(row), 0, 0));
|
||||
lenient().when(brandBatchFilter.filter(any())).thenReturn(
|
||||
new CollectDataBrandBatchFilter.BrandBatchOutcome(List.of(), List.of(), List.of(row)));
|
||||
lenient().when(resultItemBatchWriter.upsertAccepted(anyLong(), anyLong(), anyString(), anyInt(), any(), anyString()))
|
||||
.thenReturn(new CollectDataResultItemBatchWriter.UpsertCounts(1, 0, 1));
|
||||
lenient().when(resultDetailCodec.encodeChunk(any())).thenReturn("{\"chunk\":1}");
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateSameSubmissionIdAcceptedIdempotently() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest());
|
||||
|
||||
assertNotNull(vo, "重复提交幂等接受（成功返回）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateDoesNotDoublePersist() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
|
||||
verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateReturnsSuccessVo() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest());
|
||||
|
||||
assertEquals(TASK_ID, vo.getTaskId());
|
||||
assertEquals(1, vo.getChunkIndex());
|
||||
assertTrue(vo.getTaskStatus() != null);
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateManySubmissionsAllSafe() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
for (int i = 0; i < 5; i++) {
|
||||
CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest());
|
||||
assertNotNull(vo, "第 " + (i + 1) + " 次重复提交仍幂等接受");
|
||||
}
|
||||
verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void terminalTaskRejectsDuplicateSubmission() {
|
||||
FileTaskEntity done = runningTask();
|
||||
done.setStatus("SUCCESS");
|
||||
lenient().when(fileTaskMapper.selectById(TASK_ID)).thenReturn(done);
|
||||
|
||||
assertThrows(BusinessException.class, () -> service.submitResult(TASK_ID, submitRequest()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void differentChunkIndexIsNotDuplicate() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequestWithChunk(2, 2));
|
||||
|
||||
assertNotNull(vo);
|
||||
assertEquals(2, vo.getChunkIndex(), "不同 chunk 继续受理，互不视为重复");
|
||||
}
|
||||
|
||||
@Test
|
||||
void concurrentDuplicateDoesNotOverwriteWinner() {
|
||||
// 并发重复：查重未命中但 insert 撞唯一键 → 幂等接受，不覆盖 winner、不抛错
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null, null);
|
||||
org.mockito.Mockito.doThrow(new org.springframework.dao.DuplicateKeyException("dup"))
|
||||
.when(taskChunkMapper).insert(any(TaskChunkEntity.class));
|
||||
|
||||
CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest());
|
||||
|
||||
assertNotNull(vo, "并发重复幂等接受");
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateContractFrozen() {
|
||||
service.submitResult(TASK_ID, submitRequest());
|
||||
TaskChunkEntity existing = new TaskChunkEntity();
|
||||
existing.setId(702L);
|
||||
existing.setChunkIndex(1);
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing);
|
||||
|
||||
CollectDataSubmitResultVo first = service.submitResult(TASK_ID, submitRequest());
|
||||
|
||||
// 快照：重复提交返回同一 chunk 视图、不触发新的存储写入
|
||||
assertEquals(1, first.getChunkIndex());
|
||||
verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(
|
||||
anyString(), anyLong(), anyString(), anyInt(), anyString());
|
||||
}
|
||||
|
||||
private CollectDataSubmitResultRequest submitRequest() {
|
||||
return submitRequestWithChunk(1, 1);
|
||||
}
|
||||
|
||||
private CollectDataSubmitResultRequest submitRequestWithChunk(int chunkIndex, int chunkTotal) {
|
||||
CollectDataSubmitRowDto row = new CollectDataSubmitRowDto();
|
||||
row.setAsin("B0DUP1234");
|
||||
CollectDataSubmitResultRequest request = new CollectDataSubmitResultRequest();
|
||||
request.setSubmissionId(SUBMISSION_ID);
|
||||
request.setChunkIndex(chunkIndex);
|
||||
request.setChunkTotal(chunkTotal);
|
||||
request.setDone(false);
|
||||
request.setItems(List.of(row));
|
||||
return request;
|
||||
}
|
||||
|
||||
private FileTaskEntity runningTask() {
|
||||
FileTaskEntity task = new FileTaskEntity();
|
||||
task.setId(TASK_ID);
|
||||
task.setModuleType("COLLECT_DATA");
|
||||
task.setStatus("RUNNING");
|
||||
task.setUserId(USER_ID);
|
||||
task.setResultJson("{}");
|
||||
return task;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user