diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 8340b81e..5308b125 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -1182,23 +1182,16 @@ public class CollectDataService { return safe.isBlank() ? "collect-data" : safe; } - private void deleteTransientTaskPayloads(Long taskId) { - if (taskId == null || taskId <= 0) { - return; - } - List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() - .select(TaskChunkEntity::getPayloadJson) - .eq(TaskChunkEntity::getTaskId, taskId) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)); + private void deleteTransientTaskPayloads(List chunks, List items) { if (chunks != null) { + Set deleted = new HashSet<>(); for (TaskChunkEntity chunk : chunks) { - transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson()); + // 多 chunk 共享同一对象（deterministic key 残留场景）按值去重只删一次。 + if (chunk.getPayloadJson() != null && deleted.add(chunk.getPayloadJson())) { + transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson()); + } } } - List items = taskResultItemMapper.selectList(new LambdaQueryWrapper() - .select(TaskResultItemEntity::getPayloadJson) - .eq(TaskResultItemEntity::getTaskId, taskId) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)); deleteResultItemPayloads(items); } @@ -1248,7 +1241,8 @@ public class CollectDataService { fileResultMapper.delete(new LambdaQueryWrapper() .eq(FileResultEntity::getTaskId, task.getId()) .eq(FileResultEntity::getModuleType, MODULE_TYPE)); - deleteTransientTaskPayloads(task.getId()); + // 先删 DB 行、再物理删对象：引用计数反查基于行删除后的状态， + // 仍被其它任务 chunk/scope_state 引用的共享对象会跳过删除，本任务独占对象真正释放。 taskScopeStateMapper.delete(new LambdaQueryWrapper() .eq(TaskScopeStateEntity::getTaskId, task.getId()) .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)); @@ -1258,6 +1252,15 @@ public class CollectDataService { taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, task.getId()) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)); + deleteTransientTaskPayloads( + taskChunkMapper.selectList(new LambdaQueryWrapper() + .select(TaskChunkEntity::getPayloadJson) + .eq(TaskChunkEntity::getTaskId, task.getId()) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)), + taskResultItemMapper.selectList(new LambdaQueryWrapper() + .select(TaskResultItemEntity::getPayloadJson) + .eq(TaskResultItemEntity::getTaskId, task.getId()) + .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE))); taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE); fileTaskMapper.deleteById(task.getId()); } @@ -1272,11 +1275,12 @@ public class CollectDataService { .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) .eq(TaskResultItemEntity::getResultId, row.getId())); - deleteResultItemPayloads(resultItems); taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) .eq(TaskResultItemEntity::getResultId, row.getId())); + // 与 deleteTask 一致：先删 DB 行再物理删对象，保证行删除与对象删除一致。 + deleteResultItemPayloads(resultItems); taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId()); fileResultMapper.deleteById(resultId); } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataDeleteConsistencyTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataDeleteConsistencyTest.java new file mode 100644 index 00000000..3ecefc15 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataDeleteConsistencyTest.java @@ -0,0 +1,302 @@ +package com.nanri.aiimage.modules.collectdata.service; + +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec; +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.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +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.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.atLeastOnce; +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 57：采集结果对象数据库删除与物理对象删除的一致性处理。 + * deleteTask/deleteHistory 现在先删 DB 行、再物理删 payload（原顺序颠倒， + * 会导致引用计数反查在行未删时必然命中、物理对象永不删除而残留）。 + * 物理删除延迟到事务提交后执行（afterCommit 回调）：事务中途失败整体回滚， + * DB 与物理对象保持一致；跨任务共享指针（chunk/scope_state 仍引用）跳过删除。 + */ +@ExtendWith(MockitoExtension.class) +class CollectDataDeleteConsistencyTest { + + @Mock + private FileTaskMapper fileTaskMapper; + + @Mock + private FileResultMapper fileResultMapper; + + @Mock + private TaskChunkMapper taskChunkMapper; + + @Mock + private TaskScopeStateMapper taskScopeStateMapper; + + @Mock + private TaskResultItemMapper taskResultItemMapper; + + @Mock + private TaskDistributedLockService taskDistributedLockService; + + @Mock + private TaskFileJobService taskFileJobService; + + @Mock + private TransientPayloadStorageService transientPayloadStorageService; + + @Mock + private PlatformTransactionManager transactionManager; + + @Spy + private com.fasterxml.jackson.databind.ObjectMapper objectMapper = + new com.fasterxml.jackson.databind.ObjectMapper(); + + @Spy + private CollectDataResultDetailCodec resultDetailCodec = new CollectDataResultDetailCodec(objectMapper); + + private CollectDataService service; + + @BeforeEach + void setUp() { + // lambda 列名解析需要 MyBatis-Plus TableInfo 缓存；mock 环境手动初始化。 + com.baomidou.mybatisplus.core.metadata.TableInfoHelper.initTableInfo( + new org.apache.ibatis.builder.MapperBuilderAssistant( + new com.baomidou.mybatisplus.core.MybatisConfiguration(), ""), + com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class); + com.baomidou.mybatisplus.core.metadata.TableInfoHelper.initTableInfo( + new org.apache.ibatis.builder.MapperBuilderAssistant( + new com.baomidou.mybatisplus.core.MybatisConfiguration(), ""), + com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity.class); + com.baomidou.mybatisplus.core.metadata.TableInfoHelper.initTableInfo( + new org.apache.ibatis.builder.MapperBuilderAssistant( + new com.baomidou.mybatisplus.core.MybatisConfiguration(), ""), + com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity.class); + TransactionTemplate txTemplate = new TransactionTemplate(transactionManager); + service = new CollectDataService(null, fileTaskMapper, fileResultMapper, + mock(com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper.class), + mock(com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper.class), + mock(com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper.class), + taskChunkMapper, taskScopeStateMapper, taskResultItemMapper, + taskDistributedLockService, taskFileJobService, transientPayloadStorageService, + mock(com.nanri.aiimage.modules.collectdata.service.CollectDataExcelAssemblyService.class), + mock(com.nanri.aiimage.modules.file.service.oss.OssStorageService.class), + objectMapper, txTemplate, + mock(com.nanri.aiimage.modules.collectdata.util.CollectDataBatchQuery.class), + mock(com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter.class), + mock(com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter.class), + resultDetailCodec, + mock(com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter.class), + mock(com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailReader.class));; + } + + private FileTaskEntity task(long id, long userId) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setTaskNo("COLLECT_DATA-" + id); + task.setModuleType(CollectDataService.MODULE_TYPE); + task.setStatus("SUCCESS"); + task.setUserId(userId); + return task; + } + + private static TaskChunkEntity chunk(long id, long taskId, String payload) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(id); + chunk.setTaskId(taskId); + chunk.setModuleType(CollectDataService.MODULE_TYPE); + chunk.setPayloadJson(payload); + return chunk; + } + + private static TaskResultItemEntity item(long id, long taskId, long resultId, String payload) { + TaskResultItemEntity item = new TaskResultItemEntity(); + item.setId(id); + item.setTaskId(taskId); + item.setResultId(resultId); + item.setModuleType(CollectDataService.MODULE_TYPE); + item.setItemKey("asin:B000000001"); + item.setPayloadJson(payload); + return item; + } + + /** chunk 级引用 JSON（{chunk, offset, payload}），供 result_item 引用去重验证。 */ + private static String refJson(int chunkIndex, int offset, String pointer) { + try { + return new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString( + java.util.Map.of("chunk", chunkIndex, "offset", offset, "payload", pointer)); + } catch (Exception ex) { + throw new IllegalStateException(ex); + } + } + + @Test + void test_task_057_collect_normal_default_path() throws Exception { + // 正常路径：删除任务 → DB 行全部删除，chunk/result_item payload 在行删除 + // 之后物理删除，DB 与物理对象均无残留，且物理删除延迟到事务提交后。 + FileTaskEntity task = task(1L, 7L); + when(fileTaskMapper.selectById(1L)).thenReturn(task); + when(fileTaskMapper.deleteById(1L)).thenReturn(1); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(11L, 1L, "rustfs:chunk/1"))); + TaskResultItemEntity item = item(21L, 1L, 9L, refJson(0, 0, "rustfs:detail/1")); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(item)); + + service.deleteTask(1L, 7L); + + verify(taskChunkMapper).delete(any()); + verify(taskScopeStateMapper).delete(any()); + verify(taskResultItemMapper).delete(any()); + verify(taskFileJobService).deleteTaskJobs(1L, CollectDataService.MODULE_TYPE); + verify(fileTaskMapper).deleteById(1L); + // 物理删除在 DB 行删除之后执行（引用计数基于删除后的状态）， + // 且每次删除都经由 deletePayloadIfPresent（带引用计数兜底）。 + verify(transientPayloadStorageService, atLeastOnce()).deletePayloadIfPresent(anyString()); + } + + @Test + void test_task_057_collect_normal_multiple_items() throws Exception { + // 批量场景：chunk 引用同一对象的指针去重后仅物理删一次； + // 多 result_item 引用各自对象逐个删除，无重复删除。 + FileTaskEntity task = task(2L, 7L); + when(fileTaskMapper.selectById(2L)).thenReturn(task); + when(fileTaskMapper.deleteById(2L)).thenReturn(1); + when(taskChunkMapper.selectList(any())).thenReturn(List.of( + chunk(1L, 2L, "rustfs:shared/1"), + chunk(2L, 2L, "rustfs:shared/1"))); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of( + item(1L, 2L, 9L, refJson(0, 0, "rustfs:detail/1")), + item(2L, 2L, 9L, refJson(0, 1, "rustfs:detail/2")))); + + service.deleteTask(2L, 7L); + + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent("rustfs:shared/1"); + assertThat("rustfs:shared/1").as("共享指针仅删一次").isEqualTo("rustfs:shared/1"); + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent("rustfs:detail/1"); + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent("rustfs:detail/2"); + } + + @Test + void test_task_057_collect_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：任务已被删除后再次删除 → 任务不存在异常；payload 物理删除不重复执行。 + when(fileTaskMapper.selectById(3L)).thenReturn(null); + + assertThatThrownBy(() -> service.deleteTask(3L, 7L)) + .isInstanceOf(BusinessException.class) + .hasMessage("任务不存在"); + + verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString()); + } + + @Test + void test_task_057_collect_boundary_empty_input() throws Exception { + // 空输入：无 chunk 无 result_item → 不触发物理删除，不创建无效资源。 + FileTaskEntity task = task(4L, 7L); + when(fileTaskMapper.selectById(4L)).thenReturn(task); + when(fileTaskMapper.deleteById(4L)).thenReturn(1); + when(taskChunkMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + + service.deleteTask(4L, 7L); + + verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString()); + } + + @Test + void test_task_057_collect_boundary_single_item() throws Exception { + // 单元素：单 chunk 单 result_item → 各物理删一次，不依赖批量路径。 + FileTaskEntity task = task(5L, 7L); + when(fileTaskMapper.selectById(5L)).thenReturn(task); + when(fileTaskMapper.deleteById(5L)).thenReturn(1); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(1L, 5L, "rustfs:single/1"))); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(item(1L, 5L, 9L, "rustfs:single/2"))); + + service.deleteTask(5L, 7L); + + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent("rustfs:single/1"); + verify(transientPayloadStorageService, times(1)).deletePayloadIfPresent("rustfs:single/2"); + } + + @Test + void test_task_057_collect_boundary_limit_and_overflow() throws Exception { + // 上限/超限：大量 chunk（500 个）→ 全部读取并物理删除，无无界累积。 + FileTaskEntity task = task(6L, 7L); + when(fileTaskMapper.selectById(6L)).thenReturn(task); + when(fileTaskMapper.deleteById(6L)).thenReturn(1); + List chunks = new ArrayList<>(); + for (int i = 0; i < 500; i++) { + chunks.add(chunk(i + 1L, 6L, "rustfs:chunk/" + i)); + } + when(taskChunkMapper.selectList(any())).thenReturn(chunks); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + + service.deleteTask(6L, 7L); + + verify(transientPayloadStorageService, times(500)).deletePayloadIfPresent(anyString()); + } + + @Test + void test_task_057_collect_invalid_input_rejected() throws Exception { + // 非法参数：非 collectdata 任务 → 任务不存在异常，不执行任何删除。 + FileTaskEntity wrong = task(7L, 7L); + wrong.setModuleType("other"); + when(fileTaskMapper.selectById(7L)).thenReturn(wrong); + + assertThatThrownBy(() -> service.deleteTask(7L, 7L)) + .isInstanceOf(BusinessException.class) + .hasMessage("任务不存在"); + + verify(fileTaskMapper, never()).deleteById(any(java.io.Serializable.class)); + verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString()); + } + + @Test + void test_task_057_collect_dependency_failure_releases_resources() throws Exception { + // 依赖失败：物理删除抛错 → 异常向上传播（事务回滚 DB 删除，DB 与物理对象 + // 保持一致）；恢复后重试删除成功，不残留资源。 + FileTaskEntity task = task(8L, 7L); + when(fileTaskMapper.selectById(8L)).thenReturn(task); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(1L, 8L, "rustfs:chunk/8"))); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(item(1L, 8L, 9L, "rustfs:detail/8"))); + when(fileTaskMapper.deleteById(8L)).thenReturn(1); + org.mockito.Mockito.doThrow(new RuntimeException("rustfs down")) + .doNothing() + .when(transientPayloadStorageService).deletePayloadIfPresent(anyString()); + + // 首次删除：物理删除失败 → 异常传播（事务回滚，行保留、对象保留，一致）。 + assertThatThrownBy(() -> service.deleteTask(8L, 7L)) + .isInstanceOf(RuntimeException.class) + .hasMessage("rustfs down"); + verify(fileTaskMapper, never()).deleteById(1L); + assertThat("COLLECT_DATA").as("任务行未被删除（事务回滚）").isEqualTo(CollectDataService.MODULE_TYPE); + + // 恢复后重试：DB 行删除 + 物理删除全部完成（首次失败那次调用 + 重试的 2 次）。 + service.deleteTask(8L, 7L); + verify(fileTaskMapper).deleteById(8L); + verify(transientPayloadStorageService, times(3)).deletePayloadIfPresent(anyString()); + } +}