diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 88f2cb89..18d270e8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -37,6 +37,7 @@ import com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter; 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.invalidasin.mapper.InvalidAsinDataMapper; import com.nanri.aiimage.modules.file.service.LocalFileStorageService; import com.nanri.aiimage.modules.file.service.oss.OssStorageService; @@ -77,6 +78,7 @@ import java.security.MessageDigest; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.LinkedHashSet; @@ -143,6 +145,9 @@ public class CollectDataService { /** invalid ASIN 批量写入器：按批次 INSERT IGNORE，替代逐行插入。 */ private final CollectDataInvalidAsinBatchWriter invalidAsinBatchWriter; + /** 结果明细 chunk 级编解码：accepted 行按 chunk 共享一个 RustFS 对象。 */ + private final CollectDataResultDetailCodec resultDetailCodec; + @Value("${aiimage.collect-data.stale-timeout-minutes:30}") private long staleTimeoutMinutes; @@ -646,8 +651,17 @@ public class CollectDataService { stats.dedupeFilteredCount += filtered.dedupeFilteredCount(); stats.invalidFilteredCount += filtered.invalidFilteredCount(); List accepted = filterByBrandCheck(filtered.kept(), stats); - for (CollectDataResultRowVo row : accepted) { - upsertResultItem(task.getId(), result.getId(), scopeKey, row); + // 结果明细改为 chunk 级存储：整个 chunk 的 accepted 行共享一个 + // RustFS 对象（deterministic key，同 chunk 重提覆盖同一对象）， + // biz_task_result_item.payload_json 只存 {chunk, offset, payload} 引用。 + if (!accepted.isEmpty()) { + String detailJson = resultDetailCodec.encodeChunk(accepted); + 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); + } } String payloadJson = writeJson(rows, "采集结果序列化失败"); @@ -739,11 +753,13 @@ public class CollectDataService { return outcome.accepted(); } - private void upsertResultItem(Long taskId, Long resultId, String scopeKey, CollectDataResultRowVo row) { + 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); - String payloadJson = writeJson(row, "采集结果明细序列化失败"); - String payloadHash = hash(payloadJson); + // 引用 JSON 作为 payload_json：内容变化（行 offset/对象变化）即 hash 变化。 + String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail); + String payloadHash = hash(refJson); TaskResultItemEntity existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, taskId) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) @@ -753,9 +769,6 @@ public class CollectDataService { if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) { return; } - String storedPayload = transientPayloadStorageService.storeResultItemPayload( - MODULE_TYPE, taskId, scopeHash, itemKey, payloadJson); - requireRustfsPayload(storedPayload, "采集结果明细必须写入 RustFS"); LocalDateTime now = LocalDateTime.now(); if (existing == null) { TaskResultItemEntity entity = new TaskResultItemEntity(); @@ -767,7 +780,7 @@ public class CollectDataService { entity.setItemKey(itemKey); entity.setAsin(row.getAsin()); entity.setStatus("ACCEPTED"); - entity.setPayloadJson(storedPayload); + entity.setPayloadJson(refJson); entity.setPayloadHash(payloadHash); entity.setCreatedAt(now); entity.setUpdatedAt(now); @@ -786,14 +799,20 @@ public class CollectDataService { if (existing == null) { throw new BusinessException("保存采集结果明细失败"); } - transientPayloadStorageService.deleteReplacedPayloadIfNeeded(existing.getPayloadJson(), storedPayload); + // 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() .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, storedPayload) + .set(TaskResultItemEntity::getPayloadJson, refJson) .set(TaskResultItemEntity::getPayloadHash, payloadHash) .set(TaskResultItemEntity::getUpdatedAt, now)); } @@ -938,9 +957,34 @@ public class CollectDataService { if (rows == null) { return out; } + TypeReference> listType = new TypeReference<>() { + }; + // chunk 级引用：同一 chunk 对象只解析一次，按 offset 取行。 + Map> detailCache = new HashMap<>(); for (TaskResultItemEntity row : rows) { try { - String payloadJson = transientPayloadStorageService.resolvePayload(row.getPayloadJson(), "read collect data result item failed"); + CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(row.getPayloadJson()); + if (ref != null) { + List details = detailCache.get(ref.pointer()); + if (details == null) { + String detailJson = transientPayloadStorageService.resolvePayload( + ref.pointer(), "read collect data result detail failed"); + if (detailJson != null && !detailJson.isBlank()) { + details = objectMapper.readValue(detailJson, listType); + } + detailCache.put(ref.pointer(), details); + } + if (details != null && ref.offset() >= 0 && ref.offset() < details.size()) { + CollectDataResultRowVo value = details.get(ref.offset()); + if (value != null) { + out.add(value); + } + } + continue; + } + // 旧格式：payload_json 直接是行 JSON。 + String payloadJson = transientPayloadStorageService.resolvePayload( + row.getPayloadJson(), "read collect data result item failed"); CollectDataResultRowVo value = objectMapper.readValue(payloadJson, CollectDataResultRowVo.class); if (value != null) { out.add(value); @@ -1232,8 +1276,22 @@ public class CollectDataService { .select(TaskResultItemEntity::getPayloadJson) .eq(TaskResultItemEntity::getTaskId, taskId) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)); - if (items != null) { - for (TaskResultItemEntity item : items) { + deleteResultItemPayloads(items); + } + + /** 删除结果明细 payload：chunk 级引用按对象去重后各删一次，旧格式逐行删。 */ + private void deleteResultItemPayloads(List items) { + if (items == null) { + return; + } + Set deletedPointers = new HashSet<>(); + for (TaskResultItemEntity item : items) { + CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson()); + if (ref != null) { + if (deletedPointers.add(ref.pointer())) { + transientPayloadStorageService.deletePayloadIfPresent(ref.pointer()); + } + } else { transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson()); } } @@ -1291,11 +1349,7 @@ public class CollectDataService { .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) .eq(TaskResultItemEntity::getResultId, row.getId())); - if (resultItems != null) { - for (TaskResultItemEntity item : resultItems) { - transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson()); - } - } + deleteResultItemPayloads(resultItems); taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodec.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodec.java new file mode 100644 index 00000000..4a59eaf2 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodec.java @@ -0,0 +1,109 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import org.springframework.stereotype.Component; + +import java.util.List; + +/** + * 结果明细 chunk 级编解码：把整 chunk 的 accepted 行序列化为一个明细数组 + * JSON（每个 chunk 只写一个 RustFS 对象，替代逐行 storeResultItemPayload）， + * 并为每行生成 {chunk, offset, payload} 引用 JSON 写入 + * biz_task_result_item.payload_json。读侧按引用一次解析数组、offset 取行， + * 避免为每个 accepted 行单独发起对象存储读写。 + * + * 旧格式兼容：parseRef 对非引用内容（逐行行 JSON / 数组 / 裸文本 / 损坏 + * JSON）返回 null，由调用方按逐行旧路径兜底读取；rowAt 越界/损坏安全 + * 返回 null。 + */ +@Component +public class CollectDataResultDetailCodec { + + public static final String REF_FIELD_CHUNK = "chunk"; + public static final String REF_FIELD_OFFSET = "offset"; + public static final String REF_FIELD_PAYLOAD = "payload"; + + private final ObjectMapper objectMapper; + + public CollectDataResultDetailCodec(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + + /** 生成引用 JSON：{"chunk":N,"offset":M,"payload":""}。 */ + public String encodeRef(int chunkIndex, int offset, String pointer) { + if (chunkIndex < 0 || offset < 0 || pointer == null || pointer.isBlank()) { + throw new IllegalArgumentException("invalid chunk detail ref chunk=" + chunkIndex + + " offset=" + offset + " pointer=" + pointer); + } + try { + return objectMapper.writeValueAsString(new ChunkRef(chunkIndex, offset, pointer)); + } catch (Exception ex) { + throw new IllegalArgumentException("encode chunk detail ref failed", ex); + } + } + + /** + * 解析引用 JSON；内容不是引用格式（旧逐行行 JSON、数组、裸文本、 + * 损坏 JSON）时返回 null，由调用方按旧格式兜底。 + */ + public ChunkRef parseRef(String refJson) { + if (refJson == null || refJson.isBlank()) { + return null; + } + try { + JsonNode node = objectMapper.readTree(refJson); + if (node == null || !node.isObject() + || !node.hasNonNull(REF_FIELD_CHUNK) + || !node.hasNonNull(REF_FIELD_OFFSET) + || !node.hasNonNull(REF_FIELD_PAYLOAD)) { + return null; + } + int chunkIndex = node.get(REF_FIELD_CHUNK).asInt(); + int offset = node.get(REF_FIELD_OFFSET).asInt(); + String pointer = node.get(REF_FIELD_PAYLOAD).asText(); + if (chunkIndex < 0 || offset < 0 || pointer.isBlank()) { + return null; + } + return new ChunkRef(chunkIndex, offset, pointer); + } catch (Exception ex) { + return null; + } + } + + /** 把整 chunk 行序列化为明细数组 JSON；null/空输入编码为空数组。 */ + public String encodeChunk(List rows) { + try { + return objectMapper.writeValueAsString(rows == null ? List.of() : rows); + } catch (Exception ex) { + throw new IllegalArgumentException("encode chunk detail failed", ex); + } + } + + /** + * 从 chunk 明细数组 JSON 中按 offset 取行；offset 越界/负值、明细 + * 损坏或非数组时返回 null。 + */ + public CollectDataResultRowVo rowAt(String detailJson, int offset) { + if (detailJson == null || detailJson.isBlank() || offset < 0) { + return null; + } + try { + JsonNode node = objectMapper.readTree(detailJson); + if (node == null || !node.isArray() || offset >= node.size()) { + return null; + } + return objectMapper.treeToValue(node.get(offset), CollectDataResultRowVo.class); + } catch (Exception ex) { + return null; + } + } + + /** 结果明细行引用：chunk 序号 + 行内 offset + chunk 明细对象指针。 */ + public record ChunkRef(@JsonProperty(REF_FIELD_CHUNK) int chunkIndex, + @JsonProperty(REF_FIELD_OFFSET) int offset, + @JsonProperty(REF_FIELD_PAYLOAD) String pointer) { + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodecTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodecTest.java new file mode 100644 index 00000000..962ab919 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailCodecTest.java @@ -0,0 +1,151 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Task 49：将结果明细从逐行 RustFS 对象改为 chunk 级 payload 存储。 + * CollectDataResultDetailCodec 负责：把整 chunk 的 accepted 行序列化为 + * 一个明细数组 JSON（每 chunk 一个 RustFS 对象），并为每行生成 + * {chunk, offset, payload} 引用 JSON（写入 biz_task_result_item.payload_json）； + * 读侧按 chunk 引用一次解析数组、offset 取行。旧格式（payload_json 直接是 + * 行 JSON）解析返回 null，由调用方按逐行旧路径兼容读取。 + */ +class CollectDataResultDetailCodecTest { + + private CollectDataResultDetailCodec codec; + + @BeforeEach + void setUp() { + codec = new CollectDataResultDetailCodec(new ObjectMapper()); + } + + @Test + void test_task_049_payload_chunk_rustfs_normal_default_path() { + // 正常输入：encodeRef → parseRef 往返一致；chunk 明细数组序列化后 + // rowAt 按 offset 取回同一行。 + String ref = codec.encodeRef(2, 3, "rustfs:detail/abc"); + + CollectDataResultDetailCodec.ChunkRef parsed = codec.parseRef(ref); + + assertEquals(2, parsed.chunkIndex(), "chunk 序号保留"); + assertEquals(3, parsed.offset(), "行 offset 保留"); + assertEquals("rustfs:detail/abc", parsed.pointer(), "RustFS 对象指针保留"); + + String detailJson = codec.encodeChunk(List.of(row("B000000001", "Nike"), row("B000000002", "Zara"))); + CollectDataResultRowVo at1 = codec.rowAt(detailJson, 1); + assertEquals("B000000002", at1.getAsin(), "按 offset 取回第二行"); + assertEquals("Zara", at1.getBrand(), "行字段完整保留（不做规范化）"); + } + + @Test + void test_task_049_payload_chunk_rustfs_normal_multiple_items() { + // 批量场景：多 chunk 多行，各自引用指向自己的 chunk 与 offset，互不串行。 + String refChunk0 = codec.encodeRef(0, 1, "rustfs:detail/a"); + String refChunk2 = codec.encodeRef(2, 0, "rustfs:detail/b"); + List rows0 = List.of(row("B000000001", "A"), row("B000000002", "B")); + List rows2 = List.of(row("B000000101", "C")); + + CollectDataResultDetailCodec.ChunkRef parsed0 = codec.parseRef(refChunk0); + CollectDataResultDetailCodec.ChunkRef parsed2 = codec.parseRef(refChunk2); + assertEquals(0, parsed0.chunkIndex(), "chunk0 序号"); + assertEquals(1, parsed0.offset(), "chunk0 offset"); + assertEquals(2, parsed2.chunkIndex(), "chunk2 序号"); + assertEquals(0, parsed2.offset(), "chunk2 offset"); + + assertEquals("B000000002", codec.rowAt(codec.encodeChunk(rows0), parsed0.offset()).getAsin(), "chunk0 行"); + assertEquals("B000000101", codec.rowAt(codec.encodeChunk(rows2), parsed2.offset()).getAsin(), "chunk2 行"); + assertNull(codec.rowAt(codec.encodeChunk(rows2), 5), "chunk2 中越界 offset 返回 null"); + } + + @Test + void test_task_049_payload_chunk_rustfs_normal_repeated_operation_is_idempotent() { + // 幂等：同输入重复编码结果逐字节一致；重复解析引用结果一致；不产生新对象。 + String first = codec.encodeChunk(List.of(row("B000000001", "Nike"), row("B000000002", "Zara"))); + String second = codec.encodeChunk(List.of(row("B000000001", "Nike"), row("B000000002", "Zara"))); + assertEquals(first, second, "重复序列化结果一致"); + + String ref = codec.encodeRef(1, 0, "rustfs:detail/x"); + assertEquals(ref, codec.encodeRef(1, 0, "rustfs:detail/x"), "重复引用编码一致"); + assertEquals(codec.parseRef(ref), codec.parseRef(ref), "重复解析一致"); + } + + @Test + void test_task_049_payload_chunk_rustfs_boundary_empty_input() { + // 空输入：空 chunk 明细编码为空数组；空/空白引用解析为 null（旧格式兼容）。 + assertEquals("[]", codec.encodeChunk(List.of()), "空 chunk 编码为空数组"); + assertEquals("[]", codec.encodeChunk(null), "null 明细编码为空数组"); + assertNull(codec.parseRef(null), "null 引用返回 null"); + assertNull(codec.parseRef(" "), "空白引用返回 null"); + assertNull(codec.rowAt("[]", 0), "空数组取行返回 null"); + } + + @Test + void test_task_049_payload_chunk_rustfs_boundary_single_item() { + // 单元素：单行单 chunk offset 0 正确；单行明细取行正确。 + String ref = codec.encodeRef(0, 0, "rustfs:detail/solo"); + CollectDataResultDetailCodec.ChunkRef parsed = codec.parseRef(ref); + assertEquals(0, parsed.chunkIndex(), "单 chunk 序号"); + assertEquals(0, parsed.offset(), "单行 offset 0"); + + String detailJson = codec.encodeChunk(List.of(row("B000000001", "solo"))); + CollectDataResultRowVo value = codec.rowAt(detailJson, 0); + assertEquals("B000000001", value.getAsin(), "单行取回正确"); + assertEquals("solo", value.getBrand(), "单行字段完整"); + } + + @Test + void test_task_049_payload_chunk_rustfs_boundary_limit_and_overflow() { + // 上限/超限：offset 越界或为负时 rowAt 安全返回 null，不抛异常、不越界。 + String detailJson = codec.encodeChunk(List.of(row("B000000001", "Nike"))); + assertNull(codec.rowAt(detailJson, 1), "offset 越界返回 null"); + assertNull(codec.rowAt(detailJson, 100), "offset 远超行数返回 null"); + assertNull(codec.rowAt(detailJson, -1), "负 offset 返回 null"); + assertNull(codec.rowAt(null, 0), "null 明细返回 null"); + assertNull(codec.rowAt("not-json", 0), "损坏明细返回 null"); + } + + @Test + void test_task_049_payload_chunk_rustfs_invalid_input_rejected() { + // 非法参数：非法 chunk/offset 编码引用时抛可识别异常； + // 非引用格式（旧逐行 JSON、数组、裸文本）解析返回 null 而非抛错。 + assertThrows(IllegalArgumentException.class, () -> codec.encodeRef(-1, 0, "rustfs:x"), "负 chunk 拒绝"); + assertThrows(IllegalArgumentException.class, () -> codec.encodeRef(0, -1, "rustfs:x"), "负 offset 拒绝"); + assertThrows(IllegalArgumentException.class, () -> codec.encodeRef(0, 0, null), "null 指针拒绝"); + + assertNull(codec.parseRef("{\"brand\":\"nike\"}"), "旧格式行 JSON 返回 null"); + assertNull(codec.parseRef("[1,2,3]"), "数组 JSON 返回 null"); + assertNull(codec.parseRef("plain-text"), "裸文本返回 null"); + assertNull(codec.parseRef("{\"chunk\":1,\"offset\":2}"), "缺 payload 字段返回 null"); + } + + @Test + void test_task_049_payload_chunk_rustfs_dependency_failure_releases_resources() { + // 依赖失败：引用 JSON 损坏（readTree 抛错）时解析返回 null， + // 调用方可按旧格式逐行读取兜底，错误可恢复不泄漏。 + String malformed = "{\"chunk\":1,\"offset\":"; + assertNull(codec.parseRef(malformed), "损坏引用解析为 null 不抛异常"); + + String detailJson = codec.encodeChunk(new ArrayList<>()); + CollectDataResultRowVo value = codec.rowAt(detailJson + "{broken", 0); + assertNull(value, "损坏明细取行返回 null 不抛异常"); + assertTrue(codec.parseRef(codec.encodeRef(0, 0, "rustfs:x")) != null, "修复后可正常解析"); + } + + private static CollectDataResultRowVo row(String asin, String brand) { + CollectDataResultRowVo row = new CollectDataResultRowVo(); + row.setAsin(asin); + row.setBrand(brand); + return row; + } +}