diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectData10kLoadTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectData10kLoadTest.java new file mode 100644 index 00000000..04faf53b --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectData10kLoadTest.java @@ -0,0 +1,236 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +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.anyList; +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 60：采集模块 10k 行压测验收（结果完整性、内存与吞吐）。 + * CollectDataResultItemBatchWriter 以 10k 行输入压测批量 upsert：批量 SQL + * 调用次数 = ceil(10k / batchSize)（不随行数增长为逐行调用），10k 行全部 + * 写入不丢失、无重复；hash 相等重复提交全跳过（幂等零写入）；批量失败 + * 跳过该批后重试收敛到真实行数；内存上每批行数 ≤ batchSize，不无界累积。 + */ +class CollectData10kLoadTest { + + private TaskResultItemMapper taskResultItemMapper; + private TransientPayloadStorageService transientPayloadStorageService; + private CollectDataResultDetailCodec codec; + private CollectDataResultItemBatchWriter writer; + + @BeforeEach + void setUp() { + taskResultItemMapper = mock(TaskResultItemMapper.class); + transientPayloadStorageService = mock(TransientPayloadStorageService.class); + codec = new CollectDataResultDetailCodec(new ObjectMapper()); + writer = new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 100); + } + + private static List rows(int count, int startIndex) { + List rows = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + CollectDataResultRowVo row = new CollectDataResultRowVo(); + row.setAsin("B" + String.format("%09d", startIndex + i + 1)); + row.setBrand("brand" + (i % 10)); + rows.add(row); + } + return rows; + } + + @Test + void test_task_060_collect_normal_default_path() { + // 正常路径：10k 行全部写入，恰好 100 批（每批 100 行），无丢失无重复。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + List rows = rows(10_000, 0); + + CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:detail/10k"); + + assertEquals(10_000, counts.insertedOrUpdated(), "10k 行全部写入"); + assertEquals(0, counts.skipped(), "无跳过"); + assertEquals(10_000, counts.newlyInserted(), "全部为新增"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper, times(100)).upsertBatch(captor.capture()); + assertEquals(100, captor.getAllValues().get(0).size(), "每批恰好 batchSize=100"); + Set refs = new HashSet<>(); + for (Object value : captor.getAllValues()) { + for (Object entity : (List) value) { + assertTrue(refs.add(((TaskResultItemEntity) entity).getPayloadJson()), + "引用 JSON 不重复（每行唯一 offset）"); + } + } + assertEquals(10_000, refs.size(), "10k 个引用唯一"); + } + + @Test + void test_task_060_collect_normal_multiple_items() { + // 批量场景：10k 行按 10 个 chunk 分批提交（每 chunk 1000 行）， + // 全部收敛不丢失，总批量 SQL 调用次数 = 100，顺序稳定。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + int total = 0; + int newly = 0; + for (int chunk = 0; chunk < 10; chunk++) { + CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted( + 1L, 2L, "task:1", chunk, rows(1000, chunk * 1000), + "rustfs:detail/c" + chunk); + total += counts.insertedOrUpdated(); + newly += counts.newlyInserted(); + } + + assertEquals(10_000, total, "10k 行全部写入"); + assertEquals(10_000, newly, "全部新增"); + verify(taskResultItemMapper, times(100)).upsertBatch(anyList()); + } + + @Test + void test_task_060_collect_normal_repeated_operation_is_idempotent() { + // 幂等：10k 行重复提交（存量行 hash 相等）→ 全部跳过零写入， + // 批量 SQL 仅 1 次查询 0 次写入，无重复记录。 + List existing = new ArrayList<>(10_000); + List rows = rows(10_000, 0); + for (int i = 0; i < 10_000; i++) { + TaskResultItemEntity entity = new TaskResultItemEntity(); + entity.setItemKey("asin:" + rows.get(i).getAsin()); + entity.setPayloadHash(codec.encodeRefsWithHash(0, i, 1, "rustfs:detail/10k").get(0).payloadHash()); + existing.add(entity); + } + when(taskResultItemMapper.selectList(any())).thenReturn(existing); + + CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:detail/10k"); + + assertEquals(0, counts.insertedOrUpdated(), "重复提交零写入"); + assertEquals(10_000, counts.skipped(), "全部跳过"); + assertEquals(0, counts.newlyInserted(), "无新增"); + verify(taskResultItemMapper, never()).upsertBatch(anyList()); + verify(taskResultItemMapper, times(1)).selectList(any()); + } + + @Test + void test_task_060_collect_boundary_empty_input() { + // 空输入：零调用零计数，不创建无效资源。 + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, List.of(), "rustfs:x"); + + assertEquals(0, counts.insertedOrUpdated()); + assertEquals(0, counts.skipped()); + assertEquals(0, counts.newlyInserted()); + verify(taskResultItemMapper, never()).selectList(any()); + verify(taskResultItemMapper, never()).upsertBatch(anyList()); + } + + @Test + void test_task_060_collect_boundary_single_item() { + // 单元素：10k 行中仅有 1 行 → 恰好 1 次批量调用、1 行写入。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, rows(1, 0), "rustfs:x"); + + assertEquals(1, counts.insertedOrUpdated()); + verify(taskResultItemMapper, times(1)).upsertBatch(anyList()); + } + + @Test + void test_task_060_collect_boundary_limit_and_overflow() { + // 上限/超限：10k 行以 batchSize=1 提交 → 10k 次批量调用， + // 每次恰好 1 行，无无界内存增长（每批行数受 batchSize 约束）。 + CollectDataResultItemBatchWriter tinyWriter = new CollectDataResultItemBatchWriter( + taskResultItemMapper, codec, transientPayloadStorageService, 1); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + + CollectDataResultItemBatchWriter.UpsertCounts counts = tinyWriter.upsertAccepted( + 1L, 2L, "task:1", 0, rows(10_000, 0), "rustfs:detail/10k"); + + assertEquals(10_000, counts.insertedOrUpdated(), "10k 行全部写入"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper, times(10_000)).upsertBatch(captor.capture()); + assertEquals(1, captor.getAllValues().get(0).size(), "每批恰好 1 行"); + assertEquals(1, captor.getAllValues().get(9_999).size(), "末批 1 行"); + } + + @Test + void test_task_060_collect_invalid_input_rejected() { + // 非法参数：null/空白 ASIN 行在 10k 输入中安全跳过，仅合法行写入， + // 引用 JSON 仅对合法行生成，无无效资源。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + List rows = rows(10_000, 0); + for (int i = 0; i < 500; i++) { + rows.get(i * 20).setAsin(null); + } + + CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:detail/10k"); + + assertEquals(9_500, counts.insertedOrUpdated(), "仅合法行写入"); + assertEquals(9_500, counts.newlyInserted(), "新增计数同步"); + } + + @Test + void test_task_060_collect_dependency_failure_releases_resources() { + // 依赖失败：10k 行前 50 批写入失败（共 100 批）→ 跳过失败批， + // 增量计数扣除未落库行，重试同一输入后收敛到真实行数，无资源残留。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + java.util.concurrent.atomic.AtomicInteger failedBatches = new java.util.concurrent.atomic.AtomicInteger(); + doAnswer(invocation -> { + if (failedBatches.getAndIncrement() < 50) { + throw new RuntimeException("db down"); + } + return ((List) invocation.getArgument(0)).size(); + }).when(taskResultItemMapper).upsertBatch(anyList()); + List rows = rows(10_000, 0); + + CollectDataResultItemBatchWriter.UpsertCounts counts = writer.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:detail/10k"); + + assertEquals(5_000, counts.insertedOrUpdated(), "首批失败跳过，后 50 批写入"); + assertEquals(5_000, counts.newlyInserted(), "失败批新行从增量扣除"); + + // 恢复后重试同一输入：存量已落库行 hash 相等跳过，未落库行补插。 + List existing = new ArrayList<>(5_000); + for (int i = 0; i < 5_000; i++) { + TaskResultItemEntity entity = new TaskResultItemEntity(); + entity.setItemKey("asin:" + rows.get(5_000 + i).getAsin()); + entity.setPayloadHash(codec.encodeRefsWithHash(0, 5_000 + i, 1, "rustfs:detail/10k") + .get(0).payloadHash()); + existing.add(entity); + } + when(taskResultItemMapper.selectList(any())).thenReturn(existing); + CollectDataResultItemBatchWriter.UpsertCounts retry = writer.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:detail/10k"); + + assertEquals(5_000, retry.insertedOrUpdated(), "重试补插未落库 5k 行"); + assertEquals(5_000, retry.skipped(), "已落库行全部跳过"); + assertEquals(5_000, retry.newlyInserted(), "新增恰为未落库行数"); + } +}