task-13: cap chunk merge row count and payload bytes with orphan fallback

2026-08-29 17:05:49 +08:00
parent cbe0ec8bb2
commit e201b13301
3 changed files with 523 additions and 1 deletions
@@ -165,6 +165,20 @@ public class SimilarAsinProperties {
     */
    private long maxWorkbookUncompressedBytes = 512L * 1024L * 1024L;

+    /**
+     * Task 13：单次 chunk 合并后的最大行数。mergeChunkPayload 合并后总行数超过该上限时，
+     * 从最旧行开始降级到 orphan 兜底（assemble 阶段 putIfAbsent 合并回结果），chunk 不无界增长。
+     * 默认与 maxParseRows 一致（50000）。
+     */
+    private int chunkMergeMaxRows = 50000;
+
+    /**
+     * Task 13：单次 chunk 合并后 payload 的字节上限。合并后序列化字节超过该上限时，
+     * 从最旧行开始降级到 orphan 兜底；单行本身超过该上限时抛异常拒绝合并。
+     * 默认 16MB：50000 行 × 平均 300B/行 ≈ 15MB，留余量。
+     */
+    private long chunkMergePayloadMaxBytes = 16L * 1024L * 1024L;
+
    /**
     * P0-4：抢 Coze 提交锁失败后下次重试间隔（毫秒）。
     * 原硬编码 500ms，会在指数退避算法中作为基础值（500/1000/2000/4000ms 上限 4000）。
@@ -89,7 +89,9 @@ import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -1643,14 +1645,42 @@ public class SimilarAsinTaskService {
                return;
            }
            Map<String, SimilarAsinResultRowDto> persistedRows = readChunkRows(chunk);
+            int mergedRowCount = persistedRows.size() + rows.size();
+            // Task 13：单次合并后总行数超过上限时，从最旧行（存量优先）开始降级到 orphan 兜底，
+            // chunk 保持在上限内不无界增长；assemble 阶段 putIfAbsent 合并回结果不丢数据。
+            if (mergedRowCount > getChunkMergeMaxRows()) {
+                splitChunkMergeOverflow(taskId, persistedRows, rows, chunk, mergedRowCount - getChunkMergeMaxRows());
+                if (rows.isEmpty()) {
+                    return;
+                }
+            }
            for (SimilarAsinResultRowDto row : rows) {
                persistedRows.put(rowKey(row), row);
            }
            String payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败");
+            // Task 13：合并后 payload 字节超过上限时，从最旧行开始降级到 orphan 兜底；
+            // 降到只剩一行仍超上限时抛异常拒绝合并，防止无界 payload。
+            long payloadBytes = payloadJson.getBytes(StandardCharsets.UTF_8).length;
+            if (payloadBytes > getChunkMergePayloadMaxBytes()) {
+                demoteRowsToOrphan(taskId, persistedRows, payloadBytes - getChunkMergePayloadMaxBytes());
+                payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败");
+                payloadBytes = payloadJson.getBytes(StandardCharsets.UTF_8).length;
+                if (payloadBytes > getChunkMergePayloadMaxBytes() && persistedRows.size() <= 1) {
+                    throw new BusinessException("相似ASIN分片载荷超字节上限 taskId=" + taskId
+                            + " scopeHash=" + scopeHash + " chunk=" + chunkIndex
+                            + " bytes=" + payloadBytes + " limit=" + getChunkMergePayloadMaxBytes());
+                }
+            }
            String oldPayload = chunk.getPayloadJson();
            String oldPayloadHash = chunk.getPayloadHash();
            String newPayloadHash = DigestUtil.sha256Hex(payloadJson);
-            String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
+            final String storedPayload;
+            try {
+                storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
+            } catch (Exception storeEx) {
+                throw new BusinessException("相似ASIN分片载荷存储失败 taskId=" + taskId + " chunk=" + chunkIndex
+                        + ": " + (storeEx.getMessage() == null ? "" : storeEx.getMessage()), storeEx);
+            }
            int updated = taskChunkMapper.update(null, new LambdaUpdateWrapper<TaskChunkEntity>()
                    .eq(TaskChunkEntity::getId, chunk.getId())
                    .eq(TaskChunkEntity::getPayloadHash, oldPayloadHash)
@@ -1672,6 +1702,82 @@ public class SimilarAsinTaskService {
        throw new IllegalStateException("相似ASIN分片载荷更新失败");
    }

+    private int getChunkMergeMaxRows() {
+        int limit = properties.getChunkMergeMaxRows();
+        return limit > 0 ? limit : 50000;
+    }
+
+    private long getChunkMergePayloadMaxBytes() {
+        long limit = properties.getChunkMergePayloadMaxBytes();
+        return limit > 0 ? limit : 16L * 1024L * 1024L;
+    }
+
+    /**
+     * Task 13：按行数上限从最旧行开始降级到 orphan 兜底，保证合并后 chunk 行数不超过上限。
+     * 最旧优先：先降级存量行（LinkedHashMap 表头），不够再从新增行表头补齐；
+     * 传入的 rows 会被就地修改（保留未降级部分）。降级失败仅记日志，不阻断合并主流程。
+     */
+    private void splitChunkMergeOverflow(Long taskId, Map<String, SimilarAsinResultRowDto> persistedRows,
+                                         List<SimilarAsinResultRowDto> rows, TaskChunkEntity chunk,
+                                         int demoteCount) {
+        if (demoteCount <= 0 || rows.isEmpty()) {
+            return;
+        }
+        List<SimilarAsinResultRowDto> demoted = new ArrayList<>();
+        while (demoted.size() < demoteCount && !persistedRows.isEmpty()) {
+            String oldestKey = persistedRows.keySet().iterator().next();
+            SimilarAsinResultRowDto removed = persistedRows.remove(oldestKey);
+            if (removed != null) {
+                demoted.add(removed);
+            }
+        }
+        for (Iterator<SimilarAsinResultRowDto> it = rows.iterator(); it.hasNext() && demoted.size() < demoteCount; ) {
+            SimilarAsinResultRowDto row = it.next();
+            if (row != null) {
+                demoted.add(row);
+                it.remove();
+            }
+        }
+        persistOrphanCozeRows(taskId, demoted);
+        log.warn("[similar-asin] chunk merge row-limit exceeded taskId={} chunk={} mergedRows={} limit={} demoted={}",
+                taskId, chunk.getChunkIndex(), mergedRowCountOf(persistedRows, rows), getChunkMergeMaxRows(), demoted.size());
+    }
+
+    /**
+     * Task 13：按字节上限从最旧行（LinkedHashMap 表头）开始降级到 orphan 兜底，
+     * 直到 payload 字节不超过上限或只剩一行；降级失败仅记日志，不阻断合并主流程。
+     */
+    private void demoteRowsToOrphan(Long taskId, Map<String, SimilarAsinResultRowDto> persistedRows, long excessBytes) {
+        List<SimilarAsinResultRowDto> demoted = new ArrayList<>();
+        long releasedBytes = 0L;
+        while (persistedRows.size() > 1 && releasedBytes < excessBytes) {
+            String oldestKey = persistedRows.keySet().iterator().next();
+            SimilarAsinResultRowDto removed = persistedRows.remove(oldestKey);
+            if (removed != null) {
+                demoted.add(removed);
+                releasedBytes += estimateRowBytes(oldestKey, removed);
+            }
+        }
+        if (!demoted.isEmpty()) {
+            persistOrphanCozeRows(taskId, demoted);
+            log.warn("[similar-asin] chunk merge byte-limit exceeded taskId={} rows={} demoted={} releasedBytes={}",
+                    taskId, demoted.size(), demoted.size(), releasedBytes);
+        }
+    }
+
+    private static int mergedRowCountOf(Map<String, SimilarAsinResultRowDto> persistedRows, List<SimilarAsinResultRowDto> rows) {
+        return persistedRows.size() + rows.size();
+    }
+
+    private long estimateRowBytes(String rowKey, SimilarAsinResultRowDto row) {
+        try {
+            String json = objectMapper.writeValueAsString(row);
+            return json == null ? 128L : json.getBytes(StandardCharsets.UTF_8).length;
+        } catch (Exception ex) {
+            return 128L + (rowKey == null ? 0 : rowKey.getBytes(StandardCharsets.UTF_8).length);
+        }
+    }
+
    private List<SimilarAsinResultRowDto> expandRows(List<SimilarAsinResultRowDto> rows,
                                                     Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId) {
        if (rows == null || rows.isEmpty()) {