task-57: 删除任务/历史时先删 DB 行再物理删对象，引用计数反查基于删除后状态，共享对象跳过删除、独占对象真正释放
This commit is contained in:
+19
-15
@@ -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<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||
.select(TaskChunkEntity::getPayloadJson)
|
||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE));
|
||||
private void deleteTransientTaskPayloads(List<TaskChunkEntity> chunks, List<TaskResultItemEntity> items) {
|
||||
if (chunks != null) {
|
||||
Set<String> 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<TaskResultItemEntity> items = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
|
||||
.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<FileResultEntity>()
|
||||
.eq(FileResultEntity::getTaskId, task.getId())
|
||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
|
||||
deleteTransientTaskPayloads(task.getId());
|
||||
// 先删 DB 行、再物理删对象：引用计数反查基于行删除后的状态，
|
||||
// 仍被其它任务 chunk/scope_state 引用的共享对象会跳过删除，本任务独占对象真正释放。
|
||||
taskScopeStateMapper.delete(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||
.eq(TaskScopeStateEntity::getTaskId, task.getId())
|
||||
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE));
|
||||
@@ -1258,6 +1252,15 @@ public class CollectDataService {
|
||||
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
|
||||
.eq(TaskResultItemEntity::getTaskId, task.getId())
|
||||
.eq(TaskResultItemEntity::getModuleType, MODULE_TYPE));
|
||||
deleteTransientTaskPayloads(
|
||||
taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||
.select(TaskChunkEntity::getPayloadJson)
|
||||
.eq(TaskChunkEntity::getTaskId, task.getId())
|
||||
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)),
|
||||
taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
|
||||
.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<TaskResultItemEntity>()
|
||||
.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);
|
||||
}
|
||||
|
||||
+302
@@ -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<TaskChunkEntity> 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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user