From 201b5f62d2c187a8447dbb3faa33bb6e9c179cc0 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 12 Oct 2026 12:55:23 +0800 Subject: [PATCH] fix(s3): normalize PUT read quorum failures as unavailable (#8484) --- rustfs/src/app/object/put.rs | 121 ++++++++++++++++++++++++++++++- rustfs/src/error.rs | 101 ++++++++++++++++++++++++++ rustfs/src/storage/access.rs | 135 ++++++++++++++++++++++++++--------- 3 files changed, 322 insertions(+), 35 deletions(-) diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index d9a1c3477..9bff39cf2 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -18,9 +18,30 @@ 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::UploadLimitExceeded; +use crate::error::{UploadLimitExceeded, normalize_put_read_quorum_error}; static PUT_FAILURE_LOGS: rustfs_utils::LogThrottle = rustfs_utils::LogThrottle::new(5_000); +#[cfg(test)] +#[derive(Clone, Copy, PartialEq)] +enum PutReadQuorumTestStage { + DefaultSse, + Prelookup, + Commit, +} + +#[cfg(test)] +tokio::task_local! { + static PUT_READ_QUORUM_TEST_STAGE: PutReadQuorumTestStage; +} + +#[cfg(test)] +fn injected_put_read_quorum_failure(stage: PutReadQuorumTestStage) -> Option { + PUT_READ_QUORUM_TEST_STAGE + .try_with(|configured| *configured == stage) + .unwrap_or(false) + .then(|| StorageError::InsufficientReadQuorum("bucket".into(), "object".into())) +} + const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024; const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES"; @@ -1098,6 +1119,11 @@ 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_put_read_quorum_error(err) + } else { + err + }; let result: S3Result> = Err(err); let _ = completion.complete(&result); match result { @@ -1185,7 +1211,9 @@ 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 + self.execute_put_object_boxed(_fs, req) + .await + .map_err(normalize_put_read_quorum_error) } fn execute_put_object_boxed<'a>( @@ -1549,6 +1577,9 @@ impl DefaultObjectUsecase { let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now); let bucket_sse_config = load_bucket_default_sse_config(&bucket).await; + #[cfg(test)] + let bucket_sse_config = injected_put_read_quorum_failure(PutReadQuorumTestStage::DefaultSse) + .map_or(bucket_sse_config, |err| Err(ApiError::from(err).into())); rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start); let bucket_sse_config = bucket_sse_config?; debug!( @@ -1699,6 +1730,9 @@ impl DefaultObjectUsecase { // the dst xl.meta that rename_data already reads, saving a full-disk // metadata fanout per PUT. let prelookup_required = version_id.is_some() || object_lock_checks_required_for_state(&object_lock_config_state); + #[cfg(test)] + let prelookup_required = + prelookup_required || injected_put_read_quorum_failure(PutReadQuorumTestStage::Prelookup).is_some(); // Outer None = prelookup skipped (accounting comes from the commit // backfill); Some(inner) = the previous current size as observed by the // lookup, with the pre-#1009 semantics kept bit-for-bit. @@ -1713,6 +1747,9 @@ impl DefaultObjectUsecase { crate::hp_guard!("S3::put_object_prelookup"); store.get_object_info(&bucket, &key, ¤t_opts).await }; + #[cfg(test)] + let previous_current_info = + injected_put_read_quorum_failure(PutReadQuorumTestStage::Prelookup).map_or(previous_current_info, Err); Some(match previous_current_info { Ok(existing_obj_info) => { validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)?; @@ -1987,6 +2024,8 @@ impl DefaultObjectUsecase { } else { None }; + #[cfg(test)] + let injected_commit_error = injected_put_read_quorum_failure(PutReadQuorumTestStage::Commit); let put_commit = spawn_traced_join({ let store = Arc::clone(&store); let bucket = bucket.clone(); @@ -2007,6 +2046,11 @@ impl DefaultObjectUsecase { .and_then(ObjectTrafficHealth::track_write_storage); let mut reader = reader; let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now); + #[cfg(test)] + if let Some(err) = injected_commit_error { + store_put_watchdog.cancel(); + return Err(ApiError::from(err).into()); + } let (obj_info, backfilled_old_current_size) = match store .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) .await @@ -2203,6 +2247,79 @@ mod tests { use super::*; use futures::StreamExt; use http::{HeaderMap, HeaderName, HeaderValue, Method}; + + #[tokio::test] + #[serial_test::serial] + async fn put_read_quorum_failures_are_normalized_at_each_stage() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let (store, context) = real_store_test_context().await; + let bucket = format!("put-quorum-{}", Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("read-quorum PUT bucket must be created"); + let usecase = DefaultObjectUsecase::with_context(Some(context)); + + for (index, stage) in [ + PutReadQuorumTestStage::DefaultSse, + PutReadQuorumTestStage::Prelookup, + PutReadQuorumTestStage::Commit, + ] + .into_iter() + .enumerate() + { + for method in [Method::PUT, Method::POST] { + let key = format!("refused-{index}-{method}"); + let input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(key.clone()) + .body(Some(StreamingBlob::from_bytes(Bytes::from_static(b"quorum probe")))) + .content_length(Some(12)) + .build() + .expect("quorum probe input must build"); + let mut req = build_request(input, method.clone()); + if method == Method::POST { + req.extensions.insert(PostObjectRequestMarker); + } + let err = PUT_READ_QUORUM_TEST_STAGE + .scope(stage, usecase.execute_put_object(&FS::new(), req)) + .await + .expect_err("read quorum failure must refuse the upload"); + assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable, "stage {index}, method {method}"); + assert_eq!(err.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE)); + assert_eq!(err.message(), Some("The service is unavailable. Please retry.")); + assert!(matches!( + err.source().and_then(|source| source.downcast_ref::()), + Some(StorageError::InsufficientReadQuorum(_, _)) + )); + let response = err.to_http_response().expect("PUT error must serialize"); + assert_eq!(response.status(), http::StatusCode::SERVICE_UNAVAILABLE); + let xml = http_body_util::BodyExt::collect(response.into_body()) + .await + .expect("PUT error XML must be readable") + .to_bytes(); + let xml = std::str::from_utf8(&xml).expect("PUT error XML must be UTF-8"); + assert!(xml.contains("ServiceUnavailable"), "{xml}"); + assert!(!xml.contains("SlowDownRead"), "{xml}"); + let lookup = store.get_object_info(&bucket, &key, &ObjectOptions::default()).await; + assert!( + lookup.as_ref().is_err_and(is_err_object_not_found), + "failed PUT must leave no object: {lookup:?}" + ); + } + } + } + + #[test] + fn failed_put_completion_normalizes_only_s3_errors() { + 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.")); + let internal = fail_put_object(PutObjectCompletion::Internal(None), error()); + assert_eq!(internal.code(), &S3ErrorCode::Custom("SlowDownRead".into())); + } use s3s::dto::{ DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index e8cc2ae0b..e1542b685 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -173,6 +173,40 @@ pub(crate) fn slow_down_read_api_error(err: StorageError) -> ApiError { } } +/// Normalize a read-quorum failure that escaped through a PutObject boundary. +/// +/// Storage-level read quorum failures keep their `SlowDownRead` code for GET +/// compatibility. A write request that encounters the same failure must expose +/// the write-side retry contract instead, while retaining the typed source for +/// diagnostics and audit consumers. +pub(crate) fn normalize_put_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 +} + /// Marks a request body that exceeded a presigned upload size capability. /// /// This marker must survive the body-reader and storage layers so the client @@ -1602,6 +1636,73 @@ mod tests { assert_eq!(S3Error::from(api_error).status_code(), Some(StatusCode::SERVICE_UNAVAILABLE)); } + #[test] + fn put_read_quorum_error_preserves_context_and_is_idempotent() { + for source in [ + StorageError::ErasureReadQuorum, + StorageError::InsufficientReadQuorum("bucket".into(), "object".into()), + ] { + for nested in [false, true] { + let mut err: S3Error = ApiError::from(source.clone()).into(); + if nested { + err.set_source(Box::new(IoError::other(source.clone()))); + } + err.set_request_id("put-quorum-request"); + let mut headers = http::HeaderMap::new(); + headers.insert("retry-after", http::HeaderValue::from_static("2")); + err.set_headers(headers); + + let err = normalize_put_read_quorum_error(normalize_put_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.")); + assert_eq!(err.request_id(), Some("put-quorum-request")); + assert_eq!(err.headers().expect("retry headers must survive")["retry-after"], "2"); + let retained = error_chain_find(err.source().expect("typed source must survive"), |error| { + error.downcast_ref::().cloned() + }); + assert_eq!(retained, Some(source.clone())); + } + } + } + + #[test] + fn put_read_quorum_error_preserves_other_error_contracts() { + for code in [ + S3ErrorCode::SlowDown, + S3ErrorCode::AccessDenied, + S3ErrorCode::InvalidRequest, + S3ErrorCode::PreconditionFailed, + S3ErrorCode::BadDigest, + S3ErrorCode::IncompleteBody, + S3ErrorCode::NoSuchKey, + S3ErrorCode::NoSuchBucket, + S3ErrorCode::InternalError, + S3ErrorCode::ServiceUnavailable, + ] { + 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_put_read_quorum_error(err); + 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_put_read_quorum_error(err); + assert_eq!(err.code(), &S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())); + assert_eq!(err.message(), Some("original message")); + } + } + #[test] fn test_unknown_authoritative_quota_usage_maps_to_retryable_error() { let api_error = ApiError::from(QuotaError::UsageUnavailable { diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 316eee5fc..b601c9333 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -22,7 +22,7 @@ use crate::auth::{ 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; +use crate::error::{ApiError, normalize_put_read_quorum_error}; use crate::license::license_check; use crate::shared_types::RemoteAddr; use crate::storage::request_context::RequestContext; @@ -424,6 +424,11 @@ type ObjectGenerationTestHook = (String, tokio::sync::oneshot::Sender<()>, tokio #[cfg(test)] static OBJECT_GENERATION_TEST_HOOK: OnceLock>> = OnceLock::new(); +#[cfg(test)] +tokio::task_local! { + static PUT_GENERATION_READ_QUORUM_FAILURE: (); +} + #[cfg(test)] fn install_object_generation_test_hook( bucket: String, @@ -3296,40 +3301,50 @@ impl S3Access for FS { /// This method returns `Ok(())` by default. async fn put_object(&self, req: &mut S3Request) -> S3Result<()> { crate::hp_guard!("S3Access::put_object"); - let req_info = ext_req_info_mut(&mut req.extensions)?; - req_info.bucket = Some(req.input.bucket.clone()); - req_info.object = Some(req.input.key.clone()); - req_info.version_id = req.input.version_id.clone(); + async { + let req_info = ext_req_info_mut(&mut req.extensions)?; + req_info.bucket = Some(req.input.bucket.clone()); + req_info.object = Some(req.input.key.clone()); + req_info.version_id = req.input.version_id.clone(); - if has_write_offset_bytes_header(&req.headers) { - return Err(S3Error::with_message( - S3ErrorCode::NotImplemented, - ApiError::error_code_to_message(&S3ErrorCode::NotImplemented), - )); + if has_write_offset_bytes_header(&req.headers) { + return Err(S3Error::with_message( + S3ErrorCode::NotImplemented, + ApiError::error_code_to_message(&S3ErrorCode::NotImplemented), + )); + } + + let bucket = req.input.bucket.clone(); + // Snapshot before authorization, but preserve AccessDenied precedence + // by exposing any bucket-state error only after authorization succeeds. + let bucket_generation = load_bucket_generation(self, req, &bucket).await; + #[cfg(test)] + let bucket_generation = if PUT_GENERATION_READ_QUORUM_FAILURE.try_with(|_| ()).is_ok() { + Err(ApiError::from(StorageError::ErasureReadQuorum).into()) + } else { + bucket_generation + }; + let requested_retain_until = req.input.object_lock_retain_until_date.clone(); + set_requested_object_lock_retain_until(req, requested_retain_until.as_ref()); + authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?; + req.extensions.insert(bucket_generation?); + + if req.method != http::Method::POST { + authorize_replication_only_put_headers(req).await?; + } + + if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) { + authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; + } + + if retention_write_requested(req.input.object_lock_mode.as_ref(), req.input.object_lock_retain_until_date.as_ref()) { + authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?; + } + + Ok(()) } - - let bucket = req.input.bucket.clone(); - // Snapshot before authorization, but preserve AccessDenied precedence - // by exposing any bucket-state error only after authorization succeeds. - let bucket_generation = load_bucket_generation(self, req, &bucket).await; - let requested_retain_until = req.input.object_lock_retain_until_date.clone(); - set_requested_object_lock_retain_until(req, requested_retain_until.as_ref()); - authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?; - req.extensions.insert(bucket_generation?); - - if req.method != http::Method::POST { - authorize_replication_only_put_headers(req).await?; - } - - if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) { - authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; - } - - if retention_write_requested(req.input.object_lock_mode.as_ref(), req.input.object_lock_retain_until_date.as_ref()) { - authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?; - } - - Ok(()) + .await + .map_err(normalize_put_read_quorum_error) } /// Checks whether the PutObjectAcl request has accesses to the resources. @@ -5183,6 +5198,60 @@ mod tests { }); } + #[tokio::test] + #[serial] + async fn put_read_quorum_access_keeps_authorization_precedence() { + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + let server_ctx = ServerContextSlot::new(); + assert!(server_ctx.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms))))); + let fs = FS::with_server_ctx(server_ctx); + + for effect in ["Allow", "Deny"] { + let bucket = format!("put-quorum-access-{}", uuid::Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("authorization precedence bucket must be created"); + let policy_json = format!( + r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"{effect}","Principal":{{"AWS":"*"}},"Action":["s3:PutObject"],"Resource":["arn:aws:s3:::{bucket}/*"]}}]}}"# + ); + let mut metadata = (*crate::storage::get_bucket_metadata(&bucket) + .await + .expect("test bucket metadata must be cached")) + .clone(); + metadata.policy_config = Some(serde_json::from_str(&policy_json).expect("test policy must parse")); + metadata.policy_config_json = policy_json.into_bytes(); + crate::storage::storage_api::set_bucket_metadata(bucket.clone(), metadata) + .await + .expect("test policy must be installed"); + + let input = PutObjectInput::builder() + .bucket(bucket) + .key("object".into()) + .build() + .expect("access probe input must build"); + let mut req = build_request(input, Method::PUT); + ensure_req_info(&mut req); + req.extensions.insert(fs.server_ctx().clone()); + let err = super::PUT_GENERATION_READ_QUORUM_FAILURE + .scope((), fs.put_object(&mut req)) + .await + .expect_err("a generation failure must never admit a write"); + if effect == "Allow" { + assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable); + assert_eq!(err.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE)); + assert!(matches!( + err.source().and_then(|source| source.downcast_ref::()), + Some(StorageError::ErasureReadQuorum) + )); + } else { + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + assert_eq!(err.status_code(), Some(http::StatusCode::FORBIDDEN)); + } + assert!(req.extensions.get::().is_none()); + } + } + #[tokio::test] #[serial] async fn put_object_access_captures_authorized_bucket_incarnation() {