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);