From e201b13301981cbd88073837439d27ea12a51463 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 17:05:49 +0800 Subject: [PATCH] task-13: cap chunk merge row count and payload bytes with orphan fallback --- .../aiimage/config/SimilarAsinProperties.java | 14 + .../service/SimilarAsinTaskService.java | 108 ++++- ...larAsinTaskServiceChunkMergeLimitTest.java | 402 ++++++++++++++++++ 3 files changed, 523 insertions(+), 1 deletion(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceChunkMergeLimitTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index fb636031..e3df39ff 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -165,6 +165,20 @@ public class SimilarAsinProperties { */ private long maxWorkbookUncompressedBytes = 512L * 1024L * 1024L; + /** + * Task 13：单次 chunk 合并后的最大行数。mergeChunkPayload 合并后总行数超过该上限时， + * 从最旧行开始降级到 orphan 兜底（assemble 阶段 putIfAbsent 合并回结果），chunk 不无界增长。 + * 默认与 maxParseRows 一致（50000）。 + */ + private int chunkMergeMaxRows = 50000; + + /** + * Task 13：单次 chunk 合并后 payload 的字节上限。合并后序列化字节超过该上限时， + * 从最旧行开始降级到 orphan 兜底；单行本身超过该上限时抛异常拒绝合并。 + * 默认 16MB：50000 行 × 平均 300B/行 ≈ 15MB，留余量。 + */ + private long chunkMergePayloadMaxBytes = 16L * 1024L * 1024L; + /** * P0-4：抢 Coze 提交锁失败后下次重试间隔（毫秒）。 * 原硬编码 500ms，会在指数退避算法中作为基础值（500/1000/2000/4000ms 上限 4000）。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 269871c1..625d3986 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -89,7 +89,9 @@ import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -1643,14 +1645,42 @@ public class SimilarAsinTaskService { return; } Map persistedRows = readChunkRows(chunk); + int mergedRowCount = persistedRows.size() + rows.size(); + // Task 13：单次合并后总行数超过上限时，从最旧行（存量优先）开始降级到 orphan 兜底， + // chunk 保持在上限内不无界增长；assemble 阶段 putIfAbsent 合并回结果不丢数据。 + if (mergedRowCount > getChunkMergeMaxRows()) { + splitChunkMergeOverflow(taskId, persistedRows, rows, chunk, mergedRowCount - getChunkMergeMaxRows()); + if (rows.isEmpty()) { + return; + } + } for (SimilarAsinResultRowDto row : rows) { persistedRows.put(rowKey(row), row); } String payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败"); + // Task 13：合并后 payload 字节超过上限时，从最旧行开始降级到 orphan 兜底； + // 降到只剩一行仍超上限时抛异常拒绝合并，防止无界 payload。 + long payloadBytes = payloadJson.getBytes(StandardCharsets.UTF_8).length; + if (payloadBytes > getChunkMergePayloadMaxBytes()) { + demoteRowsToOrphan(taskId, persistedRows, payloadBytes - getChunkMergePayloadMaxBytes()); + payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败"); + payloadBytes = payloadJson.getBytes(StandardCharsets.UTF_8).length; + if (payloadBytes > getChunkMergePayloadMaxBytes() && persistedRows.size() <= 1) { + throw new BusinessException("相似ASIN分片载荷超字节上限 taskId=" + taskId + + " scopeHash=" + scopeHash + " chunk=" + chunkIndex + + " bytes=" + payloadBytes + " limit=" + getChunkMergePayloadMaxBytes()); + } + } String oldPayload = chunk.getPayloadJson(); String oldPayloadHash = chunk.getPayloadHash(); String newPayloadHash = DigestUtil.sha256Hex(payloadJson); - String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + final String storedPayload; + try { + storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + } catch (Exception storeEx) { + throw new BusinessException("相似ASIN分片载荷存储失败 taskId=" + taskId + " chunk=" + chunkIndex + + ": " + (storeEx.getMessage() == null ? "" : storeEx.getMessage()), storeEx); + } int updated = taskChunkMapper.update(null, new LambdaUpdateWrapper() .eq(TaskChunkEntity::getId, chunk.getId()) .eq(TaskChunkEntity::getPayloadHash, oldPayloadHash) @@ -1672,6 +1702,82 @@ public class SimilarAsinTaskService { throw new IllegalStateException("相似ASIN分片载荷更新失败"); } + private int getChunkMergeMaxRows() { + int limit = properties.getChunkMergeMaxRows(); + return limit > 0 ? limit : 50000; + } + + private long getChunkMergePayloadMaxBytes() { + long limit = properties.getChunkMergePayloadMaxBytes(); + return limit > 0 ? limit : 16L * 1024L * 1024L; + } + + /** + * Task 13：按行数上限从最旧行开始降级到 orphan 兜底，保证合并后 chunk 行数不超过上限。 + * 最旧优先：先降级存量行（LinkedHashMap 表头），不够再从新增行表头补齐； + * 传入的 rows 会被就地修改（保留未降级部分）。降级失败仅记日志，不阻断合并主流程。 + */ + private void splitChunkMergeOverflow(Long taskId, Map persistedRows, + List rows, TaskChunkEntity chunk, + int demoteCount) { + if (demoteCount <= 0 || rows.isEmpty()) { + return; + } + List demoted = new ArrayList<>(); + while (demoted.size() < demoteCount && !persistedRows.isEmpty()) { + String oldestKey = persistedRows.keySet().iterator().next(); + SimilarAsinResultRowDto removed = persistedRows.remove(oldestKey); + if (removed != null) { + demoted.add(removed); + } + } + for (Iterator it = rows.iterator(); it.hasNext() && demoted.size() < demoteCount; ) { + SimilarAsinResultRowDto row = it.next(); + if (row != null) { + demoted.add(row); + it.remove(); + } + } + persistOrphanCozeRows(taskId, demoted); + log.warn("[similar-asin] chunk merge row-limit exceeded taskId={} chunk={} mergedRows={} limit={} demoted={}", + taskId, chunk.getChunkIndex(), mergedRowCountOf(persistedRows, rows), getChunkMergeMaxRows(), demoted.size()); + } + + /** + * Task 13：按字节上限从最旧行（LinkedHashMap 表头）开始降级到 orphan 兜底， + * 直到 payload 字节不超过上限或只剩一行；降级失败仅记日志，不阻断合并主流程。 + */ + private void demoteRowsToOrphan(Long taskId, Map persistedRows, long excessBytes) { + List demoted = new ArrayList<>(); + long releasedBytes = 0L; + while (persistedRows.size() > 1 && releasedBytes < excessBytes) { + String oldestKey = persistedRows.keySet().iterator().next(); + SimilarAsinResultRowDto removed = persistedRows.remove(oldestKey); + if (removed != null) { + demoted.add(removed); + releasedBytes += estimateRowBytes(oldestKey, removed); + } + } + if (!demoted.isEmpty()) { + persistOrphanCozeRows(taskId, demoted); + log.warn("[similar-asin] chunk merge byte-limit exceeded taskId={} rows={} demoted={} releasedBytes={}", + taskId, demoted.size(), demoted.size(), releasedBytes); + } + } + + private static int mergedRowCountOf(Map persistedRows, List rows) { + return persistedRows.size() + rows.size(); + } + + private long estimateRowBytes(String rowKey, SimilarAsinResultRowDto row) { + try { + String json = objectMapper.writeValueAsString(row); + return json == null ? 128L : json.getBytes(StandardCharsets.UTF_8).length; + } catch (Exception ex) { + return 128L + (rowKey == null ? 0 : rowKey.getBytes(StandardCharsets.UTF_8).length); + } + } + private List expandRows(List rows, Map> allRowsByBaseId) { if (rows == null || rows.isEmpty()) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceChunkMergeLimitTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceChunkMergeLimitTest.java new file mode 100644 index 00000000..b19ceed3 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceChunkMergeLimitTest.java @@ -0,0 +1,402 @@ +package com.nanri.aiimage.modules.similarasin.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.SimilarAsinProperties; +import com.nanri.aiimage.modules.similarasin.client.SimilarAsinCozeClient; +import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper; +import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto; +import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +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.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.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.AfterEach; +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.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 13：chunk 合并增加单次最大行数与 payload 字节上限。 + * 校验点位于 mergeChunkPayload 单次合并入口：合并后总行数超过 chunkMergeMaxRows、 + * 或 payload 字节超过 chunkMergePayloadMaxBytes 时，从最旧行开始降级到 + * orphan 兜底（assemble 阶段 putIfAbsent 合并回结果，不丢数据）； + * 单行本身超过字节上限时抛可识别异常拒绝合并。低于上限的行为与旧路径完全一致。 + */ +@ExtendWith(MockitoExtension.class) +class SimilarAsinTaskServiceChunkMergeLimitTest { + + private static final AtomicLong NEXT_ID = new AtomicLong(90000); + + @Mock private LocalFileStorageService localFileStorageService; + @Mock private com.nanri.aiimage.modules.file.service.oss.OssStorageService ossStorageService; + @Mock private com.nanri.aiimage.config.StorageProperties storageProperties; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private SimilarAsinFilterConditionMapper filterConditionMapper; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @Mock private SimilarAsinCozeClient cozeClient; + @Mock private SimilarAsinTaskCacheService taskCacheService; + @Mock private SimilarAsinProperties properties; + @Mock private com.nanri.aiimage.modules.task.service.TaskFileJobService taskFileJobService; + @Mock private com.nanri.aiimage.modules.task.service.TaskDistributedLockService taskDistributedLockService; + @Mock private com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private org.springframework.transaction.PlatformTransactionManager transactionManager; + @Mock private com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService; + @Mock private com.nanri.aiimage.config.InstanceMetadata instanceMetadata; + @Mock private com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService cozeCredentialPoolService; + @Mock private SimilarAsinImageEmbedder imageEmbedder; + @Mock private SimilarAsinImagePrefetchService imagePrefetchService; + + @InjectMocks private SimilarAsinTaskService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(properties.isBoundedResultAssemblyEnabled()).thenReturn(true); + lenient().when(properties.getImageCacheMaxBytes()).thenReturn(256L * 1024L * 1024L); + lenient().when(properties.getResultFileTimeoutMinutes()).thenReturn(90); + lenient().when(properties.getParseResponsePreviewLimit()).thenReturn(100); + lenient().when(properties.getMaxSourceFileBytes()).thenReturn(50L * 1024L * 1024L); + lenient().when(properties.getMaxParseRows()).thenReturn(50000); + lenient().when(properties.getMaxFieldLength()).thenReturn(2000); + lenient().when(properties.getMaxWorkbookZipEntries()).thenReturn(20000); + lenient().when(properties.getMaxWorkbookUncompressedBytes()).thenReturn(512L * 1024L * 1024L); + lenient().when(properties.getChunkMergeMaxRows()).thenReturn(50000); + lenient().when(properties.getChunkMergePayloadMaxBytes()).thenReturn(16L * 1024L * 1024L); + lenient().when(transientPayloadStorageService.storeParsedPayloadFast( + eq(SimilarAsinTaskService.MODULE_TYPE), any(), anyString(), anyString(), eq(false))) + .thenReturn("rustfs:task-parsed/similar-asin/90000/payload.json"); + lenient().doAnswer(invocation -> { + FileTaskEntity task = invocation.getArgument(0); + task.setId(NEXT_ID.incrementAndGet()); + return 1; + }).when(fileTaskMapper).insert(any(FileTaskEntity.class)); + lenient().when(fileResultMapper.insert(any(com.nanri.aiimage.modules.task.model.entity.FileResultEntity.class))).thenReturn(1); + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenReturn(1); + lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of()); + } + + @AfterEach + void shutdown() { + service.shutdownAssembleExecutor(); + } + + private static SimilarAsinResultRowDto row(String rowToken, String asin, String title) { + SimilarAsinResultRowDto r = new SimilarAsinResultRowDto(); + r.setRowToken(rowToken); + r.setId(rowToken); + r.setAsin(asin); + r.setCountry("英国"); + r.setTitle(title); + return r; + } + + private static String rowsJson(List rows) throws Exception { + return new ObjectMapper().writeValueAsString(rows); + } + + private static TaskChunkEntity chunk(Long id, String scopeHash, Integer chunkIndex, String payloadJson) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(id); + chunk.setTaskId(9004L); + chunk.setModuleType(SimilarAsinTaskService.MODULE_TYPE); + chunk.setScopeHash(scopeHash); + chunk.setChunkIndex(chunkIndex); + chunk.setPayloadJson(payloadJson); + return chunk; + } + + /** 记录每次 storeChunkPayloadVersioned 收到的 payload 字符串。 */ + private void stubChunkMerge(TaskChunkEntity chunk, String payloadJson, AtomicReference storedPayload) throws Exception { + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq(chunk.getPayloadJson()), anyString())) + .thenReturn(payloadJson); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> { + storedPayload.set(invocation.getArgument(4)); + return "stored:" + invocation.getArgument(2); + }); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + } + + private static void invokeMerge(SimilarAsinTaskService service, FileTaskEntity task, + String scopeHash, Integer chunkIndex, + List cozeRows) throws Exception { + Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeCozeRowsIntoChunk", + FileTaskEntity.class, String.class, Integer.class, List.class, Map.class); + merge.setAccessible(true); + merge.invoke(service, task, scopeHash, chunkIndex, cozeRows, Map.of()); + } + + @Test + void test_task_013_payload_row_count_chunk_normal_default_path() throws Exception { + // 正常输入：行数与 payload 字节均在上限内，合并走原路径，结果完整保留。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + AtomicReference storedPayload = new AtomicReference<>(""); + stubChunkMerge(chunk, rowsJson(List.of(row("r0", "B0A0000000", "存量行"))), storedPayload); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + List cozeRows = List.of( + row("r1", "B0A0000001", "标题1"), + row("r2", "B0A0000002", "标题2"), + row("r3", "B0A0000003", "标题3")); + + invokeMerge(service, task, "hashA", 1, cozeRows); + assertNotNull(storedPayload.get()); + assertTrue(storedPayload.get().contains("\"r1\"") && storedPayload.get().contains("\"r3\""), + "上限内合并必须完整保留存量行与新增行，实际: " + storedPayload.get()); + verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskChunkMapper, times(1)).update(any(), any()); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + } + + @Test + void test_task_013_payload_row_count_chunk_normal_multiple_items() throws Exception { + // 批量场景：多 chunk 一次 merge，全部在上限内，各 chunk 分别写回、结果不丢失。 + TaskChunkEntity chunkA = chunk(1L, "hashA", 1, "ptr:chunk-A"); + TaskChunkEntity chunkB = chunk(2L, "hashB", 2, "ptr:chunk-B"); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunkA, chunkB)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r1", "B0A0000001", "标题1")))); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-B"), anyString())) + .thenReturn(rowsJson(List.of(row("r2", "B0A0000002", "标题2")))); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> "stored:" + invocation.getArgument(2)); + AtomicLong selectOneRound = new AtomicLong(0); + when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> + selectOneRound.getAndIncrement() == 0 ? chunkA : chunkB); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + invokeMerge(service, task, null, null, List.of( + row("r1", "B0A0000001", "标题1-新"), + row("r2", "B0A0000002", "标题2-新"))); + + verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskChunkMapper, times(2)).update(any(), any()); + } + + @Test + void test_task_013_payload_row_count_chunk_normal_repeated_operation_is_idempotent() throws Exception { + // 重复执行同一输入：每次 merge 恰好写一次，不产生重复对象、重复状态。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + AtomicLong storeCalls = new AtomicLong(0); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r0", "B0A0000000", "存量行")))); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> { + storeCalls.incrementAndGet(); + return "stored:" + invocation.getArgument(2); + }); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + List cozeRows = List.of(row("r1", "B0A0000001", "标题1")); + invokeMerge(service, task, "hashA", 1, cozeRows); + invokeMerge(service, task, "hashA", 1, cozeRows); + assertEquals(2, storeCalls.get(), "重复执行同一输入：每次 merge 恰好写回一次，无多余请求"); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + } + + @Test + void test_task_013_payload_row_count_chunk_boundary_empty_input() throws Exception { + // 空输入：null/空列表安全跳过，不读取 chunk、不写存储、不创建资源。 + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + invokeMerge(service, task, "hashA", 1, null); + invokeMerge(service, task, "hashA", 1, List.of()); + verify(taskChunkMapper, times(0)).selectList(any()); + verify(transientPayloadStorageService, times(0)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + } + + @Test + void test_task_013_payload_row_count_chunk_boundary_single_item() throws Exception { + // 单行：不依赖批量路径，合并后结果正确。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + AtomicReference storedPayload = new AtomicReference<>(""); + stubChunkMerge(chunk, rowsJson(List.of(row("r0", "B0A0000000", "存量行"))), storedPayload); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1"))); + + assertTrue(storedPayload.get().contains("\"r1\""), "单行合并也必须写回 chunk payload，实际: " + storedPayload.get()); + verify(taskChunkMapper, times(1)).update(any(), any()); + } + + @Test + void test_task_013_payload_row_count_chunk_boundary_limit_and_overflow() throws Exception { + // 超限场景三连：行数超限降级、字节超限降级、单行超字节上限拒绝。 + lenient().when(properties.getChunkMergeMaxRows()).thenReturn(2); + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + AtomicLong storeCalls = new AtomicLong(0); + AtomicReference lastStoredPayload = new AtomicReference<>(""); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r0", "B0A0000000", "存量行")))); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> { + storeCalls.incrementAndGet(); + lastStoredPayload.set(invocation.getArgument(4)); + return "stored:" + invocation.getArgument(2); + }); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + List cozeRows = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + cozeRows.add(row("r" + (i + 1), "B0A00000" + (i + 1), "新行" + i)); + } + + // Phase A：行数超限（上限 2，存量 1 + 新增 5）→ 只保留上限内最新行，超限部分转 orphan。 + invokeMerge(service, task, "hashA", 1, cozeRows); + assertTrue(lastStoredPayload.get().contains("\"r4\"") && lastStoredPayload.get().contains("\"r5\""), + "行数超限时保留上限内的最新行，实际: " + lastStoredPayload.get()); + assertFalse(lastStoredPayload.get().contains("\"r0\""), "行数超限时最旧行被降级，实际: " + lastStoredPayload.get()); + assertEquals(1, storeCalls.get()); + verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class)); + verify(transientPayloadStorageService, times(1)) + .storeParsedPayloadEntry(anyString(), any(), anyString(), anyString(), anyString(), eq(true)); + + // Phase B：字节超限（行数放开）→ 从最旧行降级到字节上限内，保留最新结果。 + lenient().when(properties.getChunkMergeMaxRows()).thenReturn(50000); + long oneRowBytes = rowsJson(List.of(row("r9", "B0A0000099", "样本行"))).getBytes(StandardCharsets.UTF_8).length; + lenient().when(properties.getChunkMergePayloadMaxBytes()).thenReturn(oneRowBytes + 5L); + invokeMerge(service, task, "hashA", 1, cozeRows); + assertTrue(lastStoredPayload.get().contains("\"r5\""), "字节超限时保留最新行，实际: " + lastStoredPayload.get()); + assertFalse(lastStoredPayload.get().contains("\"r0\""), "字节超限时最旧行被降级，实际: " + lastStoredPayload.get()); + assertEquals(2, storeCalls.get()); + verify(taskScopeStateMapper, times(2)).insert(any(TaskScopeStateEntity.class)); + + // Phase C：单行本身超过字节上限 → 抛可识别异常拒绝合并，不写 chunk。 + lenient().when(properties.getChunkMergePayloadMaxBytes()).thenReturn(5L); + Exception ex = assertThrows(Exception.class, () -> { + try { + invokeMerge(service, task, "hashA", 1, cozeRows); + } catch (java.lang.reflect.InvocationTargetException e) { + throw e.getCause(); + } + }); + assertNotNull(ex.getMessage()); + assertTrue(ex.getMessage().contains("字节上限"), + "超字节上限必须抛可识别异常，实际: " + ex.getMessage()); + assertEquals(2, storeCalls.get(), "拒绝合并时不写 chunk"); + } + + @Test + void test_task_013_payload_row_count_chunk_invalid_input_rejected() { + // 非法输入：chunk 载荷加载失败（resolve 抛异常）时抛出可识别异常且不写 chunk。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenThrow(new IllegalStateException("rustfs down")); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + Exception ex = assertThrows(Exception.class, () -> { + try { + invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1"))); + } catch (java.lang.reflect.InvocationTargetException e) { + throw e.getCause(); + } + }); + assertNotNull(ex.getMessage()); + assertTrue(ex.getMessage().contains("chunk"), + "chunk 读取失败消息必须可识别，实际: " + ex.getMessage()); + verify(transientPayloadStorageService, times(0)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + } + + @Test + void test_task_013_payload_row_count_chunk_dependency_failure_releases_resources() throws Exception { + // 依赖失败：payload 存储失败时抛带上下文的可识别异常、无残留状态；恢复后重试成功。 + TaskChunkEntity chunk = chunk(1L, "hashA", 1, "ptr:chunk-A"); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r0", "B0A0000000", "存量行")))); + AtomicLong storeCalls = new AtomicLong(0); + doAnswer(invocation -> { + if (storeCalls.getAndIncrement() == 0) { + throw new IllegalStateException("rustfs down"); + } + return "stored:" + invocation.getArgument(2); + }).when(transientPayloadStorageService).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + Exception ex = assertThrows(Exception.class, () -> { + try { + invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1"))); + } catch (java.lang.reflect.InvocationTargetException e) { + throw e.getCause(); + } + }); + assertTrue(ex.getMessage() != null && ex.getMessage().contains("相似ASIN分片载荷"), + "存储失败必须抛带上下文的可识别异常，实际: " + ex.getMessage()); + assertEquals(1, storeCalls.get(), "失败时只尝试一次即抛出，不静默吞错"); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + + // 依赖恢复后重试成功：结果正确、无残留状态。 + invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1"))); + assertEquals(2, storeCalls.get(), "恢复后重试成功"); + verify(taskChunkMapper, times(1)).update(any(), any()); + verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); + } +}