From 567cb1ab67b7b4f7f10de15f824538232037bf90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 17:31:21 +0800 Subject: [PATCH] =?UTF-8?q?task-58:=20collectdata=20=E7=BB=93=E6=9E=9C?= =?UTF-8?q?=E6=98=8E=E7=BB=86=E8=AF=BB=E5=8F=96=E5=A4=B1=E8=B4=A5=E9=99=8D?= =?UTF-8?q?=E7=BA=A7=EF=BC=88RustFS=20=E8=B6=85=E6=97=B6=E8=B7=B3=E8=BF=87?= =?UTF-8?q?=E8=AF=A5=20chunk=EF=BC=8C=E4=B8=8D=E4=B8=AD=E6=96=AD=E5=85=B6?= =?UTF-8?q?=E5=AE=83=20chunk=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chunk 级引用读取失败（RustFS 超时/不可用）时降级为跳过该 chunk 的引用行并记录 warn， 结果文件继续生成，缺失行由任务状态可观测；chunk 明细 JSON 损坏仍抛异常（数据问题不降级）， 旧格式单行读取失败保持抛异常。失败 chunk 在单次调用内缓存为空哨兵，避免重复 resolve。 task-052 依赖失败测试同步更新为降级语义。 --- .../util/CollectDataResultDetailReader.java | 24 +- .../CollectDataResultDetailReaderTest.java | 9 +- .../util/CollectDataRustfsDegradeTest.java | 208 ++++++++++++++++++ 3 files changed, 234 insertions(+), 7 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataRustfsDegradeTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java index fa79ba38..47247bba 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReader.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; 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.stereotype.Component; import java.util.ArrayList; @@ -19,6 +20,7 @@ import java.util.Map; * 空列表，越界/损坏行安全跳过，读取失败抛可识别异常。 */ @Component +@Slf4j public class CollectDataResultDetailReader { private static final String ERROR_DETAIL = "read collect data result detail failed"; @@ -54,10 +56,24 @@ public class CollectDataResultDetailReader { if (ref != null) { List details = detailCache.get(ref.pointer()); if (details == null) { - String detailJson = transientPayloadStorageService.resolvePayload(ref.pointer(), ERROR_DETAIL); - details = (detailJson == null || detailJson.isBlank()) - ? null - : objectMapper.readValue(detailJson, listType); + try { + String detailJson = transientPayloadStorageService.resolvePayload(ref.pointer(), ERROR_DETAIL); + // 空内容用空列表占位：与缓存 miss 区分，避免同一 chunk 重复 resolve。 + details = (detailJson == null || detailJson.isBlank()) + ? List.of() + : objectMapper.readValue(detailJson, listType); + } catch (Exception readEx) { + // RustFS 超时/不可用等读取失败：跳过该 chunk 的引用行， + // 不中断其它 chunk 的行读取（结果文件降级生成，缺失行 + // 由任务状态可观测）；chunk 明细 JSON 损坏则正常上报， + // 属于数据问题而非依赖降级。 + if (readEx instanceof com.fasterxml.jackson.core.JsonProcessingException) { + throw readEx; + } + log.warn("[collect-data] skip chunk detail read failed pointer={} err={}", + ref.pointer(), readEx.getMessage()); + details = List.of(); + } detailCache.put(ref.pointer(), details); } if (details != null && ref.offset() >= 0 && ref.offset() < details.size()) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java index 784d237e..f5d2798e 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultDetailReaderTest.java @@ -26,7 +26,8 @@ import static org.mockito.Mockito.when; * CollectDataResultDetailReader 从 biz_task_result_item 解析行：引用格式 * 按 pointer 缓存整 chunk 明细（同一 chunk 对象只 resolve/解析一次）， * 再按 offset 取行；旧格式逐行兜底。空输入返回空列表，越界/损坏行安全 - * 跳过，读取失败抛可识别异常且可恢复。 + * 跳过，chunk 级读取失败降级跳过该 chunk（不中断其它 chunk）， + * 旧格式读取失败仍抛可识别异常。 */ class CollectDataResultDetailReaderTest { @@ -163,14 +164,16 @@ class CollectDataResultDetailReaderTest { @Test void test_task_052_chunk_dependency_failure_releases_resources() { - // 依赖失败：对象存储读取抛错时抛可识别异常（不泄漏内部状态）； + // 依赖失败（task-58 起降级语义）：chunk 级读取失败 → 跳过该 chunk + // 的引用行（降级为空结果，不抛异常、不中断其它 chunk）， // 依赖恢复后同一实例再次调用成功，无资源残留。 doThrow(new IllegalStateException("object storage down")) .doReturn("[{\"asin\":\"B000000001\",\"brand\":\"back\"}]") .when(transientPayloadStorageService) .resolvePayload("rustfs:detail/down", "read collect data result detail failed"); List items = List.of(item(1L, refJson(0, 0, "rustfs:detail/down"))); - assertThrows(RuntimeException.class, () -> reader.readRows(items), "读取失败抛可识别异常"); + List degraded = reader.readRows(items); + assertEquals(0, degraded.size(), "读取失败 chunk 降级为空结果"); List rows = reader.readRows(items); assertEquals(1, rows.size(), "恢复后正常读取"); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataRustfsDegradeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataRustfsDegradeTest.java new file mode 100644 index 00000000..814a957b --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataRustfsDegradeTest.java @@ -0,0 +1,208 @@ +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.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 58：外部品牌服务不可用、RustFS 超时和重复 chunk 的降级测试。 + * 新增降级语义：chunk 级引用读取失败（RustFS 超时/不可用）时跳过该 chunk + * 的引用行并记录 warn，不中断其它 chunk 的行读取；旧格式单行读取失败仍抛 + * 可识别异常（不可降级，避免静默丢数据）。重复 chunk（多 item 引用同一 + * RustFS 对象）只 resolve 一次，失败降级对同一 chunk 只发生一次。 + */ +class CollectDataRustfsDegradeTest { + + private TransientPayloadStorageService transientPayloadStorageService; + private CollectDataResultDetailReader reader; + + @BeforeEach + void setUp() { + transientPayloadStorageService = mock(TransientPayloadStorageService.class); + reader = new CollectDataResultDetailReader( + new CollectDataResultDetailCodec(new ObjectMapper()), new ObjectMapper(), + transientPayloadStorageService); + } + + @Test + void test_task_058_chunk_brand_rustfs_normal_default_path() { + // 正常路径：无失败时行为与既有一致，多 chunk 全部取回，无降级误触发。 + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/a")), + item(2L, refJson(0, 1, "rustfs:detail/a")), + item(3L, refJson(1, 0, "rustfs:detail/b"))); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/a", "read collect data result detail failed")) + .thenReturn(rowsJson(0, 2)); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/b", "read collect data result detail failed")) + .thenReturn(rowsJson(2, 1)); + + List rows = reader.readRows(items); + + assertEquals(3, rows.size(), "3 行全部取回"); + assertEquals("B000000001", rows.get(0).getAsin(), "chunk a 首行"); + assertEquals("B000000003", rows.get(2).getAsin(), "chunk b 行"); + } + + @Test + void test_task_058_chunk_brand_rustfs_normal_multiple_items() { + // 批量场景（外部服务不可用降级）：chunk a 读取失败 → 其引用行降级跳过， + // chunk b/c 正常取回，顺序稳定不丢失。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/down", "read collect data result detail failed")) + .thenThrow(new IllegalStateException("rustfs read timeout")); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/ok1", "read collect data result detail failed")) + .thenReturn(rowsJson(0, 2)); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/ok2", "read collect data result detail failed")) + .thenReturn(rowsJson(2, 2)); + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/down")), + item(2L, refJson(0, 1, "rustfs:detail/down")), + item(3L, refJson(1, 0, "rustfs:detail/ok1")), + item(4L, refJson(1, 1, "rustfs:detail/ok1")), + item(5L, refJson(2, 0, "rustfs:detail/ok2")), + item(6L, refJson(2, 1, "rustfs:detail/ok2"))); + + List rows = reader.readRows(items); + + assertEquals(4, rows.size(), "失败 chunk 降级跳过 2 行，其余 4 行取回"); + assertEquals("B000000001", rows.get(0).getAsin(), "ok1 首行"); + assertEquals("B000000003", rows.get(2).getAsin(), "ok2 行顺序稳定"); + } + + @Test + void test_task_058_chunk_brand_rustfs_normal_repeated_operation_is_idempotent() { + // 幂等（重复 chunk）：多个 item 引用同一失败对象 → 降级只发生一次、 + // 结果稳定；重复读取结果一致，无状态残留。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/x", "read collect data result detail failed")) + .thenThrow(new IllegalStateException("rustfs down")) + .thenReturn(rowsJson(0, 2)); + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/x")), + item(2L, refJson(0, 1, "rustfs:detail/x"))); + + List first = reader.readRows(items); + + assertEquals(0, first.size(), "失败时整 chunk 降级为空"); + verify(transientPayloadStorageService, times(1)).resolvePayload(anyString(), anyString()); + + // 恢复后再次读取：本次调用重新 resolve，chunk 行全部取回。 + List second = reader.readRows(items); + assertEquals(2, second.size(), "恢复后同一实例再次读取成功"); + assertEquals("B000000002", second.get(1).getAsin(), "两行均恢复"); + } + + @Test + void test_task_058_chunk_brand_rustfs_boundary_empty_input() { + // 空输入：空列表安全跳过，不发起读取，不触发降级。 + List rows = reader.readRows(List.of()); + assertEquals(0, rows.size(), "空输入返回空列表"); + verify(transientPayloadStorageService, times(0)).resolvePayload(anyString(), anyString()); + } + + @Test + void test_task_058_chunk_brand_rustfs_boundary_single_item() { + // 单元素：单个 chunk 读取失败 → 返回空结果不抛异常（降级路径不依赖批量）。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/solo", "read collect data result detail failed")) + .thenThrow(new IllegalStateException("rustfs timeout")); + + List rows = reader.readRows(List.of(item(1L, refJson(0, 0, "rustfs:detail/solo")))); + + assertEquals(0, rows.size(), "单 chunk 失败降级为空结果"); + } + + @Test + void test_task_058_chunk_brand_rustfs_boundary_limit_and_overflow() { + // 上限/超限：全部 chunk 失败（10 个不同对象）→ 空结果，各 resolve 一次， + // 无无界累积、无重复调用。 + List items = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + items.add(item((long) i + 1, refJson(i, 0, "rustfs:detail/fail" + i))); + } + for (int i = 0; i < 10; i++) { + when(transientPayloadStorageService.resolvePayload("rustfs:detail/fail" + i, "read collect data result detail failed")) + .thenThrow(new IllegalStateException("rustfs timeout")); + } + + List rows = reader.readRows(items); + + assertEquals(0, rows.size(), "全部失败降级为空"); + verify(transientPayloadStorageService, times(10)).resolvePayload(anyString(), anyString()); + } + + @Test + void test_task_058_chunk_brand_rustfs_invalid_input_rejected() { + // 非法参数：旧格式单行读取失败不可降级（抛可识别异常，避免静默丢数据）； + // chunk 明细 JSON 损坏同样拒绝。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/broken", "read collect data result detail failed")) + .thenReturn("[{\"asin\":\"B000000001\""); + assertThrows(RuntimeException.class, + () -> reader.readRows(List.of(item(1L, refJson(0, 0, "rustfs:detail/broken")))), + "损坏 chunk JSON 仍抛异常（数据损坏不降级）"); + + when(transientPayloadStorageService.resolvePayload("{\"asin\":\"B000000001\"", "read collect data result item failed")) + .thenThrow(new IllegalStateException("rustfs down")); + assertThrows(RuntimeException.class, + () -> reader.readRows(List.of(item(2L, "{\"asin\":\"B000000001\""))), + "旧格式行读取失败抛可识别异常（不可降级）"); + } + + @Test + void test_task_058_chunk_brand_rustfs_dependency_failure_releases_resources() { + // 依赖失败（品牌服务不可用语义）：部分 chunk 失败降级、其余正常； + // 依赖恢复后同一实例再次调用全部取回，无资源残留。 + when(transientPayloadStorageService.resolvePayload("rustfs:detail/partial", "read collect data result detail failed")) + .thenThrow(new IllegalStateException("brand service down")) + .thenReturn(rowsJson(0, 2), rowsJson(0, 2)); + when(transientPayloadStorageService.resolvePayload("rustfs:detail/keep", "read collect data result detail failed")) + .thenReturn(rowsJson(2, 1)); + List items = List.of( + item(1L, refJson(0, 0, "rustfs:detail/partial")), + item(2L, refJson(0, 1, "rustfs:detail/partial")), + item(3L, refJson(1, 0, "rustfs:detail/keep"))); + + List degraded = reader.readRows(items); + assertEquals(1, degraded.size(), "失败 chunk 降级，keep 正常"); + assertTrue(degraded.get(0).getAsin().contains("B000000003"), "正常 chunk 行不丢失"); + verify(transientPayloadStorageService, times(1)).resolvePayload("rustfs:detail/partial", "read collect data result detail failed"); + + List recovered = reader.readRows(items); + assertEquals(3, recovered.size(), "恢复后全部取回，无残留"); + } + + private static TaskResultItemEntity item(Long id, String payloadJson) { + TaskResultItemEntity entity = new TaskResultItemEntity(); + entity.setId(id); + entity.setPayloadJson(payloadJson); + return entity; + } + + private static String refJson(int chunk, int offset, String pointer) { + return "{\"chunk\":" + chunk + ",\"offset\":" + offset + ",\"payload\":\"" + pointer + "\"}"; + } + + private static String rowsJson(int start, int count) { + StringBuilder sb = new StringBuilder("["); + for (int i = 0; i < count; i++) { + if (i > 0) { + sb.append(","); + } + sb.append("{\"asin\":\"B").append(String.format("%09d", start + i + 1)) + .append("\",\"brand\":\"brand").append(i).append("\"}"); + } + return sb.append("]").toString(); + } +}