fix(s3): await PUT rename completion before responding (#9229)

docs/operations/object-lock-contention.md

@@ -61,7 +61,7 @@ The `object_namespace_lock` event covers store GET/read and write acquisition, s
 
 The diagnostic retains the parent tracing span available at acquisition. When an HTTP request span carries `request_id`, guard events retain that parent even if the guard moves into a background PUT commit tail. Heal events retain the available task span. Missing parent fields must not be reconstructed from adjacent log lines.
 
-Use `read_repair_data`, `read_repair_commit`, and `heal_object` to distinguish shared read-repair reconstruction, its exclusive revalidation/rename phase, and ordinary exclusive object healing. A PUT quorum ACK can precede the end of its rename tail and lock lifetime.
+Use `read_repair_data`, `read_repair_commit`, and `heal_object` to distinguish shared read-repair reconstruction, its exclusive revalidation/rename phase, and ordinary exclusive object healing. Application PUTs use `TailDrained` completion and await rename fan-out before returning success. Completion can succeed at write quorum with minority failures routed to repair. Internal quorum-completion callers may acknowledge before their rename tail and lock lifetime end.
 
 ## Distinguish the waiting chains
 

rustfs/src/app/object/put.rs

@@ -33,6 +33,7 @@ enum PutStorageTestStage {
 #[cfg(test)]
 tokio::task_local! {
     static PUT_STORAGE_TEST_FAILURE: (PutStorageTestStage, StorageError);
+    static PUT_STORAGE_TEST_COMPLETION: std::cell::Cell<Option<crate::app::storage_api::object_usecase::WriteCompletion>>;
 }
 
 #[cfg(test)]
@@ -1571,6 +1572,8 @@ impl DefaultObjectUsecase {
         )
         .await
         .map_err(ApiError::from)?;
+        // Keep the request awaiting the rename fan-out that owns its namespace guard.
+        opts.write_completion = crate::app::storage_api::object_usecase::WriteCompletion::TailDrained;
         if let Some(etag) = preserve_etag {
             opts.preserve_etag = Some(etag);
         }
@@ -1959,6 +1962,8 @@ impl DefaultObjectUsecase {
             None
         };
         #[cfg(test)]
+        let _ = PUT_STORAGE_TEST_COMPLETION.try_with(|observed| observed.set(Some(opts.write_completion)));
+        #[cfg(test)]
         let injected_commit_error = injected_put_storage_failure(PutStorageTestStage::Commit);
         let put_commit_future = {
             let store = Arc::clone(&store);
@@ -2227,6 +2232,43 @@ mod tests {
     use futures::StreamExt;
     use http::{HeaderMap, HeaderName, HeaderValue, Method};
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn s3_put_uses_tail_drained_completion_before_acknowledgement() {
+        use crate::app::storage_api::object_usecase::WriteCompletion;
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("put-tail-completion-{}", Uuid::new_v4());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create PUT completion bucket");
+        let payload = Bytes::from_static(b"readable after PUT acknowledgement");
+        let input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key("object".to_owned())
+            .content_length(Some(i64::try_from(payload.len()).expect("small payload length")))
+            .body(Some(StreamingBlob::from_bytes(payload.clone())))
+            .build()
+            .expect("PUT completion request");
+        let usecase = DefaultObjectUsecase::with_context(Some(context));
+        let fs = FS::new();
+        let (response, observed) = PUT_STORAGE_TEST_COMPLETION
+            .scope(std::cell::Cell::new(None), async {
+                let response = Box::pin(usecase.execute_put_object(&fs, build_request(input, Method::PUT))).await;
+                (response, PUT_STORAGE_TEST_COMPLETION.with(std::cell::Cell::get))
+            })
+            .await;
+        response.expect("PUT completion response");
+        assert_eq!(observed, Some(WriteCompletion::TailDrained));
+        let object = store
+            .get_object_info(&bucket, "object", &ObjectOptions::default())
+            .await
+            .expect("acknowledged PUT metadata remains readable");
+        assert_eq!(object.size, i64::try_from(payload.len()).expect("small payload length"));
+    }
+
     #[test]
     #[serial_test::serial]
     fn put_commit_lock_failure_keeps_request_operation_in_spawned_owner() {