diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/ModuleCleanupProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/ModuleCleanupProperties.java index 2d7a4076..e71e83ec 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/ModuleCleanupProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/ModuleCleanupProperties.java @@ -12,6 +12,11 @@ public class ModuleCleanupProperties { private boolean enabled = true; private String cron = "0 0 0 * * *"; private long retentionDays = 7; + /** + * keyset 分页每批处理的任务数：每批一个短事务，批内按 id 升序， + * 批间无行级排他锁占用；0/负值回退默认 500。 + */ + private int batchSize = 500; // SHOP_DATA_CRAWL keeps one per-shop daily workbook in its task service // and must not be removed by the age-based sweep. private List moduleTypes = new ArrayList<>(List.of("DEDUPE", "SPLIT", "CONVERT", "DELETE_BRAND", "PRODUCT_RISK_RESOLVE", "PRICE_TRACK", "SHOP_MATCH", "PATROL_DELETE", "QUERY_ASIN", "WITHDRAW", "APPEARANCE_PATENT", "SIMILAR_ASIN", "COLLECT_DATA")); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java index 806387fa..1645d156 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java @@ -22,12 +22,12 @@ 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 lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; import java.time.Duration; import java.time.LocalDateTime; @@ -38,12 +38,12 @@ import java.util.Set; @Slf4j @Service -@RequiredArgsConstructor public class ModuleHistoryCleanupService { private static final Set TERMINAL_STATUSES = Set.of("SUCCESS", "FAILED", "CANCELLED", "CANCELED"); private static final Duration CLEANUP_LOCK_TTL = Duration.ofHours(2); private static final String COLLECT_DATA_MODULE_TYPE = "COLLECT_DATA"; + private static final int DEFAULT_BATCH_SIZE = 500; private final ModuleCleanupProperties moduleCleanupProperties; private final FileTaskMapper fileTaskMapper; @@ -57,15 +57,47 @@ public class ModuleHistoryCleanupService { private final CollectDataItemMapper collectDataItemMapper; private final DistributedJobLockService distributedJobLockService; private final TransientPayloadDeleteOrchestrator transientPayloadDeleteOrchestrator; + private final TransactionTemplate transactionTemplate; /** - * 单次清理最多收集的 payload 指针数：超过即截断（保底可重试）， + * 单批最多收集的 payload 指针数：超过即截断（保底可重试）， * 防止单任务行数异常巨大时无界收集造成内存增长。 */ @Value("${aiimage.module-cleanup.max-collect-payloads:10000}") private int maxCollectPayloadsPerRun = 10000; - @Transactional + public ModuleHistoryCleanupService(ModuleCleanupProperties moduleCleanupProperties, + FileTaskMapper fileTaskMapper, + FileResultMapper fileResultMapper, + TaskFileJobMapper taskFileJobMapper, + TaskResultItemMapper taskResultItemMapper, + TaskProgressSnapshotMapper taskProgressSnapshotMapper, + TaskResultPayloadMapper taskResultPayloadMapper, + TaskScopeStateMapper taskScopeStateMapper, + TaskChunkMapper taskChunkMapper, + CollectDataItemMapper collectDataItemMapper, + DistributedJobLockService distributedJobLockService, + TransientPayloadDeleteOrchestrator transientPayloadDeleteOrchestrator, + PlatformTransactionManager platformTransactionManager) { + this.moduleCleanupProperties = moduleCleanupProperties; + this.fileTaskMapper = fileTaskMapper; + this.fileResultMapper = fileResultMapper; + this.taskFileJobMapper = taskFileJobMapper; + this.taskResultItemMapper = taskResultItemMapper; + this.taskProgressSnapshotMapper = taskProgressSnapshotMapper; + this.taskResultPayloadMapper = taskResultPayloadMapper; + this.taskScopeStateMapper = taskScopeStateMapper; + this.taskChunkMapper = taskChunkMapper; + this.collectDataItemMapper = collectDataItemMapper; + this.distributedJobLockService = distributedJobLockService; + this.transientPayloadDeleteOrchestrator = transientPayloadDeleteOrchestrator; + this.transactionTemplate = new TransactionTemplate(platformTransactionManager); + } + + /** + * 按 id 升序 keyset 分页拉取过期终态任务，每页一个小批量、一个短事务； + * 页间不持有事务与行锁，失败回滚当页并中止，未处理页可下次重试。 + */ @Scheduled(cron = "${aiimage.module-cleanup.cron:0 0 0 * * *}") public void cleanupConfiguredModules() { DistributedJobLockService.LockHandle lockHandle = @@ -83,101 +115,129 @@ public class ModuleHistoryCleanupService { return; } LocalDateTime cutoff = LocalDateTime.now().minusDays(Math.max(0, moduleCleanupProperties.getRetentionDays())); + int batchSize = moduleCleanupProperties.getBatchSize(); + if (batchSize <= 0) { + batchSize = DEFAULT_BATCH_SIZE; + } - List moduleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() - .in(FileTaskEntity::getModuleType, moduleTypes) - .select(FileTaskEntity::getId, FileTaskEntity::getModuleType, FileTaskEntity::getStatus, - FileTaskEntity::getUpdatedAt, FileTaskEntity::getFinishedAt)); - - List cleanupTaskIds = new ArrayList<>(); - List cleanupCollectDataTaskIds = new ArrayList<>(); + long cursor = 0L; + int totalBatches = 0; + int totalDeletedTasks = 0; + int totalCollectedPointers = 0; List skippedActiveTaskIds = new ArrayList<>(); List skippedRetainedTaskIds = new ArrayList<>(); - for (FileTaskEntity task : moduleTasks) { - if (task == null || task.getId() == null) { - continue; + while (true) { + List page = fileTaskMapper.selectList(new LambdaQueryWrapper() + .in(FileTaskEntity::getModuleType, moduleTypes) + .gt(FileTaskEntity::getId, cursor) + .orderByAsc(FileTaskEntity::getId) + .last("LIMIT " + batchSize)); + if (page.isEmpty()) { + break; } - if (!isTerminalStatus(task.getStatus())) { - skippedActiveTaskIds.add(task.getId()); - continue; - } - if (isExpired(task, cutoff)) { - cleanupTaskIds.add(task.getId()); - if (COLLECT_DATA_MODULE_TYPE.equals(task.getModuleType())) { - cleanupCollectDataTaskIds.add(task.getId()); + long pageMaxId = cursor; + List batchTaskIds = new ArrayList<>(); + List batchCollectDataTaskIds = new ArrayList<>(); + for (FileTaskEntity task : page) { + if (task == null || task.getId() == null) { + continue; + } + pageMaxId = Math.max(pageMaxId, task.getId()); + if (!isTerminalStatus(task.getStatus())) { + skippedActiveTaskIds.add(task.getId()); + continue; + } + if (isExpired(task, cutoff)) { + batchTaskIds.add(task.getId()); + if (COLLECT_DATA_MODULE_TYPE.equals(task.getModuleType())) { + batchCollectDataTaskIds.add(task.getId()); + } + } else { + skippedRetainedTaskIds.add(task.getId()); } - } else { - skippedRetainedTaskIds.add(task.getId()); } + if (!batchTaskIds.isEmpty()) { + final List taskIds = batchTaskIds; + final List collectDataTaskIds = batchCollectDataTaskIds; + final List types = moduleTypes; + final List collected = new ArrayList<>(); + transactionTemplate.executeWithoutResult(status -> { + collected.addAll(collectPayloadPointers(types, taskIds, maxCollectPayloadsPerRun)); + int deletedRows = deleteRows(types, taskIds, collectDataTaskIds); + submitAndFlush(collected); + log.info("[module-cleanup] batch: taskIds={}, deletedRows={}, collectedPointers={}", + taskIds, deletedRows, collected.size()); + }); + totalBatches++; + totalDeletedTasks += taskIds.size(); + totalCollectedPointers += collected.size(); + } + if (pageMaxId <= cursor) { + log.warn("[module-cleanup] keyset cursor did not advance, abort loop cursor={}", cursor); + break; + } + cursor = pageMaxId; } - - if (cleanupTaskIds.isEmpty()) { - log.info("[module-cleanup] skipped: moduleTypes={}, retentionDays={}, cutoff={}, activeTaskIds={}, retainedTaskIds={}, reason=no-expired-terminal-tasks", - moduleTypes, moduleCleanupProperties.getRetentionDays(), cutoff, skippedActiveTaskIds, skippedRetainedTaskIds); - return; - } - - // 行删除前先收集将随行删除的 transient payload 指针（上限截断）， - // 全部删除成功后再提交清理队列并 flush，避免删除失败留下已提交的孤儿清理。 - List collectedPayloadPointers = collectPayloadPointers(cleanupTaskIds, maxCollectPayloadsPerRun); - - int deletedFileJobs = taskFileJobMapper.delete(new LambdaQueryWrapper() - .in(TaskFileJobEntity::getModuleType, moduleTypes) - .in(TaskFileJobEntity::getTaskId, cleanupTaskIds)); - - int deletedResultItems = taskResultItemMapper.delete(new LambdaQueryWrapper() - .in(TaskResultItemEntity::getModuleType, moduleTypes) - .in(TaskResultItemEntity::getTaskId, cleanupTaskIds)); - - int deletedProgressSnapshots = taskProgressSnapshotMapper.delete(new LambdaQueryWrapper() - .in(TaskProgressSnapshotEntity::getModuleType, moduleTypes) - .in(TaskProgressSnapshotEntity::getTaskId, cleanupTaskIds)); - - int deletedResultPayloads = taskResultPayloadMapper.delete(new LambdaQueryWrapper() - .in(TaskResultPayloadEntity::getModuleType, moduleTypes) - .in(TaskResultPayloadEntity::getTaskId, cleanupTaskIds)); - - int deletedScopeStates = taskScopeStateMapper.delete(new LambdaQueryWrapper() - .in(TaskScopeStateEntity::getModuleType, moduleTypes) - .in(TaskScopeStateEntity::getTaskId, cleanupTaskIds)); - - int deletedChunks = taskChunkMapper.delete(new LambdaQueryWrapper() - .in(TaskChunkEntity::getModuleType, moduleTypes) - .in(TaskChunkEntity::getTaskId, cleanupTaskIds)); - - int deletedCollectDataItems = 0; - if (!cleanupCollectDataTaskIds.isEmpty()) { - deletedCollectDataItems = collectDataItemMapper.delete(new LambdaQueryWrapper() - .in(CollectDataItemEntity::getTaskId, cleanupCollectDataTaskIds)); - } - - int deletedResults = fileResultMapper.delete(new LambdaQueryWrapper() - .in(FileResultEntity::getModuleType, moduleTypes) - .in(FileResultEntity::getTaskId, cleanupTaskIds)); - - int resetTasks = fileTaskMapper.update(null, new LambdaUpdateWrapper() - .in(FileTaskEntity::getId, cleanupTaskIds) - .set(FileTaskEntity::getResultJson, null) - .set(FileTaskEntity::getRequestJson, null) - .set(FileTaskEntity::getErrorMessage, null)); - - int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper() - .in(FileTaskEntity::getId, cleanupTaskIds)); - - // 行已全部删除，此时引用检查将判定这些指针不再被引用 → 提交清理队列并 flush。 - if (!collectedPayloadPointers.isEmpty()) { - transientPayloadDeleteOrchestrator.submitDeletes(collectedPayloadPointers); - transientPayloadDeleteOrchestrator.flushPendingDeletes(); - } - - log.info("[module-cleanup] completed: moduleTypes={}, retentionDays={}, cutoff={}, deletedFileJobs={}, deletedResultItems={}, deletedProgressSnapshots={}, deletedResultPayloads={}, deletedScopeStates={}, deletedChunks={}, deletedCollectDataItems={}, deletedResults={}, resetTasks={}, deletedTasks={}, collectedPayloadPointers={}, skippedActiveTaskIds={}, retainedTaskIds={}", + log.info("[module-cleanup] completed: moduleTypes={}, retentionDays={}, cutoff={}, batches={}, deletedTasks={}, collectedPayloadPointers={}, skippedActiveTaskIds={}, retainedTaskIds={}", moduleTypes, moduleCleanupProperties.getRetentionDays(), cutoff, - deletedFileJobs, deletedResultItems, deletedProgressSnapshots, deletedResultPayloads, - deletedScopeStates, deletedChunks, deletedCollectDataItems, deletedResults, resetTasks, deletedTasks, - collectedPayloadPointers.size(), skippedActiveTaskIds, skippedRetainedTaskIds); + totalBatches, totalDeletedTasks, totalCollectedPointers, + skippedActiveTaskIds, skippedRetainedTaskIds); } } + /** 一个事务内完成：收集指针 → 删除本批行 → 行删完后提交清理队列并 flush。 */ + private int deleteRows(List moduleTypes, List cleanupTaskIds, List collectDataTaskIds) { + taskFileJobMapper.delete(new LambdaQueryWrapper() + .in(TaskFileJobEntity::getModuleType, moduleTypes) + .in(TaskFileJobEntity::getTaskId, cleanupTaskIds)); + + taskResultItemMapper.delete(new LambdaQueryWrapper() + .in(TaskResultItemEntity::getModuleType, moduleTypes) + .in(TaskResultItemEntity::getTaskId, cleanupTaskIds)); + + taskProgressSnapshotMapper.delete(new LambdaQueryWrapper() + .in(TaskProgressSnapshotEntity::getModuleType, moduleTypes) + .in(TaskProgressSnapshotEntity::getTaskId, cleanupTaskIds)); + + taskResultPayloadMapper.delete(new LambdaQueryWrapper() + .in(TaskResultPayloadEntity::getModuleType, moduleTypes) + .in(TaskResultPayloadEntity::getTaskId, cleanupTaskIds)); + + taskScopeStateMapper.delete(new LambdaQueryWrapper() + .in(TaskScopeStateEntity::getModuleType, moduleTypes) + .in(TaskScopeStateEntity::getTaskId, cleanupTaskIds)); + + taskChunkMapper.delete(new LambdaQueryWrapper() + .in(TaskChunkEntity::getModuleType, moduleTypes) + .in(TaskChunkEntity::getTaskId, cleanupTaskIds)); + + if (!collectDataTaskIds.isEmpty()) { + collectDataItemMapper.delete(new LambdaQueryWrapper() + .in(CollectDataItemEntity::getTaskId, collectDataTaskIds)); + } + + fileResultMapper.delete(new LambdaQueryWrapper() + .in(FileResultEntity::getModuleType, moduleTypes) + .in(FileResultEntity::getTaskId, cleanupTaskIds)); + + fileTaskMapper.update(null, new LambdaUpdateWrapper() + .in(FileTaskEntity::getId, cleanupTaskIds) + .set(FileTaskEntity::getResultJson, null) + .set(FileTaskEntity::getRequestJson, null) + .set(FileTaskEntity::getErrorMessage, null)); + + return fileTaskMapper.delete(new LambdaQueryWrapper() + .in(FileTaskEntity::getId, cleanupTaskIds)); + } + + private void submitAndFlush(List collectedPointers) { + if (collectedPointers.isEmpty()) { + return; + } + transientPayloadDeleteOrchestrator.submitDeletes(collectedPointers); + transientPayloadDeleteOrchestrator.flushPendingDeletes(); + } + private boolean isExpired(FileTaskEntity task, LocalDateTime cutoff) { LocalDateTime completedAt = task.getFinishedAt() != null ? task.getFinishedAt() : task.getUpdatedAt(); return completedAt != null && !completedAt.isAfter(cutoff); @@ -195,16 +255,16 @@ public class ModuleHistoryCleanupService { * scope_state.parsedPayloadJson / stateJson），去重并保持稳定顺序； * 达到 {@code max} 上限即截断，防止异常巨大的任务行数引发无界收集。 */ - private List collectPayloadPointers(List cleanupTaskIds, int max) { + private List collectPayloadPointers(List moduleTypes, List cleanupTaskIds, int max) { java.util.Set pointers = new java.util.LinkedHashSet<>(); List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() - .in(TaskChunkEntity::getModuleType, moduleCleanupProperties.getModuleTypes()) + .in(TaskChunkEntity::getModuleType, moduleTypes) .in(TaskChunkEntity::getTaskId, cleanupTaskIds)); for (TaskChunkEntity chunk : chunks) { collectPointer(pointers, chunk.getPayloadJson(), max); } List scopeStates = taskScopeStateMapper.selectList(new LambdaQueryWrapper() - .in(TaskScopeStateEntity::getModuleType, moduleCleanupProperties.getModuleTypes()) + .in(TaskScopeStateEntity::getModuleType, moduleTypes) .in(TaskScopeStateEntity::getTaskId, cleanupTaskIds)); for (TaskScopeStateEntity scopeState : scopeStates) { collectPointer(pointers, scopeState.getParsedPayloadJson(), max); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java new file mode 100644 index 00000000..7d9465a0 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java @@ -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 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 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 expiredTasks(long... ids) { + List tasks = new ArrayList<>(); + for (long id : ids) { + tasks.add(expiredTask(id)); + } + return tasks; + } + + private static Answer> sequence(List rows, int rowCalls, List fallback) { + return new Answer>() { + private int calls; + + @Override + public List answer(InvocationOnMock invocation) { + return calls++ < rowCalls ? rows : fallback; + } + }; + } + + /** 按调用次数依次返回每一页（keyset 翻页，页内 id 升序且首元素大于上次游标）。 */ + private static Answer> pages(List> pages) { + return new Answer>() { + private int calls; + + @Override + public List 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> captor = + ArgumentCaptor.forClass(LambdaQueryWrapper.class); + verify(fileTaskMapper, times(3)).delete(captor.capture()); + for (LambdaQueryWrapper 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(); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java index 7435a83c..14ecbc61 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java @@ -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 tasks = new ArrayList<>(); for (long id : ids) { tasks.add(expiredTask(id)); } - when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(tasks); + List none = List.of(); + when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))) + .thenAnswer(sequence(tasks, 1, none)); + } + + /** 幂等/恢复场景：多次清理运行各返回一轮 rows。 */ + private void stubExpiredTasksRuns(int runs, long... ids) { + List tasks = new ArrayList<>(); + for (long id : ids) { + tasks.add(expiredTask(id)); + } + List 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"))