task-64: transient payload 压缩改为直接 gzip 二进制流上传

- RustfsObjectStorageService 新增 uploadBytes/readObjectBytes 二进制流 API
  (application/gzip),uploadText/readObjectAsString 改为其薄包装
- TransientPayloadStorageService 的 encodeStoredPayload 直接输出 gzip 字节,
  不再经过 gzip64 + Base64 文本编码(消除 33% 体积膨胀与中间字符串副本)
- 读取端按 gzip magic 识别二进制流解压;兼容旧 gzip64: 文本与历史裸文本
- 本地回落文件同步改为写二进制(Files.write),超限判定基于 gzip 后字节数
This commit is contained in:
2026-08-30 18:42:19 +08:00
parent f8e4360482
commit ca32cde4fe
3 changed files with 260 additions and 34 deletions
@@ -80,12 +80,17 @@ public class RustfsObjectStorageService {
} }
public String uploadText(String objectKey, String content, boolean verifyAfterUpload) { 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(); long deadlineNanos = operationDeadlineNanos();
if (!isConfigured()) { if (!isConfigured()) {
throw new IllegalStateException("transient storage is not configured"); throw new IllegalStateException("transient storage is not configured");
} }
rejectIfCircuitOpen(OP_UPLOAD, objectKey); rejectIfCircuitOpen(OP_UPLOAD, objectKey);
byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8); byte[] bytes = content == null ? new byte[0] : content;
recordPayloadBytes(bytes.length); recordPayloadBytes(bytes.length);
AtomicBoolean putCompleted = new AtomicBoolean(); AtomicBoolean putCompleted = new AtomicBoolean();
try { try {
@@ -96,7 +101,7 @@ public class RustfsObjectStorageService {
.bucket(properties.getBucket()) .bucket(properties.getBucket())
.object(objectKey) .object(objectKey)
.stream(stream, bytes.length, -1) .stream(stream, bytes.length, -1)
.contentType("application/json") .contentType("application/gzip")
.build()); .build());
putCompleted.set(true); putCompleted.set(true);
return objectKey; return objectKey;
@@ -117,6 +122,11 @@ public class RustfsObjectStorageService {
} }
public String readObjectAsString(String objectKey) { public String readObjectAsString(String objectKey) {
byte[] bytes = readObjectBytes(objectKey);
return new String(bytes, StandardCharsets.UTF_8);
}
public byte[] readObjectBytes(String objectKey) {
long deadlineNanos = operationDeadlineNanos(); long deadlineNanos = operationDeadlineNanos();
if (!isConfigured()) { if (!isConfigured()) {
throw new IllegalStateException("transient storage is not configured"); throw new IllegalStateException("transient storage is not configured");
@@ -128,7 +138,7 @@ public class RustfsObjectStorageService {
.bucket(properties.getBucket()) .bucket(properties.getBucket())
.object(objectKey) .object(objectKey)
.build())) { .build())) {
return new String(stream.readAllBytes(), StandardCharsets.UTF_8); return stream.readAllBytes();
} }
}); });
} }
@@ -145,13 +145,16 @@ public class TransientPayloadStorageService {
"transient payload only exists on instance=" + ownerInstance "transient payload only exists on instance=" + ownerInstance
+ " current=" + instanceMetadata.getInstanceId()); + " current=" + instanceMetadata.getInstanceId());
} }
return decodeStoredPayload(readLocalPayload(stripLocalInstanceId(localKey))); return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey)));
} }
if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) { 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)) { 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) { } catch (Exception ex) {
throw new IllegalStateException(errorMessage + ": " + pointer, ex); throw new IllegalStateException(errorMessage + ": " + pointer, ex);
@@ -318,8 +321,8 @@ public class TransientPayloadStorageService {
category, moduleType, taskId, objectKey, rawBytes, properties.getWarnPayloadBytes()); category, moduleType, taskId, objectKey, rawBytes, properties.getWarnPayloadBytes());
} }
boolean rawOversize = isPositiveLimit(properties.getMaxPayloadBytes()) && rawBytes > properties.getMaxPayloadBytes(); boolean rawOversize = isPositiveLimit(properties.getMaxPayloadBytes()) && rawBytes > properties.getMaxPayloadBytes();
String storedContent = encodeStoredPayload(content); byte[] storedContent = encodeStoredPayload(content);
long storedBytes = payloadBytes(storedContent); long storedBytes = storedContent.length;
boolean storedOversize = isPositiveLimit(properties.getMaxStoredPayloadBytes()) && storedBytes > properties.getMaxStoredPayloadBytes(); boolean storedOversize = isPositiveLimit(properties.getMaxStoredPayloadBytes()) && storedBytes > properties.getMaxStoredPayloadBytes();
if (rawOversize || storedOversize) { if (rawOversize || storedOversize) {
log.warn("[transient-payload] payload size exceeds rustfs limit category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} maxRawBytes={} maxStoredBytes={} fallbackToLocal={}", 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; boolean rustfsFallbackToLocal = rawOversize || storedOversize;
if (!rustfsFallbackToLocal && rustfsObjectStorageService.isConfigured()) { if (!rustfsFallbackToLocal && rustfsObjectStorageService.isConfigured()) {
try { try {
pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, storedContent, verifyAfterUpload); pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadBytes(objectKey, storedContent, verifyAfterUpload);
} catch (Exception ex) { } catch (Exception ex) {
rustfsFallbackToLocal = true; rustfsFallbackToLocal = true;
// 升级为 ERROR:rustfs 失败后只能落到本地,多实例下其他节点读不到,必须能告警。 // 升级为 ERROR:rustfs 失败后只能落到本地,多实例下其他节点读不到,必须能告警。
@@ -362,7 +365,7 @@ public class TransientPayloadStorageService {
} }
} }
private String storeLocal(String objectKey, String content) { private String storeLocal(String objectKey, byte[] content) {
try { try {
Path root = localPayloadRoot(); Path root = localPayloadRoot();
Path target = root.resolve(objectKey).normalize(); Path target = root.resolve(objectKey).normalize();
@@ -370,9 +373,8 @@ public class TransientPayloadStorageService {
throw new IllegalArgumentException("invalid local payload key: " + objectKey); throw new IllegalArgumentException("invalid local payload key: " + objectKey);
} }
Files.createDirectories(target.getParent()); Files.createDirectories(target.getParent());
Files.writeString(target, Files.write(target,
content == null ? "" : content, content == null ? new byte[0] : content,
StandardCharsets.UTF_8,
StandardOpenOption.CREATE, StandardOpenOption.CREATE,
StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.TRUNCATE_EXISTING,
StandardOpenOption.WRITE); StandardOpenOption.WRITE);
@@ -385,10 +387,10 @@ public class TransientPayloadStorageService {
} }
} }
private String readLocalPayload(String objectKey) { private byte[] readLocalPayloadBytes(String objectKey) {
try { try {
Path target = resolveLocalPayloadPath(objectKey); Path target = resolveLocalPayloadPath(objectKey);
return Files.readString(target, StandardCharsets.UTF_8); return Files.readAllBytes(target);
} catch (Exception ex) { } catch (Exception ex) {
throw new IllegalStateException("failed to read local transient payload", 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_.\\-]", "_"); return normalized.replaceAll("[^A-Za-z0-9_.\\-]", "_");
} }
private String encodeStoredPayload(String content) { /**
* 压缩 transient payload 为 gzip 二进制流(不再经过 base64 文本编码)。
* 极小内容 gzip 可能膨胀,属预期:压缩收益以可压缩内容为准,读取端按 magic 自动识别。
*/
private byte[] encodeStoredPayload(String content) {
try { try {
byte[] raw = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8); 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)) { try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
gzip.write(raw); gzip.write(raw);
} }
return "gzip64:" + Base64.getEncoder().encodeToString(baos.toByteArray()); return baos.toByteArray();
} catch (Exception ex) { } catch (Exception ex) {
throw new IllegalStateException("failed to encode transient payload", ex); throw new IllegalStateException("failed to encode transient payload", ex);
} }
@@ -456,15 +462,21 @@ public class TransientPayloadStorageService {
return value > 0L; return value > 0L;
} }
private String decodeStoredPayload(String storedContent) { private String decodeStoredPayloadBytes(byte[] storedContent) {
if (storedContent == null || storedContent.isBlank()) { if (storedContent == null || storedContent.length == 0) {
return storedContent; return "";
} }
if (!storedContent.startsWith("gzip64:")) { if (isGzip(storedContent)) {
return 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);
} }
}
String legacyText = new String(storedContent, StandardCharsets.UTF_8);
if (legacyText.startsWith("gzip64:")) {
try { try {
byte[] compressed = Base64.getDecoder().decode(storedContent.substring("gzip64:".length())); byte[] compressed = Base64.getDecoder().decode(legacyText.substring("gzip64:".length()));
try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) { try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) {
return new String(gzip.readAllBytes(), StandardCharsets.UTF_8); return new String(gzip.readAllBytes(), StandardCharsets.UTF_8);
} }
@@ -472,6 +484,15 @@ public class TransientPayloadStorageService {
throw new IllegalStateException("failed to decode transient payload", 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) { private void deleteLocalPayload(String objectKey) {
try { try {
@@ -10,17 +10,26 @@ import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir; 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.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.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue; 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.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; 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(pointer.startsWith("local:i/test-instance/task-parsed/test/1/scope/latest.json"));
assertTrue(service.wasLastStoreLocalFallback()); assertTrue(service.wasLastStoreLocalFallback());
assertEquals("abcdef", service.resolvePayload(pointer, "read failed")); 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(); verify(rustfs).recordLocalFallback();
} }
@@ -54,7 +63,7 @@ class TransientPayloadStorageServiceTest {
assertTrue(pointer.startsWith("local:")); assertTrue(pointer.startsWith("local:"));
assertTrue(service.wasLastStoreLocalFallback()); assertTrue(service.wasLastStoreLocalFallback());
verify(rustfs, never()).uploadText(anyString(), anyString(), anyBoolean()); verify(rustfs, never()).uploadBytes(anyString(), any(byte[].class), anyBoolean());
} }
@Test @Test
@@ -67,21 +76,22 @@ class TransientPayloadStorageServiceTest {
() -> service.storeParsedPayloadFast("TEST", 1L, "scope", "abcdef", false)); () -> service.storeParsedPayloadFast("TEST", 1L, "scope", "abcdef", false));
assertTrue(ex.getMessage().contains("exceeds configured size limit")); 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 @Test
void smallPayloadStillUsesRustfsAndClearsFallbackMarker() { void smallPayloadStillUsesRustfsAndClearsFallbackMarker() {
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class); RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
when(rustfs.isConfigured()).thenReturn(true); 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)); TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true));
String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "abc", false); String pointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "abc", false);
assertTrue(pointer.startsWith("rustfs:task-parsed/test/1/scope/latest.json")); assertTrue(pointer.startsWith("rustfs:task-parsed/test/1/scope/latest.json"));
assertFalse(service.wasLastStoreLocalFallback()); assertFalse(service.wasLastStoreLocalFallback());
verify(rustfs).uploadText(anyString(), anyString(), anyBoolean()); verify(rustfs).uploadBytes(anyString(), any(byte[].class), anyBoolean());
} }
private TransientPayloadStorageService newService(RustfsObjectStorageService rustfs, private TransientPayloadStorageService newService(RustfsObjectStorageService rustfs,
@@ -99,6 +109,11 @@ class TransientPayloadStorageServiceTest {
mock(TaskScopeStateMapper.class)); mock(TaskScopeStateMapper.class));
} }
/** 生成一段确定性的可压缩文本(gzip 后体积必然小于原文,便于验证压缩路径)。 */
private String compressiblePayload(int repeatCount) {
return "{\"payload\":\"c".repeat(repeatCount) + "\"}";
}
private TransientStorageProperties propertiesWithLimit(long maxPayloadBytes, private TransientStorageProperties propertiesWithLimit(long maxPayloadBytes,
long maxStoredPayloadBytes, long maxStoredPayloadBytes,
boolean fallbackToLocalOnOversize) { boolean fallbackToLocalOnOversize) {
@@ -110,4 +125,184 @@ class TransientPayloadStorageServiceTest {
properties.setFallbackToLocalOnOversize(fallbackToLocalOnOversize); properties.setFallbackToLocalOnOversize(fallbackToLocalOnOversize);
return properties; 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<byte[]> 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<String> contents = List.of(
compressiblePayload(1000), "{\"i\":1}", compressiblePayload(500), "{\"i\":2}");
List<String> 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();
}
} }