From d5952945dd90016184f53d0f541adcf4eb3bd60a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Thu, 17 Sep 2026 00:58:01 +0800 Subject: [PATCH] =?UTF-8?q?fix(rustfs):=20=E7=A1=AE=E5=AE=9A=E6=80=A7?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E4=B8=8D=E5=86=8D=E9=87=8D=E8=AF=95=EF=BC=9B?= =?UTF-8?q?=E5=9B=9E=E9=80=80=E4=B8=A4=E5=A4=84=E7=BB=8F=E5=AE=9E=E6=B5=8B?= =?UTF-8?q?=E6=97=A0=E6=95=88=E7=9A=84=E8=BF=9E=E6=8E=A5=E6=B1=A0=E5=AE=9E?= =?UTF-8?q?=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 isNonRetryable：NoSuchKey / AccessDenied / 签名错误等确定性错误立即失败， 不再白打两次请求（线上 read 一个已清理的 chunk 会连打 3 次并留下一条 ERROR） - connectionPoolMaxIdle 0→5、keepAlive 30000→300000 回退： 实测证明 unexpected end of stream 与连接复用无关——禁用复用后新容器起来的 第一次请求照样中招。至此已排除公网链路、keepAlive 过长、连接复用三项； 另用 mc 并发压 200 个小对象全部成功，说明 RustFS 服务端无问题。 剩余方向指向 MinIO SDK(8.5.17，已是最新)/OkHttp 与 RustFS 的协议细节， 需抓包对比 mc 与 Java 的请求才能定位。数据不丢（重试都能成功），影响是每次 +600ms。 --- .../config/TransientStorageProperties.java | 15 ++++++-- .../object/RustfsObjectStorageService.java | 37 +++++++++++++++++++ 2 files changed, 48 insertions(+), 4 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 aab05e41..8824cf71 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 @@ -37,13 +37,20 @@ public class TransientStorageProperties { private long failureCooldownMillis = 10000; private int dispatcherMaxRequests = 56; private int dispatcherMaxRequestsPerHost = 56; + /** + * 空闲连接保留数。 + * + *

2026-09-17 曾试过设 0（彻底不复用）来验证"unexpected end of stream 是复用死连接导致的" + * 这一假设——**实测照旧失败**（新容器起来后第一次请求就中招）。至此已排除公网链路、 + * keepAlive 过长、连接复用三项；用 mc 并发压 200 个小对象也全部成功，说明服务端没问题。 + * 剩余方向指向 MinIO Java SDK / OkHttp 与 RustFS 的协议细节，故恢复默认的连接复用。 + */ private int connectionPoolMaxIdle = 5; /** - * 空闲连接在池里的保留时长。默认 5 分钟（OkHttp 原值）会让客户端复用"已被 RustFS - * 或中间设备关掉的空闲连接"，表现为 unexpected end of stream —— 线上每天上千次， - * 全靠重试兜底。对象存储访问是突发型，连接复用率本就低，缩短保活几乎没有代价。 + * 空闲连接在池里的保留时长。曾由 300000 调到 30000 试图减少 unexpected end of stream， + * 实测无改善（该现象与连接复用无关，见 {@link #connectionPoolMaxIdle} 的排查记录），故恢复原值。 */ - private long connectionPoolKeepAliveMillis = 30000; + private long connectionPoolKeepAliveMillis = 300000; private long warnPayloadBytes = 5L * 1024 * 1024; private long maxPayloadBytes = 50L * 1024 * 1024; private long maxStoredPayloadBytes = 50L * 1024 * 1024; 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 dc7f9145..355bb1bf 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 @@ -9,6 +9,7 @@ import io.minio.MinioClient; import io.minio.PutObjectArgs; import io.minio.RemoveObjectArgs; import io.minio.StatObjectArgs; +import io.minio.errors.ErrorResponseException; import lombok.extern.slf4j.Slf4j; import okhttp3.ConnectionPool; import okhttp3.Dispatcher; @@ -21,6 +22,7 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.nio.charset.StandardCharsets; import java.util.Objects; +import java.util.Set; import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -38,6 +40,11 @@ public class RustfsObjectStorageService { private static final String OP_STAT = "stat"; private static final String OP_TOTAL = "total"; + /** S3 确定性错误码：重试不会改变结果，应立即失败而不是白打两次请求。 */ + private static final Set NON_RETRYABLE_S3_CODES = Set.of( + "NoSuchKey", "NoSuchBucket", "NoSuchVersion", + "AccessDenied", "InvalidAccessKeyId", "SignatureDoesNotMatch", "InvalidBucketName"); + private final TransientStorageProperties properties; private final ObjectProvider meterRegistryProvider; private final ObjectProvider deleteRetryServiceProvider; @@ -251,6 +258,15 @@ public class RustfsObjectStorageService { last = ex; recordOperation(operation, attempt < maxRetries ? "retry" : "failure", elapsedNanos(startedAt)); recordFailure(operation, objectKey, ex); + if (isNonRetryable(ex)) { + // 确定性错误（NoSuchKey / AccessDenied…）：重试多少次结果都一样。 + // 线上 read 一个已被清理的 chunk 就会连打 3 次请求、还被记成 ERROR。 + log.warn("[rustfs] 确定性错误，不重试 operation={} objectKey={} err={}", + operation, objectKey, ex.getMessage()); + throw ex instanceof RuntimeException runtimeException + ? runtimeException + : new IllegalStateException(ex); + } if (attempt < maxRetries) { delayMillis = retryDelayMillis(attempt); log.warn("[rustfs] operation failed, retrying operation={} objectKey={} attempt={}/{} delayMs={} err={}", @@ -279,6 +295,27 @@ public class RustfsObjectStorageService { } } + /** + * 是否为「重试也没用」的确定性错误（NoSuchKey / AccessDenied / 签名错误…）。 + * + *

只有网络类与 5xx 类失败才值得重试；确定性错误重试多少次结果都一样。 + */ + private static boolean isNonRetryable(Throwable error) { + Throwable cursor = error; + while (cursor != null) { + if (cursor instanceof ErrorResponseException responseException) { + String code = responseException.errorResponse() == null + ? null + : responseException.errorResponse().code(); + if (code != null && NON_RETRYABLE_S3_CODES.contains(code)) { + return true; + } + } + cursor = cursor.getCause(); + } + return false; + } + private void acquirePermit(String operation, String objectKey, Semaphore semaphore, long deadlineNanos) { try { checkDeadline(operation, objectKey, deadlineNanos);