fix(s3): advise retry after typed lock timeouts (#8546)

Add a one-second Retry-After hint at the canonical S3 error conversion
only for ServiceUnavailable responses retaining a typed lock timeout.
Keep existing error codes, source chains, and lock budgets unchanged.

Cover actual HTTP/XML serialization, nested causes, private-resource
redaction, and unrelated disk/quorum/text failures. Document the response
contract and the limits of log-count and five-second timeout attribution.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

docs/operations/object-lock-contention.md

@@ -2,6 +2,12 @@
 
 **Use this when:** an object GET returns `ServiceUnavailable` near the namespace lock acquisition deadline while PUT or Heal is active on the same key.
 
+## Client-visible acquisition timeouts
+
+The canonical `ApiError` conversion in `rustfs/src/error.rs` retains HTTP 503 and the S3 `ServiceUnavailable` code for typed lock acquisition timeouts, and adds `Retry-After: 1`. The header is a minimum retry delay, not a receipt that the competing owner has released its lock. Use bounded exponential backoff with jitter and keep the client's overall request budget.
+
+This advice does not change the lock acquisition budget or shorten a competing writer's commit tail. It is not added to unrelated disk timeouts, quorum failures, or errors that merely contain timeout text. A 503 near five seconds is insufficient to attribute the request to a namespace lock: disk metadata operations also have a five-second default budget. Correlate the typed lock-failure counters and acquisition events below before changing contention policy.
+
 ## Capture one acquisition and its competing owners
 
 Enable the existing diagnostic switch before starting the process and add the diagnostic target to the deployment's log filter:

rustfs/src/error.rs

@@ -599,7 +599,21 @@ fn error_chain_has_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bo
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let status = custom_error_status(&err.code);
+        // A typed lock acquisition timeout is transient contention. Keep its
+        // existing S3 code while advising a minimum delay before another attempt.
+        let lock_timeout = err.code == S3ErrorCode::ServiceUnavailable
+            && err.source.as_deref().is_some_and(|source| {
+                error_chain_find(source, |error| {
+                    matches!(error.downcast_ref::<LockError>(), Some(LockError::Timeout { .. })).then_some(())
+                })
+                .is_some()
+            });
         let mut s3e = S3Error::with_message(err.code, err.message);
+        if lock_timeout {
+            let mut headers = http::HeaderMap::new();
+            headers.insert(http::header::RETRY_AFTER, http::HeaderValue::from_static("1"));
+            s3e.set_headers(headers);
+        }
         if let Some(status) = status {
             s3e.set_status_code(status);
         }
@@ -920,6 +934,74 @@ mod tests {
     use s3s::{S3Error, S3ErrorCode};
     use std::io::{Error as IoError, ErrorKind};
 
+    #[tokio::test]
+    async fn lock_timeout_response_advises_retry_without_exposing_resource() {
+        use http_body_util::BodyExt as _;
+
+        let storage_error =
+            StorageError::Lock(LockError::timeout("private-bucket/private-key@latest", std::time::Duration::from_secs(5)));
+        let error = S3Error::from(ApiError::from(storage_error));
+        assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
+        let source = error.source().expect("retain the typed storage cause");
+        assert!(source.downcast_ref::<StorageError>().is_some());
+        assert!(error_chain_has_type::<LockError>(source));
+        let response = error.to_http_response().expect("serialize lock timeout response");
+        assert_eq!(response.status(), http::StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(response.headers()[http::header::RETRY_AFTER], "1");
+        let body = response
+            .into_body()
+            .collect()
+            .await
+            .expect("read timeout error XML")
+            .to_bytes();
+        let body = String::from_utf8(body.to_vec()).expect("S3 XML is UTF-8");
+        assert!(body.contains("<Code>ServiceUnavailable</Code>"));
+        assert!(!body.contains("private-bucket"));
+        assert!(!body.contains("private-key"));
+    }
+
+    #[test]
+    fn lock_timeout_response_preserves_wrapped_source_and_requires_503() {
+        for code in [S3ErrorCode::ServiceUnavailable, S3ErrorCode::RequestTimeout] {
+            let should_advise_retry = code == S3ErrorCode::ServiceUnavailable;
+            let error = S3Error::from(ApiError {
+                message: ApiError::error_code_to_message(&code),
+                code: code.clone(),
+                source: Some(Box::new(IoError::other(LockError::timeout(
+                    "bucket/object@latest",
+                    std::time::Duration::from_secs(5),
+                )))),
+            });
+            assert_eq!(error.code(), &code);
+            let source = error.source().expect("preserve wrapped lock timeout");
+            assert!(error_chain_has_type::<LockError>(source));
+            let response = error.to_http_response().expect("serialize wrapped timeout");
+            assert_eq!(response.headers().contains_key(http::header::RETRY_AFTER), should_advise_retry);
+        }
+    }
+
+    #[test]
+    fn unrelated_unavailable_responses_do_not_inherit_lock_retry_advice() {
+        for failure in [
+            StorageError::Timeout,
+            StorageError::DiskNotFound,
+            StorageError::FileCorrupt,
+            StorageError::NamespaceLockQuorumUnavailable {
+                mode: "read",
+                bucket: "bucket".to_owned(),
+                object: "object".to_owned(),
+                required: 2,
+                achieved: 1,
+            },
+            StorageError::Io(IoError::other("Lock acquisition timeout for bucket/object after 5s")),
+            StorageError::SlowDown,
+        ] {
+            let error = S3Error::from(ApiError::from(failure));
+            let response = error.to_http_response().expect("serialize unrelated storage failure");
+            assert!(!response.headers().contains_key(http::header::RETRY_AFTER));
+        }
+    }
+
     #[test]
     fn lock_503_mapping_records_operation_and_resource_class_without_resource_name() {
         let recorder = DebuggingRecorder::new();