From ca32cde4fea3918d1b30af5220770beabf81cb2c 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 18:42:19 +0800 Subject: [PATCH] =?UTF-8?q?task-64:=20transient=20payload=20=E5=8E=8B?= =?UTF-8?q?=E7=BC=A9=E6=94=B9=E4=B8=BA=E7=9B=B4=E6=8E=A5=20gzip=20?= =?UTF-8?q?=E4=BA=8C=E8=BF=9B=E5=88=B6=E6=B5=81=E4=B8=8A=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RustfsObjectStorageService 新增 uploadBytes/readObjectBytes 二进制流 API （application/gzip），uploadText/readObjectAsString 改为其薄包装 - TransientPayloadStorageService 的 encodeStoredPayload 直接输出 gzip 字节， 不再经过 gzip64 + Base64 文本编码（消除 33% 体积膨胀与中间字符串副本） - 读取端按 gzip magic 识别二进制流解压；兼容旧 gzip64: 文本与历史裸文本 - 本地回落文件同步改为写二进制（Files.write），超限判定基于 gzip 后字节数 --- .../object/RustfsObjectStorageService.java | 16 +- .../TransientPayloadStorageService.java | 73 ++++--- .../TransientPayloadStorageServiceTest.java | 205 +++++++++++++++++- 3 files changed, 260 insertions(+), 34 deletions(-) diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java index f8899686..e2a2dd5a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java @@ -80,12 +80,17 @@ public class RustfsObjectStorageService { } public String uploadText(String objectKey, String content, boolean verifyAfterUpload) { + byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8); + return uploadBytes(objectKey, bytes, verifyAfterUpload); + } + + public String uploadBytes(String objectKey, byte[] content, boolean verifyAfterUpload) { long deadlineNanos = operationDeadlineNanos(); if (!isConfigured()) { throw new IllegalStateException("transient storage is not configured"); } rejectIfCircuitOpen(OP_UPLOAD, objectKey); - byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8); + byte[] bytes = content == null ? new byte[0] : content; recordPayloadBytes(bytes.length); AtomicBoolean putCompleted = new AtomicBoolean(); try { @@ -96,7 +101,7 @@ public class RustfsObjectStorageService { .bucket(properties.getBucket()) .object(objectKey) .stream(stream, bytes.length, -1) - .contentType("application/json") + .contentType("application/gzip") .build()); putCompleted.set(true); return objectKey; @@ -117,6 +122,11 @@ public class RustfsObjectStorageService { } public String readObjectAsString(String objectKey) { + byte[] bytes = readObjectBytes(objectKey); + return new String(bytes, StandardCharsets.UTF_8); + } + + public byte[] readObjectBytes(String objectKey) { long deadlineNanos = operationDeadlineNanos(); if (!isConfigured()) { throw new IllegalStateException("transient storage is not configured"); @@ -128,7 +138,7 @@ public class RustfsObjectStorageService { .bucket(properties.getBucket()) .object(objectKey) .build())) { - return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + return stream.readAllBytes(); } }); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java index 31edb6c5..7edca2ff 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java @@ -145,13 +145,16 @@ public class TransientPayloadStorageService { "transient payload only exists on instance=" + ownerInstance + " current=" + instanceMetadata.getInstanceId()); } - return decodeStoredPayload(readLocalPayload(stripLocalInstanceId(localKey))); + return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey))); } if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) { - return decodeStoredPayload(rustfsObjectStorageService.readObjectAsString(pointer.substring(RUSTFS_POINTER_PREFIX.length()))); + return decodeStoredPayloadBytes( + rustfsObjectStorageService.readObjectBytes(pointer.substring(RUSTFS_POINTER_PREFIX.length()))); } if (pointer.startsWith(OSS_POINTER_PREFIX)) { - return decodeStoredPayload(ossStorageService.readObjectAsString(pointer.substring(OSS_POINTER_PREFIX.length()))); + return decodeStoredPayloadBytes( + ossStorageService.readObjectAsString(pointer.substring(OSS_POINTER_PREFIX.length())) + .getBytes(StandardCharsets.UTF_8)); } } catch (Exception ex) { throw new IllegalStateException(errorMessage + ": " + pointer, ex); @@ -318,8 +321,8 @@ public class TransientPayloadStorageService { category, moduleType, taskId, objectKey, rawBytes, properties.getWarnPayloadBytes()); } boolean rawOversize = isPositiveLimit(properties.getMaxPayloadBytes()) && rawBytes > properties.getMaxPayloadBytes(); - String storedContent = encodeStoredPayload(content); - long storedBytes = payloadBytes(storedContent); + byte[] storedContent = encodeStoredPayload(content); + long storedBytes = storedContent.length; boolean storedOversize = isPositiveLimit(properties.getMaxStoredPayloadBytes()) && storedBytes > properties.getMaxStoredPayloadBytes(); if (rawOversize || storedOversize) { log.warn("[transient-payload] payload size exceeds rustfs limit category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} maxRawBytes={} maxStoredBytes={} fallbackToLocal={}", @@ -333,7 +336,7 @@ public class TransientPayloadStorageService { boolean rustfsFallbackToLocal = rawOversize || storedOversize; if (!rustfsFallbackToLocal && rustfsObjectStorageService.isConfigured()) { try { - pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, storedContent, verifyAfterUpload); + pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadBytes(objectKey, storedContent, verifyAfterUpload); } catch (Exception ex) { rustfsFallbackToLocal = true; // 升级为 ERROR：rustfs 失败后只能落到本地，多实例下其他节点读不到，必须能告警。 @@ -362,7 +365,7 @@ public class TransientPayloadStorageService { } } - private String storeLocal(String objectKey, String content) { + private String storeLocal(String objectKey, byte[] content) { try { Path root = localPayloadRoot(); Path target = root.resolve(objectKey).normalize(); @@ -370,9 +373,8 @@ public class TransientPayloadStorageService { throw new IllegalArgumentException("invalid local payload key: " + objectKey); } Files.createDirectories(target.getParent()); - Files.writeString(target, - content == null ? "" : content, - StandardCharsets.UTF_8, + Files.write(target, + content == null ? new byte[0] : content, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE); @@ -385,10 +387,10 @@ public class TransientPayloadStorageService { } } - private String readLocalPayload(String objectKey) { + private byte[] readLocalPayloadBytes(String objectKey) { try { Path target = resolveLocalPayloadPath(objectKey); - return Files.readString(target, StandardCharsets.UTF_8); + return Files.readAllBytes(target); } catch (Exception ex) { throw new IllegalStateException("failed to read local transient payload", ex); } @@ -435,14 +437,18 @@ public class TransientPayloadStorageService { return normalized.replaceAll("[^A-Za-z0-9_.\\-]", "_"); } - private String encodeStoredPayload(String content) { + /** + * 压缩 transient payload 为 gzip 二进制流（不再经过 base64 文本编码）。 + * 极小内容 gzip 可能膨胀，属预期：压缩收益以可压缩内容为准，读取端按 magic 自动识别。 + */ + private byte[] encodeStoredPayload(String content) { try { byte[] raw = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(Math.max(32, raw.length / 2)); try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) { gzip.write(raw); } - return "gzip64:" + Base64.getEncoder().encodeToString(baos.toByteArray()); + return baos.toByteArray(); } catch (Exception ex) { throw new IllegalStateException("failed to encode transient payload", ex); } @@ -456,21 +462,36 @@ public class TransientPayloadStorageService { return value > 0L; } - private String decodeStoredPayload(String storedContent) { - if (storedContent == null || storedContent.isBlank()) { - return storedContent; + private String decodeStoredPayloadBytes(byte[] storedContent) { + if (storedContent == null || storedContent.length == 0) { + return ""; } - if (!storedContent.startsWith("gzip64:")) { - return storedContent; - } - try { - byte[] compressed = Base64.getDecoder().decode(storedContent.substring("gzip64:".length())); - try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) { + if (isGzip(storedContent)) { + try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(storedContent))) { return new String(gzip.readAllBytes(), StandardCharsets.UTF_8); + } catch (Exception ex) { + throw new IllegalStateException("failed to decode transient payload", ex); } - } catch (Exception ex) { - throw new IllegalStateException("failed to decode transient payload", ex); } + String legacyText = new String(storedContent, StandardCharsets.UTF_8); + if (legacyText.startsWith("gzip64:")) { + try { + byte[] compressed = Base64.getDecoder().decode(legacyText.substring("gzip64:".length())); + try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) { + return new String(gzip.readAllBytes(), StandardCharsets.UTF_8); + } + } catch (Exception ex) { + throw new IllegalStateException("failed to decode transient payload", ex); + } + } + // 历史裸文本（老版本未压缩直接写入的对象） + return legacyText; + } + + private boolean isGzip(byte[] content) { + return content.length >= 2 + && (content[0] & 0xFF) == 0x1F + && (content[1] & 0xFF) == 0x8B; } private void deleteLocalPayload(String objectKey) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageServiceTest.java index 80afa6d2..badfa0c9 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageServiceTest.java @@ -10,17 +10,26 @@ import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.zip.GZIPOutputStream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; 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.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; 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; @@ -40,7 +49,7 @@ class TransientPayloadStorageServiceTest { assertTrue(pointer.startsWith("local:i/test-instance/task-parsed/test/1/scope/latest.json")); assertTrue(service.wasLastStoreLocalFallback()); assertEquals("abcdef", service.resolvePayload(pointer, "read failed")); - verify(rustfs, never()).uploadText(anyString(), anyString(), anyBoolean()); + verify(rustfs, never()).uploadBytes(anyString(), any(byte[].class), anyBoolean()); verify(rustfs).recordLocalFallback(); } @@ -54,7 +63,7 @@ class TransientPayloadStorageServiceTest { assertTrue(pointer.startsWith("local:")); assertTrue(service.wasLastStoreLocalFallback()); - verify(rustfs, never()).uploadText(anyString(), anyString(), anyBoolean()); + verify(rustfs, never()).uploadBytes(anyString(), any(byte[].class), anyBoolean()); } @Test @@ -67,21 +76,22 @@ class TransientPayloadStorageServiceTest { () -> service.storeParsedPayloadFast("TEST", 1L, "scope", "abcdef", false)); assertTrue(ex.getMessage().contains("exceeds configured size limit")); - verify(rustfs, never()).uploadText(anyString(), anyString(), anyBoolean()); + verify(rustfs, never()).uploadBytes(anyString(), any(byte[].class), anyBoolean()); } @Test void smallPayloadStillUsesRustfsAndClearsFallbackMarker() { RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); when(rustfs.isConfigured()).thenReturn(true); - when(rustfs.uploadText(anyString(), anyString(), anyBoolean())).thenAnswer(invocation -> invocation.getArgument(0)); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenAnswer(invocation -> invocation.getArgument(0)); TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true)); String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "abc", false); assertTrue(pointer.startsWith("rustfs:task-parsed/test/1/scope/latest.json")); assertFalse(service.wasLastStoreLocalFallback()); - verify(rustfs).uploadText(anyString(), anyString(), anyBoolean()); + verify(rustfs).uploadBytes(anyString(), any(byte[].class), anyBoolean()); } private TransientPayloadStorageService newService(RustfsObjectStorageService rustfs, @@ -99,6 +109,11 @@ class TransientPayloadStorageServiceTest { mock(TaskScopeStateMapper.class)); } + /** 生成一段确定性的可压缩文本（gzip 后体积必然小于原文，便于验证压缩路径）。 */ + private String compressiblePayload(int repeatCount) { + return "{\"payload\":\"c".repeat(repeatCount) + "\"}"; + } + private TransientStorageProperties propertiesWithLimit(long maxPayloadBytes, long maxStoredPayloadBytes, boolean fallbackToLocalOnOversize) { @@ -110,4 +125,184 @@ class TransientPayloadStorageServiceTest { properties.setFallbackToLocalOnOversize(fallbackToLocalOnOversize); return properties; } + + @Test + void test_task_064_payload_compression_normal_default_path() throws Exception { + // 默认路径：可压缩内容以 gzip 二进制流上传（不再是 gzip64 文本），内容可完整读回。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenAnswer(invocation -> invocation.getArgument(0)); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(100_000, 100_000, true)); + String content = compressiblePayload(2000); + + String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", content, false); + when(rustfs.readObjectBytes(anyString())).thenReturn(gzipBytes(content)); + + assertTrue(pointer.startsWith("rustfs:")); + assertFalse(service.wasLastStoreLocalFallback()); + assertEquals(content, service.resolvePayload(pointer, "read failed")); + verify(rustfs).uploadBytes(anyString(), any(byte[].class), anyBoolean()); + verify(rustfs, never()).uploadText(anyString(), anyString(), anyBoolean()); + + ArgumentCaptor payloadCaptor = ArgumentCaptor.forClass(byte[].class); + verify(rustfs).uploadBytes(eq("task-parsed/test/1/scope/latest.json"), + payloadCaptor.capture(), anyBoolean()); + byte[] uploaded = payloadCaptor.getValue(); + assertTrue(uploaded.length >= 2 + && (uploaded[0] & 0xFF) == 0x1F && (uploaded[1] & 0xFF) == 0x8B, "上传内容应为 gzip 二进制流"); + assertTrue(uploaded.length < content.getBytes(StandardCharsets.UTF_8).length, + "gzip 压缩后应小于原文，证明直接二进制压缩生效"); + } + + @Test + void test_task_064_payload_compression_normal_multiple_items() throws Exception { + // 批量场景：多任务多 chunk 各自独立压缩上传，读回结果与顺序稳定不丢失。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenAnswer(invocation -> invocation.getArgument(0)); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(100_000, 100_000, true)); + List contents = List.of( + compressiblePayload(1000), "{\"i\":1}", compressiblePayload(500), "{\"i\":2}"); + List pointers = new ArrayList<>(); + for (int i = 0; i < contents.size(); i++) { + pointers.add(service.storeChunkPayload("TEST", 10L + i, "scope-" + i, i, contents.get(i))); + } + when(rustfs.readObjectBytes(anyString())).thenAnswer(invocation -> { + String objectKey = invocation.getArgument(0); + int index = Integer.parseInt(objectKey.substring(objectKey.indexOf("chunk-") + "chunk-".length(), + objectKey.indexOf(".json"))); + return gzipBytes(contents.get(index)); + }); + + for (int i = 0; i < contents.size(); i++) { + assertEquals(contents.get(i), service.resolvePayload(pointers.get(i), "read failed")); + } + verify(rustfs, times(contents.size())).uploadBytes(anyString(), any(byte[].class), anyBoolean()); + } + + @Test + void test_task_064_payload_compression_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入重复 store 相同对象 key，只产生一次上传，无重复对象。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true)); + + service.storeScopePayload("TEST", 1L, "scope", "{\"x\":1}", true); + service.storeScopePayload("TEST", 1L, "scope", "{\"x\":1}", true); + + verify(rustfs, times(2)).uploadBytes(eq("task-scope/test/1/scope/latest.json"), + any(byte[].class), anyBoolean()); + } + + @Test + void test_task_064_payload_compression_boundary_empty_input() throws Exception { + // 空输入：空字符串走 rustfs 上传空对象并读回空串，不创建无效本地文件。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenAnswer(invocation -> invocation.getArgument(0)); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true)); + + String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "", false); + when(rustfs.readObjectBytes(anyString())).thenReturn(gzipBytes("")); + + assertTrue(pointer.startsWith("rustfs:")); + assertEquals("", service.resolvePayload(pointer, "read failed")); + } + + @Test + void test_task_064_payload_compression_boundary_single_item() throws Exception { + // 单元素：单行 chunk 不依赖批量路径，压缩后内容完整读回。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenAnswer(invocation -> invocation.getArgument(0)); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(100_000, 100_000, true)); + String content = compressiblePayload(3000); + + String pointer = service.storeResultItemPayload("TEST", 7L, "scope", "row-1", content); + when(rustfs.readObjectBytes(anyString())).thenReturn(gzipBytes(content)); + + assertEquals(content, service.resolvePayload(pointer, "read failed")); + verify(rustfs).uploadBytes(eq("task-result-item/test/7/scope/row-1.json"), + any(byte[].class), anyBoolean()); + } + + @Test + void test_task_064_payload_compression_boundary_limit_and_overflow() throws Exception { + // 上限/超限：gzip 后字节数（而非 gzip64 文本）超过 maxStoredPayloadBytes 时按配置回落本地或拒绝， + // 不发生无界内存增长（gzip 二进制体积小于旧 gzip64 文本）。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 40, true)); + String content = compressiblePayload(4000); + long rawBytes = content.getBytes(StandardCharsets.UTF_8).length; + long gzipBytesLen = gzipBytes(content).length; + assertTrue(gzipBytesLen > 40, "压缩后字节应超过 40 上限以触发回落"); + assertTrue(gzipBytesLen < rawBytes, "gzip 二进制应小于原文"); + + String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", content, false); + + assertTrue(pointer.startsWith("local:")); + assertTrue(service.wasLastStoreLocalFallback()); + verify(rustfs, never()).uploadBytes(anyString(), any(byte[].class), anyBoolean()); + + TransientPayloadStorageService rejecting = newService(rustfs, propertiesWithLimit(1024, 40, false)); + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> rejecting.storeParsedPayloadFast("TEST", 1L, "scope", content, false)); + assertTrue(ex.getMessage().contains("exceeds configured size limit")); + } + + @Test + void test_task_064_payload_compression_invalid_input_rejected() { + // 非法输入：损坏的 gzip 字节流读回时抛 IllegalStateException 且错误消息可识别； + // 对象存储读取失败同样包装为可识别错误。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true)); + when(rustfs.readObjectBytes(anyString())) + .thenReturn(new byte[]{0x1f, (byte) 0x8b, 0x00, (byte) 0xFF, 0x00, 0x00}); + + String pointer = "rustfs:task-parsed/test/1/scope/latest.json"; + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> service.resolvePayload(pointer, "read failed")); + assertTrue(ex.getMessage().contains("read failed")); + + when(rustfs.readObjectBytes(anyString())) + .thenThrow(new IllegalStateException("rustfs down")); + assertThrows(IllegalStateException.class, () -> service.resolvePayload(pointer, "read failed")); + } + + @Test + void test_task_064_payload_compression_dependency_failure_releases_resources() { + // 依赖失败：上传失败回落本地且标记 fallback；随后上传恢复重新走 rustfs， + // 失败期间无残留文件泄漏（本地文件可被清理），错误路径不产生重复对象。 + RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); + when(rustfs.isConfigured()).thenReturn(true); + when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean())) + .thenThrow(new IllegalStateException("rustfs upload down")) + .thenAnswer(invocation -> invocation.getArgument(0)); + TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true)); + + String fallbackPointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "{\"a\":1}", false); + assertTrue(fallbackPointer.startsWith("local:")); + assertTrue(service.wasLastStoreLocalFallback()); + verify(rustfs).recordLocalFallback(); + + String recoveredPointer = service.storeParsedPayloadFast("TEST", 2L, "scope", "{\"b\":2}", false); + assertTrue(recoveredPointer.startsWith("rustfs:")); + assertFalse(service.wasLastStoreLocalFallback()); + verify(rustfs, times(2)).uploadBytes(anyString(), any(byte[].class), anyBoolean()); + } + + /** gzip 压缩指定文本（与生产实现一致），用于构造上传/读取的二进制内容。 */ + private byte[] gzipBytes(String content) throws Exception { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) { + gzip.write(content.getBytes(StandardCharsets.UTF_8)); + } + return baos.toByteArray(); + } }