From 2de58b7afc4f353d9bea8d7181ad3ed0eebf2492 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:13:43 +0800 Subject: [PATCH] =?UTF-8?q?task-68:=20payload=20=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E6=94=B9=E4=B8=BA=E6=89=B9=E9=87=8F=E5=BC=95?= =?UTF-8?q?=E7=94=A8=E6=A3=80=E6=9F=A5=E4=B8=8E=E5=BC=82=E6=AD=A5=E7=89=A9?= =?UTF-8?q?=E7=90=86=E5=88=A0=E9=99=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 TransientPayloadDeleteOrchestrator：提交去重入队（maxPendingDeletes 上限）， flush 时一次性 IN 批量反查 biz_task_chunk / biz_task_scope_state， 未引用对象交由后台线程池异步物理删除（剥 rustfs: 前缀） - 幂等：重复提交只入队一次；保守：查询失败整批保留可重试，不误删 - 空值/非指针忽略，本地/OSS 指针不在批量删除范围（各有归属与清理路径） --- .../TransientPayloadDeleteOrchestrator.java | 205 ++++++++++++++++ ...ransientPayloadDeleteOrchestratorTest.java | 221 ++++++++++++++++++ 2 files changed, 426 insertions(+) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestratorTest.java 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 new file mode 100644 index 00000000..8efc6d2b --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestrator.java @@ -0,0 +1,205 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.file.service.object.RustfsObjectStorageService; +import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.function.Supplier; + +/** + * P2-9 扩展（task-68）：payload 引用删除改为批量引用检查与异步物理删除。 + *

调用方把待删 payload（指针或 JSON 编码指针）批量提交，本组件去重入队； + * {@link #flushPendingDeletes()} 对整批 pending 一次性 IN 反查 + * biz_task_chunk / biz_task_scope_state（比逐条两次查询少一个数量级的 DB 往返）， + * 确认不再被引用的对象交由后台线程池异步物理删除，不阻塞业务线程。 + *

幂等：重复提交相同对象只入队一次；保守：引用检查失败时本批保留， + * 调用方（如周期清理任务）可再次 flush 重试。 + */ +@Service +@Slf4j +public class TransientPayloadDeleteOrchestrator { + + private static final String RUSTFS_POINTER_PREFIX = "rustfs:"; + + private final TransientPayloadStorageService transientPayloadStorageService; + private final RustfsObjectStorageService rustfsObjectStorageService; + private final TaskChunkMapper taskChunkMapper; + private final TaskScopeStateMapper taskScopeStateMapper; + private final ObjectMapper objectMapper; + private final ExecutorService asyncDeleteExecutor; + + private final Set pendingPointers = new LinkedHashSet<>(); + + @Value("${aiimage.transient-storage.max-pending-deletes:1000}") + private long maxPendingDeletes = 1000; + + public TransientPayloadDeleteOrchestrator(TransientPayloadStorageService transientPayloadStorageService, + RustfsObjectStorageService rustfsObjectStorageService, + TaskChunkMapper taskChunkMapper, + TaskScopeStateMapper taskScopeStateMapper, + ObjectMapper objectMapper, + ExecutorService asyncDeleteExecutor) { + this.transientPayloadStorageService = transientPayloadStorageService; + this.rustfsObjectStorageService = rustfsObjectStorageService; + this.taskChunkMapper = taskChunkMapper; + this.taskScopeStateMapper = taskScopeStateMapper; + this.objectMapper = objectMapper; + this.asyncDeleteExecutor = asyncDeleteExecutor; + } + + /** + * 批量提交待删 payload 值（指针或 JSON 编码指针）。空值与非指针值忽略； + * 已在 pending 中的对象幂等跳过；pending 达到上限后拒绝新提交并返回实际入队数。 + */ + public int submitDeletes(List values) { + if (values == null || values.isEmpty()) { + return 0; + } + synchronized (pendingPointers) { + int accepted = 0; + for (String value : values) { + String pointer = transientPayloadStorageService.extractPointer(value); + if (pointer == null) { + continue; + } + if (pendingPointers.size() >= maxPendingDeletes) { + log.warn("[transient-payload] delete queue full, drop {} pointer={}", "submit", pointer); + break; + } + if (pendingPointers.add(pointer)) { + accepted++; + } + } + return accepted; + } + } + + /** + * 对 pending 中的对象批量做引用检查，未引用的异步物理删除，返回本批删除数。 + * 引用检查异常时保守跳过整批（不删），调用方可再次 flush 重试。 + */ + public int flushPendingDeletes() { + List batch; + synchronized (pendingPointers) { + if (pendingPointers.isEmpty()) { + return 0; + } + batch = new ArrayList<>(pendingPointers); + pendingPointers.clear(); + } + try { + Set stillReferenced = batchReferencedPointers(batch); + List toDelete = new ArrayList<>(batch); + toDelete.removeAll(stillReferenced); + if (!toDelete.isEmpty()) { + asyncDeleteExecutor.submit(() -> deleteObjects(toDelete)); + } + if (!stillReferenced.isEmpty()) { + log.info("[transient-payload] skip delete, still referenced count={}", stillReferenced.size()); + } + return toDelete.size(); + } catch (Exception ex) { + log.warn("[transient-payload] batch reference check failed, keep pending count={} err={}", + batch.size(), ex.getMessage()); + synchronized (pendingPointers) { + pendingPointers.addAll(batch); + } + return 0; + } + } + + public int pendingCount() { + synchronized (pendingPointers) { + return pendingPointers.size(); + } + } + + private void deleteObjects(List pointers) { + for (String pointer : pointers) { + if (pointer == null || !pointer.startsWith(RUSTFS_POINTER_PREFIX)) { + // 本地/OSS 指针有实例归属与其它清理路径，不在此批量删除范围。 + continue; + } + String objectKey = pointer.substring(RUSTFS_POINTER_PREFIX.length()); + try { + rustfsObjectStorageService.deleteObject(objectKey); + } catch (Exception ex) { + log.warn("[transient-payload] async delete failed objectKey={} err={}", + objectKey, ex.getMessage()); + } + } + } + + /** 批量 IN 反查两张引用表，返回仍被引用的指针集合（查询异常抛给调用方）。 */ + private Set batchReferencedPointers(List pointers) { + List jsonEncoded = pointers.stream() + .map(this::jsonEncodePointer) + .filter(java.util.Objects::nonNull) + .toList(); + List candidates = new ArrayList<>(pointers); + candidates.addAll(jsonEncoded); + + Set referenced = new LinkedHashSet<>(); + List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() + .in(TaskChunkEntity::getPayloadJson, candidates)); + for (TaskChunkEntity chunk : chunks) { + referenced.addAll(matchingPointers(chunk.getPayloadJson(), pointers)); + } + List scopeStates = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .and(w -> w.in(TaskScopeStateEntity::getParsedPayloadJson, candidates) + .or() + .in(TaskScopeStateEntity::getStateJson, candidates))); + for (TaskScopeStateEntity scopeState : scopeStates) { + referenced.addAll(matchingPointers(scopeState.getParsedPayloadJson(), pointers)); + referenced.addAll(matchingPointers(scopeState.getStateJson(), pointers)); + } + return referenced; + } + + private Set matchingPointers(String dbValue, List pointers) { + Set matches = new LinkedHashSet<>(); + if (dbValue == null || dbValue.isBlank()) { + return matches; + } + String candidate = dbValue.trim(); + for (int i = 0; i < 4; i++) { + String pointer = transientPayloadStorageService.extractPointer(candidate); + if (pointer != null && pointers.contains(pointer)) { + matches.add(pointer); + } + String decoded; + try { + decoded = objectMapper.readValue(candidate, String.class); + } catch (Exception ex) { + break; + } + if (decoded == null || decoded.equals(candidate)) { + break; + } + candidate = decoded.trim(); + } + return matches; + } + + private String jsonEncodePointer(String pointer) { + try { + return objectMapper.writeValueAsString(pointer); + } catch (Exception ex) { + return null; + } + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestratorTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestratorTest.java new file mode 100644 index 00000000..cecc425d --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadDeleteOrchestratorTest.java @@ -0,0 +1,221 @@ +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.config.InstanceMetadata; +import com.nanri.aiimage.config.StorageProperties; +import com.nanri.aiimage.config.TransientStorageProperties; +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.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +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.springframework.test.util.ReflectionTestUtils; + +import java.nio.file.Path; +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.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 68：payload 引用删除改为批量引用检查与异步物理删除。 + * TransientPayloadDeleteOrchestrator 把待删 pointer 去重入队，flush 时一次性 + * IN 批量反查 biz_task_chunk / biz_task_scope_state（而非逐条两次查询）， + * 未引用的对象交由后台线程池异步物理删除；重复提交幂等、失败保守保留。 + */ +class TransientPayloadDeleteOrchestratorTest { + + @TempDir + Path tempDir; + + private TaskChunkMapper chunkMapper; + private TaskScopeStateMapper scopeStateMapper; + private RustfsObjectStorageService rustfs; + private OssStorageService oss; + private TransientPayloadStorageService storage; + private TransientPayloadDeleteOrchestrator orchestrator; + private ExecutorService executor; + + @BeforeAll + static void initializeMybatisMetadata() { + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new MybatisConfiguration(), ""), + TaskChunkEntity.class); + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new MybatisConfiguration(), ""), + TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + rustfs = mock(RustfsObjectStorageService.class); + oss = mock(OssStorageService.class); + chunkMapper = mock(TaskChunkMapper.class); + scopeStateMapper = mock(TaskScopeStateMapper.class); + TransientStorageProperties properties = new TransientStorageProperties(); + properties.setEnabled(true); + StorageProperties storageProperties = new StorageProperties(); + storageProperties.setLocalTempDir(tempDir.toString()); + storage = new TransientPayloadStorageService( + properties, storageProperties, rustfs, oss, + new ObjectMapper(), new InstanceMetadata("test-instance"), + chunkMapper, scopeStateMapper); + executor = Executors.newFixedThreadPool(2); + orchestrator = new TransientPayloadDeleteOrchestrator( + storage, rustfs, chunkMapper, scopeStateMapper, new ObjectMapper(), executor); + ReflectionTestUtils.setField(orchestrator, "maxPendingDeletes", 100L); + } + + private static String pointer(int taskId) { + return "rustfs:task-parsed/test/" + taskId + "/scope/latest.json"; + } + + private void stubNoReferences() { + when(chunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + when(scopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of()); + } + + 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_068_payload_normal_default_path() throws Exception { + // 默认路径：批量提交未引用 pointer，flush 后全部异步物理删除。 + stubNoReferences(); + int accepted = orchestrator.submitDeletes(List.of(pointer(1), pointer(2))); + assertEquals(2, accepted); + + CountDownLatch done = latchOnDelete(2); + assertEquals(2, orchestrator.flushPendingDeletes(), "两个对象提交物理删除"); + assertTrue(done.await(2, TimeUnit.SECONDS), "异步删除完成"); + verify(rustfs, times(2)).deleteObject(anyString()); + verify(chunkMapper).selectList(any(LambdaQueryWrapper.class)); + verify(scopeStateMapper).selectList(any(LambdaQueryWrapper.class)); + } + + @Test + void test_task_068_payload_normal_multiple_items() throws Exception { + // 批量场景：5 个对象批量检查，1 个仍被 scope_state 引用则跳过，其余 4 个删除。 + stubNoReferences(); + TaskScopeStateEntity referenced = new TaskScopeStateEntity(); + referenced.setParsedPayloadJson("\"rustfs:task-parsed/test/3/scope/latest.json\""); + when(scopeStateMapper.selectList(any(LambdaQueryWrapper.class))) + .thenReturn(List.of(referenced)); + + List values = new ArrayList<>(); + for (int i = 1; i <= 5; i++) { + values.add(pointer(i)); + } + assertEquals(5, orchestrator.submitDeletes(values)); + CountDownLatch done = latchOnDelete(4); + assertEquals(4, orchestrator.flushPendingDeletes(), "仅删除未引用对象"); + assertTrue(done.await(2, TimeUnit.SECONDS)); + verify(rustfs, times(4)).deleteObject(anyString()); + verify(rustfs, never()).deleteObject("task-parsed/test/3/scope/latest.json"); + } + + @Test + void test_task_068_payload_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：重复提交同一对象只入队一次、只删除一次，不产生重复物理删除请求。 + stubNoReferences(); + assertEquals(1, orchestrator.submitDeletes(List.of(pointer(1)))); + assertEquals(0, orchestrator.submitDeletes(List.of(pointer(1))), "重复提交被去重"); + + CountDownLatch done = latchOnDelete(1); + assertEquals(1, orchestrator.flushPendingDeletes()); + assertTrue(done.await(2, TimeUnit.SECONDS)); + verify(rustfs, times(1)).deleteObject(anyString()); + } + + @Test + void test_task_068_payload_boundary_empty_input() throws Exception { + // 空输入：空集合/含空值集合不入队、不查询、不删除。 + assertEquals(0, orchestrator.submitDeletes(List.of())); + assertEquals(0, orchestrator.submitDeletes(null)); + assertEquals(0, orchestrator.submitDeletes(java.util.Arrays.asList("", " ", null))); + + assertEquals(0, orchestrator.flushPendingDeletes()); + verify(chunkMapper, never()).selectList(any()); + verify(scopeStateMapper, never()).selectList(any()); + verify(rustfs, never()).deleteObject(anyString()); + } + + @Test + void test_task_068_payload_boundary_single_item() throws Exception { + // 单元素：单个对象不依赖批量路径，检查一次删除一次。 + stubNoReferences(); + assertEquals(1, orchestrator.submitDeletes(List.of(pointer(7)))); + CountDownLatch done = latchOnDelete(1); + assertEquals(1, orchestrator.flushPendingDeletes()); + assertTrue(done.await(2, TimeUnit.SECONDS)); + verify(rustfs).deleteObject("task-parsed/test/7/scope/latest.json"); + } + + @Test + void test_task_068_payload_boundary_limit_and_overflow() throws Exception { + // 上限/超限：pending 队列达到 maxPendingDeletes 后拒绝新提交，不发生无界内存增长。 + stubNoReferences(); + ReflectionTestUtils.setField(orchestrator, "maxPendingDeletes", 3L); + List values = List.of(pointer(1), pointer(2), pointer(3), pointer(4), pointer(5)); + assertEquals(3, orchestrator.submitDeletes(values), "超过上限的提交被拒绝"); + + CountDownLatch done = latchOnDelete(3); + assertEquals(3, orchestrator.flushPendingDeletes()); + assertTrue(done.await(2, TimeUnit.SECONDS)); + verify(rustfs, times(3)).deleteObject(anyString()); + } + + @Test + void test_task_068_payload_invalid_input_rejected() { + // 非法参数：非指针值不入队，不发起引用检查。 + stubNoReferences(); + assertEquals(0, orchestrator.submitDeletes(List.of("not-a-pointer", "plain text"))); + assertEquals(0, orchestrator.flushPendingDeletes()); + verify(chunkMapper, never()).selectList(any()); + } + + @Test + void test_task_068_payload_dependency_failure_releases_resources() throws Exception { + // 依赖失败：引用检查查询异常时保守不删、pending 保留可重试；恢复后删除成功。 + when(chunkMapper.selectList(any(LambdaQueryWrapper.class))) + .thenThrow(new RuntimeException("db down")); + assertEquals(1, orchestrator.submitDeletes(List.of(pointer(1)))); + + assertEquals(0, orchestrator.flushPendingDeletes(), "查询失败保守不删"); + verify(rustfs, never()).deleteObject(anyString()); + + stubNoReferences(); + CountDownLatch done = latchOnDelete(1); + assertEquals(1, orchestrator.flushPendingDeletes(), "恢复后重试成功"); + assertTrue(done.await(2, TimeUnit.SECONDS)); + verify(rustfs).deleteObject("task-parsed/test/1/scope/latest.json"); + } +}