fix(s3): normalize PUT read quorum failures as unavailable (#8484)

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<StorageError> {
+    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<S3Response<()>> = 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<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
-        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, &current_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::<StorageError>()),
+                    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("<Code>ServiceUnavailable</Code>"), "{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,

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::<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
+}
+
 /// 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::<StorageError>().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 {

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<std::sync::Mutex<Option<ObjectGenerationTestHook>>> = 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<PutObjectInput>) -> 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::<StorageError>()),
+                    Some(StorageError::ErasureReadQuorum)
+                ));
+            } else {
+                assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+                assert_eq!(err.status_code(), Some(http::StatusCode::FORBIDDEN));
+            }
+            assert!(req.extensions.get::<BucketGenerationGuard>().is_none());
+        }
+    }
+
     #[tokio::test]
     #[serial]
     async fn put_object_access_captures_authorized_bucket_incarnation() {