task-69: 历史清理任务收集 transient payload 指针并提交删除队列

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<String> collectedPayloadPointers = collectPayloadPointers(cleanupTaskIds, maxCollectPayloadsPerRun);
+
             int deletedFileJobs = taskFileJobMapper.delete(new LambdaQueryWrapper<TaskFileJobEntity>()
                     .in(TaskFileJobEntity::getModuleType, moduleTypes)
                     .in(TaskFileJobEntity::getTaskId, cleanupTaskIds));
@@ -151,11 +164,17 @@ public class ModuleHistoryCleanupService {
             int deletedTasks = fileTaskMapper.delete(new LambdaQueryWrapper<FileTaskEntity>()
                     .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<String> collectPayloadPointers(List<Long> cleanupTaskIds, int max) {
+        java.util.Set<String> pointers = new java.util.LinkedHashSet<>();
+        List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
+                .in(TaskChunkEntity::getModuleType, moduleCleanupProperties.getModuleTypes())
+                .in(TaskChunkEntity::getTaskId, cleanupTaskIds));
+        for (TaskChunkEntity chunk : chunks) {
+            collectPointer(pointers, chunk.getPayloadJson(), max);
+        }
+        List<TaskScopeStateEntity> scopeStates = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
+                .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<String> 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);
+        }
+    }
 }

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<String> pointers) {
         for (String pointer : pointers) {
             if (pointer == null || !pointer.startsWith(RUSTFS_POINTER_PREFIX)) {

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<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);
+        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<FileTaskEntity> 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 <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;
+            }
+        };
+    }
+
+    private <T> Answer<List<T>> sequenceThenFail(List<T> rows, int rowCalls) {
+        return new Answer<List<T>>() {
+            private int calls;
+
+            @Override
+            public List<T> 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<TaskChunkEntity> chunks = List.of(chunk(11, jsonPointer(11)));
+        List<TaskScopeStateEntity> 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<TaskChunkEntity> chunks = List.of(
+                chunk(21, jsonPointer(21)), chunk(21, jsonPointer(22)), chunk(22, pointer(23)));
+        List<TaskScopeStateEntity> 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<TaskChunkEntity> 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<TaskChunkEntity> chunks = List.of(chunk(61, null), chunk(61, "not-a-pointer"), chunk(61, " "));
+        List<TaskScopeStateEntity> 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<TaskChunkEntity> 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");
+    }
+}