task-70: 历史清理改为 keyset 分页、小批量和短事务
This commit is contained in:
+406
@@ -0,0 +1,406 @@
|
||||
package com.nanri.aiimage.modules.task.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
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.ModuleCleanupProperties;
|
||||
import com.nanri.aiimage.config.StorageProperties;
|
||||
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||
import com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper;
|
||||
import com.nanri.aiimage.modules.collectdata.model.entity.CollectDataItemEntity;
|
||||
import com.nanri.aiimage.modules.file.service.object.RustfsObjectStorageService;
|
||||
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||
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.TaskFileJobMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
|
||||
import com.nanri.aiimage.modules.task.mapper.TaskResultPayloadMapper;
|
||||
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.TaskFileJobEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskResultPayloadEntity;
|
||||
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||
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.io.TempDir;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.ArgumentMatchers;
|
||||
import org.mockito.invocation.InvocationOnMock;
|
||||
import org.mockito.stubbing.Answer;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.ValueOperations;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
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.anyString;
|
||||
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 70：将历史清理改为 keyset 分页、小批量和短事务。
|
||||
* ModuleHistoryCleanupService 按 id 升序分页拉取过期任务（keyset，页大小
|
||||
* batchSize），每页一个短事务执行收集、删除与 flush；下一页游标用当前页
|
||||
* 最大 id 推进，无重复无遗漏；页内异常回滚并中止，未处理页可下次重试。
|
||||
*/
|
||||
class ModuleHistoryCleanupKeysetPaginationTest {
|
||||
|
||||
@TempDir
|
||||
Path tempDir;
|
||||
|
||||
private static final LocalDateTime OLD_FINISHED_AT = LocalDateTime.of(2026, 1, 1, 0, 0);
|
||||
|
||||
private ModuleCleanupProperties cleanupProperties;
|
||||
private FileTaskMapper fileTaskMapper;
|
||||
private FileResultMapper fileResultMapper;
|
||||
private TaskFileJobMapper taskFileJobMapper;
|
||||
private TaskResultItemMapper taskResultItemMapper;
|
||||
private TaskProgressSnapshotMapper taskProgressSnapshotMapper;
|
||||
private TaskResultPayloadMapper taskResultPayloadMapper;
|
||||
private TaskScopeStateMapper taskScopeStateMapper;
|
||||
private TaskChunkMapper taskChunkMapper;
|
||||
private CollectDataItemMapper collectDataItemMapper;
|
||||
private DistributedJobLockService lockService;
|
||||
private TransientPayloadStorageService storage;
|
||||
private RustfsObjectStorageService rustfs;
|
||||
private OssStorageService oss;
|
||||
private TransientPayloadDeleteOrchestrator orchestrator;
|
||||
private ExecutorService executor;
|
||||
private PlatformTransactionManager transactionManager;
|
||||
private ModuleHistoryCleanupService cleanupService;
|
||||
|
||||
/** 记录一次事务 begin（快照状态）→ 删除 mappers → commit 的调用轨迹。 */
|
||||
private final List<String> trace = new ArrayList<>();
|
||||
|
||||
@BeforeAll
|
||||
static void initializeMybatisMetadata() {
|
||||
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskResultItemEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskProgressSnapshotEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskResultPayloadEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||
TableInfoHelper.initTableInfo(assistant, CollectDataItemEntity.class);
|
||||
}
|
||||
|
||||
@BeforeEach
|
||||
@SuppressWarnings("unchecked")
|
||||
void setUp() {
|
||||
cleanupProperties = new ModuleCleanupProperties();
|
||||
fileTaskMapper = mock(FileTaskMapper.class);
|
||||
fileResultMapper = mock(FileResultMapper.class);
|
||||
taskFileJobMapper = mock(TaskFileJobMapper.class);
|
||||
taskResultItemMapper = mock(TaskResultItemMapper.class);
|
||||
taskProgressSnapshotMapper = mock(TaskProgressSnapshotMapper.class);
|
||||
taskResultPayloadMapper = mock(TaskResultPayloadMapper.class);
|
||||
taskScopeStateMapper = mock(TaskScopeStateMapper.class);
|
||||
taskChunkMapper = mock(TaskChunkMapper.class);
|
||||
collectDataItemMapper = mock(CollectDataItemMapper.class);
|
||||
StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class);
|
||||
ValueOperations<String, String> valueOperations = mock(ValueOperations.class);
|
||||
when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
when(valueOperations.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(Boolean.TRUE);
|
||||
lockService = new DistributedJobLockService(redisTemplate);
|
||||
|
||||
rustfs = mock(RustfsObjectStorageService.class);
|
||||
oss = mock(OssStorageService.class);
|
||||
TransientStorageProperties transientProperties = new TransientStorageProperties();
|
||||
transientProperties.setEnabled(true);
|
||||
StorageProperties storageProperties = new StorageProperties();
|
||||
storageProperties.setLocalTempDir(tempDir.toString());
|
||||
storage = new TransientPayloadStorageService(
|
||||
transientProperties, storageProperties, rustfs, oss,
|
||||
new ObjectMapper(), new InstanceMetadata("test-instance"),
|
||||
taskChunkMapper, taskScopeStateMapper);
|
||||
executor = Executors.newFixedThreadPool(2);
|
||||
orchestrator = new TransientPayloadDeleteOrchestrator(
|
||||
storage, rustfs, taskChunkMapper, taskScopeStateMapper, new ObjectMapper(), executor);
|
||||
ReflectionTestUtils.setField(orchestrator, "maxPendingDeletes", 1000L);
|
||||
transactionManager = mock(PlatformTransactionManager.class);
|
||||
doAnswer(invocation -> {
|
||||
trace.add("begin");
|
||||
return mock(TransactionStatus.class);
|
||||
}).when(transactionManager).getTransaction(ArgumentMatchers.any(TransactionDefinition.class));
|
||||
doAnswer(invocation -> {
|
||||
trace.add("commit");
|
||||
return null;
|
||||
}).when(transactionManager).commit(any(TransactionStatus.class));
|
||||
doAnswer(invocation -> {
|
||||
trace.add("rollback");
|
||||
return null;
|
||||
}).when(transactionManager).rollback(any(TransactionStatus.class));
|
||||
cleanupService = new ModuleHistoryCleanupService(
|
||||
cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper,
|
||||
taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper,
|
||||
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator,
|
||||
transactionManager);
|
||||
}
|
||||
|
||||
private static FileTaskEntity expiredTask(long id) {
|
||||
FileTaskEntity task = new FileTaskEntity();
|
||||
task.setId(id);
|
||||
task.setModuleType("DEDUPE");
|
||||
task.setStatus("SUCCESS");
|
||||
task.setFinishedAt(OLD_FINISHED_AT);
|
||||
return task;
|
||||
}
|
||||
|
||||
private static List<FileTaskEntity> expiredTasks(long... ids) {
|
||||
List<FileTaskEntity> tasks = new ArrayList<>();
|
||||
for (long id : ids) {
|
||||
tasks.add(expiredTask(id));
|
||||
}
|
||||
return tasks;
|
||||
}
|
||||
|
||||
private static <T> Answer<List<T>> sequence(List<T> rows, int rowCalls, List<T> fallback) {
|
||||
return new Answer<List<T>>() {
|
||||
private int calls;
|
||||
|
||||
@Override
|
||||
public List<T> answer(InvocationOnMock invocation) {
|
||||
return calls++ < rowCalls ? rows : fallback;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/** 按调用次数依次返回每一页（keyset 翻页，页内 id 升序且首元素大于上次游标）。 */
|
||||
private static <T> Answer<List<T>> pages(List<List<T>> pages) {
|
||||
return new Answer<List<T>>() {
|
||||
private int calls;
|
||||
|
||||
@Override
|
||||
public List<T> answer(InvocationOnMock invocation) {
|
||||
int idx = calls++;
|
||||
return idx < pages.size() ? pages.get(idx) : List.of();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private void stubDeletes() {
|
||||
when(taskFileJobMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
when(taskResultItemMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
when(taskProgressSnapshotMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
when(taskResultPayloadMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
when(taskScopeStateMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
when(taskChunkMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
lenient().when(fileResultMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
lenient().when(fileTaskMapper.update(any(), any())).thenReturn(0);
|
||||
lenient().when(fileTaskMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(1);
|
||||
}
|
||||
|
||||
private static String pointer(int taskId) {
|
||||
return "rustfs:task-parsed/test/" + taskId + "/scope/latest.json";
|
||||
}
|
||||
|
||||
private static String jsonPointer(int taskId) {
|
||||
return "\"" + pointer(taskId) + "\"";
|
||||
}
|
||||
|
||||
private static TaskChunkEntity chunk(long taskId, String payloadJson) {
|
||||
TaskChunkEntity chunk = new TaskChunkEntity();
|
||||
chunk.setTaskId(taskId);
|
||||
chunk.setModuleType("DEDUPE");
|
||||
chunk.setPayloadJson(payloadJson);
|
||||
return chunk;
|
||||
}
|
||||
|
||||
private static TaskScopeStateEntity scopeState(long taskId, String parsedPayloadJson, String stateJson) {
|
||||
TaskScopeStateEntity state = new TaskScopeStateEntity();
|
||||
state.setTaskId(taskId);
|
||||
state.setModuleType("DEDUPE");
|
||||
state.setParsedPayloadJson(parsedPayloadJson);
|
||||
state.setStateJson(stateJson);
|
||||
return state;
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_normal_default_path() {
|
||||
// 默认路径：单页完成清理，事务 begin→删除→commit 成对出现。
|
||||
cleanupProperties.setBatchSize(100);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(expiredTasks(1, 2), 1, List.of()));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(List.of("begin", "commit"), trace, "单页一事务，先提交后删除");
|
||||
verify(fileTaskMapper).delete(any(LambdaQueryWrapper.class));
|
||||
assertEquals(1, traceCount("commit"), "恰好一次提交");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_normal_multiple_items() {
|
||||
// 批量场景：3 页（每页 2 条）翻页清理，页内删除不丢行，总删除等于页和。
|
||||
cleanupProperties.setBatchSize(2);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenAnswer(pages(
|
||||
List.of(expiredTasks(1, 2), expiredTasks(3, 4), expiredTasks(5, 6))));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(3, traceCount("begin"), "每页一个事务");
|
||||
assertEquals(3, traceCount("commit"));
|
||||
ArgumentCaptor<LambdaQueryWrapper<FileTaskEntity>> captor =
|
||||
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
|
||||
verify(fileTaskMapper, times(3)).delete(captor.capture());
|
||||
for (LambdaQueryWrapper<FileTaskEntity> wrapper : captor.getAllValues()) {
|
||||
assertTrue(wrapper.getCustomSqlSegment().contains("IN"), "每页按任务 id 批量删除");
|
||||
}
|
||||
assertEquals(3, captor.getAllValues().size(), "三页三次小批量删除，非一次大删");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_normal_repeated_operation_is_idempotent() {
|
||||
// 幂等：翻页到页尾后第二轮无行，事务数不随重复运行增加。
|
||||
cleanupProperties.setBatchSize(100);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(expiredTasks(1), 2, List.of()));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(2, traceCount("begin"), "每轮一个事务，页尾后无多余事务");
|
||||
assertEquals(2, traceCount("commit"));
|
||||
verify(fileTaskMapper, times(2)).delete(any(LambdaQueryWrapper.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_boundary_empty_input() {
|
||||
// 空输入：无过期任务时一次事务都不开，不查询子表、不删除。
|
||||
cleanupProperties.setBatchSize(100);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(0, traceCount("begin"), "空输入不开事务");
|
||||
verify(taskChunkMapper, never()).selectList(any(LambdaQueryWrapper.class));
|
||||
verify(taskChunkMapper, never()).delete(any(LambdaQueryWrapper.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_boundary_single_item() {
|
||||
// 单元素：单任务单页单事务，不依赖批量路径，行删除一次。
|
||||
cleanupProperties.setBatchSize(100);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(expiredTasks(9), 1, List.of()));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(List.of("begin", "commit"), trace, "单任务一事务");
|
||||
verify(fileTaskMapper).delete(any(LambdaQueryWrapper.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_boundary_limit_and_overflow() {
|
||||
// 上限/超限：batchSize 为 0/负值时按默认 500 处理，不因非法配置死循环或崩溃。
|
||||
cleanupProperties.setBatchSize(-1);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(expiredTasks(1), 1, List.of()));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(1, traceCount("commit"), "非法 batchSize 走默认值，仍正常完成");
|
||||
verify(fileTaskMapper).delete(any(LambdaQueryWrapper.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_invalid_input_rejected() {
|
||||
// 非法参数：缺少必填模块类型配置时拒绝执行，不开事务、不查询、不删除。
|
||||
cleanupProperties.setModuleTypes(null);
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertEquals(0, traceCount("begin"), "非法配置不开事务");
|
||||
verify(fileTaskMapper, never()).selectList(any(LambdaQueryWrapper.class));
|
||||
verify(fileTaskMapper, never()).delete(any(LambdaQueryWrapper.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_070_cleanup_dependency_failure_releases_resources() throws Exception {
|
||||
// 依赖失败：页内删除抛异常时回滚事务并中止，未处理页不继续；
|
||||
// 已收集指针不提交、不 flush；恢复后重跑成功提交。
|
||||
cleanupProperties.setBatchSize(2);
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenAnswer(pages(
|
||||
List.of(expiredTasks(1, 2), expiredTasks(3, 4))));
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(List.of(chunk(1, jsonPointer(1))), 1, List.of()));
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskChunkMapper.delete(any(LambdaQueryWrapper.class)))
|
||||
.thenThrow(new RuntimeException("db down"));
|
||||
|
||||
assertThrows(RuntimeException.class, () -> cleanupService.cleanupConfiguredModules());
|
||||
|
||||
assertEquals(List.of("begin", "rollback"), trace, "页内异常回滚事务");
|
||||
verify(fileTaskMapper, never()).delete(any(LambdaQueryWrapper.class));
|
||||
assertEquals(0, orchestrator.pendingCount(), "指针不提交");
|
||||
verify(rustfs, never()).deleteObject(anyString());
|
||||
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(List.of(chunk(1, jsonPointer(1))), 1, List.of()));
|
||||
when(taskChunkMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
doAnswer(invocation -> {
|
||||
done.countDown();
|
||||
return null;
|
||||
}).when(rustfs).deleteObject(anyString());
|
||||
|
||||
cleanupService.cleanupConfiguredModules();
|
||||
|
||||
assertTrue(done.await(2, TimeUnit.SECONDS), "恢复后指针异步删除完成");
|
||||
assertEquals(0, orchestrator.pendingCount());
|
||||
assertEquals(1, traceCount("commit"), "恢复后提交成功");
|
||||
}
|
||||
|
||||
private int traceCount(String event) {
|
||||
return (int) trace.stream().filter(event::equals).count();
|
||||
}
|
||||
}
|
||||
+22
-4
@@ -39,6 +39,7 @@ import org.mockito.stubbing.Answer;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.ValueOperations;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
@@ -92,6 +93,7 @@ class ModuleHistoryCleanupPayloadCleanupTest {
|
||||
private OssStorageService oss;
|
||||
private TransientPayloadDeleteOrchestrator orchestrator;
|
||||
private ExecutorService executor;
|
||||
private PlatformTransactionManager transactionManager;
|
||||
private ModuleHistoryCleanupService cleanupService;
|
||||
|
||||
@BeforeAll
|
||||
@@ -141,10 +143,12 @@ class ModuleHistoryCleanupPayloadCleanupTest {
|
||||
orchestrator = new TransientPayloadDeleteOrchestrator(
|
||||
storage, rustfs, taskChunkMapper, taskScopeStateMapper, new ObjectMapper(), executor);
|
||||
ReflectionTestUtils.setField(orchestrator, "maxPendingDeletes", 1000L);
|
||||
transactionManager = mock(PlatformTransactionManager.class);
|
||||
cleanupService = new ModuleHistoryCleanupService(
|
||||
cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper,
|
||||
taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper,
|
||||
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator);
|
||||
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator,
|
||||
transactionManager);
|
||||
}
|
||||
|
||||
private static FileTaskEntity expiredTask(long id) {
|
||||
@@ -156,12 +160,26 @@ class ModuleHistoryCleanupPayloadCleanupTest {
|
||||
return task;
|
||||
}
|
||||
|
||||
/** 任务查询返回一次 rows 后变为空（keyset 分页到页尾），runs 次完整清理各查一轮。 */
|
||||
private void stubExpiredTasks(long... ids) {
|
||||
List<FileTaskEntity> tasks = new ArrayList<>();
|
||||
for (long id : ids) {
|
||||
tasks.add(expiredTask(id));
|
||||
}
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(tasks);
|
||||
List<FileTaskEntity> none = List.of();
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(tasks, 1, none));
|
||||
}
|
||||
|
||||
/** 幂等/恢复场景：多次清理运行各返回一轮 rows。 */
|
||||
private void stubExpiredTasksRuns(int runs, long... ids) {
|
||||
List<FileTaskEntity> tasks = new ArrayList<>();
|
||||
for (long id : ids) {
|
||||
tasks.add(expiredTask(id));
|
||||
}
|
||||
List<FileTaskEntity> none = List.of();
|
||||
when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class)))
|
||||
.thenAnswer(sequence(tasks, runs, none));
|
||||
}
|
||||
|
||||
/** 收集与 flush 都走同一 mapper.selectList：前 rowCalls 次返回 rows（行未删），之后返回 fallback。 */
|
||||
@@ -287,7 +305,7 @@ class ModuleHistoryCleanupPayloadCleanupTest {
|
||||
@Test
|
||||
void test_task_069_payload_cleanup_normal_repeated_operation_is_idempotent() throws Exception {
|
||||
// 幂等：行已删除后再次运行不收集指针、不提交、不发起物理删除。
|
||||
stubExpiredTasks(31);
|
||||
stubExpiredTasksRuns(2, 31);
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
|
||||
@@ -378,7 +396,7 @@ class ModuleHistoryCleanupPayloadCleanupTest {
|
||||
void test_task_069_payload_cleanup_dependency_failure_releases_resources() throws Exception {
|
||||
// 依赖失败：行删除抛异常时中止，指针不提交、不 flush、不删除对象；
|
||||
// 恢复后再次运行，收集、删除与清理队列全部正常完成。
|
||||
stubExpiredTasks(71);
|
||||
stubExpiredTasksRuns(2, 71);
|
||||
stubDeletes();
|
||||
when(taskChunkMapper.delete(any(LambdaQueryWrapper.class)))
|
||||
.thenThrow(new RuntimeException("db down"))
|
||||
|
||||
Reference in New Issue
Block a user