From 5749a421cbcc61a39cdacabbe2c84aedfb7869b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 19:29:32 +0800 Subject: [PATCH] =?UTF-8?q?task-69:=20=E5=8E=86=E5=8F=B2=E6=B8=85=E7=90=86?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E6=94=B6=E9=9B=86=20transient=20payload=20?= =?UTF-8?q?=E6=8C=87=E9=92=88=E5=B9=B6=E6=8F=90=E4=BA=A4=E5=88=A0=E9=99=A4?= =?UTF-8?q?=E9=98=9F=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/ModuleHistoryCleanupService.java | 59 ++- .../TransientPayloadDeleteOrchestrator.java | 8 + ...oduleHistoryCleanupPayloadCleanupTest.java | 404 ++++++++++++++++++ 3 files changed, 469 insertions(+), 2 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java 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 0e8572d2..806387fa 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 @@ -24,6 +24,7 @@ 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; @@ -55,6 +56,14 @@ public class ModuleHistoryCleanupService { private final TaskChunkMapper taskChunkMapper; private final CollectDataItemMapper collectDataItemMapper; private final DistributedJobLockService distributedJobLockService; + private final TransientPayloadDeleteOrchestrator transientPayloadDeleteOrchestrator; + + /** + * 单次清理最多收集的 payload 指针数：超过即截断（保底可重试）， + * 防止单任务行数异常巨大时无界收集造成内存增长。 + */ + @Value("${aiimage.module-cleanup.max-collect-payloads:10000}") + private int maxCollectPayloadsPerRun = 10000; @Transactional @Scheduled(cron = "${aiimage.module-cleanup.cron:0 0 0 * * *}") @@ -108,6 +117,10 @@ public class ModuleHistoryCleanupService { return; } + // 行删除前先收集将随行删除的 transient payload 指针（上限截断）， + // 全部删除成功后再提交清理队列并 flush，避免删除失败留下已提交的孤儿清理。 + List collectedPayloadPointers = collectPayloadPointers(cleanupTaskIds, maxCollectPayloadsPerRun); + int deletedFileJobs = taskFileJobMapper.delete(new LambdaQueryWrapper() .in(TaskFileJobEntity::getModuleType, moduleTypes) .in(TaskFileJobEntity::getTaskId, cleanupTaskIds)); @@ -151,11 +164,17 @@ public class ModuleHistoryCleanupService { int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper() .in(FileTaskEntity::getId, cleanupTaskIds)); - log.info("[module-cleanup] completed: moduleTypes={}, retentionDays={}, cutoff={}, deletedFileJobs={}, deletedResultItems={}, deletedProgressSnapshots={}, deletedResultPayloads={}, deletedScopeStates={}, deletedChunks={}, deletedCollectDataItems={}, deletedResults={}, resetTasks={}, deletedTasks={}, skippedActiveTaskIds={}, retainedTaskIds={}", + // 行已全部删除，此时引用检查将判定这些指针不再被引用 → 提交清理队列并 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={}", moduleTypes, moduleCleanupProperties.getRetentionDays(), cutoff, deletedFileJobs, deletedResultItems, deletedProgressSnapshots, deletedResultPayloads, deletedScopeStates, deletedChunks, deletedCollectDataItems, deletedResults, resetTasks, deletedTasks, - skippedActiveTaskIds, skippedRetainedTaskIds); + collectedPayloadPointers.size(), skippedActiveTaskIds, skippedRetainedTaskIds); } } @@ -170,4 +189,40 @@ public class ModuleHistoryCleanupService { } return TERMINAL_STATUSES.contains(status.trim().toUpperCase(Locale.ROOT)); } + + /** + * 删除前批量收集将随行删除的 payload 指针（chunk.payloadJson、 + * scope_state.parsedPayloadJson / stateJson），去重并保持稳定顺序； + * 达到 {@code max} 上限即截断，防止异常巨大的任务行数引发无界收集。 + */ + private List collectPayloadPointers(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::getTaskId, cleanupTaskIds)); + for (TaskChunkEntity chunk : chunks) { + collectPointer(pointers, chunk.getPayloadJson(), max); + } + List scopeStates = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .in(TaskScopeStateEntity::getModuleType, moduleCleanupProperties.getModuleTypes()) + .in(TaskScopeStateEntity::getTaskId, cleanupTaskIds)); + for (TaskScopeStateEntity scopeState : scopeStates) { + collectPointer(pointers, scopeState.getParsedPayloadJson(), max); + collectPointer(pointers, scopeState.getStateJson(), max); + } + if (pointers.size() >= max) { + log.warn("[module-cleanup] payload pointer collection truncated at max={}", max); + } + return new ArrayList<>(pointers); + } + + private void collectPointer(java.util.Set pointers, String value, int max) { + if (pointers.size() >= max || value == null || value.isBlank()) { + return; + } + String pointer = transientPayloadDeleteOrchestrator.extractPointer(value); + if (pointer != null) { + pointers.add(pointer); + } + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java index 8efc6d2b..e6dc36b6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java @@ -128,6 +128,14 @@ public class TransientPayloadDeleteOrchestrator { } } + /** + * 从任意存储值（指针或 JSON 编码指针）中提取 rustfs 指针，非指针返回 null。 + * 供调用方在提交前过滤待删值（如清理任务收集 payload 字段）。 + */ + public String extractPointer(String value) { + return transientPayloadStorageService.extractPointer(value); + } + private void deleteObjects(List pointers) { for (String pointer : pointers) { if (pointer == null || !pointer.startsWith(RUSTFS_POINTER_PREFIX)) { 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 new file mode 100644 index 00000000..7435a83c --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java @@ -0,0 +1,404 @@ +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.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 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.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 69：为数据库删除任务补充 transient payload 指针收集和清理队列。 + * ModuleHistoryCleanupService 在删除 chunk / scope_state 行之前先批量收集 + * payloadJson / parsedPayloadJson / stateJson 中的指针并提交给 + * TransientPayloadDeleteOrchestrator，全部删除完成后 flush 一次， + * 使清理任务删除的 DB 行不再遗留孤儿对象；删除失败不提交、行删完才提交。 + */ +class ModuleHistoryCleanupPayloadCleanupTest { + + @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 ModuleHistoryCleanupService cleanupService; + + @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); + cleanupService = new ModuleHistoryCleanupService( + cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper, + taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper, + taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator); + } + + 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 void stubExpiredTasks(long... ids) { + List tasks = new ArrayList<>(); + for (long id : ids) { + tasks.add(expiredTask(id)); + } + when(fileTaskMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(tasks); + } + + /** 收集与 flush 都走同一 mapper.selectList：前 rowCalls 次返回 rows（行未删），之后返回 fallback。 */ + private 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; + } + }; + } + + private Answer> sequenceThenFail(List rows, int rowCalls) { + return new Answer>() { + private int calls; + + @Override + public List answer(InvocationOnMock invocation) { + if (calls++ < rowCalls) { + return rows; + } + throw new RuntimeException("db down"); + } + }; + } + + 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; + } + + 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 CountDownLatch latchOnDelete(int count) { + CountDownLatch latch = new CountDownLatch(count); + doAnswer(invocation -> { + latch.countDown(); + return null; + }).when(rustfs).deleteObject(anyString()); + return latch; + } + + @Test + void test_task_069_payload_cleanup_normal_default_path() throws Exception { + // 默认路径：清理删除 chunk / scope_state 行前收集 payload 指针提交队列， + // 行全部删除后 flush，未引用对象异步物理删除一次。 + stubExpiredTasks(11); + stubDeletes(); + List chunks = List.of(chunk(11, jsonPointer(11))); + List states = List.of(scopeState(11, jsonPointer(12), pointer(13))); + doAnswer(sequence(chunks, 1, List.of())).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + doAnswer(sequence(states, 1, List.of())).when(taskScopeStateMapper).selectList(any(LambdaQueryWrapper.class)); + + CountDownLatch done = latchOnDelete(3); + cleanupService.cleanupConfiguredModules(); + + assertTrue(done.await(2, TimeUnit.SECONDS), "三个收集的指针异步删除完成"); + verify(taskChunkMapper).delete(any(LambdaQueryWrapper.class)); + verify(taskScopeStateMapper).delete(any(LambdaQueryWrapper.class)); + verify(rustfs, times(3)).deleteObject(anyString()); + verify(rustfs).deleteObject("task-parsed/test/11/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/12/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/13/scope/latest.json"); + } + + @Test + void test_task_069_payload_cleanup_normal_multiple_items() throws Exception { + // 批量场景：多个任务、多个 chunk / scope_state 行的指针全部收集， + // 去重后顺序稳定、无丢失；flush 后每个对象恰好删除一次。 + stubExpiredTasks(21, 22); + stubDeletes(); + List chunks = List.of( + chunk(21, jsonPointer(21)), chunk(21, jsonPointer(22)), chunk(22, pointer(23))); + List states = List.of( + scopeState(21, pointer(24), jsonPointer(21)), scopeState(22, jsonPointer(25), null)); + doAnswer(sequence(chunks, 1, List.of())).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + doAnswer(sequence(states, 1, List.of())).when(taskScopeStateMapper).selectList(any(LambdaQueryWrapper.class)); + + CountDownLatch done = latchOnDelete(5); + cleanupService.cleanupConfiguredModules(); + + assertTrue(done.await(2, TimeUnit.SECONDS), "五个收集的指针异步删除完成"); + verify(rustfs, times(5)).deleteObject(anyString()); + verify(rustfs).deleteObject("task-parsed/test/21/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/22/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/23/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/24/scope/latest.json"); + verify(rustfs).deleteObject("task-parsed/test/25/scope/latest.json"); + } + + @Test + void test_task_069_payload_cleanup_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：行已删除后再次运行不收集指针、不提交、不发起物理删除。 + stubExpiredTasks(31); + stubDeletes(); + when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + CountDownLatch done = latchOnDelete(1); + + cleanupService.cleanupConfiguredModules(); + cleanupService.cleanupConfiguredModules(); + + assertEquals(1, done.getCount(), "两次运行均无对象进入删除队列"); + verify(rustfs, never()).deleteObject(anyString()); + verify(taskChunkMapper, times(2)).selectList(any(LambdaQueryWrapper.class)); + verify(taskScopeStateMapper, times(2)).selectList(any(LambdaQueryWrapper.class)); + } + + @Test + void test_task_069_payload_cleanup_boundary_empty_input() { + // 空输入：无过期任务时安全跳过，不查询 rows、不提交、不 flush、不删除任何对象。 + stubExpiredTasks(); + + cleanupService.cleanupConfiguredModules(); + + verify(taskChunkMapper, never()).selectList(any(LambdaQueryWrapper.class)); + verify(taskScopeStateMapper, never()).selectList(any(LambdaQueryWrapper.class)); + verify(taskChunkMapper, never()).delete(any(LambdaQueryWrapper.class)); + verify(taskScopeStateMapper, never()).delete(any(LambdaQueryWrapper.class)); + verify(rustfs, never()).deleteObject(anyString()); + } + + @Test + void test_task_069_payload_cleanup_boundary_single_item() throws Exception { + // 单元素：单任务单 chunk 单指针，收集一次提交一次，删除一次。 + stubExpiredTasks(41); + stubDeletes(); + doAnswer(sequence(List.of(chunk(41, jsonPointer(41))), 1, List.of())) + .when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + doAnswer(sequence(List.of(), 1, List.of())) + .when(taskScopeStateMapper).selectList(any(LambdaQueryWrapper.class)); + + CountDownLatch done = latchOnDelete(1); + cleanupService.cleanupConfiguredModules(); + + assertTrue(done.await(2, TimeUnit.SECONDS), "单指针异步删除完成"); + verify(rustfs, times(1)).deleteObject(anyString()); + verify(rustfs).deleteObject("task-parsed/test/41/scope/latest.json"); + } + + @Test + void test_task_069_payload_cleanup_boundary_limit_and_overflow() throws Exception { + // 上限/超限：收集达到上限即停止，不发生无界收集； + // flush 时引用检查失败则本批保留（保守），可下次重试。 + stubExpiredTasks(51); + stubDeletes(); + ReflectionTestUtils.setField(cleanupService, "maxCollectPayloadsPerRun", 3); + List chunks = new ArrayList<>(); + for (int i = 1; i <= 5; i++) { + chunks.add(chunk(51, jsonPointer(50 + i))); + } + doAnswer(sequenceThenFail(chunks, 1)).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + + cleanupService.cleanupConfiguredModules(); + + assertEquals(3, orchestrator.pendingCount(), "仅保留上限内的指针"); + verify(rustfs, never()).deleteObject(anyString()); + verify(taskChunkMapper).delete(any(LambdaQueryWrapper.class)); + } + + @Test + void test_task_069_payload_cleanup_invalid_input_rejected() throws Exception { + // 非法参数：空值与纯空白不收集，非指针文本由清理队列过滤， + // 行正常删除、无物理删除；flush 引用检查失败时有效指针保守保留。 + stubExpiredTasks(61); + stubDeletes(); + List chunks = List.of(chunk(61, null), chunk(61, "not-a-pointer"), chunk(61, " ")); + List states = List.of(scopeState(61, null, pointer(63))); + doAnswer(sequenceThenFail(chunks, 1)).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + doAnswer(sequenceThenFail(states, 1)).when(taskScopeStateMapper).selectList(any(LambdaQueryWrapper.class)); + + cleanupService.cleanupConfiguredModules(); + + assertEquals(1, orchestrator.pendingCount(), "仅有效指针入队"); + verify(taskChunkMapper).delete(any(LambdaQueryWrapper.class)); + verify(taskScopeStateMapper).delete(any(LambdaQueryWrapper.class)); + verify(rustfs, never()).deleteObject(anyString()); + } + + @Test + void test_task_069_payload_cleanup_dependency_failure_releases_resources() throws Exception { + // 依赖失败：行删除抛异常时中止，指针不提交、不 flush、不删除对象； + // 恢复后再次运行，收集、删除与清理队列全部正常完成。 + stubExpiredTasks(71); + stubDeletes(); + when(taskChunkMapper.delete(any(LambdaQueryWrapper.class))) + .thenThrow(new RuntimeException("db down")) + .thenReturn(0); + List chunks = List.of(chunk(71, jsonPointer(71))); + doAnswer(sequence(chunks, 2, List.of())).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class)); + when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + + org.junit.jupiter.api.Assertions.assertThrows(RuntimeException.class, + () -> cleanupService.cleanupConfiguredModules()); + + assertEquals(0, orchestrator.pendingCount(), "删除失败时指针不提交"); + verify(rustfs, never()).deleteObject(anyString()); + + CountDownLatch done = latchOnDelete(1); + cleanupService.cleanupConfiguredModules(); + + assertTrue(done.await(2, TimeUnit.SECONDS), "恢复后指针异步删除完成"); + assertEquals(0, orchestrator.pendingCount(), "恢复后清理队列清空"); + verify(rustfs, times(1)).deleteObject(anyString()); + verify(rustfs).deleteObject("task-parsed/test/71/scope/latest.json"); + } +}