From c448f49e3088988dacdac7d49a824b6c7099c3e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Mon, 14 Sep 2026 02:31:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E4=B8=B4=E6=97=B6=E5=AD=98=E5=82=A8):=20Ru?= =?UTF-8?q?stFS=20=E4=B8=8A=E4=BC=A0=E5=A4=B1=E8=B4=A5=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E5=A4=B1=E8=B4=A5=EF=BC=8C=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E9=9D=99=E9=BB=98=E5=9B=9E=E8=90=BD=20local=20=E6=8C=87?= =?UTF-8?q?=E9=92=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 多实例/容器化部署下 local 指针只有写入它的实例能读（跨节点读直接报错、容器重建即丢）， 此前上传失败静默降级会把跨节点不可读的脏指针落库，故障延后到其它节点的合并/组装才爆。 现改为在写失败点抛 BusinessException（中文原因透传调用方），并新增 fallback-to-local-on-error 开关（默认 false）供单机部署回退旧行为； 跨实例读 local 指针的报错改中文并带 objectKey；超限回落策略不变。 --- .../config/TransientStorageProperties.java | 6 ++++ .../TransientPayloadStorageService.java | 23 ++++++++++---- .../src/main/resources/application.yml | 1 + .../TransientPayloadStorageServiceTest.java | 30 +++++++++++++++++-- 4 files changed, 51 insertions(+), 9 deletions(-) diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java index f9f2cbd1..7c6fc927 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java @@ -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; 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 8336abff..69d3a3e1 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 @@ -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) { diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index f89eaa83..043bbb4f 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -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} 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 badfa0c9..f10c7cce 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 @@ -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:"));