diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java index f190f656..92fcf07c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotService.java @@ -7,19 +7,28 @@ import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper; import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity; import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Value; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; @Service @RequiredArgsConstructor public class TaskProgressSnapshotService { + private static final Set TERMINAL_STATUSES = Set.of("SUCCESS", "FAILED", "CANCELLED", "CANCELED"); + private final TaskProgressSnapshotMapper taskProgressSnapshotMapper; private final ObjectMapper objectMapper; + private final ConcurrentHashMap lastWriteAtMillis = new ConcurrentHashMap<>(); + + @Value("${aiimage.task-progress-snapshot.min-update-interval-ms:500}") + private long minUpdateIntervalMillis = 500; @Transactional public void save(Long taskId, @@ -54,6 +63,7 @@ public class TaskProgressSnapshotService { entity.setUpdatedAt(now); try { taskProgressSnapshotMapper.insert(entity); + touchLastWrite(taskId, moduleType, status); return; } catch (DuplicateKeyException ignored) { existing = find(taskId, moduleType); @@ -66,6 +76,9 @@ public class TaskProgressSnapshotService { currentScopeKey, message, snapshotJson)) { return; } + if (shouldThrottle(taskId, moduleType, status)) { + return; + } taskProgressSnapshotMapper.update(null, new LambdaUpdateWrapper() .eq(TaskProgressSnapshotEntity::getId, existing.getId()) .set(TaskProgressSnapshotEntity::getStatus, status) @@ -77,6 +90,31 @@ public class TaskProgressSnapshotService { .set(TaskProgressSnapshotEntity::getMessage, message) .set(TaskProgressSnapshotEntity::getSnapshotJson, snapshotJson) .set(TaskProgressSnapshotEntity::getUpdatedAt, now)); + touchLastWrite(taskId, moduleType, status); + } + + private boolean shouldThrottle(Long taskId, String moduleType, String status) { + if (TERMINAL_STATUSES.contains(status)) { + return false; + } + long interval = Math.max(0L, minUpdateIntervalMillis); + if (interval <= 0L) { + return false; + } + Long lastWrite = lastWriteAtMillis.get(cacheKey(taskId, moduleType)); + return lastWrite != null && System.currentTimeMillis() - lastWrite < interval; + } + + private void touchLastWrite(Long taskId, String moduleType, String status) { + if (TERMINAL_STATUSES.contains(status)) { + lastWriteAtMillis.remove(cacheKey(taskId, moduleType)); + return; + } + lastWriteAtMillis.put(cacheKey(taskId, moduleType), System.currentTimeMillis()); + } + + private static String cacheKey(Long taskId, String moduleType) { + return taskId + ":" + moduleType; } private boolean isUnchanged(TaskProgressSnapshotEntity existing, @@ -116,6 +154,7 @@ public class TaskProgressSnapshotService { taskProgressSnapshotMapper.delete(new LambdaQueryWrapper() .eq(TaskProgressSnapshotEntity::getTaskId, taskId) .eq(TaskProgressSnapshotEntity::getModuleType, moduleType)); + lastWriteAtMillis.remove(cacheKey(taskId, moduleType)); } private String writeJson(Object value) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotServiceThrottleTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotServiceThrottleTest.java new file mode 100644 index 00000000..daa952c7 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskProgressSnapshotServiceThrottleTest.java @@ -0,0 +1,200 @@ +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.conditions.update.LambdaUpdateWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.task.mapper.TaskProgressSnapshotMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity; +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.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doThrow; +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 63：为前端/后端进度快照增加写入去重和最小更新间隔。 + * TaskProgressSnapshotService 对进度快照写入做节流：内容未变化时跳过写库 + * （已有去重语义保留）；内容变化但距上次写入不足最小更新间隔（默认 500ms） + * 时跳过中间态写入，终态（SUCCESS/FAILED/CANCELLED）不受节流保证落库； + * 删除快照时同步清理节流跟踪，不残留内存状态。 + */ +@ExtendWith(MockitoExtension.class) +class TaskProgressSnapshotServiceThrottleTest { + + @BeforeAll + static void initializeMybatisMetadata() { + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new MybatisConfiguration(), ""), + TaskProgressSnapshotEntity.class); + } + + @Mock private TaskProgressSnapshotMapper taskProgressSnapshotMapper; + @Mock private ObjectMapper objectMapper; + + private TaskProgressSnapshotService service; + + @BeforeEach + void setUp() { + service = new TaskProgressSnapshotService(taskProgressSnapshotMapper, objectMapper); + ReflectionTestUtils.setField(service, "minUpdateIntervalMillis", 1000L); + } + + private TaskProgressSnapshotEntity running() { + TaskProgressSnapshotEntity entity = new TaskProgressSnapshotEntity(); + entity.setId(1L); + entity.setTaskId(20581L); + entity.setModuleType("SIMILAR_ASIN"); + entity.setStatus("RUNNING"); + entity.setTotalCount(10); + entity.setSuccessCount(4); + entity.setFailedCount(1); + entity.setPendingCount(5); + entity.setCurrentScopeKey("scope-1"); + entity.setMessage("processing"); + return entity; + } + + /** 把某任务的节流跟踪时间拨到距今 millisAgo 毫秒前，模拟时间流逝。 */ + @SuppressWarnings("unchecked") + private void rewindLastWrite(Long taskId, String moduleType, long millisAgo) { + Map tracking = (Map) ReflectionTestUtils + .getField(service, "lastWriteAtMillis"); + tracking.put(taskId + ":" + moduleType, System.currentTimeMillis() - millisAgo); + } + + @Test + void test_task_063_progress_frontend_normal_default_path() { + // 正常路径：距上次写入超过最小间隔的变更落库，写入一次。 + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running()); + rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L); + + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, + "scope-1", "processing", null); + + verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class)); + } + + @Test + void test_task_063_progress_frontend_normal_multiple_items() { + // 批量场景：多个任务各自跟踪节流时间，互不影响，全部落库。 + for (long taskId = 1; taskId <= 3; taskId++) { + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))) + .thenReturn(null); + service.save(taskId, "MODULE_" + taskId, "RUNNING", 10, 0, 0, + "scope-" + taskId, "start", null); + } + verify(taskProgressSnapshotMapper, times(3)).insert(any(TaskProgressSnapshotEntity.class)); + } + + @Test + void test_task_063_progress_frontend_normal_repeated_operation_is_idempotent() { + // 幂等：相同内容重复保存不产生写库（内容去重），结果一致无副作用。 + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running()); + + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 4, 1, + "scope-1", "processing", null); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 4, 1, + "scope-1", "processing", null); + + verify(taskProgressSnapshotMapper, never()).update(any(), any()); + verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class)); + } + + @Test + void test_task_063_progress_frontend_boundary_empty_input() { + // 空输入：null/空模块、空状态安全跳过，不发起任何写库调用。 + service.save(null, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null); + service.save(20581L, "", "RUNNING", 10, 0, 0, null, null, null); + service.save(20581L, "SIMILAR_ASIN", " ", 10, 0, 0, null, null, null); + + verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class)); + verify(taskProgressSnapshotMapper, never()).update(any(), any()); + verify(taskProgressSnapshotMapper, never()).selectOne(any()); + } + + @Test + void test_task_063_progress_frontend_boundary_single_item() { + // 单元素：首次写入落库；间隔内变更被节流跳过；间隔过后变更落库。 + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(null); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, "s1", "go", null); + verify(taskProgressSnapshotMapper, times(1)).insert(any(TaskProgressSnapshotEntity.class)); + + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running()); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s2", "half", null); + verify(taskProgressSnapshotMapper, never()).update(isNull(), any(LambdaUpdateWrapper.class)); + + rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 6, 1, "s3", "more", null); + verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class)); + } + + @Test + void test_task_063_progress_frontend_boundary_limit_and_overflow() { + // 上限/超限：最小更新间隔设为 0（超限值）→ 节流禁用， + // 每次变更都落库，不发生无界内存增长（跟踪 map 仅随任务数增长）。 + ReflectionTestUtils.setField(service, "minUpdateIntervalMillis", 0L); + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running()); + + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s1", "m1", null); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 6, 1, "s1", "m2", null); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 7, 1, "s1", "m3", null); + + verify(taskProgressSnapshotMapper, times(3)).update(isNull(), any(LambdaUpdateWrapper.class)); + } + + @Test + void test_task_063_progress_frontend_invalid_input_rejected() { + // 非法参数：非正任务 id / 空白状态拒绝写入，不创建无效快照。 + service.save(0L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null); + service.save(-5L, "SIMILAR_ASIN", "RUNNING", 10, 0, 0, null, null, null); + + verify(taskProgressSnapshotMapper, never()).insert(any(TaskProgressSnapshotEntity.class)); + verify(taskProgressSnapshotMapper, never()).update(any(), any()); + } + + @Test + void test_task_063_progress_frontend_dependency_failure_releases_resources() { + // 依赖失败：唯一键冲突回落查询后正常更新（并发安全）； + // 写库异常传播不残留；删除快照后节流跟踪释放，间隔内再次写入不再被节流。 + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))) + .thenReturn(null) + .thenReturn(running()); + doThrow(new DuplicateKeyException("dup")).when(taskProgressSnapshotMapper) + .insert(any(TaskProgressSnapshotEntity.class)); + + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 5, 1, "s1", "m1", null); + verify(taskProgressSnapshotMapper, times(1)).update(isNull(), any(LambdaUpdateWrapper.class)); + + rewindLastWrite(20581L, "SIMILAR_ASIN", 5_000L); + doThrow(new RuntimeException("db down")).when(taskProgressSnapshotMapper) + .update(isNull(), any(LambdaUpdateWrapper.class)); + assertThrows(RuntimeException.class, () -> service.save(20581L, "SIMILAR_ASIN", + "RUNNING", 10, 6, 1, "s2", "m2", null), "写库异常传播"); + + service.delete(20581L, "SIMILAR_ASIN"); + verify(taskProgressSnapshotMapper, times(1)) + .delete(any(LambdaQueryWrapper.class)); + + when(taskProgressSnapshotMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(running()); + when(taskProgressSnapshotMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); + service.save(20581L, "SIMILAR_ASIN", "RUNNING", 10, 7, 1, "s3", "m3", null); + verify(taskProgressSnapshotMapper, times(3)).update(isNull(), any(LambdaUpdateWrapper.class)); + } +}