fix(临时存储): RustFS 上传失败改为直接失败，不再静默回落 local 指针
多实例/容器化部署下 local 指针只有写入它的实例能读（跨节点读直接报错、容器重建即丢）， 此前上传失败静默降级会把跨节点不可读的脏指针落库，故障延后到其它节点的合并/组装才爆。 现改为在写失败点抛 BusinessException（中文原因透传调用方），并新增 fallback-to-local-on-error 开关（默认 false）供单机部署回退旧行为； 跨实例读 local 指针的报错改中文并带 objectKey；超限回落策略不变。
This commit is contained in:
@@ -48,6 +48,12 @@ public class TransientStorageProperties {
|
||||
*/
|
||||
private long maxDecompressedPayloadBytes = 100L * 1024 * 1024;
|
||||
private boolean fallbackToLocalOnOversize = true;
|
||||
/**
|
||||
* 上传失败（已重试+熔断）时是否回落到本机磁盘。默认 false：
|
||||
* 多实例/容器化部署下 local 指针只有写入它的那个实例能读，宁可让本次写入失败，
|
||||
* 也不要把跨节点不可读的脏指针交给调用方；仅单机部署才应打开。
|
||||
*/
|
||||
private boolean fallbackToLocalOnError = false;
|
||||
private boolean deleteRetryEnabled = true;
|
||||
private String deleteRetryCron = "0 */5 * * * *";
|
||||
private int deleteRetryQueueCapacity = 10000;
|
||||
|
||||
+17
-6
@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.task.service;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.common.exception.BusinessException;
|
||||
import com.nanri.aiimage.config.InstanceMetadata;
|
||||
import com.nanri.aiimage.config.StorageProperties;
|
||||
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||
@@ -142,8 +143,9 @@ public class TransientPayloadStorageService {
|
||||
if (ownerInstance != null && !ownerInstance.equals(instanceMetadata.getInstanceId())) {
|
||||
// 跨实例消费 local 指针时立即报错，避免悄悄读出空内容造成下游错位。
|
||||
throw new IllegalStateException(
|
||||
"transient payload only exists on instance=" + ownerInstance
|
||||
+ " current=" + instanceMetadata.getInstanceId());
|
||||
"临时载荷仅存在于实例 " + ownerInstance + "（当前实例 "
|
||||
+ instanceMetadata.getInstanceId() + "），本地兜底指针跨节点不可读; objectKey="
|
||||
+ stripLocalInstanceId(localKey));
|
||||
}
|
||||
return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey)));
|
||||
}
|
||||
@@ -312,6 +314,9 @@ public class TransientPayloadStorageService {
|
||||
String content,
|
||||
boolean encodeAsJsonString,
|
||||
boolean verifyAfterUpload) {
|
||||
// 标记位必须先重置：本方法会在 isWriteEnabled 不满足时提前返回，
|
||||
// 否则调用方可能读到同一线程上一次 store 的陈旧值。
|
||||
LAST_STORE_LOCAL_FALLBACK.set(Boolean.FALSE);
|
||||
if (!isWriteEnabled()) {
|
||||
return content;
|
||||
}
|
||||
@@ -339,10 +344,16 @@ public class TransientPayloadStorageService {
|
||||
try {
|
||||
pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadBytes(objectKey, storedContent, verifyAfterUpload);
|
||||
} catch (Exception ex) {
|
||||
rustfsFallbackToLocal = true;
|
||||
// 升级为 ERROR：rustfs 失败后只能落到本地，多实例下其他节点读不到，必须能告警。
|
||||
log.error("[transient-payload] rustfs upload failed, fallback to local store instanceId={} category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} err={}",
|
||||
instanceMetadata.getInstanceId(), category, moduleType, taskId, objectKey, rawBytes, storedBytes, ex.getMessage());
|
||||
// 多实例/容器化部署下 local 指针只有写入它的那个实例能读，跨节点消费会直接报错；
|
||||
// 默认（fallbackToLocalOnError=false）在写失败点直接抛错，失败归属清晰；
|
||||
// 仅单机部署才应把开关打开，回退到"写本地磁盘"的旧行为。
|
||||
rustfsFallbackToLocal = properties.isFallbackToLocalOnError();
|
||||
log.error("[transient-payload] rustfs upload failed, fallbackToLocal={} instanceId={} category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} err={}",
|
||||
rustfsFallbackToLocal, instanceMetadata.getInstanceId(), category, moduleType, taskId, objectKey, rawBytes, storedBytes, ex.getMessage());
|
||||
if (!rustfsFallbackToLocal) {
|
||||
throw new BusinessException("临时存储(RustFS)不可用，载荷写入失败，请稍后重试: moduleType="
|
||||
+ moduleType + " taskId=" + taskId + " objectKey=" + objectKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (pointer == null) {
|
||||
|
||||
@@ -155,6 +155,7 @@ aiimage:
|
||||
max-payload-bytes: ${AIIMAGE_TRANSIENT_STORAGE_MAX_PAYLOAD_BYTES:52428800}
|
||||
max-stored-payload-bytes: ${AIIMAGE_TRANSIENT_STORAGE_MAX_STORED_PAYLOAD_BYTES:52428800}
|
||||
fallback-to-local-on-oversize: ${AIIMAGE_TRANSIENT_STORAGE_FALLBACK_TO_LOCAL_ON_OVERSIZE:true}
|
||||
fallback-to-local-on-error: ${AIIMAGE_TRANSIENT_STORAGE_FALLBACK_TO_LOCAL_ON_ERROR:false}
|
||||
delete-retry-enabled: ${AIIMAGE_TRANSIENT_STORAGE_DELETE_RETRY_ENABLED:true}
|
||||
delete-retry-cron: ${AIIMAGE_TRANSIENT_STORAGE_DELETE_RETRY_CRON:0 */5 * * * *}
|
||||
delete-retry-queue-capacity: ${AIIMAGE_TRANSIENT_STORAGE_DELETE_RETRY_QUEUE_CAPACITY:10000}
|
||||
|
||||
+27
-3
@@ -1,6 +1,7 @@
|
||||
package com.nanri.aiimage.modules.task.service;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.common.exception.BusinessException;
|
||||
import com.nanri.aiimage.config.InstanceMetadata;
|
||||
import com.nanri.aiimage.config.StorageProperties;
|
||||
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||
@@ -275,16 +276,39 @@ class TransientPayloadStorageServiceTest {
|
||||
assertThrows(IllegalStateException.class, () -> service.resolvePayload(pointer, "read failed"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void uploadFailureFailsFastByDefaultWithoutLocalPointer() {
|
||||
// 默认（fallbackToLocalOnError=false）：上传失败直接抛业务异常，不写本地文件、
|
||||
// 不置 fallback 标记、不记降级指标——避免产生跨实例不可读的 local 脏指针。
|
||||
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||
when(rustfs.isConfigured()).thenReturn(true);
|
||||
when(rustfs.uploadBytes(anyString(), any(byte[].class), anyBoolean()))
|
||||
.thenThrow(new IllegalStateException("rustfs upload down"));
|
||||
TransientPayloadStorageService service = newService(rustfs, propertiesWithLimit(1024, 4096, true));
|
||||
|
||||
BusinessException ex = assertThrows(BusinessException.class,
|
||||
() -> service.storeChunkPayload("TEST", 9L, "scope", 1, "{\"a\":1}"));
|
||||
|
||||
assertTrue(ex.getMessage().contains("临时存储(RustFS)不可用"));
|
||||
assertTrue(ex.getMessage().contains("objectKey=task-chunk/test/9/scope/chunk-1.json"));
|
||||
assertFalse(service.wasLastStoreLocalFallback());
|
||||
verify(rustfs, never()).recordLocalFallback();
|
||||
assertFalse(java.nio.file.Files.exists(tempDir.resolve("transient-payload")),
|
||||
"失败时不应落任何本地临时文件");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_064_payload_compression_dependency_failure_releases_resources() {
|
||||
// 依赖失败：上传失败回落本地且标记 fallback；随后上传恢复重新走 rustfs，
|
||||
// 失败期间无残留文件泄漏（本地文件可被清理），错误路径不产生重复对象。
|
||||
// 依赖失败：开关显式打开（单机降级模式）时上传失败回落本地且标记 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));
|
||||
TransientStorageProperties properties = propertiesWithLimit(1024, 4096, true);
|
||||
properties.setFallbackToLocalOnError(true);
|
||||
TransientPayloadStorageService service = newService(rustfs, properties);
|
||||
|
||||
String fallbackPointer = service.storeParsedPayloadFast("TEST", 1L, "scope", "{\"a\":1}", false);
|
||||
assertTrue(fallbackPointer.startsWith("local:"));
|
||||
|
||||
Reference in New Issue
Block a user