fix(rustfs): 确定性错误不再重试;回退两处经实测无效的连接池实验

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