task-27: chunk 接收改为原子插入/幂等 upsert，去掉先查后插

persistResultChunk 去除提交前的 findResultChunk 预查，直接 store payload + insert，唯一索引 uk_task_scope_chunk 兜底幂等：重复提交命中 DuplicateKeyException 后重查 winner 校验并清理本次重存 payload；空分片（无可处理数据）在落库前拒绝，不创建 RustFS payload 与分片行。
2026-08-29 19:11:10 +08:00
parent 0092092dbd
commit 23eddde770
3 changed files with 644 additions and 8 deletions
@@ -1257,15 +1257,11 @@ public class ShopDataCrawlTaskService {
        validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);

        List<ShopDataCrawlCountryResultDto> countryResults = copyCountryResults(incoming.getCountryResults());
+        if (!hasProcessableChunkData(incoming.getCountryResults())) {
+            throw new BusinessException("店铺数据抓取结果分片内容为空，拒绝接收");
+        }
        String payloadJson = writeJson(countryResults, "序列化店铺数据抓取结果分片失败");
        String payloadHash = DigestUtil.sha256Hex(payloadJson);
-        TaskChunkEntity existing = findResultChunk(taskId, scopeHash, chunkIndex);
-        if (existing != null) {
-            validateExistingChunk(existing, chunkTotal, payloadHash);
-            int receivedChunkCount = countResultChunks(taskId, scopeHash);
-            persistResultScope(taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount);
-            return new ResultChunkReceipt(scopeHash, chunkTotal, receivedChunkCount >= chunkTotal);
-        }

        ensureRustfsPayloadStorageEnabled();
        String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(
@@ -2349,6 +2345,30 @@ public class ShopDataCrawlTaskService {
        return copy;
    }

+    /**
+     * 分片是否含可处理数据：与 copyCountryResults 的拷贝语义一致 ——
+     * 任一国家含至少一个非空行（国家名非空白且行内容非空）即视为有数据。
+     */
+    private boolean hasProcessableChunkData(List<ShopDataCrawlCountryResultDto> results) {
+        if (results == null || results.isEmpty()) {
+            return false;
+        }
+        for (ShopDataCrawlCountryResultDto source : results) {
+            if (source == null || blank(source.getCountry())) {
+                continue;
+            }
+            if (source.getItems() == null) {
+                continue;
+            }
+            for (ShopDataCrawlRowDto sourceRow : source.getItems()) {
+                if (sourceRow != null && !rowEmpty(sourceRow)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
    private ShopDataCrawlRowDto copyRow(ShopDataCrawlRowDto source) {
        ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
        row.setDate(trim(source.getDate()));
@@ -0,0 +1,603 @@
+package com.nanri.aiimage.modules.shopdatacrawl.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nanri.aiimage.common.exception.BusinessException;
+import com.nanri.aiimage.config.InstanceMetadata;
+import com.nanri.aiimage.config.TaskPressureProperties;
+import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
+import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlCountryResultDto;
+import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
+import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
+import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest;
+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.TaskProgressSnapshotService;
+import com.nanri.aiimage.modules.task.service.TaskResultItemService;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.apache.ibatis.session.Configuration;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.dao.DuplicateKeyException;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+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.ArgumentMatchers.nullable;
+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 27：将 chunk 接收改为原子插入/幂等 upsert，减少先查后插。
+ * persistResultChunk 去掉首包提交前的 findResultChunk 预查，改为直接
+ * store payload + insert；唯一索引 uk_task_scope_chunk 兜底幂等 ——
+ * 重复提交（同内容）命中 DuplicateKeyException 后重查 winner 校验，
+ * 并清理本次重存的 payload；空分片（无可处理数据）在落库前被拒绝，
+ * 不创建 RustFS payload 与分片行。
+ */
+@ExtendWith(MockitoExtension.class)
+class ShopDataCrawlChunkUpsertTest {
+
+    private static final String MODULE_TYPE = "SHOP_DATA_CRAWL";
+    private static final String SHOP_NAME = "Demo Shop";
+
+    @BeforeAll
+    static void initializeMybatisMetadata() {
+        MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), "");
+        TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
+        TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
+    }
+
+    @Mock private FileTaskMapper fileTaskMapper;
+    @Mock private FileResultMapper fileResultMapper;
+    @Mock private ShopDataCrawlResolveService shopDataCrawlResolveService;
+    @Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService;
+    @Mock private ShopDataCrawlTaskCacheService taskCacheService;
+    @Mock private OssStorageService ossStorageService;
+    @Mock private ZiniaoShopSwitchService ziniaoShopSwitchService;
+    @Mock private TaskPressureProperties taskPressureProperties;
+    @Mock private TaskFileJobService taskFileJobService;
+    @Mock private TaskResultItemService taskResultItemService;
+    @Mock private TaskProgressSnapshotService taskProgressSnapshotService;
+    @Mock private TaskDistributedLockService taskDistributedLockService;
+    @Mock private TaskChunkMapper taskChunkMapper;
+    @Mock private TaskScopeStateMapper taskScopeStateMapper;
+    @Mock private TransientPayloadStorageService transientPayloadStorageService;
+    @Mock private InstanceMetadata instanceMetadata;
+    @Mock private ShopDataCrawlDailyFileService dailyFileService;
+    @Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules();
+
+    private ShopDataCrawlTaskService service;
+
+    private final List<TaskChunkEntity> storedChunks = new ArrayList<>();
+    private final List<TaskScopeStateEntity> storedScopes = new ArrayList<>();
+    private final Map<String, String> rustfsPayloads = new LinkedHashMap<>();
+    private FileTaskEntity task;
+    private FileResultEntity result;
+    private int nextPayloadId;
+    /** 模拟依赖故障开关：RustFS 存储失败 / 数据库插入失败 / 唯一键竞态失败。 */
+    private boolean storeFails;
+    private boolean insertFails;
+    private boolean insertDuplicate;
+
+    @BeforeEach
+    void configureStorage() {
+        service = new ShopDataCrawlTaskService(
+                fileTaskMapper,
+                fileResultMapper,
+                shopDataCrawlResolveService,
+                excelAssemblyService,
+                taskCacheService,
+                ossStorageService,
+                ziniaoShopSwitchService,
+                objectMapper,
+                taskPressureProperties,
+                taskFileJobService,
+                taskResultItemService,
+                taskProgressSnapshotService,
+                taskDistributedLockService,
+                taskChunkMapper,
+                taskScopeStateMapper,
+                transientPayloadStorageService,
+                instanceMetadata,
+                dailyFileService,
+                null);
+
+        storedChunks.clear();
+        storedScopes.clear();
+        rustfsPayloads.clear();
+        nextPayloadId = 0;
+        storeFails = false;
+        insertFails = false;
+        insertDuplicate = false;
+
+        lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
+        lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong()))
+                .thenReturn(mock(TaskDistributedLockService.LockHandle.class));
+        lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of());
+        lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
+        lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of());
+        lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null);
+        lenient().when(excelAssemblyService.countRows(any())).thenReturn(1);
+
+        lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> {
+            Long taskId = invocation.getArgument(0);
+            return task != null && Objects.equals(taskId, task.getId()) ? task : null;
+        });
+        lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
+        lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> {
+            Long resultId = invocation.getArgument(0);
+            return result != null && Objects.equals(resultId, result.getId()) ? result : null;
+        });
+        lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<FileResultEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of();
+        });
+        lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1);
+
+        lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> {
+            String value = invocation.getArgument(0);
+            return value == null ? "" : value.trim();
+        });
+
+        configureTransientPayloadStorage();
+        configureChunkMapper();
+        configureScopeMapper();
+    }
+
+    @Test
+    void test_task_027_chunk_normal_default_path() {
+        // 正常输入：首个分片直接原子插入，不再先查后插——
+        // taskChunkMapper 不出现预查 selectOne，分片行与 RustFS payload 各落一份。
+        givenRunningTask(1001L, 2001L);
+
+        service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
+
+        assertEquals(1, storedChunks.size());
+        assertEquals("rustfs:payload-1", storedChunks.get(0).getPayloadJson());
+        assertEquals(1, rustfsPayloads.size());
+        assertEquals(-1, result.getSuccess(), "1/2 未齐，任务继续运行");
+        assertEquals("RUNNING", task.getStatus());
+        verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class));
+        verify(taskChunkMapper, never()).selectOne(any());
+        verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString());
+    }
+
+    @Test
+    void test_task_027_chunk_normal_multiple_items() {
+        // 批量场景：乱序分片（2/2 先到、1/2 后到）全部原子插入，合并结果按 chunk_index 稳定，
+        // 全程无预查 selectOne；齐集时触发组装任务。
+        givenRunningTask(1002L, 2002L);
+
+        service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
+
+        assertEquals(-1, result.getSuccess());
+        assertEquals(1, storedChunks.size());
+        verify(taskFileJobService, never()).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString());
+
+        service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
+
+        assertEquals(1, result.getSuccess());
+        assertEquals(2, storedChunks.size());
+        assertEquals(2, rustfsPayloads.size());
+        assertTrue(task.getResultJson().indexOf("B001") < task.getResultJson().indexOf("B002"),
+                "合并结果按 chunk_index 顺序稳定");
+        verify(taskChunkMapper, never()).selectOne(any());
+        verify(taskFileJobService).enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(),
+                "task:" + task.getId() + ":owner:instance-a");
+        verify(excelAssemblyService, never()).writeWorkbook(any(), any());
+        verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any());
+        verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE));
+    }
+
+    @Test
+    void test_task_027_chunk_normal_repeated_operation_is_idempotent() {
+        // 重复执行：同一分片重试靠唯一索引 uk_task_scope_chunk 兜底幂等——
+        // 不产生重复分片行、重试重存的 payload 立即清理，insert 共两次（首次成功+重试被唯一键拒绝）。
+        givenRunningTask(1003L, 2003L);
+        ShopDataCrawlSubmitResultRequest request = request(chunk(1, 2, "DE", row("2026-07-25", "B001")));
+
+        service.submitResult(task.getId(), request);
+        service.submitResult(task.getId(), request);
+
+        assertEquals(1, storedChunks.size(), "重复提交不产生重复分片行");
+        assertEquals(1, rustfsPayloads.size(), "重试重存的 payload 被清理");
+        verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class));
+        verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2");
+        assertEquals(-1, result.getSuccess());
+    }
+
+    @Test
+    void test_task_027_chunk_boundary_empty_input() {
+        // 空输入：无可处理数据的分片（空 items / 空白国家）在落库前拒绝，不创建 payload 与分片行。
+        givenRunningTask(1004L, 2004L);
+
+        ShopDataCrawlShopPayloadDto emptyItems = legacyChunk(false, "DE", null);
+        emptyItems.setChunkIndex(1);
+        emptyItems.setChunkTotal(1);
+        emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of())));
+
+        BusinessException error = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(emptyItems)));
+        assertTrue(error.getMessage().contains("内容为空"));
+
+        ShopDataCrawlShopPayloadDto blankCountry = chunk(1, 1, "DE", row("2026-07-25", "B001"));
+        blankCountry.setCountryResults(List.of(countryWithItems("   ", List.of(row("2026-07-25", "B001")))));
+
+        assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(blankCountry)));
+
+        assertEquals(0, storedChunks.size(), "空分片不落库");
+        assertEquals(0, rustfsPayloads.size(), "空分片不写 RustFS");
+        verify(taskChunkMapper, never()).insert(any(TaskChunkEntity.class));
+    }
+
+    @Test
+    void test_task_027_chunk_boundary_single_item() {
+        // 单元素：单分片 1/1 原子插入即齐集，直接合并成功，无预查 selectOne。
+        givenRunningTask(1005L, 2005L);
+
+        service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001"))));
+
+        assertEquals(1, result.getSuccess());
+        assertEquals(1, storedChunks.size());
+        assertEquals(1, rustfsPayloads.size());
+        assertEquals(1, storedScopes.size());
+        assertEquals(1, storedScopes.get(0).getReceivedChunkCount());
+        assertEquals(1, storedScopes.get(0).getCompleted());
+        assertTrue(task.getResultJson().contains("B001"));
+        verify(taskChunkMapper, never()).selectOne(any());
+    }
+
+    @Test
+    void test_task_027_chunk_boundary_limit_and_overflow() {
+        // 上限/超限：chunk_index 超 chunk_total 拒绝；5/5 分片全量到达时计数一致、顺序稳定、任务完成，
+        // 期间全部为原子插入（无预查），无重复对象累积。
+        givenRunningTask(1006L, 2006L);
+
+        BusinessException overflow = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(2, 1, "DE", row("2026-07-25", "B001")))));
+        assertTrue(overflow.getMessage().contains("chunk_index"));
+
+        String[] countries = {"UK", "DE", "FR", "ES", "IT"};
+        String[] dates = {"2026-07-25", "2026-07-26", "2026-07-27", "2026-07-28", "2026-07-29"};
+        for (int i = 1; i <= 5; i++) {
+            service.submitResult(task.getId(), request(chunk(i, 5, countries[i - 1], row(dates[i - 1], "B00" + i))));
+            if (i < 5) {
+                assertEquals(-1, result.getSuccess(), i + "/5 未齐");
+            }
+        }
+
+        assertEquals(1, result.getSuccess(), "5/5 齐集后任务完成");
+        assertEquals(5, storedChunks.size());
+        assertEquals(5, storedScopes.get(0).getReceivedChunkCount());
+        assertEquals(1, storedScopes.get(0).getCompleted());
+        String json = task.getResultJson();
+        for (int i = 1; i < 5; i++) {
+            assertTrue(json.indexOf("B00" + i) < json.indexOf("B00" + (i + 1)), "合并顺序按 chunk_index 稳定");
+        }
+        verify(taskChunkMapper, times(5)).insert(any(TaskChunkEntity.class));
+        verify(taskChunkMapper, never()).selectOne(any());
+    }
+
+    @Test
+    void test_task_027_chunk_invalid_input_rejected() {
+        // 非法参数：非正 chunk_index/chunk_total、跨分片改 chunk_total、同 index 不同内容 →
+        // 项目约定异常及可识别消息；重存 payload 不残留。
+        givenRunningTask(1007L, 2007L);
+
+        BusinessException zeroIndex = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001")))));
+        assertTrue(zeroIndex.getMessage().contains("chunk_index"));
+
+        BusinessException zeroTotal = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001")))));
+        assertTrue(zeroTotal.getMessage().contains("chunk_total"));
+
+        service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
+
+        BusinessException changedTotal = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(2, 3, "UK", row("2026-07-26", "B002")))));
+        assertTrue(changedTotal.getMessage().contains("chunk_total"));
+
+        BusinessException differentContent = assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B099")))));
+        assertTrue(differentContent.getMessage().contains("不同内容"));
+
+        assertEquals(1, storedChunks.size(), "非法输入不产生额外分片行");
+        assertEquals(1, rustfsPayloads.size(), "同 index 不同内容的重存 payload 被清理");
+        verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class));
+        verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2");
+    }
+
+    @Test
+    void test_task_027_chunk_dependency_failure_releases_resources() {
+        // 依赖失败：RustFS 存储异常/插入异常/唯一键竞态 winner 丢失 —— 临时 payload 清理、
+        // 锁释放（故障后可重试成功）、不残留分片行。
+        givenRunningTask(1008L, 2008L);
+
+        storeFails = true;
+        assertThrows(RuntimeException.class, () ->
+                service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))));
+        assertEquals(0, storedChunks.size());
+        assertEquals(0, rustfsPayloads.size());
+        storeFails = false;
+
+        insertFails = true;
+        assertThrows(RuntimeException.class, () ->
+                service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))));
+        assertEquals(0, storedChunks.size(), "插入失败不残留分片行");
+        assertEquals(0, rustfsPayloads.size(), "插入失败清理已存 payload");
+        verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-1");
+        insertFails = false;
+
+        service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
+        assertEquals(1, storedChunks.size(), "故障恢复后重试成功");
+        assertEquals(1, rustfsPayloads.size());
+        assertEquals(-1, result.getSuccess());
+
+        insertDuplicate = true;
+        assertThrows(BusinessException.class, () ->
+                service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002")))));
+        insertDuplicate = false;
+
+        service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
+
+        assertEquals(1, result.getSuccess(), "竞态失败后可重试成功");
+        assertEquals(2, storedChunks.size());
+        assertEquals(2, rustfsPayloads.size());
+    }
+
+    private void configureTransientPayloadStorage() {
+        lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
+        lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
+                anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> {
+            if (storeFails) {
+                throw new RuntimeException("rustfs store down");
+            }
+            String pointer = "rustfs:payload-" + (++nextPayloadId);
+            rustfsPayloads.put(pointer, invocation.getArgument(4));
+            return pointer;
+        });
+        lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> {
+            String value = invocation.getArgument(0);
+            if (value == null) {
+                return null;
+            }
+            return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:")
+                    ? value : null;
+        });
+        lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> {
+            String pointer = invocation.getArgument(0);
+            String payload = rustfsPayloads.get(pointer);
+            if (payload == null) {
+                throw new IllegalStateException("missing test RustFS payload: " + pointer);
+            }
+            return payload;
+        });
+        lenient().doAnswer(invocation -> {
+            rustfsPayloads.remove(invocation.getArgument(0));
+            return null;
+        }).when(transientPayloadStorageService).deletePayloadIfPresent(any());
+    }
+
+    private void configureChunkMapper() {
+        lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> {
+            if (insertDuplicate) {
+                throw new DuplicateKeyException("duplicate chunk key (race)");
+            }
+            if (insertFails) {
+                throw new RuntimeException("db down");
+            }
+            TaskChunkEntity chunk = invocation.getArgument(0);
+            boolean duplicate = storedChunks.stream().anyMatch(existing ->
+                    Objects.equals(existing.getTaskId(), chunk.getTaskId())
+                            && Objects.equals(existing.getScopeHash(), chunk.getScopeHash())
+                            && Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex()));
+            if (duplicate) {
+                throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex());
+            }
+            chunk.setId((long) storedChunks.size() + 1L);
+            storedChunks.add(chunk);
+            return 1;
+        });
+        lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            String scopeHash = queryScopeHash(query);
+            Integer chunkIndex = queryInteger(query);
+            return storedChunks.stream()
+                    .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex))
+                    .findFirst()
+                    .orElse(null);
+        });
+        lenient().when(taskChunkMapper.selectCount(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            String scopeHash = queryScopeHash(query);
+            return storedChunks.stream()
+                    .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null))
+                    .count();
+        });
+        lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            String scopeHash = queryScopeHash(query);
+            return storedChunks.stream()
+                    .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null))
+                    .sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex))
+                    .toList();
+        });
+    }
+
+    private void configureScopeMapper() {
+        lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
+            TaskScopeStateEntity scope = invocation.getArgument(0);
+            scope.setId((long) storedScopes.size() + 1L);
+            storedScopes.add(scope);
+            return 1;
+        });
+        lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<TaskScopeStateEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            String scopeHash = queryScopeHash(query);
+            return storedScopes.stream()
+                    .filter(scope -> Objects.equals(taskId, scope.getTaskId()))
+                    .filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash()))
+                    .findFirst()
+                    .orElse(null);
+        });
+        lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenReturn(1);
+        lenient().when(taskScopeStateMapper.delete(any())).thenAnswer(invocation -> {
+            LambdaQueryWrapper<TaskScopeStateEntity> query = invocation.getArgument(0);
+            Long taskId = queryLong(query);
+            int before = storedScopes.size();
+            storedScopes.removeIf(scope -> taskId == null || Objects.equals(taskId, scope.getTaskId()));
+            return before - storedScopes.size();
+        });
+    }
+
+    private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) {
+        return (taskId == null || Objects.equals(taskId, chunk.getTaskId()))
+                && (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash()))
+                && (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex()));
+    }
+
+    private Long queryLong(LambdaQueryWrapper<?> query) {
+        query.getSqlSegment();
+        return query.getParamNameValuePairs().values().stream()
+                .filter(Long.class::isInstance)
+                .map(Long.class::cast)
+                .findFirst()
+                .orElse(null);
+    }
+
+    private Integer queryInteger(LambdaQueryWrapper<?> query) {
+        query.getSqlSegment();
+        return query.getParamNameValuePairs().values().stream()
+                .filter(Integer.class::isInstance)
+                .map(Integer.class::cast)
+                .findFirst()
+                .orElse(null);
+    }
+
+    private String queryScopeHash(LambdaQueryWrapper<?> query) {
+        query.getSqlSegment();
+        return query.getParamNameValuePairs().values().stream()
+                .filter(String.class::isInstance)
+                .map(String.class::cast)
+                .filter(value -> value.length() == 64)
+                .findFirst()
+                .orElse(null);
+    }
+
+    private void givenRunningTask(long taskId, long resultId) {
+        task = new FileTaskEntity();
+        task.setId(taskId);
+        task.setUserId(7L);
+        task.setModuleType(MODULE_TYPE);
+        task.setStatus("RUNNING");
+        task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
+        task.setResultJson("[]");
+        task.setCreatedAt(LocalDateTime.now());
+        task.setUpdatedAt(LocalDateTime.now());
+
+        result = new FileResultEntity();
+        result.setId(resultId);
+        result.setTaskId(taskId);
+        result.setUserId(7L);
+        result.setModuleType(MODULE_TYPE);
+        result.setSourceFilename(SHOP_NAME);
+        result.setSourceFileUrl("shop-1");
+        result.setSuccess(-1);
+        result.setCreatedAt(LocalDateTime.now());
+    }
+
+    private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) {
+        ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest();
+        request.setShops(List.of(payload));
+        return request;
+    }
+
+    private ShopDataCrawlShopPayloadDto chunk(int chunkIndex,
+                                              int chunkTotal,
+                                              String country,
+                                              ShopDataCrawlRowDto row) {
+        ShopDataCrawlShopPayloadDto payload = legacyChunk(false, country, row);
+        payload.setChunkIndex(chunkIndex);
+        payload.setChunkTotal(chunkTotal);
+        return payload;
+    }
+
+    private ShopDataCrawlShopPayloadDto legacyChunk(boolean shopDone,
+                                                    String country,
+                                                    ShopDataCrawlRowDto row) {
+        ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto();
+        payload.setShopName(SHOP_NAME);
+        if (row != null) {
+            payload.setCountryResults(List.of(country(country, row)));
+        }
+        payload.setShopDone(shopDone);
+        return payload;
+    }
+
+    private ShopDataCrawlCountryResultDto country(String country, ShopDataCrawlRowDto row) {
+        return countryWithItems(country, List.of(row));
+    }
+
+    private ShopDataCrawlCountryResultDto countryWithItems(String country, List<ShopDataCrawlRowDto> items) {
+        ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto();
+        result.setCountry(country);
+        result.setItems(items);
+        return result;
+    }
+
+    private ShopDataCrawlRowDto row(String date, String asin) {
+        ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
+        row.setDate(date);
+        row.setAsin(asin);
+        row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg");
+        row.setInventorySales("10");
+        row.setSalesRank("20");
+        row.setPageViews("30");
+        row.setUnitsSold("40");
+        row.setPrice("50");
+        row.setRecommendedOffer("60");
+        return row;
+    }
+}
@@ -35,6 +35,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.dao.DuplicateKeyException;

 import java.time.LocalDateTime;
 import java.util.ArrayList;
@@ -213,7 +214,10 @@ class ShopDataCrawlTaskServiceChunkTest {

        assertEquals(1, storedChunks.size());
        assertEquals(1, rustfsPayloads.size());
-        verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class));
+        // 原子插入路径：首次 insert 成功，重试命中唯一索引 uk_task_scope_chunk 抛键冲突被吸收；
+        // 重试重存的 payload 立即清理，不产生重复分片。
+        verify(taskChunkMapper, times(2)).insert(any(TaskChunkEntity.class));
+        verify(transientPayloadStorageService).deletePayloadIfPresent("rustfs:payload-2");
        assertEquals(-1, result.getSuccess());
    }

@@ -388,6 +392,15 @@ class ShopDataCrawlTaskServiceChunkTest {
    private void configureChunkMapper() {
        lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> {
            TaskChunkEntity chunk = invocation.getArgument(0);
+            // 与 V30 的 uk_task_scope_chunk 唯一索引一致：同 (task_id, scope_hash, chunk_index) 重复插入抛键冲突。
+            boolean duplicate = storedChunks.stream().anyMatch(existing ->
+                    Objects.equals(existing.getTaskId(), chunk.getTaskId())
+                            && Objects.equals(existing.getScopeHash(), chunk.getScopeHash())
+                            && Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex()));
+            if (duplicate) {
+                throw new DuplicateKeyException("duplicate chunk key: "
+                        + chunk.getScopeHash() + "/" + chunk.getChunkIndex());
+            }
            chunk.setId((long) storedChunks.size() + 1L);
            storedChunks.add(chunk);
            return 1;