gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-11-08 03:23:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: merge main into buffer pool branch

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser Zhi
2026-11-04 22:39:39 +08:00
parent cc7fe612f0 fe3ada0689
commit 2d488da93f
10 changed files with 71 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+14 -16
View File
@@ -4731,7 +4731,6 @@ version = "1.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e"
dependencies = [
"generic-array 0.14.7",
"rustversion",
"typenum",
]
@@ -9349,9 +9348,9 @@ dependencies = [
[[package]]
name = "russh"
version = "0.65.0"
version = "0.65.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3bb77a4b259312b2abbc541bca1c13cb4e9b61c560a7fd894068c3b92e0c9dc3"
checksum = "94be3d86c71b9ed220a675c405e35c7a02565a009bf8da2e9cc13a700320a5c5"
dependencies = [
"aes 0.9.4",
"aws-lc-rs",
@@ -10116,9 +10115,9 @@ dependencies = [
[[package]]
name = "rustfs-kafka"
version = "1.4.0"
version = "1.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bd2f8b46ec4dd462c2235948d1efa2103957230c171e118c6ff8e5d1a845ab48"
checksum = "82dc39ba1a5a9f6b2e49decb0b54c23ebaebdfb08a899307c266694bfd269bf9"
dependencies = [
"base64 0.23.1",
"bytes",
@@ -10141,9 +10140,9 @@ dependencies = [
[[package]]
name = "rustfs-kafka-async"
version = "1.4.0"
version = "1.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b3b98705f6f9f5071b8793d29935de764c31668c6e9877553b229eb67127af6"
checksum = "684f5e78ea1421bbf915e4cd0515bdefd587cfa281da000e41ae4a434f90195a"
dependencies = [
"base64 0.23.1",
"bytes",
@@ -11202,15 +11201,14 @@ dependencies = [
[[package]]
name = "rustls-connector"
version = "0.23.9"
version = "0.23.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "524994fd6957827c775abc941f41a146ce351bbd0a9b8cf632ee37f43a13c205"
checksum = "f799ef81c26fda57609b9456ae1e6750e2ca3dafc2dc5478232c0705f1458fbf"
dependencies = [
"futures-io",
"futures-rustls",
"log",
"rustls",
"rustls-pki-types",
"rustls-platform-verifier",
"rustls-webpki",
]
@@ -12861,9 +12859,9 @@ dependencies = [
[[package]]
name = "toml"
version = "1.1.8+spec-1.1.0"
version = "1.2.0+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "20489e00e4d8741d6be680764cc12e270655e375a20d1011e844a9c3379e678d"
checksum = "c5ffcdf8f50436dbbfd675c1fe1b8e0854f42b7defea8584bdebe1aee900e561"
dependencies = [
"indexmap 2.14.2",
"serde_core",
@@ -12885,9 +12883,9 @@ dependencies = [
[[package]]
name = "toml_edit"
version = "0.25.17+spec-1.1.0"
version = "0.25.18+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3641d5bbb5349a79e1020a242d251efbc546ad8048d133958323ce9c40a9c9c"
checksum = "1ec6e78bd70db9420d8d212fa9dc97f116448a1ceff5d00eabb991d407e21396"
dependencies = [
"indexmap 2.14.2",
"toml_datetime",
@@ -12906,9 +12904,9 @@ dependencies = [
[[package]]
name = "toml_writer"
version = "1.1.3+spec-1.1.0"
version = "1.2.0+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
checksum = "009dbb7700f11c55ef83aebbe4c0e9cf1f1ac0844bd8d6f00dc38f302026e13a"
[[package]]
name = "tonic"
Cargo.toml
+3 -3
View File
@@ -168,7 +168,7 @@ http-body-util = "0.1.5"
memchr = "2.8.3"
minlz = "1.2.3"
reqwest = "0.13.5"
rustfs-kafka-async = { version = "1.4.0" }
rustfs-kafka-async = { version = "1.4.1" }
socket2 = { version = "0.6.5" }
tokio = { version = "1.54.1" }
tokio-rustls = { default-features = false, version = "0.26.6" }
@@ -212,7 +212,7 @@ serde_urlencoded = "0.7.1"
# releases.
aes-gcm = { version = "0.11.1" }
argon2 = { version = "0.6.0" }
aws-lc-rs = { version = "1.18.0", default-features = false }
aws-lc-rs = { version = "1.19.0", default-features = false }
blake2 = "0.11.0"
chacha20poly1305 = { version = "0.11.0" }
crc-fast = "1.10.0"
@@ -378,7 +378,7 @@ libunftp = { version = "0.24.0" }
unftp-core = "0.1.0"
suppaftp = { version = "12.2.1" }
rcgen = { version = "0.14.10", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
russh = { version = "0.65.0" }
russh = { version = "0.65.1" }
russh-sftp = "3.0.1"
# WebDAV
crates/zstd-compat/Cargo.toml
+1 -1
View File
@@ -30,7 +30,7 @@ zstdmt = ["zstd_fixed/zstdmt"]
[dev-dependencies]
zstd-safe = { version = "8.2.0", features = ["seekable"] }
toml = "1.1.8"
toml = "1.2.0"
semver = "1.0.28"
apache-avro.workspace = true
arrow-array = "60.1.0"
rustfs/src/app/metadata_route.rs
+2 -6
View File
@@ -689,18 +689,14 @@ async fn call_list_object_versions(req: S3Request<Body>, bucket: String) -> S3Re
let request = req.map_input(|_| input);
let output = DefaultBucketUsecase::from_global()
.execute_list_object_versions_m(request)
.await
.map_err(crate::error::normalize_read_quorum_error)?;
.await?;
xml_response(&output.output)
}
async fn call_list_objects_v2(req: S3Request<Body>, bucket: String) -> S3Result<S3Response<Body>> {
let input = list_objects_v2_input(bucket, &req.uri, &req.headers)?;
let request = req.map_input(|_| input);
let output = DefaultBucketUsecase::from_global()
.execute_list_objects_v2m(request)
.await
.map_err(crate::error::normalize_read_quorum_error)?;
let output = DefaultBucketUsecase::from_global().execute_list_objects_v2m(request).await?;
xml_response(&output.output)
}
rustfs/src/app/object/put.rs
+5 -11
View File
@@ -18,7 +18,7 @@ use super::*;
use crate::app::trailer_adapter::trailer_source;
use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length};
use crate::error::{ClientBodyReadTimeout, UploadLimitExceeded, normalize_read_quorum_error};
use crate::error::{ClientBodyReadTimeout, UploadLimitExceeded};
static PUT_FAILURE_LOGS: rustfs_utils::LogThrottle = rustfs_utils::LogThrottle::new(5_000);
#[cfg(test)]
@@ -1117,11 +1117,6 @@ impl PutObjectCompletion {
/// Record the failed completion and hand the error back to the caller.
fn fail_put_object(completion: PutObjectCompletion, err: S3Error) -> S3Error {
let err = if matches!(&completion, PutObjectCompletion::S3(_)) {
normalize_read_quorum_error(err)
} else {
err
};
let result: S3Result<S3Response<()>> = Err(err);
let _ = completion.complete(&result);
match result {
@@ -1209,9 +1204,7 @@ impl DefaultObjectUsecase {
#[instrument(name = "execute_put_object", level = "info", skip(self, _fs, req))]
#[hotpath::measure(impl_type = "DefaultObjectUsecase", future = true)]
pub async fn execute_put_object(&self, _fs: &FS, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
self.execute_put_object_boxed(_fs, req)
.await
.map_err(normalize_read_quorum_error)
self.execute_put_object_boxed(_fs, req).await
}
fn execute_put_object_boxed<'a>(
@@ -2526,13 +2519,14 @@ mod tests {
}
#[test]
fn failed_put_completion_normalizes_only_s3_errors() {
fn failed_put_completion_keeps_write_safe_quorum_code() {
let error = || S3Error::from(ApiError::from(StorageError::ErasureReadQuorum));
let err = fail_put_object(PutObjectCompletion::S3(OperationHelper::Disabled), error());
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
assert_eq!(err.message(), Some("The service is unavailable. Please retry."));
// Internal completions share the write-safe default; SlowDownRead is GET-only.
let internal = fail_put_object(PutObjectCompletion::Internal(None), error());
assert_eq!(internal.code(), &S3ErrorCode::Custom("SlowDownRead".into()));
assert_eq!(internal.code(), &S3ErrorCode::ServiceUnavailable);
}
use s3s::dto::{
DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule, ServerSideEncryptionByDefault,
rustfs/src/app/object/shared.rs
+2 -1
View File
@@ -479,7 +479,8 @@ mod bucket_default_sse_lookup_tests {
let err = classify_bucket_default_sse_lookup("bucket", Err(StorageError::ErasureReadQuorum))
.expect_err("an unreadable metadata subsystem must never degrade to plaintext");
assert_eq!(err.code(), &S3ErrorCode::Custom("SlowDownRead".into()));
// Write path: retryable ServiceUnavailable, not the GET-only SlowDownRead.
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
}
#[test]
rustfs/src/error.rs
+35 -53
View File
@@ -207,37 +207,14 @@ pub(crate) fn slow_down_read_api_error(err: StorageError) -> ApiError {
}
}
/// Normalize a storage read-quorum failure at PUT and LIST boundaries.
///
/// Storage-level read quorum failures keep their `SlowDownRead` code for GET
/// compatibility. PUT and LIST expose storage unavailability instead of
/// throttling, while retaining the typed source for diagnostics and audit consumers.
pub(crate) fn normalize_read_quorum_error(mut err: S3Error) -> S3Error {
let is_slow_down_read = matches!(
err.code(),
S3ErrorCode::Custom(code) if &**code == SLOW_DOWN_READ_CODE
);
if !is_slow_down_read {
return err;
}
let has_read_quorum_source = err.source().is_some_and(|source| {
error_chain_find(source, |error| {
matches!(
error.downcast_ref::<StorageError>(),
Some(StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _))
)
.then_some(())
})
.is_some()
});
if has_read_quorum_source {
err.set_code(S3ErrorCode::ServiceUnavailable);
err.set_message(ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable));
err.set_status_code(StatusCode::SERVICE_UNAVAILABLE);
}
err
/// True when a storage failure means the object is not readable right now
/// (GET/HEAD `SlowDownRead`), as opposed to write-path unavailability.
/// Writes and listings never use this predicate; they take `ApiError::from`.
pub(crate) fn is_object_read_unavailable(err: &StorageError) -> bool {
matches!(
err,
StorageError::PartMissingOrCorrupt | StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _)
)
}
/// Marks a request body that exceeded a presigned upload size capability.
@@ -826,9 +803,10 @@ impl From<StorageError> for ApiError {
| StorageError::DiskNotFound
| StorageError::Timeout
| StorageError::TooManyOpenFiles => S3ErrorCode::ServiceUnavailable,
StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => {
S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())
}
// Write-safe default: a read-quorum failure on PUT/multipart/LIST
// must not look like a GET "resource is unreadable" throttle. GET
// object reads opt into `SlowDownRead` via `map_get_object_reader_error`.
StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable,
StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest,
@@ -2332,7 +2310,7 @@ mod tests {
}
#[test]
fn put_read_quorum_error_preserves_context_and_is_idempotent() {
fn write_path_quorum_errors_are_service_unavailable() {
for source in [
StorageError::ErasureReadQuorum,
StorageError::InsufficientReadQuorum("bucket".into(), "object".into()),
@@ -2347,7 +2325,6 @@ mod tests {
headers.insert("retry-after", http::HeaderValue::from_static("2"));
err.set_headers(headers);
let err = normalize_read_quorum_error(normalize_read_quorum_error(err));
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
assert_eq!(err.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
assert_eq!(err.message(), Some("The service is unavailable. Please retry."));
@@ -2361,6 +2338,21 @@ mod tests {
}
}
#[test]
fn get_object_reader_keeps_slow_down_read_for_quorum() {
for source in [
StorageError::ErasureReadQuorum,
StorageError::InsufficientReadQuorum("bucket".into(), "object".into()),
StorageError::PartMissingOrCorrupt,
] {
assert!(is_object_read_unavailable(&source));
let api_error = slow_down_read_api_error(source.clone());
assert_eq!(api_error.code, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()));
assert_eq!(api_error.message, SLOW_DOWN_READ_MESSAGE);
}
assert!(!is_object_read_unavailable(&StorageError::ErasureWriteQuorum));
}
#[test]
fn put_read_quorum_error_preserves_other_error_contracts() {
for code in [
@@ -2378,24 +2370,11 @@ mod tests {
let mut err = S3Error::with_message(code.clone(), "original message");
err.set_source(Box::new(StorageError::ErasureReadQuorum));
let original_status = err.status_code();
let err = normalize_read_quorum_error(err);
// Non-quorum codes must not be rewritten by the write-path mapping.
assert_eq!(err.code(), &code);
assert_eq!(err.message(), Some("original message"));
assert_eq!(err.status_code(), original_status);
}
for source in [
None,
Some(StorageError::ErasureWriteQuorum),
Some(StorageError::PartMissingOrCorrupt),
] {
let mut err = S3Error::with_message(S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()), "original message");
if let Some(source) = source {
err.set_source(Box::new(source));
}
let err = normalize_read_quorum_error(err);
assert_eq!(err.code(), &S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()));
assert_eq!(err.message(), Some("original message"));
}
}
#[test]
@@ -2490,10 +2469,11 @@ mod tests {
(StorageError::FaultyRemoteDisk, S3ErrorCode::ServiceUnavailable),
(StorageError::DiskNotFound, S3ErrorCode::ServiceUnavailable),
(StorageError::TooManyOpenFiles, S3ErrorCode::ServiceUnavailable),
(StorageError::ErasureReadQuorum, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())),
// Write-safe default; GET object reads opt into SlowDownRead.
(StorageError::ErasureReadQuorum, S3ErrorCode::ServiceUnavailable),
(
StorageError::InsufficientReadQuorum("test".into(), "test".into()),
S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()),
S3ErrorCode::ServiceUnavailable,
),
(StorageError::ErasureWriteQuorum, S3ErrorCode::ServiceUnavailable),
(
@@ -2668,7 +2648,9 @@ mod tests {
StorageError::ErasureReadQuorum,
StorageError::InsufficientReadQuorum("bucket".into(), "object".into()),
] {
let api_error = ApiError::from(error);
// MinIO-compatible GET object-read response.
assert!(is_object_read_unavailable(&error));
let api_error = slow_down_read_api_error(error);
assert_eq!(api_error.code, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()));
assert_eq!(api_error.message, SLOW_DOWN_READ_MESSAGE);
rustfs/src/storage/access.rs
+1 -2
View File
@@ -22,7 +22,7 @@ use crate::auth::{
get_request_auth_type_with_query, get_session_token, parse_presigned_multipart_max_total_object_size,
parse_presigned_put_max_content_length, reject_unsigned_amz_headers_on_sigv4_request,
};
use crate::error::{ApiError, normalize_read_quorum_error};
use crate::error::ApiError;
use crate::license::license_check;
use crate::shared_types::RemoteAddr;
use crate::storage::denial_audit::DenialAuditor;
@@ -3579,7 +3579,6 @@ impl S3Access for FS {
Ok(())
}
.await
.map_err(normalize_read_quorum_error)
}
/// Checks whether the PutObjectAcl request has accesses to the resources.
rustfs/src/storage/ecfs.rs
+3 -12
View File
@@ -1539,20 +1539,14 @@ impl S3 for FS {
) -> S3Result<S3Response<ListObjectVersionsOutput>> {
record_s3_op(S3Operation::ListObjectVersions);
let usecase = s3_api::bucket_usecase_for(self);
usecase
.execute_list_object_versions(req)
.await
.map_err(crate::error::normalize_read_quorum_error)
usecase.execute_list_object_versions(req).await
}
#[instrument(level = "debug", skip(self, req))]
async fn list_objects(&self, req: S3Request<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
record_s3_op(S3Operation::ListObjects);
let usecase = s3_api::bucket_usecase_for(self);
usecase
.execute_list_objects(req)
.await
.map_err(crate::error::normalize_read_quorum_error)
usecase.execute_list_objects(req).await
}
#[instrument(level = "debug", skip(self, req))]
@@ -1560,10 +1554,7 @@ impl S3 for FS {
crate::hp_guard!("S3::list_objects_v2");
record_s3_op(S3Operation::ListObjectsV2);
let usecase = s3_api::bucket_usecase_for(self);
usecase
.execute_list_objects_v2(req)
.await
.map_err(crate::error::normalize_read_quorum_error)
usecase.execute_list_objects_v2(req).await
}
#[instrument(level = "debug", skip(self, req))]
rustfs/src/storage/sse.rs
+5 -3
View File
@@ -159,7 +159,7 @@ fn md5_base64(input: impl AsRef<[u8]>) -> String {
use super::Error;
use super::get_bucket_sse_config;
use crate::error::{ApiError, slow_down_read_api_error};
use crate::error::ApiError;
use rustfs_utils::http::headers::{
AMZ_ENCRYPTION_AES, AMZ_ENCRYPTION_KMS, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT,
@@ -712,8 +712,10 @@ pub(crate) fn validate_sse_headers_for_read(metadata: &HashMap<String, String>,
}
pub(crate) fn map_get_object_reader_error(err: StorageError) -> ApiError {
if matches!(err, StorageError::PartMissingOrCorrupt) {
return slow_down_read_api_error(err);
// GET/HEAD object-read contract: these failures mean "this object is not
// readable right now" (S3 `SlowDownRead`), not a write admission reject.
if crate::error::is_object_read_unavailable(&err) {
return crate::error::slow_down_read_api_error(err);
}
if let StorageError::Io(io_error) = &err