diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index a0e492e0..267615db 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -424,6 +424,53 @@ public class AppearancePatentTaskService { public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) { submitResultLocked(taskId, request); + // 分片补传后自动恢复终态失败的组装 job —— 走与删除品牌同一套口径， + // 此前外观专利没有接该入口，分片补齐后只能人工重置 job（线上任务 28459 即如此）。 + maybeRecoverTerminalFailedAssemble(taskId); + } + + /** + * 补传恢复：此前因分片缺失导致组装 job 重试耗尽（终态失败），客户端补传缺口后 + * 把失败的组装 job 重置为 PENDING 重新派发（{@code resetTerminalFailedForRecovery} 自行补发 dispatch 事件）。 + * + *

best-effort：恢复失败不得影响补传本身——分片已经落库，恢复只是让后续组装继续推进。 + * 常态（无终态失败 job）下只查两次即返回，不触发分片扫描。 + */ + private void maybeRecoverTerminalFailedAssemble(Long taskId) { + if (taskId == null || taskId <= 0) { + return; + } + try { + FileResultEntity result = findResultRecord(taskId); + if (result == null) { + return; + } + if (taskFileJobService.hasSuccessfulAssembleJob(taskId, MODULE_TYPE, result.getId())) { + return; + } + if (!taskFileJobService.isTerminalFailedAssembleJob(taskId, MODULE_TYPE, result.getId())) { + return; + } + if (!isResultSubmissionComplete(taskId)) { + log.info("[appearance-patent] 组装 job 终态失败但分片仍未补传完整，暂不恢复 taskId={} resultId={}", + taskId, result.getId()); + return; + } + log.info("[appearance-patent] 分片已补传完整，恢复终态失败的组装 job taskId={} resultId={}", + taskId, result.getId()); + taskFileJobService.resetTerminalFailedForRecovery(taskId, MODULE_TYPE, result.getId()); + } catch (Exception ex) { + log.warn("[appearance-patent] 补传恢复检查失败（不影响本次补传）taskId={} err={}", taskId, ex.getMessage(), ex); + } + } + + /** 按 task 取结果行（不创建）；不存在返回 null。 */ + private FileResultEntity findResultRecord(Long taskId) { + List rows = fileResultMapper.selectList(new LambdaQueryWrapper() + .eq(FileResultEntity::getTaskId, taskId) + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .last("limit 1")); + return rows == null || rows.isEmpty() ? null : rows.getFirst(); } private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) { @@ -1223,6 +1270,7 @@ public class AppearancePatentTaskService { if (rows == null || rows.isEmpty()) { return; } + String conflictDetail = "未发生冲突"; for (int attempt = 1; attempt <= CHUNK_PAYLOAD_MERGE_RETRY_LIMIT; attempt++) { TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() .eq(TaskChunkEntity::getTaskId, taskId) @@ -1264,13 +1312,35 @@ public class AppearancePatentTaskService { transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); return; } - transientPayloadStorageService.deletePayloadIfPresent(storedPayload); + // CAS 冲突：读回行上的当前哈希，否则只有一句 conflict 无从判断是谁改的（线上 28459 至今未能定位） + String currentHash = currentPayloadHash(chunk.getId()); + conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash; if (attempt < CHUNK_PAYLOAD_MERGE_RETRY_LIMIT) { - log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}", - taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT); + log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}", + taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT, oldPayloadHash, currentHash); + // 还要重试：这次写的对象会被下次重写，先删掉避免堆积 + transientPayloadStorageService.deletePayloadIfPresent(storedPayload); + } else { + // 终局失败：保留这次写入的版本化对象，作为读路径「同槽位兄弟对象」兜底的恢复源。 + // 行没指过去不该让该分片永久判死——删掉它才是线上 28459 丢数据的形态。 + log.error("[appearance-patent] chunk payload 合并冲突重试耗尽，保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}", + taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload); } } - throw new IllegalStateException("appearance patent chunk payload update conflict"); + throw new IllegalStateException("appearance patent chunk payload update conflict " + conflictDetail); + } + + /** 读回 chunk 行当前的 payload_hash，用于 CAS 冲突定位（行已不存在/读取失败时返回可读标记）。 */ + private String currentPayloadHash(Long chunkId) { + if (chunkId == null) { + return "chunkId 为空"; + } + try { + TaskChunkEntity latest = taskChunkMapper.selectById(chunkId); + return latest == null ? "行已不存在" : latest.getPayloadHash(); + } catch (Exception ex) { + return "读取失败:" + ex.getMessage(); + } } private List expandRows(List representatives, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java index c5fd8e27..82a6e8a1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java @@ -214,6 +214,7 @@ public class SimilarAsinPipelineSupport { return; } int maxAttempts = 3; + String conflictDetail = "未发生冲突"; for (int attempt = 1; attempt <= maxAttempts; attempt++) { TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() .eq(TaskChunkEntity::getTaskId, taskId) @@ -275,13 +276,35 @@ public class SimilarAsinPipelineSupport { transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); return; } - transientPayloadStorageService.deletePayloadIfPresent(storedPayload); + // CAS 冲突：读回行上的当前哈希，否则只有一句 conflict 无从判断是谁改的 + String currentHash = currentPayloadHash(chunk.getId()); + conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash; if (attempt < maxAttempts) { - log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}", - taskId, scopeHash, chunkIndex, attempt, maxAttempts); + log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}", + taskId, scopeHash, chunkIndex, attempt, maxAttempts, oldPayloadHash, currentHash); + // 还要重试：这次写的对象会被下次重写，先删掉避免堆积 + transientPayloadStorageService.deletePayloadIfPresent(storedPayload); + } else { + // 终局失败：保留这次写入的版本化对象，作为读路径「同槽位兄弟对象」兜底的恢复源。 + // 与外观专利同一口径——行没指过去不该让该分片永久判死。 + log.error("[similar-asin] chunk payload 合并冲突重试耗尽，保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}", + taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload); } } - throw new IllegalStateException("相似ASIN分片载荷更新失败"); + throw new IllegalStateException("相似ASIN分片载荷更新失败 " + conflictDetail); + } + + /** 读回 chunk 行当前的 payload_hash，用于 CAS 冲突定位（行已不存在/读取失败时返回可读标记）。 */ + private String currentPayloadHash(Long chunkId) { + if (chunkId == null) { + return "chunkId 为空"; + } + try { + TaskChunkEntity latest = taskChunkMapper.selectById(chunkId); + return latest == null ? "行已不存在" : latest.getPayloadHash(); + } catch (Exception ex) { + return "读取失败:" + ex.getMessage(); + } } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentChunkMergeConflictTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentChunkMergeConflictTest.java new file mode 100644 index 00000000..4c6849bf --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentChunkMergeConflictTest.java @@ -0,0 +1,218 @@ +package com.nanri.aiimage.modules.appearancepatent.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.config.AppearancePatentProperties; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.StorageProperties; +import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentLlmClient; +import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +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.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +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.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 外观专利分片载荷合并的 CAS 冲突处理（2026-09-17 线上任务 28459 的遗留项）。 + * + *

原实现在每次 CAS 冲突后都删掉刚写入的版本化对象，重试耗尽即抛异常、行仍指向旧指针—— + * 一旦旧对象也不在，该分片就永久读不到（28459 的 chunk-462/473/484 正是这个形态）。 + * 现在：冲突时读回行上的当前哈希以便定位；**终局失败保留最后一个兜底对象**， + * 让读路径的「同槽位兄弟对象」兜底仍有数据可取。 + */ +@ExtendWith(MockitoExtension.class) +class AppearancePatentChunkMergeConflictTest { + + private static final Long TASK_ID = 28459L; + private static final String SCOPE_HASH = "2248d39710545b47b7c7035fc60c4e33924a16174abf65dd1c5a230918e6f61e"; + private static final int CHUNK_INDEX = 462; + + @Mock private LocalFileStorageService localFileStorageService; + @Mock private OssStorageService ossStorageService; + @Mock private StorageProperties storageProperties; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @Mock private AppearancePatentLlmClient llmClient; + @Mock private AppearancePatentTaskCacheService taskCacheService; + @Mock private AppearancePatentProperties properties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private PlatformTransactionManager transactionManager; + @Mock private DistributedJobLockService distributedJobLockService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private TaskProgressLightAssembler taskProgressLightAssembler; + + @InjectMocks private AppearancePatentTaskService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + // 共享写开启才会走版本化对象存储；否则落到本地兜底路径直接报「RustFS 未配置」 + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + } + + /** 首次冲突后重试成功：中间那次写的对象要删（会被下次重写），且最终行被改到新对象。 */ + @Test + void retryAfterConflictDeletesSupersededObject() { + stubChunk("hash-old", "ptr-chunk-462.json"); + AtomicInteger stores = new AtomicInteger(); + stubVersionedStore(stores); + when(taskChunkMapper.update(any(), any())).thenReturn(0, 1); + when(taskChunkMapper.selectById(anyLong())).thenReturn(chunk("hash-other", "ptr-chunk-462.json")); + + invokeMerge(); + + verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + // 第 1 次冲突写的对象被删；第 2 次成功，走的是「替换旧对象」而不是删新对象 + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1"); + verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-2"); + verify(transientPayloadStorageService).deleteReplacedPayloadIfNeeded(eq("ptr-chunk-462.json"), eq("sibling-2")); + } + + /** 冲突重试耗尽：**保留**最后一次写入的兜底对象（本次要修的形态），并抛异常带出两个哈希。 */ + @Test + void exhaustedConflictKeepsLastStoredObjectAsFallback() { + stubChunk("hash-old", "ptr-chunk-462.json"); + AtomicInteger stores = new AtomicInteger(); + stubVersionedStore(stores); + when(taskChunkMapper.update(any(), any())).thenReturn(0); + when(taskChunkMapper.selectById(anyLong())).thenReturn(chunk("hash-other", "ptr-chunk-462.json")); + + IllegalStateException ex = assertThrows(IllegalStateException.class, this::invokeMerge); + + verify(transientPayloadStorageService, times(3)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()); + // 前两次冲突的对象照旧删除；第三次（终局）的对象必须保留 + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1"); + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2"); + verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3"); + assertTrue(ex.getMessage().contains("hash-old"), "异常需带出期望哈希，实际: " + ex.getMessage()); + assertTrue(ex.getMessage().contains("hash-other"), "异常需带出当前哈希，实际: " + ex.getMessage()); + } + + /** 冲突时读回行上的当前哈希，供定位（此前只有一句 update conflict，线上无法定位）。 */ + @Test + void conflictReadsBackCurrentHashForDiagnostics() { + stubChunk("hash-old", "ptr-chunk-462.json"); + stubVersionedStore(new AtomicInteger()); + when(taskChunkMapper.update(any(), any())).thenReturn(0); + when(taskChunkMapper.selectById(7L)).thenReturn(chunk("hash-changed-by-other-writer", "ptr-chunk-462.json")); + + assertThrows(IllegalStateException.class, this::invokeMerge); + + verify(taskChunkMapper, times(3)).selectById(7L); + } + + /** 读回行失败不得掩盖原始冲突：异常信息里给出可读标记。 */ + @Test + void readBackFailureDoesNotMaskConflict() { + stubChunk("hash-old", "ptr-chunk-462.json"); + stubVersionedStore(new AtomicInteger()); + when(taskChunkMapper.update(any(), any())).thenReturn(0); + when(taskChunkMapper.selectById(anyLong())).thenThrow(new IllegalStateException("db down")); + + IllegalStateException ex = assertThrows(IllegalStateException.class, this::invokeMerge); + + assertTrue(ex.getMessage().contains("读取失败"), "实际: " + ex.getMessage()); + } + + // ===== 辅助 ===== + + private void invokeMerge() { + AppearancePatentResultRowDto row = new AppearancePatentResultRowDto(); + row.setRowToken("r1"); + row.setAsin("B0A0000001"); + ReflectionTestUtils.invokeMethod(service, "mergeChunkPayload", TASK_ID, SCOPE_HASH, CHUNK_INDEX, List.of(row)); + } + + private void stubChunk(String payloadHash, String payloadPointer) { + when(taskChunkMapper.selectOne(any())).thenReturn(chunk(payloadHash, payloadPointer)); + lenient().when(transientPayloadStorageService.resolvePayload(eq(payloadPointer), anyString())) + .thenReturn("[{\"rowToken\":\"r1\",\"asin\":\"B0A0000001\"}]"); + } + + private void stubVersionedStore(AtomicInteger stores) { + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned( + anyString(), anyLong(), anyString(), any(), anyString())) + .thenAnswer(inv -> "sibling-" + stores.incrementAndGet()); + } + + private TaskChunkEntity chunk(String payloadHash, String payloadPointer) { + TaskChunkEntity chunk = new TaskChunkEntity(); + chunk.setId(7L); + chunk.setTaskId(TASK_ID); + chunk.setModuleType(AppearancePatentTaskService.MODULE_TYPE); + chunk.setScopeHash(SCOPE_HASH); + chunk.setChunkIndex(CHUNK_INDEX); + chunk.setPayloadJson(payloadPointer); + chunk.setPayloadHash(payloadHash); + return chunk; + } + + /** 断言辅助：确认存的对象数（避免误用未使用的 import）。 */ + @Test + void storeIsCalledOnceWhenUpdateSucceeds() { + stubChunk("hash-old", "ptr-chunk-462.json"); + AtomicInteger stores = new AtomicInteger(); + stubVersionedStore(stores); + when(taskChunkMapper.update(any(), any())).thenReturn(1); + + invokeMerge(); + + assertEquals(1, stores.get()); + verify(taskChunkMapper, never()).selectById(anyLong()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTerminalFailedRecoveryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTerminalFailedRecoveryTest.java new file mode 100644 index 00000000..690b27bc --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTerminalFailedRecoveryTest.java @@ -0,0 +1,243 @@ +package com.nanri.aiimage.modules.appearancepatent.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.config.AppearancePatentProperties; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.StorageProperties; +import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentLlmClient; +import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentSubmitResultRequest; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +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.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +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.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +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.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 外观专利「补传恢复」入口（2026-09-17 线上任务 28459 的遗留项）。 + * + *

分片缺失导致组装 job 重试耗尽后，客户端补传缺口应能自动把该 job 重置重跑。 + * 该能力在 {@code TaskFileJobService} 里早就有了，但只有删除品牌模块接了入口， + * 外观专利没有——28459 补齐分片后仍需人工重置 job 才能出结果。 + */ +@ExtendWith(MockitoExtension.class) +class AppearancePatentTerminalFailedRecoveryTest { + + private static final Long TASK_ID = 28459L; + private static final Long RESULT_ID = 31490L; + private static final String MODULE_TYPE = "APPEARANCE_PATENT"; + + @Mock private LocalFileStorageService localFileStorageService; + @Mock private OssStorageService ossStorageService; + @Mock private StorageProperties storageProperties; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @Mock private AppearancePatentLlmClient llmClient; + @Mock private AppearancePatentTaskCacheService taskCacheService; + @Mock private AppearancePatentProperties properties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private PlatformTransactionManager transactionManager; + @Mock private DistributedJobLockService distributedJobLockService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private TaskProgressLightAssembler taskProgressLightAssembler; + @Mock private TransactionStatus transactionStatus; + + @InjectMocks private AppearancePatentTaskService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(transactionManager.getTransaction(any(TransactionDefinition.class))).thenReturn(transactionStatus); + lenient().doAnswer(inv -> null).when(transactionManager).commit(transactionStatus); + lenient().doAnswer(inv -> null).when(transactionManager).rollback(transactionStatus); + lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of()); + lenient().when(taskChunkMapper.selectCount(any())).thenReturn(0L); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null); + lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + lenient().when(transientPayloadStorageService.storeChunkPayload( + eq(MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString())) + .thenReturn("\"rustfs:task-chunk/appearance_patent/28459/hash/chunk-11.json\""); + } + + // ===== 直接覆盖恢复判定 ===== + + /** 已有成功的组装 job → 不恢复。 */ + @Test + void successfulAssembleJobSkipsRecovery() { + stubResultRow(); + when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true); + + invokeRecovery(); + + verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong()); + } + + /** 组装 job 不是「重试耗尽的终态失败」→ 不恢复。 */ + @Test + void nonTerminalFailedJobSkipsRecovery() { + stubResultRow(); + when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false); + when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false); + + invokeRecovery(); + + verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong()); + } + + /** 终态失败 + 分片已补传完整 → 重置 job 重新派发（本次要修的场景）。 */ + @Test + void terminalFailedJobWithCompleteChunksIsReset() { + stubResultRow(); + when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false); + when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true); + when(taskScopeStateMapper.selectCount(any())).thenReturn(1L); + + invokeRecovery(); + + verify(taskFileJobService).resetTerminalFailedForRecovery(TASK_ID, MODULE_TYPE, RESULT_ID); + } + + /** 终态失败但分片尚未补齐 → 不恢复（否则又会读到缺失分片再失败一次）。 */ + @Test + void terminalFailedJobWithIncompleteChunksIsNotReset() { + stubResultRow(); + when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false); + when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true); + when(taskScopeStateMapper.selectCount(any())).thenReturn(0L); + + invokeRecovery(); + + verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong()); + } + + /** 还没有结果行 → 不恢复。 */ + @Test + void missingResultRowSkipsRecovery() { + when(fileResultMapper.selectList(any())).thenReturn(List.of()); + + invokeRecovery(); + + verify(taskFileJobService, never()).isTerminalFailedAssembleJob(anyLong(), anyString(), anyLong()); + } + + /** taskId 非法 → 直接返回，不查库。 */ + @Test + void invalidTaskIdSkipsLookup() { + ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", 0L); + ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", (Object) null); + + verify(fileResultMapper, never()).selectList(any()); + } + + /** 恢复检查自身抛异常 → 吞掉，不影响补传结果（best-effort）。 */ + @Test + void recoveryFailureDoesNotPropagate() { + when(fileResultMapper.selectList(any())).thenThrow(new IllegalStateException("db down")); + + assertDoesNotThrow(this::invokeRecovery); + } + + // ===== 走完整提交路径 ===== + + /** 分片提交成功后触发恢复检查（接线正确）。 */ + @Test + void submitResultTriggersRecoveryCheck() { + stubRunningTask(); + stubResultRow(); + when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false); + when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true); + when(taskScopeStateMapper.selectCount(any())).thenReturn(1L); + + service.submitResult(TASK_ID, request()); + + verify(taskFileJobService).resetTerminalFailedForRecovery(TASK_ID, MODULE_TYPE, RESULT_ID); + } + + // ===== 辅助 ===== + + private void invokeRecovery() { + ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", TASK_ID); + } + + private void stubResultRow() { + FileResultEntity result = new FileResultEntity(); + result.setId(RESULT_ID); + result.setTaskId(TASK_ID); + result.setModuleType(MODULE_TYPE); + lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of(result)); + } + + private void stubRunningTask() { + FileTaskEntity task = new FileTaskEntity(); + task.setId(TASK_ID); + task.setModuleType(MODULE_TYPE); + task.setStatus("RUNNING"); + task.setUserId(1121L); + task.setResultJson("{\"parsedPayloadRef\":\"rustfs:task-parsed/x.json\",\"ownerInstanceId\":\"instance-a\"}"); + when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task); + } + + private AppearancePatentSubmitResultRequest request() { + AppearancePatentSubmitResultRequest request = new AppearancePatentSubmitResultRequest(); + request.setSubmissionId("appearance-patent-" + TASK_ID); + request.setChunkIndex(11); + request.setChunkTotal(500); + request.setDone(false); + return request; + } +} 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 index d70c3854..4b28ce67 100644 --- 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 @@ -33,6 +33,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -42,10 +43,12 @@ 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.anyLong; 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.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -397,4 +400,98 @@ class SimilarAsinTaskServiceChunkMergeLimitTest { verify(taskChunkMapper, times(1)).update(any(), any()); verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class)); } + + // ===== CAS 冲突处理（2026-09-17 线上任务 28459 遗留项，与外观专利同一口径） ===== + + /** 冲突重试耗尽：**保留**最后一次写入的对象作读兜底，异常带出期望/当前哈希。 */ + @Test + void casConflictExhaustedKeepsLastStoredObjectAsFallback() throws Exception { + TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A"); + chunk.setPayloadHash("hashA"); + stubConflictMerge(chunk, 0); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + Throwable ex = invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1"))); + + assertTrue(ex instanceof IllegalStateException, "实际: " + ex); + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1"); + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2"); + verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3"); + assertTrue(ex.getMessage().contains("hashA"), "需带出期望哈希，实际: " + ex.getMessage()); + assertTrue(ex.getMessage().contains("hashB"), "需带出当前哈希，实际: " + ex.getMessage()); + } + + /** 冲突后重试成功：被顶替的中间对象删除，最终对象写入行（不删）。 */ + @Test + void casConflictThenSuccessDeletesSupersededObjects() throws Exception { + TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A"); + chunk.setPayloadHash("hashA"); + stubConflictMerge(chunk, 0, 0, 1); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1"))); + + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1"); + verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2"); + verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3"); + verify(transientPayloadStorageService).deleteReplacedPayloadIfNeeded(eq("ptr:chunk-A"), eq("sibling-3")); + } + + /** 冲突时读回行上的当前哈希供定位（此前只有一句 conflict，线上无法定位）。 */ + @Test + void casConflictReadsBackCurrentHashForDiagnostics() throws Exception { + TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A"); + chunk.setPayloadHash("hashA"); + stubConflictMerge(chunk, 0); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1"))); + + verify(taskChunkMapper, times(3)).selectById(7L); + } + + /** 读回行失败不得掩盖原始冲突：异常信息里给出可读标记。 */ + @Test + void casConflictReadBackFailureDoesNotMaskConflict() throws Exception { + TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A"); + chunk.setPayloadHash("hashA"); + stubConflictMerge(chunk, 0); + when(taskChunkMapper.selectById(anyLong())).thenThrow(new IllegalStateException("db down")); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(9004L); + Throwable ex = invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1"))); + + assertTrue(ex.getMessage().contains("读取失败"), "实际: " + ex.getMessage()); + } + + /** 反射调用会包一层 InvocationTargetException，取根因以便断言业务异常。 */ + private Throwable invokeMergeExpectingFailure(FileTaskEntity task, List rows) throws Exception { + try { + invokeMerge(service, task, "hashA", 1, rows); + return null; + } catch (java.lang.reflect.InvocationTargetException ex) { + return ex.getCause(); + } + } + + /** 冲突场景公共 stub：每次尝试写一个不同的版本化对象，update 按传入序列返回。 */ + private void stubConflictMerge(TaskChunkEntity chunk, Integer... updateResults) throws Exception { + when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk)); + when(taskChunkMapper.selectOne(any())).thenReturn(chunk); + when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString())) + .thenReturn(rowsJson(List.of(row("r0", "B0A0000000", "存量行")))); + AtomicInteger stores = new AtomicInteger(); + when(transientPayloadStorageService.storeChunkPayloadVersioned( + anyString(), any(), anyString(), any(), anyString())) + .thenAnswer(inv -> "sibling-" + stores.incrementAndGet()); + when(taskChunkMapper.update(any(), any())).thenReturn(updateResults[0], java.util.Arrays.copyOfRange(updateResults, 1, updateResults.length)); + + TaskChunkEntity current = chunk(7L, "hashB", 1, "ptr:chunk-A"); + current.setPayloadHash("hashB"); + lenient().when(taskChunkMapper.selectById(7L)).thenReturn(current); + } }