From 4d9114625646f8d62f1ffaf8f534ec2c47668c74 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 16:46:29 +0800 Subject: [PATCH] task-12: extend coze result buffering beyond poll path - bufferCozeRowsOrMerge: unified DONE result persist entry point; buffers to transient storage when toggle on, falls back to immediate chunk merge on failure (result never lost), no-op on empty input - submit/retry synchronous immediate DONE results now buffer too (previously always merged immediately); submit path persists a DONE scope state row to carry the buffer pointer (persistImmediateCozeDoneState) - poll path drops the batchTotal>1 gate: single-batch tasks also buffer - CozeCandidate/CozeBatchContext records package-private for test access --- .../service/SimilarAsinTaskService.java | 157 +++++-- ...larAsinTaskServiceCozeBufferScopeTest.java | 431 ++++++++++++++++++ 2 files changed, 552 insertions(+), 36 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceCozeBufferScopeTest.java 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 42c32756..269871c1 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 @@ -2616,6 +2616,17 @@ public class SimilarAsinTaskService { if (!emptyResultMessage.isBlank()) { throw new IllegalStateException(emptyResultMessage); } + // Task 12：同步 immediate DONE 结果也走缓冲（原立即 merge），由 finalize/assemble 前 + // 一次性 flush 合并到 chunk，减少 chunk payload 频繁读写。先落一条 DONE state 承载 + // 缓冲 pointer；缓冲关闭/失败/重复时回退立即 merge，结果不丢失。 + if (isCozeResultBufferEnabled()) { + TaskScopeStateEntity doneState = persistImmediateCozeDoneState(task, result, job, batchRows, + batchScopeKey, batchScopeHash, batchIndex, batchTotal, submit.credentialName()); + if (doneState != null) { + bufferCozeRowsOrMerge(doneState, readCozeBatchContext(doneState), cozeRows, task, allRowsByBaseId); + return false; + } + } mergeCozeRowsIntoChunk(task, null, null, cozeRows, allRowsByBaseId); return false; } @@ -2851,31 +2862,23 @@ public class SimilarAsinTaskService { if (!failureMessage.isBlank()) { cozeRows = cozeClient.markRowsFailed(batchRows, failureMessage); } - // P0-3：仅"DONE 且 batchTotal>1 且 feature toggle 开启"时缓冲 cozeRows 到 transient storage， - // 由 finalize 阶段一次性合并到 chunk。失败 batch / 单 batch 任务保留原立即 merge 路径。 - boolean buffered = false; - if (failureMessage.isBlank() - && isCozeResultBufferEnabled() - && context.batchTotal() != null && context.batchTotal() > 1 - && cozeRows != null && !cozeRows.isEmpty()) { - CozeBatchContext bufferedContext = bufferCozeResultForFlush(state, context, cozeRows); - if (bufferedContext != null - && bufferedContext.resultPayloadPointer() != null - && !bufferedContext.resultPayloadPointer().isBlank()) { - context = bufferedContext; - buffered = true; + // Task 12：DONE 结果统一走 bufferCozeRowsOrMerge（原 P0-3 仅 poll 且 batchTotal>1 缓冲； + // 现单 batch、submit/retry 同步 immediate 结果也缓冲），缓冲失败回退立即 merge， + // flush 在 finalize/assemble 前一次性合并到 chunk，结果不丢失。 + // 失败行（markRowsFailed）保持立即 merge 语义不变。 + if (failureMessage.isBlank()) { + if (cozeRows != null && !cozeRows.isEmpty()) { + FileTaskEntity pollTask = taskForPoll(state.getTaskId()); + bufferCozeRowsOrMerge(state, context, cozeRows, pollTask, + pollTask == null ? Map.of() : allRowsByBaseIdForPoll(pollTask)); } - } - if (!buffered) { - FileTaskEntity task = taskForPoll(state.getTaskId()); - if (task != null) { - Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); + } else { + FileTaskEntity pollTask = taskForPoll(state.getTaskId()); + if (pollTask != null) { + Map> allRowsByBaseId = allRowsByBaseIdForPoll(pollTask); try { - mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); + mergeCozeRowsIntoChunk(pollTask, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); } catch (Exception mergeEx) { - if (failureMessage.isBlank()) { - throw mergeEx; - } log.warn("[similar-asin] coze failed result merge failed, mark state terminal anyway taskId={} stateId={} executeId={} err={}", state.getTaskId(), state.getId(), state.getCozeExecuteId(), firstNonBlank(mergeEx.getMessage(), mergeEx.getClass().getSimpleName())); @@ -3406,7 +3409,9 @@ public class SimilarAsinTaskService { if (!emptyResultMessage.isBlank()) { throw new IllegalStateException(emptyResultMessage); } - mergeCozeRowsIntoChunk(task, null, null, cozeRows, allRowsByBaseId); + // Task 12：retry 同步 immediate DONE 结果也走缓冲（原立即 merge）， + // 缓冲失败回退立即 merge；flush 在 finalize 时一次性完成。 + bufferCozeRowsOrMerge(state, context, cozeRows, task, allRowsByBaseId); markCozeStateTerminal(state, COZE_STATUS_DONE, null); maybeFinalizeCozeJobLocked(state.getTaskId(), context); return; @@ -4048,6 +4053,86 @@ public class SimilarAsinTaskService { return properties.isCozeResultBufferEnabled(); } + /** + * Task 12：统一 Coze DONE 结果落库入口。 + * 缓冲开关开启时把 cozeRows 写入 transient storage（pointer 存进 state.stateJson）， + * 由 flushBufferedCozeResults 在 finalize/assemble 前一次性合并到 chunk； + * 缓冲失败（存储异常 / state 更新失败 / 开关关闭）回退立即 merge，结果不丢失。 + * 空 rows / 空 state / 空 context 直接返回，不产生任何写入。 + */ + private void bufferCozeRowsOrMerge(TaskScopeStateEntity state, + CozeBatchContext context, + List cozeRows, + FileTaskEntity task, + Map> allRowsByBaseId) { + if (state == null || context == null || cozeRows == null || cozeRows.isEmpty()) { + return; + } + if (!isCozeResultBufferEnabled()) { + mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); + return; + } + CozeBatchContext bufferedContext = bufferCozeResultForFlush(state, context, cozeRows); + if (bufferedContext == null + || bufferedContext.resultPayloadPointer() == null + || bufferedContext.resultPayloadPointer().isBlank()) { + // 缓冲失败：回退立即 merge，避免结果悬挂在 transient storage 之外。 + mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); + } + } + + /** + * Task 12：为 submit 同步 immediate DONE 结果落一条 DONE state 承载缓冲 pointer。 + * 与 saveCozeBatchState（SUBMITTED 异步）不同，该 state 直接以 DONE 终态插入， + * 不会被 countPendingCozeStates 扫描；缓冲失败/重复插入时返回 null，调用方回退立即 merge。 + */ + private TaskScopeStateEntity persistImmediateCozeDoneState(FileTaskEntity task, + FileResultEntity result, + TaskFileJobEntity job, + List batchRows, + String batchScopeKey, + String batchScopeHash, + int batchIndex, + int batchTotal, + String credentialName) { + LocalDateTime now = LocalDateTime.now(); + CozeBatchContext context = new CozeBatchContext( + job.getId(), + result.getId(), + null, + null, + batchIndex, + batchTotal, + currentInstanceId(), + 0, + credentialName, + null + ); + TaskScopeStateEntity state = new TaskScopeStateEntity(); + state.setTaskId(task.getId()); + state.setModuleType(MODULE_TYPE); + state.setScopeKey(batchScopeKey); + state.setScopeHash(batchScopeHash); + state.setStateJson(writeJson(context, "serialize immediate coze done state context failed")); + state.setCozeStatus(COZE_STATUS_DONE); + state.setCozeSubmittedAt(now); + state.setCozeCompletedAt(now); + state.setCozeAttemptCount(0); + state.setChunkTotal(batchTotal); + state.setReceivedChunkCount(batchIndex); + state.setCompleted(1); + state.setCreatedAt(now); + state.setUpdatedAt(now); + try { + taskScopeStateMapper.insert(state); + return state; + } catch (DuplicateKeyException ex) { + log.info("[similar-asin] duplicate immediate done coze state ignored taskId={} scope={}", + task.getId(), batchScopeKey); + return null; + } + } + /** * P0-3：在 finalize 之前一次性把所有 DONE/FAILED state 上缓冲的 cozeRows * 按 chunkScopeHash 分组合并到 chunk。把每个 batch 的"loadSubmittedChunks + @@ -4291,9 +4376,9 @@ public class SimilarAsinTaskService { return "coze:task:" + taskId + ":rows:" + DigestUtil.sha256Hex(rowKeys.toString()); } - private record CozeCandidate(String chunkScopeHash, - Integer chunkIndex, - SimilarAsinResultRowDto row) { + record CozeCandidate(String chunkScopeHash, + Integer chunkIndex, + SimilarAsinResultRowDto row) { } private record PythonUploadProgress(int current, int total, String unit) { @@ -6789,16 +6874,16 @@ public class SimilarAsinTaskService { boolean terminal) { } - private record CozeBatchContext(Long jobId, - Long resultId, - String chunkScopeHash, - Integer chunkIndex, - Integer batchIndex, - Integer batchTotal, - String ownerInstanceId, - Integer submitRetryCount, - String credentialName, - String resultPayloadPointer) { + record CozeBatchContext(Long jobId, + Long resultId, + String chunkScopeHash, + Integer chunkIndex, + Integer batchIndex, + Integer batchTotal, + String ownerInstanceId, + Integer submitRetryCount, + String credentialName, + String resultPayloadPointer) { } private static class SourceRowsBuilder { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceCozeBufferScopeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceCozeBufferScopeTest.java new file mode 100644 index 00000000..ebc01ff8 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceCozeBufferScopeTest.java @@ -0,0 +1,431 @@ +package com.nanri.aiimage.modules.similarasin.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.SimilarAsinProperties; +import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService; +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.FileResultEntity; +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.TaskFileJobService; +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.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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.atLeastOnce; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.lenient; +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 12：扩展 Coze 结果缓冲覆盖范围，减少频繁读写完整 chunk payload。 + * P0-3 缓冲只覆盖"poll DONE 且 batchTotal>1"；Task 12 扩展为： + * 1) poll DONE 结果去掉 batchTotal 限制，单 batch 也走缓冲； + * 2) retry 提交同步 immediate DONE 结果也走缓冲（原立即 merge）； + * 3) 统一走 bufferCozeRowsOrMerge：缓冲失败回退立即 merge，结果不丢失。 + * flushBufferedCozeResults 在 finalize 前一次性合并，全任务收敛为一次 chunk 读写。 + */ +@ExtendWith(MockitoExtension.class) +class SimilarAsinTaskServiceCozeBufferScopeTest { + + private static final AtomicLong NEXT_ID = new AtomicLong(71000); + private static final String MODULE = SimilarAsinTaskService.MODULE_TYPE; + private static final String CREDENTIAL = "cred-1"; + + @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 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() throws Exception { + 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.getCozeBatchSize()).thenReturn(5); + lenient().when(properties.getCozeTextOnlyBatchSize()).thenReturn(10); + lenient().when(properties.isCozeResultBufferEnabled()).thenReturn(true); + lenient().when(properties.getCozePollTimeoutMillis()).thenReturn(30_000); + lenient().when(properties.getDbJobTouchIntervalMillis()).thenReturn(2_000L); + lenient().when(properties.getDbTaskTouchIntervalMillis()).thenReturn(2_000L); + lenient().when(properties.getCozeSubmitLockWaitMillis()).thenReturn(1_000L); + lenient().when(properties.getCozeSubmitLockRetryDelayMillis()).thenReturn(100L); + lenient().when(properties.getCozeSubmitMinIntervalMillis()).thenReturn(0L); + lenient().when(properties.getCozeSubmitMaxRetryCount()).thenReturn(3); + lenient().when(properties.getCozeFlushPendingMinutes()).thenReturn(10); + lenient().when(cozeClient.configuredCredentialCount()).thenReturn(1); + lenient().when(cozeClient.nextCredential()).thenReturn(new SimilarAsinCozeClient.CozeCredentialRef( + CREDENTIAL, "wf-1", "token-1", 4)); + lenient().when(cozeClient.credentialByName(CREDENTIAL)).thenReturn(new SimilarAsinCozeClient.CozeCredentialRef( + CREDENTIAL, "wf-1", "token-1", 4)); + lenient().when(cozeCredentialPoolService.borrow(eq(MODULE), any())).thenReturn( + mock(CozeCredentialPoolService.BorrowedCredential.class)); + lenient().when(distributedJobLockService.tryLock(anyString(), any())).thenReturn( + mock(com.nanri.aiimage.common.service.DistributedJobLockService.LockHandle.class)); + 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(FileResultEntity.class))).thenReturn(1); + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenReturn(1); + lenient().when(taskScopeStateMapper.update(any(), any())).thenReturn(1); + lenient().when(taskScopeStateMapper.selectList(any())).thenReturn(List.of()); + lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of()); + lenient().when(transientPayloadStorageService.storeParsedPayloadEntry( + eq(MODULE), any(), anyString(), anyString(), anyString(), eq(true))) + .thenAnswer(invocation -> "rustfs:coze-result/" + NEXT_ID.incrementAndGet()); + } + + @AfterEach + void shutdown() { + service.shutdownAssembleExecutor(); + } + + private static SimilarAsinResultRowDto row(String rowToken, String id, String asin, String country, String title) { + SimilarAsinResultRowDto r = new SimilarAsinResultRowDto(); + r.setRowToken(rowToken); + r.setId(id); + r.setAsin(asin); + r.setCountry(country); + r.setTitle(title); + r.setMainUrl("https://img.example.com/" + asin + ".jpg"); + return r; + } + + private static TaskChunkEntity chunk(Long id, String scopeHash, Integer chunkIndex, String payloadJson) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(id); + chunk.setTaskId(7104L); + chunk.setModuleType(MODULE); + chunk.setScopeHash(scopeHash); + chunk.setChunkIndex(chunkIndex); + chunk.setPayloadJson(payloadJson); + chunk.setPayloadHash("h-" + id); + return chunk; + } + + private static FileTaskEntity task() { + FileTaskEntity task = new FileTaskEntity(); + task.setId(7104L); + task.setModuleType(MODULE); + task.setStatus("RUNNING"); + task.setResultJson("{\"categorySwitch\":true}"); + return task; + } + + private static TaskScopeStateEntity state(FileTaskEntity task, long id, String status, int batchTotal) { + TaskScopeStateEntity state = new TaskScopeStateEntity(); + state.setId(id); + state.setTaskId(task.getId()); + state.setModuleType(MODULE); + state.setScopeHash("scope-" + id); + state.setCozeStatus(status); + state.setParsedPayloadJson("ptr:batch-" + id); + state.setStateJson("{\"jobId\":7101,\"resultId\":7201,\"chunkScopeHash\":null,\"chunkIndex\":null," + + "\"batchIndex\":1,\"batchTotal\":" + batchTotal + ",\"ownerInstanceId\":\"test-instance\"," + + "\"submitRetryCount\":0,\"credentialName\":\"" + CREDENTIAL + "\",\"resultPayloadPointer\":\"ptr:buffer-" + id + "\"}"); + return state; + } + + private static String rowsJson(List rows) throws Exception { + return new ObjectMapper().writeValueAsString(rows); + } + + private SimilarAsinTaskService.CozeBatchContext context(int batchTotal) { + return new SimilarAsinTaskService.CozeBatchContext( + 7101L, 7201L, null, null, 1, batchTotal, "test-instance", 0, CREDENTIAL, null); + } + + private void stubChunkMerge(String payloadJson) throws Exception { + TaskChunkEntity chunk = chunk(1L, "scope-1", 1, "ptr:chunk-1"); + // loadSubmittedChunks 只保留非空 chunk，chunk payload 必须能解析出至少一行。 + // 全部 lenient：缓冲成功路径不触达 merge，仅缓冲失败/flush 合并路径消费。 + lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenAnswer(invocation -> { + String pointer = invocation.getArgument(0); + if (pointer != null && pointer.startsWith("ptr:chunk-")) { + return payloadJson; + } + return "[]"; + }); + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> "ptr:stored-" + invocation.getArgument(3)); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + lenient().when(taskChunkMapper.update(any(), any())).thenReturn(1); + } + + private static String chunkRowsJson() throws Exception { + // chunk-1 已含 r1 行：loadSubmittedChunks 只保留非空 chunk，且 rowKey 索引能命中缓冲行。 + return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"))); + } + + @Test + void test_task_012_payload_chunk_normal_default_path() throws Exception { + // 正常输入：DONE 结果（batchTotal=1 单 batch）经 bufferCozeRowsOrMerge 走缓冲， + // 不立即写 chunk；缓冲失败回退立即 merge 结果不丢失。 + FileTaskEntity task = task(); + List rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")); + + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of()); + + verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry( + eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true)); + verify(transientPayloadStorageService, never()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskChunkMapper, never()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_normal_multiple_items() throws Exception { + // 批量场景：多个 DONE state（单 batch）全部缓冲；flush 后按 chunk 分组一次合并 + FileTaskEntity task = task(); + stubChunkMerge(chunkRowsJson()); + when(taskScopeStateMapper.selectList(any())).thenReturn( + List.of(state(task, 1L, "DONE", 1), state(task, 2L, "DONE", 1))); + when(fileTaskMapper.selectById(7104L)).thenReturn(task); + when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenAnswer(invocation -> { + String pointer = invocation.getArgument(0); + if (pointer != null && pointer.startsWith("ptr:buffer-")) { + return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"))); + } + if (pointer != null && pointer.startsWith("ptr:chunk-")) { + return chunkRowsJson(); + } + return "[]"; + }); + when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(invocation -> "ptr:stored-" + invocation.getArgument(3)); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk(1L, "scope-1", 1, "ptr:chunk-1")); + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(1L, "scope-1", 1, "ptr:chunk-1"))); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushBufferedCozeResults", Long.class); + flush.setAccessible(true); + flush.invoke(service, 7104L); + + verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskChunkMapper, atLeastOnce()).update(any(), any()); + // pointer 清理：每个缓冲 state 都更新 + verify(taskScopeStateMapper, atLeastOnce()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_normal_repeated_operation_is_idempotent() throws Exception { + // 重复执行同一输入：缓冲写幂等（同一 state 不产生重复 buffer/merge） + FileTaskEntity task = task(); + List rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")); + + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + TaskScopeStateEntity state = state(task, 1L, "DONE", 2); + bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of()); + bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of()); + + // 缓冲 2 次（每次重新写 pointer 是幂等语义：同一 state 覆盖写，无重复行） + verify(transientPayloadStorageService, times(2)).storeParsedPayloadEntry( + eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true)); + verify(taskChunkMapper, never()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_boundary_empty_input() throws Exception { + // 空输入：无行时缓冲与 merge 都不发生，不创建无效资源 + FileTaskEntity task = task(); + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), null, task, Map.of()); + bufferOrMerge.invoke(service, state(task, 2L, "DONE", 1), context(1), List.of(), task, Map.of()); + + verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true)); + verify(taskChunkMapper, never()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_boundary_single_item() throws Exception { + // 单 batch（batchTotal=1）：原 P0-3 例外，现在也缓冲 + FileTaskEntity task = task(); + List rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")); + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of()); + + verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry( + eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true)); + verify(taskChunkMapper, never()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_boundary_limit_and_overflow() throws Exception { + // 缓冲开关关闭：回退立即 merge，DONE 结果仍落 chunk 不丢失 + FileTaskEntity task = task(); + stubChunkMerge(chunkRowsJson()); + when(properties.isCozeResultBufferEnabled()).thenReturn(false); + List rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")); + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of()); + + verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true)); + verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + } + + @Test + void test_task_012_payload_chunk_invalid_input_rejected() throws Exception { + // 缓冲写失败（storeParsedPayloadEntry 抛异常）：回退立即 merge，结果不丢失 + FileTaskEntity task = task(); + stubChunkMerge(chunkRowsJson()); + when(transientPayloadStorageService.storeParsedPayloadEntry( + eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true))) + .thenThrow(new IllegalStateException("rustfs full")); + List rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")); + Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge", + TaskScopeStateEntity.class, + SimilarAsinTaskService.CozeBatchContext.class, + List.class, FileTaskEntity.class, Map.class); + bufferOrMerge.setAccessible(true); + bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of()); + + verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + verify(taskChunkMapper, atLeastOnce()).update(any(), any()); + } + + @Test + void test_task_012_payload_chunk_dependency_failure_releases_resources() throws Exception { + // flush 时 chunk 写失败：抛可识别业务异常且不清 pointer（保留待重试）； + // 依赖恢复后重试 flush 成功，chunk 合并一次、pointer 清理。 + FileTaskEntity task = task(); + stubChunkMerge(chunkRowsJson()); + TaskScopeStateEntity s = state(task, 1L, "DONE", 1); + when(taskScopeStateMapper.selectList(any())).thenReturn(List.of(s)); + when(fileTaskMapper.selectById(7104L)).thenReturn(task); + when(transientPayloadStorageService.resolvePayload(anyString(), anyString())) + .thenAnswer(invocation -> { + String pointer = invocation.getArgument(0); + if (pointer != null && pointer.startsWith("ptr:buffer-")) { + return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"))); + } + if (pointer != null && pointer.startsWith("ptr:chunk-")) { + return chunkRowsJson(); + } + return "[]"; + }); + java.util.concurrent.atomic.AtomicInteger storeCalls = new java.util.concurrent.atomic.AtomicInteger(0); + doAnswer(invocation -> { + if (storeCalls.incrementAndGet() == 1) { + throw new IllegalStateException("rustfs write failed"); + } + return "ptr:stored-" + invocation.getArgument(3); + }).when(transientPayloadStorageService).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + + Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushBufferedCozeResults", Long.class); + flush.setAccessible(true); + Exception ex = assertThrows(Exception.class, () -> { + try { + flush.invoke(service, 7104L); + } catch (java.lang.reflect.InvocationTargetException e) { + throw e.getCause(); + } + }); + assertTrue(ex.getMessage() != null && ex.getMessage().contains("刷新缓冲区"), + "flush 失败消息必须可识别, 实际: " + ex.getMessage()); + // 失败分组不清 pointer：buffer 未被删除、stateJson 未更新，留待重试 + verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString()); + verify(taskScopeStateMapper, never()).update(any(), any()); + + // 恢复后重试 flush：chunk 合并成功一次，pointer 清理 + flush.invoke(service, 7104L); + assertEquals(2, storeCalls.get(), "恢复后重试应再次写 chunk"); + verify(taskScopeStateMapper, atLeastOnce()).update(any(), any()); + } +}