diff --git a/Cargo.lock b/Cargo.lock index a9e69a614c..65af81ec0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index 0a1e9549fe..723cdc9deb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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 diff --git a/crates/zstd-compat/Cargo.toml b/crates/zstd-compat/Cargo.toml index 9244aff11b..ee8428a17f 100644 --- a/crates/zstd-compat/Cargo.toml +++ b/crates/zstd-compat/Cargo.toml @@ -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" diff --git a/rustfs/src/app/metadata_route.rs b/rustfs/src/app/metadata_route.rs index d52e1ab0ae..53615a8bb7 100644 --- a/rustfs/src/app/metadata_route.rs +++ b/rustfs/src/app/metadata_route.rs @@ -689,18 +689,14 @@ async fn call_list_object_versions(req: S3Request, 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, bucket: String) -> S3Result> { 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) } diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index c3912ff610..77a3db5be5 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -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> = 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) -> S3Result> { - 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, diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index 3e36bd7cac..c2af412902 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -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] diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 6e2b3de15e..64c7c87ae2 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -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::(), - 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 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); diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index ec5a4b987e..dc9e7ccdce 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -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. diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index d0afd65c9a..bdda391be9 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -1539,20 +1539,14 @@ impl S3 for FS { ) -> S3Result> { 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) -> S3Result> { 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))] diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index ee081e25be..a0a34f76a5 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -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, } 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