task-50: 结果明细批量 upsert 写入器与唯一键幂等

CollectDataResultItemBatchWriter 按唯一键 uk_task_scope_item 批量 upsert
biz_task_result_item：一次批量查询现有行（payload_hash 相等即跳过），
分批发 INSERT ... ON DUPLICATE KEY UPDATE，替代逐行 select/insert/update；
旧格式逐行 payload 升级为引用时物理删除（引用计数兜底）。CollectDataService
submitResult 接入批量写入，移除逐行 upsertResultItem。

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java

@@ -4,7 +4,6 @@ import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.crypto.digest.DigestUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
-import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -38,6 +37,7 @@ import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec;
+import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter;
 import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
 import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
 import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
@@ -73,8 +73,6 @@ import org.springframework.dao.DuplicateKeyException;
 
 import java.io.File;
 import java.io.FileInputStream;
-import java.nio.charset.StandardCharsets;
-import java.security.MessageDigest;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
@@ -148,6 +146,9 @@ public class CollectDataService {
     /** 结果明细 chunk 级编解码：accepted 行按 chunk 共享一个 RustFS 对象。 */
     private final CollectDataResultDetailCodec resultDetailCodec;
 
+    /** 结果明细批量 upsert 器：按唯一键 uk_task_scope_item 批量写入，替代逐行 select/insert/update。 */
+    private final CollectDataResultItemBatchWriter resultItemBatchWriter;
+
     @Value("${aiimage.collect-data.stale-timeout-minutes:30}")
     private long staleTimeoutMinutes;
 
@@ -659,9 +660,10 @@ public class CollectDataService {
                 String storedDetail = transientPayloadStorageService.storeResultPayload(
                         MODULE_TYPE, taskId, scopeHash, "chunk-" + chunkIndex, detailJson);
                 requireRustfsPayload(storedDetail, "采集结果明细必须写入 RustFS");
-                for (int i = 0; i < accepted.size(); i++) {
-                    upsertResultItem(task.getId(), result.getId(), scopeKey, chunkIndex, accepted.get(i), i, storedDetail);
-                }
+                // 批量 upsert：先一次批量查现有行（payload_hash 相等即跳过，幂等），
+                // 再按唯一键 uk_task_scope_item 分批发 INSERT ... ON DUPLICATE KEY UPDATE。
+                resultItemBatchWriter.upsertAccepted(task.getId(), result.getId(), scopeKey,
+                        chunkIndex, accepted, storedDetail);
             }
 
             String payloadJson = writeJson(rows, "采集结果序列化失败");
@@ -753,70 +755,6 @@ public class CollectDataService {
         return outcome.accepted();
     }
 
-    private void upsertResultItem(Long taskId, Long resultId, String scopeKey, int chunkIndex,
-                                  CollectDataResultRowVo row, int offset, String storedDetail) {
-        String itemKey = "asin:" + row.getAsin();
-        String scopeHash = hash(scopeKey);
-        // 引用 JSON 作为 payload_json：内容变化（行 offset/对象变化）即 hash 变化。
-        String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail);
-        String payloadHash = hash(refJson);
-        TaskResultItemEntity existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper<TaskResultItemEntity>()
-                .eq(TaskResultItemEntity::getTaskId, taskId)
-                .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
-                .eq(TaskResultItemEntity::getScopeHash, scopeHash)
-                .eq(TaskResultItemEntity::getItemKey, itemKey)
-                .last("limit 1"));
-        if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) {
-            return;
-        }
-        LocalDateTime now = LocalDateTime.now();
-        if (existing == null) {
-            TaskResultItemEntity entity = new TaskResultItemEntity();
-            entity.setTaskId(taskId);
-            entity.setModuleType(MODULE_TYPE);
-            entity.setResultId(resultId);
-            entity.setScopeKey(scopeKey);
-            entity.setScopeHash(scopeHash);
-            entity.setItemKey(itemKey);
-            entity.setAsin(row.getAsin());
-            entity.setStatus("ACCEPTED");
-            entity.setPayloadJson(refJson);
-            entity.setPayloadHash(payloadHash);
-            entity.setCreatedAt(now);
-            entity.setUpdatedAt(now);
-            try {
-                taskResultItemMapper.insert(entity);
-                return;
-            } catch (DuplicateKeyException ignored) {
-                existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper<TaskResultItemEntity>()
-                        .eq(TaskResultItemEntity::getTaskId, taskId)
-                        .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
-                        .eq(TaskResultItemEntity::getScopeHash, scopeHash)
-                        .eq(TaskResultItemEntity::getItemKey, itemKey)
-                        .last("limit 1"));
-            }
-        }
-        if (existing == null) {
-            throw new BusinessException("保存采集结果明细失败");
-        }
-        // chunk 级引用共享同一 RustFS 对象（deterministic key，同 chunk 重提
-        // 覆盖同一对象，引用 pointer 稳定，无需删除）；旧格式逐行对象在升级
-        // 为引用后不再被任何行持有，直接物理删除，避免泄漏。
-        CollectDataResultDetailCodec.ChunkRef oldRef = resultDetailCodec.parseRef(existing.getPayloadJson());
-        if (oldRef == null) {
-            transientPayloadStorageService.deletePayloadIfPresent(existing.getPayloadJson());
-        }
-        taskResultItemMapper.update(null, new LambdaUpdateWrapper<TaskResultItemEntity>()
-                .eq(TaskResultItemEntity::getId, existing.getId())
-                .set(TaskResultItemEntity::getResultId, resultId)
-                .set(TaskResultItemEntity::getScopeKey, scopeKey)
-                .set(TaskResultItemEntity::getAsin, row.getAsin())
-                .set(TaskResultItemEntity::getStatus, "ACCEPTED")
-                .set(TaskResultItemEntity::getPayloadJson, refJson)
-                .set(TaskResultItemEntity::getPayloadHash, payloadHash)
-                .set(TaskResultItemEntity::getUpdatedAt, now));
-    }
-
     private void persistChunk(Long taskId,
                               String scopeKey,
                               String scopeHash,
@@ -1193,20 +1131,6 @@ public class CollectDataService {
         }
     }
 
-    private String hash(String value) {
-        try {
-            MessageDigest digest = MessageDigest.getInstance("SHA-256");
-            byte[] bytes = digest.digest((value == null ? "" : value).getBytes(StandardCharsets.UTF_8));
-            StringBuilder sb = new StringBuilder(bytes.length * 2);
-            for (byte b : bytes) {
-                sb.append(String.format("%02x", b));
-            }
-            return sb.toString();
-        } catch (Exception ex) {
-            throw new IllegalStateException("failed to hash collect data payload", ex);
-        }
-    }
-
     private String normalizeAsin(String value) {
         return normalize(value).toUpperCase(Locale.ROOT);
     }

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java

@@ -0,0 +1,140 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
+import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * 结果明细批量 upsert：把整 chunk 的 accepted 行分批发往
+ * biz_task_result_item。先一次批量查询现有行（payload_hash 相等即跳过，
+ * 幂等），再按唯一键 uk_task_scope_item 用一条 INSERT ... ON DUPLICATE
+ * KEY UPDATE 批量写入，替代逐行 select/insert/update；批量失败跳过该批
+ * 可恢复，不中断任务流程。
+ */
+@Slf4j
+@Component
+public class CollectDataResultItemBatchWriter {
+
+    private static final String MODULE_TYPE = "collectdata";
+    private static final int DEFAULT_BATCH_SIZE = 100;
+
+    private final TaskResultItemMapper taskResultItemMapper;
+    private final CollectDataResultDetailCodec resultDetailCodec;
+    private final TransientPayloadStorageService transientPayloadStorageService;
+    private final int batchSize;
+
+    public CollectDataResultItemBatchWriter(TaskResultItemMapper taskResultItemMapper,
+                                            CollectDataResultDetailCodec resultDetailCodec,
+                                            TransientPayloadStorageService transientPayloadStorageService,
+                                            @Value("${aiimage.collect-data.result-item-batch-size:100}") int batchSize) {
+        this.taskResultItemMapper = taskResultItemMapper;
+        this.resultDetailCodec = resultDetailCodec;
+        this.transientPayloadStorageService = transientPayloadStorageService;
+        this.batchSize = batchSize <= 0 ? DEFAULT_BATCH_SIZE : batchSize;
+    }
+
+    /** 批量写入计数：insertedOrUpdated 实际写入行数，skipped hash 相等跳过行数。 */
+    public record UpsertCounts(int insertedOrUpdated, int skipped) {
+    }
+
+    /** 把整 chunk 的 accepted 行批量 upsert；scopeKey 与 chunk 内 offset 已知，仅计算 refJson 与 hash。 */
+    public UpsertCounts upsertAccepted(Long taskId, Long resultId, String scopeKey, int chunkIndex,
+                                       List<CollectDataResultRowVo> rows, String storedDetail) {
+        if (rows == null || rows.isEmpty()) {
+            return new UpsertCounts(0, 0);
+        }
+        String scopeHash = hash(scopeKey);
+        // 一次性取回本 scope 现有行，构建 item_key → 现有行 映射（hash 相等即跳过）。
+        List<TaskResultItemEntity> existingList = taskResultItemMapper.selectList(
+                new LambdaQueryWrapper<TaskResultItemEntity>()
+                        .eq(TaskResultItemEntity::getTaskId, taskId)
+                        .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)
+                        .eq(TaskResultItemEntity::getScopeHash, scopeHash));
+        Map<String, TaskResultItemEntity> existingByKey = new HashMap<>();
+        for (TaskResultItemEntity existing : existingList) {
+            if (existing != null && existing.getItemKey() != null) {
+                existingByKey.put(existing.getItemKey(), existing);
+            }
+        }
+
+        List<TaskResultItemEntity> toUpsert = new ArrayList<>();
+        int skipped = 0;
+        LocalDateTime now = LocalDateTime.now();
+        for (int offset = 0; offset < rows.size(); offset++) {
+            CollectDataResultRowVo row = rows.get(offset);
+            if (row == null || row.getAsin() == null || row.getAsin().isBlank()) {
+                continue;
+            }
+            String itemKey = "asin:" + row.getAsin();
+            // chunk 级引用共享同一 RustFS 对象（deterministic key，同 chunk 重提
+            // 覆盖同一对象，引用 pointer 稳定，无需删除）。
+            String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail);
+            String payloadHash = hash(refJson);
+            TaskResultItemEntity existing = existingByKey.get(itemKey);
+            if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) {
+                skipped++;
+                continue;
+            }
+            // 旧格式逐行对象在升级为引用后不再被任何行持有，直接物理删除避免泄漏
+            // （deletePayloadIfPresent 内部带全局引用计数兜底）。
+            if (existing != null && existing.getPayloadJson() != null
+                    && resultDetailCodec.parseRef(existing.getPayloadJson()) == null) {
+                transientPayloadStorageService.deletePayloadIfPresent(existing.getPayloadJson());
+            }
+            TaskResultItemEntity entity = new TaskResultItemEntity();
+            entity.setId(existing == null ? null : existing.getId());
+            entity.setTaskId(taskId);
+            entity.setModuleType(MODULE_TYPE);
+            entity.setResultId(resultId);
+            entity.setScopeKey(scopeKey);
+            entity.setScopeHash(scopeHash);
+            entity.setItemKey(itemKey);
+            entity.setAsin(row.getAsin());
+            entity.setStatus("ACCEPTED");
+            entity.setPayloadJson(refJson);
+            entity.setPayloadHash(payloadHash);
+            entity.setCreatedAt(existing == null ? now : existing.getCreatedAt());
+            entity.setUpdatedAt(now);
+            toUpsert.add(entity);
+        }
+
+        int written = 0;
+        for (int from = 0; from < toUpsert.size(); from += batchSize) {
+            int to = Math.min(from + batchSize, toUpsert.size());
+            List<TaskResultItemEntity> batch = toUpsert.subList(from, to);
+            try {
+                written += taskResultItemMapper.upsertBatch(batch);
+            } catch (RuntimeException ex) {
+                log.warn("[collect-data] upsert result item batch failed, skip batch {}..{} taskId={}",
+                        from, to, taskId, ex);
+            }
+        }
+        return new UpsertCounts(written, skipped);
+    }
+
+    private String hash(String value) {
+        try {
+            java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
+            byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            StringBuilder sb = new StringBuilder(bytes.length * 2);
+            for (byte b : bytes) {
+                sb.append(String.format("%02x", b));
+            }
+            return sb.toString();
+        } catch (Exception ex) {
+            throw new IllegalStateException("结果明细 hash 计算失败", ex);
+        }
+    }
+}

backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskResultItemMapper.java

@@ -2,8 +2,39 @@ package com.nanri.aiimage.modules.task.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
+import org.apache.ibatis.annotations.Insert;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
 
 @Mapper
 public interface TaskResultItemMapper extends BaseMapper<TaskResultItemEntity> {
+
+    /**
+     * 按唯一键 uk_task_scope_item (task_id, module_type, scope_hash, item_key)
+     * 批量 upsert：命中唯一键时更新行内容（含 id 保留），否则插入。
+     */
+    @Insert("""
+            <script>
+            INSERT INTO biz_task_result_item
+                (task_id, module_type, result_id, scope_key, scope_hash, item_key, asin, status,
+                 payload_json, payload_hash, created_at, updated_at)
+            VALUES
+            <foreach collection="rows" item="row" separator=",">
+                (#{row.taskId}, #{row.moduleType}, #{row.resultId}, #{row.scopeKey}, #{row.scopeHash},
+                 #{row.itemKey}, #{row.asin}, #{row.status}, #{row.payloadJson}, #{row.payloadHash},
+                 #{row.createdAt}, #{row.updatedAt})
+            </foreach>
+            ON DUPLICATE KEY UPDATE
+                result_id = VALUES(result_id),
+                scope_key = VALUES(scope_key),
+                asin = VALUES(asin),
+                status = VALUES(status),
+                payload_json = VALUES(payload_json),
+                payload_hash = VALUES(payload_hash),
+                updated_at = VALUES(updated_at)
+            </script>
+            """)
+    int upsertBatch(@Param("rows") List<TaskResultItemEntity> rows);
 }

backend-java/src/main/resources/application.yml

@@ -264,6 +264,7 @@ aiimage:
     brand-check-batch-size: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_BATCH_SIZE:10}
     brand-check-cache-capacity: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_CACHE_CAPACITY:512}
     invalid-asin-batch-size: ${AIIMAGE_COLLECT_DATA_INVALID_ASIN_BATCH_SIZE:100}
+    result-item-batch-size: ${AIIMAGE_COLLECT_DATA_RESULT_ITEM_BATCH_SIZE:100}
   image-video:
     coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn}
     coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu}

backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java

@@ -0,0 +1,239 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper;
+import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
+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 50：结果明细批量 upsert mapper 与幂等唯一键。
+ * CollectDataResultItemBatchWriter 把整 chunk 的 accepted 行批量 upsert 到
+ * biz_task_result_item：先一次批量查询现有行（payload_hash 相等即跳过，
+ * 幂等），再按唯一键 uk_task_scope_item 用一条 INSERT ... ON DUPLICATE
+ * KEY UPDATE 批量写入，替代逐行 select/insert/update。空输入零调用，
+ * 批量失败跳过该批可恢复，无无界内存增长。
+ */
+class CollectDataResultItemBatchWriterTest {
+
+    private TaskResultItemMapper taskResultItemMapper;
+    private TransientPayloadStorageService transientPayloadStorageService;
+    private CollectDataResultDetailCodec codec;
+    private CollectDataResultItemBatchWriter writer;
+
+    @BeforeEach
+    void setUp() {
+        taskResultItemMapper = mock(TaskResultItemMapper.class);
+        transientPayloadStorageService = mock(TransientPayloadStorageService.class);
+        codec = new CollectDataResultDetailCodec(new ObjectMapper());
+        writer = new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 10);
+    }
+
+    @Test
+    void test_task_050_task_normal_default_path() {
+        // 正常输入：无现有行 → 一次 upsertBatch 全量写入，引用字段完整。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(2);
+        List<CollectDataResultRowVo> rows = List.of(
+                row("B000000001", "Nike"), row("B000000002", "Zara"));
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:detail/abc");
+
+        assertEquals(2, counts.insertedOrUpdated(), "两行均写入");
+        assertEquals(0, counts.skipped(), "无跳过");
+        verify(taskResultItemMapper).selectList(any());
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(taskResultItemMapper).upsertBatch(captor.capture());
+        List<TaskResultItemEntity> entities = captor.getValue();
+        assertEquals(2, entities.size(), "批量写入两行");
+        assertEquals("asin:B000000001", entities.get(0).getItemKey(), "item_key 语义键");
+        assertEquals("ACCEPTED", entities.get(0).getStatus(), "状态保持 ACCEPTED");
+        assertTrue(entities.get(0).getPayloadJson().contains("rustfs:detail/abc"), "引用含对象指针");
+        assertTrue(entities.get(0).getPayloadJson().contains("\"offset\":0"), "引用含 offset");
+        assertEquals(2L, entities.get(0).getResultId(), "result_id 保留");
+        assertEquals(1L, entities.get(0).getTaskId(), "task_id 保留");
+    }
+
+    @Test
+    void test_task_050_task_normal_multiple_items() {
+        // 批量场景：75 行（8 批次），顺序稳定不丢失，每批数量正确。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 75; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand-" + (i % 7)));
+        }
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 3, rows, "rustfs:detail/x");
+
+        assertEquals(75, counts.insertedOrUpdated(), "全部写入");
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(taskResultItemMapper, times(8)).upsertBatch(captor.capture());
+        for (int i = 0; i < 8; i++) {
+            assertEquals(i < 7 ? 10 : 5, captor.getAllValues().get(i).size(), "批次 " + i + " 数量");
+        }
+        assertEquals("asin:B000000001", ((TaskResultItemEntity) captor.getAllValues().get(0).get(0)).getItemKey(),
+                "顺序稳定");
+        assertEquals("asin:B000000075", ((TaskResultItemEntity) captor.getAllValues().get(7).get(4)).getItemKey(),
+                "末批末行不丢失");
+    }
+
+    @Test
+    void test_task_050_task_normal_repeated_operation_is_idempotent() {
+        // 幂等：同一 chunk 重复提交时 payload_hash 相等 → 全部跳过不写入，
+        // 无重复记录、无重复写入调用。
+        String refJson = codec.encodeRef(0, 0, "rustfs:detail/x");
+        TaskResultItemEntity existing = new TaskResultItemEntity();
+        existing.setId(100L);
+        existing.setItemKey("asin:B000000001");
+        existing.setPayloadJson(refJson);
+        existing.setPayloadHash(sha256(refJson));
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing));
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0,
+                        List.of(row("B000000001", "Nike")), "rustfs:detail/x");
+
+        assertEquals(0, counts.insertedOrUpdated(), "hash 相等全部跳过");
+        assertEquals(1, counts.skipped(), "1 行跳过");
+        verify(taskResultItemMapper, never()).upsertBatch(anyList());
+    }
+
+    @Test
+    void test_task_050_task_boundary_empty_input() {
+        // 空输入：空列表零调用返回零计数。
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, List.of(), "rustfs:x");
+
+        assertEquals(0, counts.insertedOrUpdated(), "空输入返回 0");
+        assertEquals(0, counts.skipped(), "空输入无跳过");
+        verify(taskResultItemMapper, never()).selectList(any());
+        verify(taskResultItemMapper, never()).upsertBatch(anyList());
+    }
+
+    @Test
+    void test_task_050_task_boundary_single_item() {
+        // 单元素：单行正常 upsert；hash 不同时更新，唯一键幂等；
+        // 旧格式逐行 payload（非引用）在升级为引用时物理删除，避免泄漏。
+        TaskResultItemEntity existing = new TaskResultItemEntity();
+        existing.setId(100L);
+        existing.setItemKey("asin:B000000001");
+        existing.setPayloadJson("{\"brand\":\"nike\",\"asin\":\"B000000001\"}");
+        existing.setPayloadHash("old-hash");
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing));
+        when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1);
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 1,
+                        List.of(row("B000000001", "Nike")), "rustfs:new");
+
+        assertEquals(1, counts.insertedOrUpdated(), "hash 不同触发 upsert 更新");
+        verify(transientPayloadStorageService).deletePayloadIfPresent("{\"brand\":\"nike\",\"asin\":\"B000000001\"}");
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(taskResultItemMapper).upsertBatch(captor.capture());
+        TaskResultItemEntity entity = (TaskResultItemEntity) captor.getValue().get(0);
+        assertEquals(100L, entity.getId(), "复用现有行 id（ON DUPLICATE 命中唯一键）");
+        assertTrue(entity.getPayloadJson().contains("rustfs:new"), "引用更新为新对象");
+    }
+
+    @Test
+    void test_task_050_task_boundary_limit_and_overflow() {
+        // 上限/超限：批次大小 3，10 行 = 4 批（3+3+3+1），无无界积累。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        CollectDataResultItemBatchWriter smallWriter =
+                new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 3);
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand"));
+        }
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts = smallWriter.upsertAccepted(
+                1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(10, counts.insertedOrUpdated(), "全部写入");
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(taskResultItemMapper, times(4)).upsertBatch(captor.capture());
+        assertEquals(3, captor.getAllValues().get(0).size(), "首批 3 行");
+        assertEquals(1, captor.getAllValues().get(3).size(), "末批 1 行");
+    }
+
+    @Test
+    void test_task_050_task_invalid_input_rejected() {
+        // 非法参数：null/空白 ASIN 的行跳过不写入；null 行安全跳过。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        rows.add(null);
+        rows.add(row(null, "nike"));
+        rows.add(row("B000000001", "nike"));
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(1, counts.insertedOrUpdated(), "仅合法行写入");
+        verify(taskResultItemMapper).upsertBatch(anyList());
+    }
+
+    @Test
+    void test_task_050_task_dependency_failure_releases_resources() {
+        // 依赖失败：批量 upsert 抛错时跳过该批不中断，恢复后继续，无资源泄漏。
+        when(taskResultItemMapper.selectList(any())).thenReturn(List.of());
+        doThrow(new RuntimeException("db down"))
+                .doAnswer(invocation -> ((List<?>) invocation.getArgument(0)).size())
+                .when(taskResultItemMapper).upsertBatch(anyList());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 20; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand"));
+        }
+
+        CollectDataResultItemBatchWriter.UpsertCounts counts =
+                writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x");
+
+        assertEquals(10, counts.insertedOrUpdated(), "首批失败跳过，第二批 10 行写入");
+        verify(taskResultItemMapper, times(2)).upsertBatch(anyList());
+    }
+
+    private static CollectDataResultRowVo row(String asin, String brand) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setAsin(asin);
+        row.setBrand(brand);
+        return row;
+    }
+
+    private static String sha256(String value) {
+        try {
+            java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
+            byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            StringBuilder sb = new StringBuilder(bytes.length * 2);
+            for (byte b : bytes) {
+                sb.append(String.format("%02x", b));
+            }
+            return sb.toString();
+        } catch (Exception ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+}