feat(gateway): serve owned multipart completion

Use shared request authorization and native completion responses across both S3 stacks. Keep storage responsible for final commit and replay decisions.

rustfs/src/app/lifecycle_transition_api_test/multipart.rs

@@ -61,11 +61,11 @@ async fn complete_multipart_upload_transitions_immediately_via_usecase() {
         .build()
         .unwrap();
 
-    let complete_output = Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST)))
+    let complete_output = Box::pin(usecase.complete_for_s3s_request(build_request(complete_input, Method::POST)))
         .await
         .expect("Failed to complete multipart upload through usecase");
     assert!(
-        complete_output.output.version_id.is_some(),
+        complete_output.version_id.is_some(),
         "versioned CompleteMultipartUpload should return the created version ID"
     );
 
@@ -140,7 +140,7 @@ async fn compensation_driven_complete_multipart_upload_still_transitions_inner()
         .unwrap();
 
     with_forced_immediate_enqueue_timeout(&backend, || async {
-        Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST)))
+        Box::pin(usecase.complete_for_s3s_request(build_request(complete_input, Method::POST)))
             .await
             .expect("Failed to complete multipart upload through usecase");
     })

rustfs/src/app/multipart_usecase.rs

@@ -42,9 +42,7 @@ use super::storage_api::multipart_usecase::data_usage::{
     quota_object_size, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
     record_bucket_object_write_unknown_previous_memory,
 };
-use super::storage_api::multipart_usecase::error::{
-    StorageError, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found,
-};
+use super::storage_api::multipart_usecase::error::{StorageError, is_err_object_not_found, is_err_version_not_found};
 use super::storage_api::multipart_usecase::helper::{OperationHelper, run_owned_mutation};
 #[cfg(test)]
 use super::storage_api::multipart_usecase::io::{DecryptReader, EncryptReader, HardLimitReader, boxed_reader, wrap_reader};
@@ -546,12 +544,13 @@ mod tests {
                 .bucket("bucket".to_owned())
                 .key(key.to_owned())
                 .upload_id("upload".to_owned())
+                .multipart_upload(Some(CompletedMultipartUpload::default()))
                 .build()
                 .expect("valid complete request DTO");
-            let error = Box::pin(make_usecase().execute_complete_multipart_upload(build_request(complete, Method::POST)))
+            let error = Box::pin(make_usecase().complete_for_s3s_request(build_request(complete, Method::POST)))
                 .await
                 .expect_err("uploads created before the fix must not publish reserved keys");
-            assert_eq!(error.code(), &S3ErrorCode::InvalidArgument);
+            assert_eq!(error.code(), &rustfs_s3_types::S3ErrorCode::InvalidArgument);
         }
     }
 
@@ -710,10 +709,10 @@ mod tests {
                 .expect("complete");
             let complete_req = denied_request(complete, authorizer.clone());
             let error = usecase
-                .execute_complete_multipart_upload(complete_req)
+                .complete_for_s3s_request(complete_req)
                 .await
                 .expect_err("denied completion");
-            assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+            assert_eq!(error.code(), &rustfs_s3_types::S3ErrorCode::AccessDenied);
             assert_eq!(authorizer.calls(), vec![(KmsAction::GenerateDataKeyAction, "finance-key".to_string()); 3]);
             assert!(
                 store
@@ -756,10 +755,10 @@ mod tests {
                 .build()
                 .expect("replay");
             let error = usecase
-                .execute_complete_multipart_upload(denied_request(replay, authorizer.clone()))
+                .complete_for_s3s_request(denied_request(replay, authorizer.clone()))
                 .await
                 .expect_err("replay must authorize before comparing guessed part metadata");
-            assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+            assert_eq!(error.code(), &rustfs_s3_types::S3ErrorCode::AccessDenied);
             assert_eq!(authorizer.calls().len(), 4);
         });
     }

rustfs/src/app/multipart_usecase/complete.rs

@@ -15,33 +15,38 @@
 //! CompleteMultipartUpload.
 
 use super::*;
-use crate::app::storage_api::s3::dto::{
-    ChecksumType, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, CompletedPart, ETag, ServerSideEncryption,
-};
+use crate::app::RequestEnvelope;
+use crate::app::dto_values::stored_etag;
 use crate::app::storage_api::s3::header::{
     X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
 };
-use crate::app::storage_api::s3::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
+use http::StatusCode;
+use rustfs_gateway_types::dto::{
+    ChecksumType, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, CompletedPart, ServerSideEncryption,
+};
+use rustfs_gateway_types::{ETag, RecordedUpload, resolve_upload_async};
+use rustfs_s3_types::{S3Error, S3ErrorCode, S3Result, s3_error};
+use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_MD5_HEADER};
 
-fn validate_complete_multipart_parts(parts: &[CompletePart]) -> rustfs_s3_types::S3Result<()> {
+fn validate_complete_multipart_parts(parts: &[CompletePart]) -> S3Result<()> {
     if let Some(part) = parts
         .iter()
         .find(|part| !(1..=MAX_MULTIPART_PART_NUMBER as usize).contains(&part.part_num))
     {
-        return Err(rustfs_s3_types::S3Error::with_message(
-            rustfs_s3_types::S3ErrorCode::InvalidPart,
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidPart,
             format!("Part number {} must be between 1 and {MAX_MULTIPART_PART_NUMBER}", part.part_num),
         ));
     }
 
     if parts.windows(2).any(|window| window[0].part_num >= window[1].part_num) {
-        return Err(rustfs_s3_types::s3_error!(InvalidPartOrder, "Part numbers must be strictly increasing"));
+        return Err(s3_error!(InvalidPartOrder, "Part numbers must be strictly increasing"));
     }
 
     Ok(())
 }
 
-fn normalize_complete_multipart_parts(parts: Vec<CompletePart>) -> rustfs_s3_types::S3Result<Vec<CompletePart>> {
+fn normalize_complete_multipart_parts(parts: Vec<CompletePart>) -> S3Result<Vec<CompletePart>> {
     // For duplicate part numbers, keep the last occurrence from the request.
     // This matches retry/resend semantics where later uploads override earlier ones.
     let mut seen = HashSet::with_capacity(parts.len());
@@ -57,13 +62,12 @@ fn normalize_complete_multipart_parts(parts: Vec<CompletePart>) -> rustfs_s3_typ
     Ok(deduped_reversed)
 }
 
-fn complete_part_from_s3(value: CompletedPart) -> CompletePart {
+/// A part as storage completes it: a number outside `usize` is part `0`, which the range check
+/// refuses, and the entity tag is the opaque tag the request named.
+fn complete_part(value: CompletedPart) -> CompletePart {
     CompletePart {
-        part_num: value
-            .part_number
-            .and_then(|part_num| usize::try_from(part_num).ok())
-            .unwrap_or_default(),
-        etag: value.e_tag.map(|v| v.value().to_owned()),
+        part_num: usize::try_from(value.part_number).unwrap_or_default(),
+        etag: value.e_tag.map(|etag| etag.opaque_tag().to_owned()),
         checksum_crc32: value.checksum_crc32,
         checksum_crc32c: value.checksum_crc32c,
         checksum_sha1: value.checksum_sha1,
@@ -175,44 +179,104 @@ fn build_complete_multipart_location(headers: &HeaderMap, uri: &Uri, bucket: &st
     }
 }
 
+/// Whether `If-Match`, as the header spells it, admits the current object tagged `stored`: `*`
+/// always, a strong tag only when it is the stored one, a weak or unreadable tag never.
+fn if_match_admits(condition: &str, stored: Option<&str>) -> bool {
+    match ETag::parse_http_header(condition) {
+        Ok(tag) if tag.is_any() => true,
+        Ok(tag) if !tag.is_weak() => stored == Some(tag.opaque_tag()),
+        _ => false,
+    }
+}
+
+/// The value of the request's `x-amz-checksum-<algorithm>` header, as the legacy decoder read it
+/// into its per-algorithm input member: what the answer echoes when storage records no checksum.
+fn requested_checksum(headers: &HeaderMap, algorithm: rustfs_rio::ChecksumType) -> Option<String> {
+    let name = algorithm.key()?;
+    headers.get(name).and_then(|value| value.to_str().ok()).map(str::to_owned)
+}
+
+/// Storage-observed ownership and optional live metadata; never publication authority.
+struct CompletionUpload {
+    bucket: String,
+    key: String,
+    user_defined: Option<HashMap<String, String>>,
+}
+
+impl RecordedUpload for CompletionUpload {
+    fn bucket(&self) -> &str {
+        &self.bucket
+    }
+    fn key(&self) -> &str {
+        &self.key
+    }
+}
+
 impl DefaultMultipartUsecase {
-    #[instrument(level = "debug", skip(self, req))]
+    /// Completes a multipart upload using owned request facts.
+    ///
+    /// Callers must finish shared request authorization before invoking this method.
+    /// Once the owned task starts, dropping the caller's future does not cancel it.
+    /// Storage rechecks the upload under its commit locks and determines whether a retry can replay a retained completion.
+    ///
+    /// # Errors
+    ///
+    /// Returns the S3 error from request, encryption, quota, object-lock, or storage checks when completion is rejected.
+    #[instrument(level = "debug", skip(self, envelope, input))]
     #[hotpath::measure(impl_type = "MultipartUsecase")]
     pub async fn execute_complete_multipart_upload(
         &self,
-        req: S3Request<CompleteMultipartUploadInput>,
-    ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
+        envelope: RequestEnvelope,
+        input: CompleteMultipartUploadInput,
+    ) -> S3Result<CompleteMultipartUploadOutput> {
+        let mut input = input;
+        let multipart_upload = Some(std::mem::take(&mut input.multipart_upload));
+        self.execute_complete_multipart_upload_parts(envelope, input, multipart_upload)
+            .await
+    }
+
+    pub(crate) async fn execute_complete_multipart_upload_parts(
+        &self,
+        envelope: RequestEnvelope,
+        input: CompleteMultipartUploadInput,
+        multipart_upload: Option<rustfs_gateway_types::dto::CompletedMultipartUpload>,
+    ) -> S3Result<CompleteMultipartUploadOutput> {
         let owner = self.clone();
-        run_owned_mutation(async move { Box::pin(owner.execute_complete_multipart_upload_owned(req)).await }).await
+        run_owned_mutation(async move {
+            Box::pin(owner.execute_complete_multipart_upload_owned(envelope, input, multipart_upload)).await
+        })
+        .await
     }
 
     async fn execute_complete_multipart_upload_owned(
         &self,
-        mut req: S3Request<CompleteMultipartUploadInput>,
-    ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
-        validate_new_object_key(&req.input.key, "CompleteMultipartUpload")?;
+        envelope: RequestEnvelope,
+        input: CompleteMultipartUploadInput,
+        multipart_upload: Option<rustfs_gateway_types::dto::CompletedMultipartUpload>,
+    ) -> S3Result<CompleteMultipartUploadOutput> {
+        let headers = envelope.headers();
+        validate_new_object_key(input.key.as_str(), "CompleteMultipartUpload")?;
         reject_presigned_multipart_max_total_object_size_for_other_operation(
-            &req.headers,
-            req.uri.query(),
-            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+            headers,
+            envelope.uri().query(),
+            envelope.verified_sigv4().is_some(),
         )?;
         reject_presigned_put_max_content_length_for_other_operation(
-            &req.headers,
-            req.uri.query(),
-            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+            headers,
+            envelope.uri().query(),
+            envelope.verified_presigned().is_some(),
         )?;
-        let mut helper = OperationHelper::new(
-            &req,
+        let mut helper = OperationHelper::from_view(
+            &envelope.audit_view(),
             EventName::ObjectCreatedCompleteMultipartUpload,
             S3Operation::CompleteMultipartUpload,
         );
-        let replication_authorized = replication_request_authorized(&req);
-        let input = std::mem::take(&mut req.input);
+        let replication_authorized = replication_request_authorized(&envelope);
         let CompleteMultipartUploadInput {
-            multipart_upload,
-            bucket,
-            key,
+            bucket: bucket_name,
+            key: object_key,
             upload_id,
+            checksum_type: requested_checksum_type,
             if_match,
             if_none_match,
             sse_customer_algorithm,
@@ -220,10 +284,18 @@ impl DefaultMultipartUsecase {
             sse_customer_key_md5,
             ..
         } = input;
+        let bucket = bucket_name.as_str().to_owned();
+        let key = object_key.as_str().to_owned();
+        let sse_customer_key = sse_customer_key.map(|key| key.expose_secret().to_owned());
 
         let customer_key_provided = sse_customer_key.is_some();
         reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
-        reject_non_wildcard_write_if_none_match(if_none_match.as_ref(), &req.headers)?;
+        // The input member and the header are one value on the wire; both are checked, as the
+        // legacy guard checks both.
+        if if_none_match.as_deref().is_some_and(|condition| condition.trim() != "*") {
+            return Err(S3Error::new(S3ErrorCode::NotImplemented));
+        }
+        reject_non_wildcard_write_if_none_match(None, headers)?;
         validate_table_catalog_object_mutation(&bucket, &key).await?;
 
         if if_match.is_some() || if_none_match.is_some() {
@@ -238,12 +310,8 @@ impl DefaultMultipartUsecase {
                             return Err(ApiError::from(StorageError::ObjectNotFound(bucket.clone(), key.clone())).into());
                         }
                     } else {
-                        if let Some(ifmatch) = &if_match
-                            && !ifmatch.is_any()
-                            && ifmatch
-                                .as_etag()
-                                .and_then(ETag::as_strong)
-                                .is_none_or(|expected| info.etag.as_deref() != Some(expected))
+                        if let Some(condition) = if_match.as_deref()
+                            && !if_match_admits(condition, info.etag.as_deref())
                         {
                             return Err(s3_error!(PreconditionFailed));
                         }
@@ -266,16 +334,16 @@ impl DefaultMultipartUsecase {
         }
 
         let Some(multipart_upload) = multipart_upload else { return Err(s3_error!(InvalidPart)) };
-        let Some(parts) = multipart_upload.parts.filter(|parts| !parts.is_empty()) else {
+        if multipart_upload.parts.is_empty() {
             return Err(S3Error::with_message(
                 S3ErrorCode::InvalidRequest,
                 "You must specify at least one part".to_string(),
             ));
-        };
+        }
 
-        let mut opts = get_complete_multipart_upload_opts_with_replication_authorization(&req.headers, replication_authorized)
+        let mut opts = get_complete_multipart_upload_opts_with_replication_authorization(headers, replication_authorized)
             .map_err(ApiError::from)?;
-        apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
+        apply_bucket_generation_guard(&envelope, &bucket, &mut opts)?;
         let (versioned, version_suspended) = BucketVersioningSys::write_state(&bucket, &key)
             .await
             .map_err(ApiError::from)?;
@@ -284,11 +352,11 @@ impl DefaultMultipartUsecase {
         let capacity_scope_token = Uuid::new_v4();
         opts.capacity_scope_token = Some(capacity_scope_token);
 
-        let uploaded_parts_vec = parts.into_iter().map(complete_part_from_s3).collect::<Vec<_>>();
+        let uploaded_parts_vec = multipart_upload.parts.into_iter().map(complete_part).collect::<Vec<_>>();
 
         let uploaded_parts = normalize_complete_multipart_parts(uploaded_parts_vec)?;
 
-        if has_complete_multipart_object_lock_headers(&req.headers) {
+        if has_complete_multipart_object_lock_headers(headers) {
             return Err(S3Error::with_message(
                 S3ErrorCode::InvalidRequest,
                 "CompleteMultipartUpload does not accept object lock or governance bypass headers.".to_string(),
@@ -299,15 +367,45 @@ impl DefaultMultipartUsecase {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
-        // A lost Complete response retries the same upload id after staging is
-        // gone. `get_multipart_info` is NoSuchUpload in that case; completion
-        // below replays the committed object instead of failing the retry.
-        let multipart_info = match store.get_multipart_info(&bucket, &key, &upload_id, &opts).await {
-            Ok(info) => Some(info),
-            Err(err) if is_err_invalid_upload_id(&err) => None,
-            Err(err) => return Err(multipart_upload_api_error(err)),
+        let lookup_store = &store;
+        let lookup_opts = &opts;
+        let (bucket_ref, key_ref) = (bucket.as_str(), key.as_str());
+        let mut absent_upload_id = None;
+        let absent = &mut absent_upload_id;
+        let resolved = resolve_upload_async(&upload_id, &bucket_name, &object_key, |id| async move {
+            let binding = lookup_store
+                .lookup_multipart_completion_binding(bucket_ref, key_ref, id, lookup_opts)
+                .await
+                .map_err(multipart_upload_api_error)?;
+            Ok::<_, S3Error>(match binding {
+                Some(binding) => Some(CompletionUpload {
+                    bucket: binding.bucket().to_owned(),
+                    key: binding.object().to_owned(),
+                    user_defined: binding.active_info().map(|info| info.user_defined.clone()),
+                }),
+                None => {
+                    // Preserve quota and object-lock refusal order without claiming ownership.
+                    // Final storage completion still resolves and checks the original request.
+                    *absent = Some(id.to_owned());
+                    None
+                }
+            })
+        })
+        .await?;
+        let (upload_id, upload_user_defined) = match resolved {
+            Ok((id, upload)) => (id.id().to_owned(), upload.user_defined),
+            Err(_) => match absent_upload_id {
+                Some(id) => (id, None),
+                None => {
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::NoSuchUpload,
+                        "The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.",
+                    ));
+                }
+            },
         };
-        let completion_principal = SseKmsPrincipal::from_request(&req);
+
+        let completion_principal = SseKmsPrincipal::from_request(&envelope);
         let completion_authorization = multipart_completion_authorization(
             EncryptionRequest {
                 bucket: &bucket,
@@ -324,12 +422,12 @@ impl DefaultMultipartUsecase {
             replication_authorized,
         );
         opts.multipart_completion_authorization = Some(completion_authorization.clone());
-        if let Some(info) = multipart_info.as_ref() {
+        if let Some(user_defined) = upload_user_defined.as_ref() {
             completion_authorization
-                .authorize(&info.user_defined)
+                .authorize(user_defined)
                 .await
                 .map_err(map_get_object_reader_error)?;
-            validate_complete_multipart_checksum_type(&req.headers, &info.user_defined)?;
+            validate_complete_multipart_checksum_type(headers, user_defined)?;
         }
 
         let quota_metadata_sys = self.bucket_metadata_sys();
@@ -348,11 +446,8 @@ impl DefaultMultipartUsecase {
             || (!super::super::object::shared::put_like_write_creates_new_version(&opts)
                 && super::super::object::shared::object_lock_checks_required_for_state(&object_lock_config_state));
         let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
-            let current_opts = internal_object_info_lookup_opts(
-                get_opts(&bucket, &key, None, None, &req.headers)
-                    .await
-                    .map_err(ApiError::from)?,
-            );
+            let current_opts =
+                internal_object_info_lookup_opts(get_opts(&bucket, &key, None, None, headers).await.map_err(ApiError::from)?);
             match store.get_object_info(&bucket, &key, &current_opts).await {
                 Ok(existing_obj_info) => {
                     validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)?;
@@ -376,8 +471,7 @@ impl DefaultMultipartUsecase {
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
 
-        let upload_user_defined = multipart_info.as_ref().map(|info| info.user_defined.clone());
-        let table_data_plane_publication_guards = retained_table_data_plane_publication_guards(&req);
+        let table_data_plane_publication_guards = retained_table_data_plane_publication_guards(&envelope);
 
         // Replication admission is decided at the object-creation commit, not
         // at multipart-session creation. Configuration and matching targets
@@ -496,9 +590,9 @@ impl DefaultMultipartUsecase {
         let encryption_metadata = upload_user_defined.as_ref().unwrap_or(obj_info.user_defined.as_ref());
         let server_side_encryption = encryption_metadata
             .get("x-amz-server-side-encryption")
-            .map(|s| ServerSideEncryption::from(s.clone()));
+            .map(|s| ServerSideEncryption::custom(s.clone()));
         let ssekms_key_id = match server_side_encryption.as_ref() {
-            Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS => encryption_metadata
+            Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS.as_str() => encryption_metadata
                 .get("x-amz-server-side-encryption-aws-kms-key-id")
                 .cloned(),
             _ => None,
@@ -517,9 +611,9 @@ impl DefaultMultipartUsecase {
             None
         };
         let mpu_version_for_event = mpu_version.clone();
-        // checksum: stored (decrypted) values take precedence over the request input;
-        // additional algorithms (XXHash3/64/128, SHA-512, MD5), which have no typed
-        // CompleteMultipartUploadOutput field, are echoed as raw response headers (#1261).
+        // checksum: stored (decrypted) values take precedence over the request's;
+        // additional algorithms (XXHash3/64/128, SHA-512, MD5) are echoed as
+        // response headers, not as body members (#1261).
         // Keyless completion remains supported, but plaintext checksums are
         // returned only after verifying that key (backlog#3050). The customer-key
         // MD5 is a fingerprint of the key, not of the plaintext, and the S3
@@ -528,49 +622,77 @@ impl DefaultMultipartUsecase {
         let disclose_checksums = (ssec_algorithm.is_none() && ssec_key_md5.is_none()) || customer_key_provided;
         let classified = if disclose_checksums {
             let (checksums, is_multipart) = obj_info
-                .decrypt_checksums(opts.part_number.unwrap_or(0), &req.headers)
+                .decrypt_checksums(opts.part_number.unwrap_or(0), headers)
                 .map_err(ApiError::from)?;
             crate::app::object_usecase::classify_response_checksums(checksums, is_multipart)
         } else {
             Default::default()
         };
         let checksum_crc32 = if disclose_checksums {
-            classified.crc32.or(input.checksum_crc32)
+            classified
+                .crc32
+                .or_else(|| requested_checksum(headers, rustfs_rio::ChecksumType::CRC32))
         } else {
             None
         };
         let checksum_crc32c = if disclose_checksums {
-            classified.crc32c.or(input.checksum_crc32c)
+            classified
+                .crc32c
+                .or_else(|| requested_checksum(headers, rustfs_rio::ChecksumType::CRC32C))
         } else {
             None
         };
         let checksum_sha1 = if disclose_checksums {
-            classified.sha1.or(input.checksum_sha1)
+            classified
+                .sha1
+                .or_else(|| requested_checksum(headers, rustfs_rio::ChecksumType::SHA1))
         } else {
             None
         };
         let checksum_sha256 = if disclose_checksums {
-            classified.sha256.or(input.checksum_sha256)
+            classified
+                .sha256
+                .or_else(|| requested_checksum(headers, rustfs_rio::ChecksumType::SHA256))
         } else {
             None
         };
         let checksum_crc64nvme = if disclose_checksums {
-            classified.crc64nvme.or(input.checksum_crc64nvme)
+            classified
+                .crc64nvme
+                .or_else(|| requested_checksum(headers, rustfs_rio::ChecksumType::CRC64_NVME))
         } else {
             None
         };
         let checksum_type = if disclose_checksums {
-            classified.checksum_type.map(ChecksumType::from).or(input.checksum_type)
+            classified.checksum_type.map(ChecksumType::custom).or(requested_checksum_type)
         } else {
             None
         };
         let complete_extra_checksum_headers = classified.extra;
 
-        let location = build_complete_multipart_location(&req.headers, &req.uri, &bucket, &key);
+        let mut answer_headers = HeaderMap::new();
+        crate::app::object_usecase::inject_additional_checksum_headers(&mut answer_headers, &complete_extra_checksum_headers);
+        if let Some(algorithm) = ssec_algorithm.as_deref() {
+            let value = HeaderValue::from_str(algorithm)
+                .map_err(|_| s3_error!(InternalError, "Invalid stored SSE-C algorithm metadata"))?;
+            answer_headers.insert(SSEC_ALGORITHM_HEADER, value);
+        }
+        if let Some(key_md5) = ssec_key_md5.as_deref() {
+            let value =
+                HeaderValue::from_str(key_md5).map_err(|_| s3_error!(InternalError, "Invalid stored SSE-C key metadata"))?;
+            answer_headers.insert(SSEC_KEY_MD5_HEADER, value);
+        }
+        let header = |name: &str| {
+            answer_headers
+                .get(name)
+                .and_then(|value| value.to_str().ok())
+                .map(str::to_owned)
+        };
+        let location = build_complete_multipart_location(headers, envelope.uri(), &bucket, &key);
         let output = CompleteMultipartUploadOutput {
-            bucket: Some(bucket.clone()),
-            key: Some(key.clone()),
-            e_tag: obj_info.etag.clone().map(|etag| to_s3s_xml_etag(&etag)),
+            bucket: Some(bucket_name),
+            key: Some(object_key),
+            e_tag: obj_info.etag.as_deref().map(stored_etag),
             location: Some(location),
             server_side_encryption,
             ssekms_key_id,
@@ -581,6 +703,13 @@ impl DefaultMultipartUsecase {
             checksum_crc64nvme,
             checksum_type,
             version_id: mpu_version,
+            sse_customer_algorithm: header(SSEC_ALGORITHM_HEADER),
+            sse_customer_key_md5: header(SSEC_KEY_MD5_HEADER),
+            checksum_header_md5: header("x-amz-checksum-md5"),
+            checksum_header_sha512: header("x-amz-checksum-sha512"),
+            checksum_header_xxhash3: header("x-amz-checksum-xxhash3"),
+            checksum_header_xxhash64: header("x-amz-checksum-xxhash64"),
+            checksum_header_xxhash128: header("x-amz-checksum-xxhash128"),
             ..Default::default()
         };
         // Set object info for event notification
@@ -588,39 +717,48 @@ impl DefaultMultipartUsecase {
         if let Some(version_id) = &mpu_version_for_event {
             helper = helper.version_id(version_id.clone());
         }
-
-        let mut response = S3Response::new(output);
-        crate::app::object_usecase::inject_additional_checksum_headers(&mut response.headers, &complete_extra_checksum_headers);
-        if let Some(algorithm) = ssec_algorithm.as_deref() {
-            let value = HeaderValue::from_str(algorithm)
-                .map_err(|_| s3_error!(InternalError, "Invalid stored SSE-C algorithm metadata"))?;
-            response
-                .headers
-                .insert("x-amz-server-side-encryption-customer-algorithm", value);
-        }
-        if let Some(key_md5) = ssec_key_md5.as_deref() {
-            let value =
-                HeaderValue::from_str(key_md5).map_err(|_| s3_error!(InternalError, "Invalid stored SSE-C key metadata"))?;
-            response
-                .headers
-                .insert("x-amz-server-side-encryption-customer-key-md5", value);
-        }
         if replayed {
             // The object-created event was emitted by the commit that published
             // the version. The retry is still a successful API call.
             helper = helper.suppress_event();
         }
-        let result = Ok(response);
-        let _ = helper.complete(&result);
+        let result = Ok(output);
+        let _ = helper.complete_owned(&result, StatusCode::OK, &answer_headers);
         result
     }
 }
 
+#[cfg(test)]
+impl DefaultMultipartUsecase {
+    /// CompleteMultipartUpload for an s3s request: its facts in an envelope and its input across
+    /// the compatibility seam, as the tests of the legacy request shape reach the owned use case.
+    pub(crate) async fn complete_for_s3s_request(
+        &self,
+        req: crate::app::storage_api::s3::S3Request<crate::app::storage_api::s3::dto::CompleteMultipartUploadInput>,
+    ) -> S3Result<CompleteMultipartUploadOutput> {
+        let envelope = RequestEnvelope::from_s3s(&req);
+        let (input, parts) =
+            crate::storage::ecfs::complete_multipart_upload_input(req.input).map_err(crate::app::dto_values::seam_error)?;
+        match parts {
+            Some(parts) => {
+                let mut input = input;
+                input.multipart_upload = parts;
+                self.execute_complete_multipart_upload(envelope, input).await
+            }
+            None => self.execute_complete_multipart_upload_parts(envelope, input, None).await,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::super::test_support::*;
     use super::*;
-    use crate::app::storage_api::s3::dto::{ChecksumAlgorithm, CreateMultipartUploadInput, StreamingBlob, UploadPartInput};
+    use crate::app::storage_api::s3::dto::{
+        ChecksumAlgorithm, CompleteMultipartUploadInput, CompletedPart, CreateMultipartUploadInput, ETag, ServerSideEncryption,
+        StreamingBlob, UploadPartInput,
+    };
+    use rustfs_gateway_types::compat::s3s_5761ddfe::leaf::etag_to_s3s;
 
     fn upload_metadata_with_checksum_type(recorded: &str) -> HashMap<String, String> {
         HashMap::from([(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE.to_string(), recorded.to_string())])
@@ -774,9 +912,7 @@ mod tests {
             .unwrap();
         let req = build_request(input, Method::POST);
 
-        let err = Box::pin(make_usecase().execute_complete_multipart_upload(req))
-            .await
-            .unwrap_err();
+        let err = Box::pin(make_usecase().complete_for_s3s_request(req)).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InvalidPart);
     }
 
@@ -801,7 +937,7 @@ mod tests {
             .expect("complete multipart input should build");
         let req = build_request(input, Method::POST);
 
-        let err = Box::pin(make_usecase().execute_complete_multipart_upload(req))
+        let err = Box::pin(make_usecase().complete_for_s3s_request(req))
             .await
             .expect_err("missing parts list must be rejected");
         assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
@@ -850,7 +986,7 @@ mod tests {
                 .build()
                 .expect("complete multipart input should build");
             usecase
-                .execute_complete_multipart_upload(build_request(input, Method::POST))
+                .complete_for_s3s_request(build_request(input, Method::POST))
                 .await
                 .expect("compressed multipart completion should succeed");
 
@@ -924,7 +1060,7 @@ mod tests {
             ..Default::default()
         });
         usecase
-            .execute_complete_multipart_upload(complete_request)
+            .complete_for_s3s_request(complete_request)
             .await
             .expect("authorized REPLICA-only completion should succeed without the source-request marker");
 
@@ -1046,21 +1182,43 @@ mod tests {
                     part.e_tag = Some(ETag::Strong("guessed-etag".to_string()));
                     part.checksum_sha256 = Some("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=".to_string());
                 }
-                let result = usecase
-                    .execute_complete_multipart_upload(build_request(input, Method::POST))
-                    .await;
+                let result = usecase.complete_for_s3s_request(build_request(input, Method::POST)).await;
                 if keyed {
                     let response = result.expect("verified key completes and replays");
-                    assert!(response.output.checksum_sha256.is_some(), "verified key retains checksum response");
-                    assert_eq!(
-                        response
-                            .headers
-                            .get("x-amz-server-side-encryption-customer-key-md5")
-                            .expect("key MD5"),
-                        customer_md5
-                    );
-                    let etag = response.output.e_tag.expect("completed encrypted object ETag");
-                    let version_id = response.output.version_id.expect("completed object version");
+                    assert!(response.checksum_sha256.is_some(), "verified key retains checksum response");
+                    assert_eq!(response.sse_customer_key_md5.as_deref(), Some(customer_md5));
+                    #[cfg(feature = "gateway-stack")]
+                    {
+                        use rustfs_gateway::{Limits, MetaView, OperationCodec, TargetKind, WireRequest};
+                        let request = http::Request::builder()
+                            .method("POST")
+                            .uri("http://localhost/bucket/object")
+                            .header("host", "localhost")
+                            .body(bytes::Bytes::new())
+                            .expect("wire request");
+                        let wire = WireRequest::accept(request, &Limits::default()).expect("accepted wire request");
+                        let view = MetaView::of(&wire, TargetKind::Object).expect("object view");
+                        let encoded = rustfs_gateway::dto::CompleteMultipartUpload::encode(response.clone(), &view, 200)
+                            .expect("completed output encodes");
+                        assert_eq!(encoded.status.as_u16(), 200);
+                        assert_eq!(encoded.headers.get(SSEC_ALGORITHM_HEADER).and_then(|v| v.to_str().ok()), Some("AES256"));
+                        assert!(
+                            encoded
+                                .headers
+                                .get(SSEC_KEY_MD5_HEADER)
+                                .is_some_and(|v| v.as_bytes() == customer_md5.as_bytes())
+                        );
+                        let rustfs_gateway::ResponseBody::Complete(body) = encoded.body else {
+                            panic!("completion must encode a complete XML document");
+                        };
+                        let xml = std::str::from_utf8(&body).expect("UTF-8 completion XML");
+                        assert!(xml.contains(&format!(
+                            "<ChecksumSHA256>{}</ChecksumSHA256>",
+                            response.checksum_sha256.as_deref().expect("completed checksum")
+                        )));
+                    }
+                    let etag = etag_to_s3s(&response.e_tag.expect("completed encrypted object ETag"));
+                    let version_id = response.version_id.expect("completed object version");
                     assert_ne!(version_id, "null");
                     assert_multipart_etag_listings(&fs, &bucket, key, &etag, &version_id).await;
                     let head = crate::app::storage_api::s3::S3::head_object(
@@ -1387,7 +1545,7 @@ mod tests {
             .headers
             .insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("*"));
         usecase
-            .execute_complete_multipart_upload(first_complete_req)
+            .complete_for_s3s_request(first_complete_req)
             .await
             .expect("initial conditional complete multipart upload should succeed");
 
@@ -1406,7 +1564,7 @@ mod tests {
             .execute_create_multipart_upload(second_create_req)
             .await
             .expect_err("create multipart upload with If-None-Match: * on existing key must fail");
-        assert_eq!(create_err.code(), &S3ErrorCode::PreconditionFailed);
+        assert_eq!(create_err.code(), &crate::app::storage_api::s3::S3ErrorCode::PreconditionFailed);
 
         let second_upload_id = usecase
             .execute_create_multipart_upload(build_request(create_input, Method::POST))
@@ -1438,7 +1596,7 @@ mod tests {
             .build()
             .expect("DTO-only complete multipart input should build");
         let dto_complete_err = usecase
-            .execute_complete_multipart_upload(build_request(dto_only_complete_input, Method::POST))
+            .complete_for_s3s_request(build_request(dto_only_complete_input, Method::POST))
             .await
             .expect_err("DTO-only If-None-Match: * on existing key must fail");
         assert_eq!(dto_complete_err.code(), &S3ErrorCode::PreconditionFailed);
@@ -1453,7 +1611,7 @@ mod tests {
                 .build()
                 .expect("If-Match complete input should build");
             let err = usecase
-                .execute_complete_multipart_upload(build_request(input, Method::POST))
+                .complete_for_s3s_request(build_request(input, Method::POST))
                 .await
                 .expect_err("weak or nonmatching If-Match must reject the overwrite");
             assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed);
@@ -1471,7 +1629,7 @@ mod tests {
             .build()
             .expect("weak If-None-Match complete multipart input should build");
         let weak_complete_err = usecase
-            .execute_complete_multipart_upload(build_request(weak_if_none_match_input, Method::POST))
+            .complete_for_s3s_request(build_request(weak_if_none_match_input, Method::POST))
             .await
             .expect_err("weak If-None-Match ETag must be rejected as unsupported");
         assert_eq!(weak_complete_err.code(), &S3ErrorCode::NotImplemented);
@@ -1489,7 +1647,7 @@ mod tests {
             .headers
             .insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("*"));
         let complete_err = usecase
-            .execute_complete_multipart_upload(second_complete_req)
+            .complete_for_s3s_request(second_complete_req)
             .await
             .expect_err("complete multipart upload with If-None-Match: * on existing key must fail");
         assert_eq!(complete_err.code(), &S3ErrorCode::PreconditionFailed);
@@ -1651,7 +1809,7 @@ mod tests {
             .headers
             .insert(http::header::IF_MATCH, HeaderValue::from_static("*"));
         let dm_if_match_err = usecase
-            .execute_complete_multipart_upload(dm_if_match_req)
+            .complete_for_s3s_request(dm_if_match_req)
             .await
             .expect_err("If-Match against a current delete marker must fail with NoSuchKey");
         assert_eq!(dm_if_match_err.code(), &S3ErrorCode::NoSuchKey);
@@ -1669,7 +1827,7 @@ mod tests {
             .headers
             .insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("*"));
         usecase
-            .execute_complete_multipart_upload(dm_if_none_match_req)
+            .complete_for_s3s_request(dm_if_none_match_req)
             .await
             .expect("If-None-Match: * completion must succeed past a delete marker");
     }
@@ -1710,8 +1868,8 @@ mod tests {
         }
 
         let first_usecase = usecase.clone();
-        let first = first_usecase.execute_complete_multipart_upload(build_request(inputs.remove(0), Method::POST));
-        let second = usecase.execute_complete_multipart_upload(build_request(inputs.remove(0), Method::POST));
+        let first = first_usecase.complete_for_s3s_request(build_request(inputs.remove(0), Method::POST));
+        let second = usecase.complete_for_s3s_request(build_request(inputs.remove(0), Method::POST));
         let (first, second) = tokio::join!(first, second);
 
         assert_eq!(usize::from(first.is_ok()) + usize::from(second.is_ok()), 1);
@@ -1848,11 +2006,8 @@ mod tests {
             request
         };
         let first = request(input.clone());
-        let completion = tokio::spawn(async move {
-            DefaultMultipartUsecase::from_global()
-                .execute_complete_multipart_upload(first)
-                .await
-        });
+        let completion =
+            tokio::spawn(async move { DefaultMultipartUsecase::from_global().complete_for_s3s_request(first).await });
         tokio::time::timeout(std::time::Duration::from_secs(10), entered.wait())
             .await
             .expect("committed MPU");
@@ -1869,7 +2024,7 @@ mod tests {
             4
         );
         DefaultMultipartUsecase::from_global()
-            .execute_complete_multipart_upload(request(input))
+            .complete_for_s3s_request(request(input))
             .await
             .expect("idempotent replay");
         let audits = probe.audits.lock().expect("audits");
@@ -1920,7 +2075,7 @@ mod tests {
             .build()
             .expect("complete multipart input should build");
         let err = DefaultMultipartUsecase::from_global()
-            .execute_complete_multipart_upload(build_request(input, Method::POST))
+            .complete_for_s3s_request(build_request(input, Method::POST))
             .await
             .expect_err("empty parts list must be rejected");
         assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
@@ -1955,17 +2110,17 @@ mod tests {
             .build()
             .expect("complete multipart input should build");
         let completed = DefaultMultipartUsecase::from_global()
-            .execute_complete_multipart_upload(build_request(input, Method::POST))
+            .complete_for_s3s_request(build_request(input, Method::POST))
             .await
             .expect("staged part should remain completable after empty request rejection");
-        assert!(completed.output.e_tag.is_some());
+        assert!(completed.e_tag.is_some());
         let completed_info = store
             .get_object_info(
                 completed
-                    .output
                     .bucket
-                    .as_deref()
-                    .expect("complete response should include bucket"),
+                    .as_ref()
+                    .expect("complete response should include bucket")
+                    .as_str(),
                 object,
                 &ObjectOptions::default(),
             )
@@ -1997,9 +2152,7 @@ mod tests {
             .unwrap();
         let req = build_request(input, Method::POST);
 
-        let err = Box::pin(make_usecase().execute_complete_multipart_upload(req))
-            .await
-            .unwrap_err();
+        let err = Box::pin(make_usecase().complete_for_s3s_request(req)).await.unwrap_err();
         assert_ne!(err.code(), &S3ErrorCode::InvalidPartOrder);
     }
 
@@ -2026,9 +2179,7 @@ mod tests {
             .unwrap();
         let req = build_request(input, Method::POST);
 
-        let err = Box::pin(make_usecase().execute_complete_multipart_upload(req))
-            .await
-            .unwrap_err();
+        let err = Box::pin(make_usecase().complete_for_s3s_request(req)).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InvalidPartOrder);
     }
 
@@ -2046,7 +2197,7 @@ mod tests {
                 ..Default::default()
             }])
             .expect_err("out-of-range complete part number must be rejected");
-            assert_eq!(err.code(), &rustfs_s3_types::S3ErrorCode::InvalidPart);
+            assert_eq!(err.code(), &S3ErrorCode::InvalidPart);
             let expected = format!("Part number {part_num} must be between 1 and {MAX_MULTIPART_PART_NUMBER}");
             assert_eq!(err.message(), Some(expected.as_str()));
         }
@@ -2099,10 +2250,197 @@ mod tests {
             let mut req = build_request(input, Method::POST);
             req.headers.insert(header_name, HeaderValue::from_str(header_value).unwrap());
 
-            let err = Box::pin(make_usecase().execute_complete_multipart_upload(req))
-                .await
-                .unwrap_err();
+            let err = Box::pin(make_usecase().complete_for_s3s_request(req)).await.unwrap_err();
             assert_eq!(err.code(), &S3ErrorCode::InvalidRequest, "header {header_name} should be rejected");
         }
     }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn completion_absent_binding_refusal_order_baseline() {
+        use crate::app::storage_api::test::bucket::quota::BucketQuota;
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+        for (shape, raw_id) in [("absent", None), ("dot", Some("a/../b")), ("absolute", Some("/upload"))] {
+            for (label, full_quota, locked) in [
+                ("absent-quota", true, false),
+                ("absent", false, false),
+                ("absent-lock", false, true),
+                ("absent-quota-lock", true, true),
+            ] {
+                let bucket = format!("{shape}-{label}-{}", Uuid::new_v4().simple());
+                let key = "object";
+                store
+                    .make_bucket(&bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("create bucket");
+                let upload = store
+                    .new_multipart_upload(&bucket, key, &ObjectOptions::default())
+                    .await
+                    .expect("mint upload");
+                store
+                    .abort_multipart_upload(&bucket, key, &upload.upload_id, &ObjectOptions::default())
+                    .await
+                    .expect("abort upload");
+                assert!(
+                    store
+                        .lookup_multipart_completion_binding(&bucket, key, &upload.upload_id, &ObjectOptions::default())
+                        .await
+                        .expect("observe binding")
+                        .is_none()
+                );
+                if locked {
+                    metadata_sys::update(
+                        &bucket,
+                        crate::app::storage_api::test::bucket::metadata::OBJECT_LOCK_CONFIG,
+                        b"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>"
+                            .to_vec(),
+                    )
+                    .await
+                    .expect("configure object lock");
+                    let mut opts = ObjectOptions::default();
+                    opts.user_defined
+                        .insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_owned(), "ON".to_owned());
+                    let mut reader = PutObjReader::from_vec(b"existing".to_vec());
+                    store
+                        .clone()
+                        .put_object(&bucket, key, &mut reader, &opts)
+                        .await
+                        .expect("write held object");
+                }
+                crate::app::storage_api::test::data_usage::seed_bucket_usage_memory_for_test(&bucket, 2).await;
+                if full_quota {
+                    let metadata = DefaultMultipartUsecase::from_global()
+                        .bucket_metadata_sys()
+                        .expect("metadata context");
+                    QuotaChecker::new(metadata)
+                        .set_quota_config(
+                            &bucket,
+                            BucketQuota {
+                                quota: Some(1),
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("set legacy quota");
+                }
+                let probe = crate::app::storage_api::multipart_usecase::helper::MutationCompletionProbe::install(&bucket);
+                let request_id = raw_id.unwrap_or(&upload.upload_id).to_owned();
+                let input = CompleteMultipartUploadInput::builder()
+                    .bucket(bucket)
+                    .key(key.to_owned())
+                    .upload_id(request_id)
+                    .multipart_upload(Some(CompletedMultipartUpload {
+                        parts: Some(vec![CompletedPart {
+                            part_number: Some(1),
+                            e_tag: Some(ETag::Strong("part".to_owned())),
+                            ..Default::default()
+                        }]),
+                    }))
+                    .build()
+                    .expect("complete input");
+                let error = DefaultMultipartUsecase::from_global()
+                    .complete_for_s3s_request(build_request(input, Method::POST))
+                    .await
+                    .expect_err("absent upload cannot complete");
+                assert_eq!(probe.audits.lock().expect("audits").len(), 0, "{shape}/{label}");
+                assert_eq!(probe.events.lock().expect("events").len(), 0, "{shape}/{label}");
+                let (code, status, message, storage_cause) = if raw_id.is_some() {
+                    (
+                        S3ErrorCode::NoSuchUpload,
+                        404,
+                        "The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.",
+                        false,
+                    )
+                } else if full_quota {
+                    (
+                        S3ErrorCode::InvalidRequest,
+                        400,
+                        "Bucket quota exceeded. Current usage: 2 bytes, limit: 1 bytes",
+                        false,
+                    )
+                } else if locked {
+                    (
+                        S3ErrorCode::AccessDenied,
+                        403,
+                        "Object has a legal hold and cannot be overwritten. Remove the legal hold first.",
+                        false,
+                    )
+                } else {
+                    (
+                        S3ErrorCode::NoSuchUpload,
+                        404,
+                        "The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
+                        true,
+                    )
+                };
+                assert_eq!(error.source().is_some(), storage_cause, "{shape}/{label}");
+                assert_eq!(error.code(), &code, "{shape}/{label}");
+                assert_eq!(error.status_code().map(|status| status.as_u16()), Some(status), "{label}");
+                assert_eq!(error.message(), Some(message), "{label}");
+                assert_eq!(
+                    error
+                        .source()
+                        .is_some_and(|cause| cause.downcast_ref::<StorageError>().is_some()),
+                    storage_cause,
+                    "{label}"
+                );
+            }
+        }
+    }
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn live_completion_checks_recorded_checksum_type_before_parts() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+        let bucket = format!("complete-type-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket");
+        let mut opts = ObjectOptions::default();
+        opts.user_defined
+            .insert(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE.to_owned(), "COMPOSITE".to_owned());
+        let upload = store
+            .new_multipart_upload(&bucket, "object", &opts)
+            .await
+            .expect("active upload");
+        let binding = store
+            .lookup_multipart_completion_binding(&bucket, "object", &upload.upload_id, &opts)
+            .await
+            .expect("binding lookup")
+            .expect("active binding");
+        assert!(binding.active_info().is_some());
+        let input = CompleteMultipartUploadInput::builder()
+            .bucket(bucket)
+            .key("object".to_owned())
+            .upload_id(upload.upload_id)
+            .multipart_upload(Some(CompletedMultipartUpload {
+                parts: Some(vec![CompletedPart {
+                    part_number: Some(1),
+                    e_tag: Some(ETag::Strong("not-uploaded".to_owned())),
+                    ..Default::default()
+                }]),
+            }))
+            .build()
+            .expect("request");
+        let mut request = build_request(input, Method::POST);
+        request
+            .headers
+            .insert(AMZ_CHECKSUM_TYPE, HeaderValue::from_static("FULL_OBJECT"));
+        let error = DefaultMultipartUsecase::from_global()
+            .complete_for_s3s_request(request)
+            .await
+            .expect_err("recorded checksum type is checked first");
+        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(
+            error.message(),
+            Some(
+                "The upload was created with checksum type COMPOSITE. The complete request must use the same checksum type, got FULL_OBJECT."
+            )
+        );
+    }
 }

rustfs/src/app/storage_api.rs

@@ -1607,7 +1607,7 @@ pub(crate) mod ecfs {
 
 pub(crate) mod error {
     pub(crate) use crate::storage::storage_api::{
-        StorageError, is_err_bucket_not_found, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found,
+        StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
     };
 
     pub(crate) type Error = StorageError;

rustfs/src/server/gateway_handlers/object_commit.rs

@@ -12,7 +12,39 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! The gateway operations of migration batch E1 (rustfs/backlog#2765): CopyObject, UploadPartCopy and CompleteMultipartUpload.
-//! None is served yet; the batch appends its rows here.
+//! CompleteMultipartUpload gateway registration and its shared authorization.
 
-pub(super) const ROWS: &[super::OpRow] = &[];
+use super::{OpRow, Served, row};
+use crate::app::{RequestEnvelope, multipart_usecase::DefaultMultipartUsecase};
+use crate::runtime_sources::AppContext;
+use crate::storage_api::server::http::access::access_owned::object_commit::authorize_complete_multipart_upload;
+use rustfs_gateway::{Req, Resp, dto};
+use rustfs_s3_ops::S3Operation;
+use rustfs_s3_types::{S3Result, s3_error};
+use std::sync::Arc;
+
+pub(super) const ROWS: &[OpRow] = &[row::<dto::CompleteMultipartUpload>()];
+
+impl Served for dto::CompleteMultipartUpload {
+    const METRIC: S3Operation = S3Operation::CompleteMultipartUpload;
+
+    async fn authorize(envelope: &mut RequestEnvelope, request: &Req<Self>) -> S3Result<()> {
+        let server = envelope
+            .server_context()
+            .cloned()
+            .ok_or_else(|| s3_error!(InternalError, "server context is not initialized"))?;
+        let input = request.input();
+        let specific_etag = input.if_match.as_deref().is_some_and(|value| value.trim() != "*");
+        Ok(
+            authorize_complete_multipart_upload(envelope, &server, input.bucket.as_str(), input.key.as_str(), specific_etag)
+                .await?,
+        )
+    }
+
+    async fn execute(context: Option<Arc<AppContext>>, envelope: RequestEnvelope, request: Req<Self>) -> S3Result<Resp<Self>> {
+        let output = DefaultMultipartUsecase::with_context(context)
+            .execute_complete_multipart_upload(envelope, request.into_input())
+            .await?;
+        Ok(Resp::new(output))
+    }
+}

rustfs/src/storage/access.rs

@@ -3348,17 +3348,10 @@ impl S3Access for FS {
     /// This method returns `Ok(())` by default.
     async fn complete_multipart_upload(&self, req: &mut S3Request<CompleteMultipartUploadInput>) -> S3Result<()> {
         let bucket = req.input.bucket.clone();
-        let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
-        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());
-
-        authorize_request(req, complete_multipart_upload_authorize_action()).await?;
+        let key = req.input.key.clone();
         let specific_etag = req.input.if_match.as_ref().is_some_and(|etag| !etag.is_any());
-        authorize_write_etag_condition(req, specific_etag).await?;
-        authorize_replication_only_put_headers(req).await?;
-        req.install_bucket_generation(bucket_generation?);
-        Ok(())
+        access_owned::object_commit::authorize_complete_multipart_upload(req, self.server_ctx(), &bucket, &key, specific_etag)
+            .await
     }
 
     /// Checks whether the CopyObject request has accesses to the resources.

rustfs/src/storage/access_owned/object_commit.rs

@@ -27,6 +27,29 @@ use super::super::{
 };
 use crate::error::ApiError;
 
+/// Authorizes completion and records the bucket generation before publication.
+pub(crate) async fn authorize_complete_multipart_upload<R: AuthorizationRequest + ?Sized>(
+    req: &mut R,
+    server: &ServerContextSlot,
+    bucket: &str,
+    key: &str,
+    specific_etag: bool,
+) -> S3Result<()> {
+    use super::super::{
+        authorize_replication_only_put_headers, authorize_write_etag_condition, complete_multipart_upload_authorize_action,
+        load_bucket_generation,
+    };
+    let generation = load_bucket_generation(server, req, bucket).await;
+    let info = req.req_info_mut().ok_or_else(missing_req_info_error)?;
+    info.bucket = Some(bucket.to_owned());
+    info.object = Some(key.to_owned());
+    authorize_request(req, complete_multipart_upload_authorize_action()).await?;
+    authorize_write_etag_condition(req, specific_etag).await?;
+    authorize_replication_only_put_headers(req).await?;
+    req.install_bucket_generation(generation?);
+    Ok(())
+}
+
 /// The object a copy names as its source, as decoded from `x-amz-copy-source`.
 pub(crate) struct CopySourceObject<'a> {
     pub(crate) bucket: &'a str,

rustfs/src/storage/access_owned/object_commit_test.rs

@@ -308,3 +308,45 @@ async fn copy_hooks_source_first_inner() {
         assert_eq!(part_error.code(), &part_expected, "UploadPartCopy: {case}");
     }
 }
+
+#[tokio::test]
+#[serial]
+async fn completion_authorization_requires_write_read_and_replication_permissions() {
+    let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+    let server = ServerContextSlot::new();
+    assert!(server.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms)))));
+    for (actions, specific_etag, replica, allowed) in [
+        (vec!["s3:PutObject"], false, false, true),
+        (vec![], false, false, false),
+        (vec!["s3:PutObject"], true, false, false),
+        (vec!["s3:PutObject"], false, true, false),
+        (vec!["s3:PutObject", "s3:GetObject"], true, false, true),
+        (vec!["s3:GetObject"], false, false, false),
+    ] {
+        let bucket = source_bucket(&store, &actions).await;
+        let mut request = build_request((), Method::POST);
+        request.extensions.insert(Arc::clone(&server));
+        request.extensions.insert(ReqInfo::default());
+        if replica {
+            request
+                .headers
+                .insert("x-amz-replication-status", http::HeaderValue::from_static("REPLICA"));
+        }
+        let mut envelope = RequestEnvelope::from_s3s(&request);
+        for owned in [false, true] {
+            let result = if owned {
+                super::authorize_complete_multipart_upload(&mut envelope, &server, &bucket, "object", specific_etag).await
+            } else {
+                super::authorize_complete_multipart_upload(&mut request, &server, &bucket, "object", specific_etag).await
+            };
+            if allowed {
+                result.expect("permitted completion");
+            } else {
+                assert_eq!(result.expect_err("completion refused").code(), &S3ErrorCode::AccessDenied);
+            }
+        }
+        assert_eq!(envelope.bucket_generation().is_some(), allowed);
+        assert_eq!(envelope.req_info().unwrap().bucket.as_deref(), Some(bucket.as_str()));
+        assert_eq!(envelope.req_info().unwrap().object.as_deref(), Some("object"));
+    }
+}

rustfs/src/storage/ecfs.rs

@@ -302,6 +302,9 @@ fn seam_error(error: ConversionError) -> S3Error {
 // The bodies of the methods below, one file per migration batch (rustfs/backlog#2765).
 #[path = "legacy_edge/mod.rs"]
 mod legacy_edge;
+
+#[cfg(test)]
+pub(crate) use legacy_edge::complete_multipart_upload_input;
 #[cfg(test)]
 pub(crate) use legacy_edge::list_objects_v2_through_legacy_edge;
 

rustfs/src/storage/legacy_edge/mod.rs

@@ -27,6 +27,9 @@ mod listing_multipart;
 pub(crate) use listing_multipart::list_objects_v2_through_legacy_edge;
 mod object_meta;
 mod object_misc;
+
+#[cfg(test)]
+pub(crate) use object_misc::complete_multipart_upload_input;
 mod object_read;
 mod object_write;
 

rustfs/src/storage/legacy_edge/object_misc.rs

@@ -21,13 +21,95 @@
 //! owns this file and is the only one that edits it.
 
 use crate::storage::ecfs::*;
+use crate::storage::ecfs_extend::validate_new_object_key;
 use crate::storage::helper::audit_only_operation;
 use crate::storage::s3_api;
+use http::{HeaderMap, HeaderValue};
+use rustfs_gateway_types::compat::s3s_5761ddfe::generated::ops::complete_multipart_upload;
 use rustfs_gateway_types::compat::s3s_5761ddfe::generated::ops::delete_object;
+use rustfs_gateway_types::dto as gateway_dto;
 use rustfs_io_metrics::record_s3_op;
 use rustfs_s3_ops::S3Operation;
 use tracing::instrument;
 
+/// CompleteMultipartUpload's input as its use case reads it, from what the legacy decoder read.
+///
+/// The parts cross with their numbers and entity tags as the use case reads them: an absent number
+/// as part `0`, which the use case refuses as it always has, and a tag as its opaque bytes, which
+/// the gateway's request grammar might refuse. The per-algorithm checksum members stay behind: the
+/// use case reads the `x-amz-checksum-*` headers they were decoded from, and the gateway input
+/// holds only one checksum.
+pub(crate) fn complete_multipart_upload_input(
+    mut input: CompleteMultipartUploadInput,
+) -> Result<(gateway_dto::CompleteMultipartUploadInput, Option<gateway_dto::CompletedMultipartUpload>), ConversionError> {
+    let has_upload = input.multipart_upload.is_some();
+    input.multipart_upload.get_or_insert_with(Default::default);
+    let tags: Vec<Option<String>> = input
+        .multipart_upload
+        .iter_mut()
+        .flat_map(|upload| upload.parts.iter_mut().flatten())
+        .map(|part| {
+            part.part_number.get_or_insert(0);
+            part.e_tag.take().map(|tag| tag.value().to_owned())
+        })
+        .collect();
+    for checksum in [
+        &mut input.checksum_crc32,
+        &mut input.checksum_crc32c,
+        &mut input.checksum_crc64nvme,
+        &mut input.checksum_md5,
+        &mut input.checksum_sha1,
+        &mut input.checksum_sha256,
+        &mut input.checksum_sha512,
+        &mut input.checksum_xxhash128,
+        &mut input.checksum_xxhash3,
+        &mut input.checksum_xxhash64,
+    ] {
+        *checksum = None;
+    }
+    let mut converted = complete_multipart_upload::input_from_s3s(input)?;
+    for (part, tag) in converted.multipart_upload.parts.iter_mut().zip(tags) {
+        part.e_tag = tag.map(rustfs_gateway_types::ETag::from_stored);
+    }
+    let parts = std::mem::take(&mut converted.multipart_upload);
+    Ok((converted, has_upload.then_some(parts)))
+}
+
+/// CompleteMultipartUpload's answer as the legacy stack writes it: the members its output has no
+/// carrier for are the response headers it has always written them as.
+pub(super) fn complete_multipart_upload_answer(
+    mut output: gateway_dto::CompleteMultipartUploadOutput,
+) -> Result<(CompleteMultipartUploadOutput, HeaderMap), ConversionError> {
+    let mut headers = HeaderMap::new();
+    for (name, value) in [
+        ("x-amz-checksum-md5", output.checksum_header_md5.take()),
+        ("x-amz-checksum-sha512", output.checksum_header_sha512.take()),
+        ("x-amz-checksum-xxhash3", output.checksum_header_xxhash3.take()),
+        ("x-amz-checksum-xxhash64", output.checksum_header_xxhash64.take()),
+        ("x-amz-checksum-xxhash128", output.checksum_header_xxhash128.take()),
+        ("x-amz-server-side-encryption-customer-algorithm", output.sse_customer_algorithm.take()),
+        ("x-amz-server-side-encryption-customer-key-md5", output.sse_customer_key_md5.take()),
+    ] {
+        if let Some(value) = value {
+            let value = HeaderValue::from_str(&value).map_err(|_| ConversionError {
+                field: name,
+                reason: "not a header value",
+            })?;
+            headers.insert(name, value);
+        }
+    }
+    let mut output = complete_multipart_upload::output_to_s3s(output)?;
+    output.e_tag = output.e_tag.map(xml_escaped_etag);
+    Ok((output, headers))
+}
+
+/// Prepares the entity tag for legacy XML serialization.
+fn xml_escaped_etag(mut etag: ETag) -> ETag {
+    let (ETag::Strong(value) | ETag::Weak(value)) = &mut etag;
+    *value = quick_xml::escape::escape(value.as_str()).into_owned();
+    etag
+}
+
 impl FS {
     #[instrument(name = "complete_multipart_upload", level = "debug", skip(self, req))]
     pub(in crate::storage::ecfs) async fn complete_multipart_upload_edge(
@@ -35,8 +117,31 @@ impl FS {
         req: S3Request<CompleteMultipartUploadInput>,
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
         crate::hp_guard!("S3::complete_multipart_upload");
+        record_s3_op(S3Operation::CompleteMultipartUpload);
+        // The key's own refusal and message, which the seam's key floor would replace.
+        validate_new_object_key(&req.input.key, "CompleteMultipartUpload")?;
         let usecase = s3_api::multipart_usecase_for(self);
-        Box::pin(usecase.execute_complete_multipart_upload(req)).await
+        let answer = through_gateway_seam(
+            req,
+            complete_multipart_upload_input,
+            complete_multipart_upload_answer,
+            |envelope, (mut input, parts)| {
+                Box::pin(async move {
+                    match parts {
+                        Some(parts) => {
+                            input.multipart_upload = parts;
+                            usecase.execute_complete_multipart_upload(envelope, input).await
+                        }
+                        None => usecase.execute_complete_multipart_upload_parts(envelope, input, None).await,
+                    }
+                })
+            },
+        )
+        .await?;
+        let (output, headers) = answer.output;
+        let mut response = S3Response::new(output);
+        response.headers = headers;
+        Ok(response)
     }
 
     /// Copy an object from one location to another

rustfs/src/storage/legacy_edge/tests/mod.rs

@@ -51,5 +51,6 @@ fn store_uninitialized_premise_holds() -> bool {
 mod bucket_access;
 mod bucket_replication_lifecycle;
 mod listing_multipart;
+mod object_commit;
 mod object_meta;
 mod object_read;

rustfs/src/storage/legacy_edge/tests/object_commit.rs

@@ -0,0 +1,137 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Tests of the `object_commit` legacy-edge conversions (batch E1): what crosses the gateway seam
+//! for CompleteMultipartUpload, in both directions.
+
+use super::super::object_misc::{complete_multipart_upload_answer, complete_multipart_upload_input};
+use crate::storage::ecfs::*;
+use crate::storage::storage_api::to_s3s_xml_etag;
+use rustfs_gateway_types::dto as gateway_dto;
+
+fn input_with_parts(parts: Vec<CompletedPart>) -> CompleteMultipartUploadInput {
+    CompleteMultipartUploadInput {
+        bucket: "bucket".to_owned(),
+        key: "object".to_owned(),
+        upload_id: "upload".to_owned(),
+        multipart_upload: Some(CompletedMultipartUpload { parts: Some(parts) }),
+        ..Default::default()
+    }
+}
+
+/// Preserve part numbers and entity tags across the conversion.
+#[test]
+fn complete_input_preserves_part_fields() {
+    let parts = vec![
+        CompletedPart {
+            part_number: None,
+            e_tag: Some(ETag::Strong(String::new())),
+            ..Default::default()
+        },
+        CompletedPart {
+            part_number: Some(2),
+            e_tag: Some(ETag::Weak("abc".to_owned())),
+            ..Default::default()
+        },
+        CompletedPart {
+            part_number: Some(3),
+            e_tag: None,
+            ..Default::default()
+        },
+    ];
+
+    let (_, converted) = complete_multipart_upload_input(input_with_parts(parts)).expect("every part crosses");
+    let converted = converted.expect("multipart payload");
+
+    let read: Vec<(i32, Option<&str>)> = converted
+        .parts
+        .iter()
+        .map(|part| (part.part_number, part.e_tag.as_ref().map(|tag| tag.opaque_tag())))
+        .collect();
+    assert_eq!(read, [(0, Some("")), (2, Some("abc")), (3, None)]);
+}
+
+/// Several per-algorithm checksum members, and one of the wrong width, do not stop the request:
+/// the use case reads the headers they came from.
+#[test]
+fn complete_input_leaves_the_checksum_members_behind() {
+    let mut input = input_with_parts(Vec::new());
+    input.checksum_crc32 = Some("AAAA".to_owned());
+    input.checksum_sha256 = Some("47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=".to_owned());
+
+    let (converted, _) = complete_multipart_upload_input(input).expect("the checksum members stay behind");
+
+    assert!(converted.checksum_spec.is_none());
+}
+
+/// The members the legacy output cannot carry become the headers the legacy stack wrote, and the
+/// entity tag reaches the XML writer escaped as main escapes a stored tag.
+#[test]
+fn complete_answer_writes_header_members_as_headers_and_escapes_the_tag() {
+    let stored = "a&b<c";
+    let output = gateway_dto::CompleteMultipartUploadOutput {
+        e_tag: Some(rustfs_gateway_types::ETag::from_stored(stored)),
+        checksum_header_xxhash3: Some("AAAAAAAAAAA=".to_owned()),
+        sse_customer_algorithm: Some("AES256".to_owned()),
+        sse_customer_key_md5: Some("SaaDheEjzuynJH8eW6AEpQ==".to_owned()),
+        ..Default::default()
+    };
+
+    #[cfg(feature = "gateway-stack")]
+    {
+        use rustfs_gateway::{Limits, MetaView, OperationCodec, TargetKind, WireRequest};
+        let request = http::Request::builder()
+            .method("POST")
+            .uri("http://localhost/bucket/object")
+            .header("host", "localhost")
+            .body(bytes::Bytes::new())
+            .expect("wire request");
+        let wire = WireRequest::accept(request, &Limits::default()).expect("accepted wire request");
+        let view = MetaView::of(&wire, TargetKind::Object).expect("object view");
+        let encoded = rustfs_gateway::dto::CompleteMultipartUpload::encode(output.clone(), &view, 200).expect("encoded response");
+        assert_eq!(
+            encoded.headers.get("x-amz-checksum-xxhash3").and_then(|v| v.to_str().ok()),
+            Some("AAAAAAAAAAA=")
+        );
+    }
+
+    let (legacy, headers) = complete_multipart_upload_answer(output).expect("the answer crosses");
+
+    assert_eq!(legacy.e_tag, Some(to_s3s_xml_etag(stored)));
+    let written: Vec<(&str, &str)> = headers
+        .iter()
+        .map(|(name, value)| (name.as_str(), value.to_str().expect("ASCII")))
+        .collect();
+    assert_eq!(
+        written,
+        [
+            ("x-amz-checksum-xxhash3", "AAAAAAAAAAA="),
+            ("x-amz-server-side-encryption-customer-algorithm", "AES256"),
+            ("x-amz-server-side-encryption-customer-key-md5", "SaaDheEjzuynJH8eW6AEpQ=="),
+        ]
+    );
+}
+
+/// A header member no header can carry is refused by name, never written or dropped.
+#[test]
+fn complete_answer_refuses_a_header_member_that_is_not_a_header_value() {
+    let output = gateway_dto::CompleteMultipartUploadOutput {
+        checksum_header_md5: Some("line\nbreak".to_owned()),
+        ..Default::default()
+    };
+
+    let refused = complete_multipart_upload_answer(output).expect_err("not a header value");
+
+    assert_eq!(refused.field, "x-amz-checksum-md5");
+}

rustfs/src/storage/storage_api.rs

@@ -568,8 +568,8 @@ pub(crate) mod ecstore_error {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::error::PoolMetadataFailure;
     pub(crate) use rustfs_ecstore::api::error::{
-        Error, PoolMetadataError, Result, StorageError, is_err_bucket_not_found, is_err_invalid_upload_id,
-        is_err_object_not_found, is_err_version_not_found,
+        Error, PoolMetadataError, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found,
+        is_err_version_not_found,
     };
 }
 
@@ -2368,10 +2368,6 @@ pub(crate) fn is_err_bucket_not_found(err: &Error) -> bool {
     ecstore_error::is_err_bucket_not_found(err)
 }
 
-pub(crate) fn is_err_invalid_upload_id(err: &Error) -> bool {
-    ecstore_error::is_err_invalid_upload_id(err)
-}
-
 pub(crate) fn is_err_object_not_found(err: &Error) -> bool {
     ecstore_error::is_err_object_not_found(err)
 }

rustfs/tests/embedded_gateway_stack_test/object_commit.rs

@@ -12,6 +12,434 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Refusal-matrix rows of migration batch E1 (rustfs/backlog#2765). None is served yet.
+//! Refusal-matrix rows of migration batch E1 (rustfs/backlog#2765), and how the legacy stack
+//! answers CompleteMultipartUpload for every request shape whose conversion onto the gateway's
+//! types needs care. CompleteMultipartUpload is served on both stacks.
 
-pub(crate) const MATRIX: &[super::MatrixRow] = &[];
+use super::{LEGACY, create_bucket_in_the_store, s3_client, with_stack};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+
+pub(crate) const MATRIX: &[super::MatrixRow] = &[super::MatrixRow {
+    operation: "CompleteMultipartUpload",
+    metric: rustfs_s3_ops::S3Operation::CompleteMultipartUpload,
+    has_error_body: true,
+    call: |client, bucket| {
+        Box::pin(async move {
+            use rustfs_ecstore::api::object::{ObjectOptions, PutObjReader};
+            use rustfs_storage_api::MultipartOperations as _;
+            let context = rustfs::app::context::get_global_app_context().expect("embedded context");
+            let store = context.object_store();
+            let opts = ObjectOptions::default();
+            let upload = store
+                .new_multipart_upload(bucket, "complete-matrix", &opts)
+                .await
+                .expect("matrix upload");
+            let mut body = PutObjReader::from_vec(b"matrix part".to_vec());
+            let stored = store
+                .put_object_part(bucket, "complete-matrix", &upload.upload_id, 1, &mut body, &opts)
+                .await
+                .expect("matrix part");
+            super::outcome(
+                client
+                    .complete_multipart_upload()
+                    .bucket(bucket)
+                    .key("complete-matrix")
+                    .upload_id(upload.upload_id)
+                    .multipart_upload(
+                        CompletedMultipartUpload::builder()
+                            .parts(part(Some(1), stored.etag.as_deref().expect("stored part ETag")))
+                            .build(),
+                    )
+                    .send()
+                    .await,
+            )
+        })
+    },
+}];
+
+const ORDINARY_BUCKET: &str = "e1-complete";
+/// Names RustFS creates and the AWS bucket grammar refuses.
+const RESERVED_SHAPED_BUCKETS: [&str; 2] = ["sthree-e1", "e1-s3alias"];
+
+/// What one CompleteMultipartUpload answered: `200` with the members a client reads, or the
+/// refusal's status, code and message.
+async fn complete(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    upload_id: &str,
+    part: CompletedPart,
+    checksums: &[(&str, &str)],
+) -> String {
+    let mut request = client
+        .complete_multipart_upload()
+        .bucket(bucket)
+        .key(key)
+        .upload_id(upload_id)
+        .multipart_upload(CompletedMultipartUpload::builder().parts(part).build());
+    for (algorithm, value) in checksums {
+        request = match *algorithm {
+            "crc32" => request.checksum_crc32(*value),
+            "sha256" => request.checksum_sha256(*value),
+            other => panic!("no setter for {other}"),
+        };
+    }
+    match request.send().await {
+        Ok(output) => format!(
+            "200 bucket={:?} key_len={:?} etag={:?} crc32={:?} sha256={:?} type={:?}",
+            output.bucket(),
+            output.key().map(str::len),
+            output.e_tag().map(|etag| etag.ends_with("-1\"")),
+            output.checksum_crc32(),
+            output.checksum_sha256(),
+            output.checksum_type(),
+        ),
+        Err(error) => format!(
+            "{:?} {:?} {:?}",
+            error.raw_response().map(|response| response.status().as_u16()),
+            error.code(),
+            error.message()
+        ),
+    }
+}
+
+/// Creates an upload of one part on `key` and returns its id and the part's entity tag.
+async fn one_part_upload(client: &Client, bucket: &str, key: &str) -> (String, String) {
+    let upload_id = client
+        .create_multipart_upload()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("CreateMultipartUpload")
+        .upload_id()
+        .expect("an upload id")
+        .to_owned();
+    let etag = client
+        .upload_part()
+        .bucket(bucket)
+        .key(key)
+        .upload_id(&upload_id)
+        .part_number(1)
+        .body(ByteStream::from_static(b"one part"))
+        .send()
+        .await
+        .expect("UploadPart")
+        .e_tag()
+        .expect("a part entity tag")
+        .to_owned();
+    (upload_id, etag)
+}
+
+fn part(number: Option<i32>, etag: &str) -> CompletedPart {
+    CompletedPart::builder().set_part_number(number).e_tag(etag).build()
+}
+
+/// CompleteMultipartUpload on the legacy stack answers every request shape as before its use case
+/// moved onto the gateway's types. Every answer was measured on `main` before the move.
+#[test]
+fn the_legacy_stack_completes_every_upload_as_before() {
+    with_stack(LEGACY, |server| async move {
+        let client = s3_client(&server.endpoint(), server.access_key(), server.secret_key());
+        client
+            .create_bucket()
+            .bucket(ORDINARY_BUCKET)
+            .send()
+            .await
+            .expect("CreateBucket");
+        for bucket in RESERVED_SHAPED_BUCKETS {
+            create_bucket_in_the_store(bucket).await;
+        }
+        let mut answered = Vec::new();
+
+        for bucket in [ORDINARY_BUCKET].into_iter().chain(RESERVED_SHAPED_BUCKETS) {
+            let (upload_id, etag) = one_part_upload(&client, bucket, "object").await;
+            let answer = complete(&client, bucket, "object", &upload_id, part(Some(1), &etag), &[]).await;
+            answered.push((format!("bucket {bucket}"), answer));
+        }
+
+        let too_long = "k".repeat(1025);
+        for (label, key) in [("NUL key", "nul\0key"), ("1025-byte key", too_long.as_str())] {
+            let answer = complete(&client, ORDINARY_BUCKET, key, "no-such-upload", part(Some(1), "\"0\""), &[]).await;
+            answered.push((label.to_owned(), answer));
+        }
+
+        type ShapeCase<'a> = (&'a str, Option<i32>, fn(&str) -> String, &'a [(&'a str, &'a str)]);
+        let shapes: [ShapeCase<'_>; 10] = [
+            ("empty part ETag", Some(1), |_| String::new(), &[]),
+            ("weak spelling of the stored part ETag", Some(1), |etag| format!("W/{etag}"), &[]),
+            ("unquoted stored part ETag", Some(1), |etag| etag.trim_matches('"').to_owned(), &[]),
+            (
+                "quoted other part ETag",
+                Some(1),
+                |_| "\"00000000000000000000000000000000\"".to_owned(),
+                &[],
+            ),
+            (
+                "unquoted other part ETag",
+                Some(1),
+                |_| "00000000000000000000000000000000".to_owned(),
+                &[],
+            ),
+            (
+                "weak other part ETag",
+                Some(1),
+                |_| "W/\"00000000000000000000000000000000\"".to_owned(),
+                &[],
+            ),
+            ("part without a number", None, str::to_owned, &[]),
+            (
+                "two checksum headers",
+                Some(1),
+                str::to_owned,
+                &[
+                    ("crc32", "AAAAAA=="),
+                    ("sha256", "47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU="),
+                ],
+            ),
+            ("checksum of the wrong width", Some(1), str::to_owned, &[("crc32", "AAAA")]),
+            ("ordinary", Some(1), str::to_owned, &[("crc32", "AAAAAA==")]),
+        ];
+        for (index, (label, number, etag_for, checksums)) in shapes.into_iter().enumerate() {
+            let key = format!("shape-{index}");
+            let (upload_id, etag) = one_part_upload(&client, ORDINARY_BUCKET, &key).await;
+            let answer = complete(&client, ORDINARY_BUCKET, &key, &upload_id, part(number, &etag_for(&etag)), checksums).await;
+            answered.push((label.to_owned(), answer));
+        }
+
+        for (label, upload_id) in [
+            ("upload id with a dot-dot segment", "a/../b"),
+            ("upload id never minted", "bm90LWFuLXVwbG9hZA"),
+        ] {
+            let answer = complete(&client, ORDINARY_BUCKET, "object", upload_id, part(Some(1), "\"0\""), &[]).await;
+            answered.push((label.to_owned(), answer));
+        }
+
+        // No body at all: the legacy decoder refuses it before the operation.
+        let (upload_id, _) = one_part_upload(&client, ORDINARY_BUCKET, "no-body").await;
+        let answer = match client
+            .complete_multipart_upload()
+            .bucket(ORDINARY_BUCKET)
+            .key("no-body")
+            .upload_id(&upload_id)
+            .send()
+            .await
+        {
+            Ok(_) => "200".to_owned(),
+            Err(error) => format!(
+                "{:?} {:?} {:?}",
+                error.raw_response().map(|response| response.status().as_u16()),
+                error.code(),
+                error.message()
+            ),
+        };
+        answered.push(("no body".to_owned(), answer));
+
+        let expected = [
+            (
+                "bucket e1-complete",
+                "200 bucket=Some(\"e1-complete\") key_len=Some(6) etag=Some(true) crc32=None sha256=None type=None",
+            ),
+            (
+                "bucket sthree-e1",
+                "200 bucket=Some(\"sthree-e1\") key_len=Some(6) etag=Some(true) crc32=None sha256=None type=None",
+            ),
+            (
+                "bucket e1-s3alias",
+                "200 bucket=Some(\"e1-s3alias\") key_len=Some(6) etag=Some(true) crc32=None sha256=None type=None",
+            ),
+            (
+                "NUL key",
+                "Some(400) Some(\"InvalidArgument\") Some(\"Object key contains invalid control characters: \\\"nul\\\\0key\\\"\")",
+            ),
+            ("1025-byte key", "Some(400) Some(\"KeyTooLongError\") Some(\"Your key is too long.\")"),
+            (
+                "empty part ETag",
+                "Some(400) Some(\"InvalidPart\") Some(\"One or more of the specified parts could not be found.  The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.\")",
+            ),
+            (
+                "weak spelling of the stored part ETag",
+                "200 bucket=Some(\"e1-complete\") key_len=Some(7) etag=Some(true) crc32=None sha256=None type=None",
+            ),
+            (
+                "unquoted stored part ETag",
+                "200 bucket=Some(\"e1-complete\") key_len=Some(7) etag=Some(true) crc32=None sha256=None type=None",
+            ),
+            (
+                "quoted other part ETag",
+                "Some(400) Some(\"InvalidPart\") Some(\"One or more of the specified parts could not be found.  The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.\")",
+            ),
+            (
+                "unquoted other part ETag",
+                "Some(400) Some(\"InvalidPart\") Some(\"One or more of the specified parts could not be found.  The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.\")",
+            ),
+            (
+                "weak other part ETag",
+                "Some(400) Some(\"InvalidPart\") Some(\"One or more of the specified parts could not be found.  The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.\")",
+            ),
+            (
+                "part without a number",
+                "Some(400) Some(\"InvalidPart\") Some(\"Part number 0 must be between 1 and 10000\")",
+            ),
+            (
+                "two checksum headers",
+                "200 bucket=Some(\"e1-complete\") key_len=Some(7) etag=Some(true) crc32=Some(\"AAAAAA==\") sha256=Some(\"47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=\") type=None",
+            ),
+            (
+                "checksum of the wrong width",
+                "Some(400) Some(\"BadDigest\") Some(\"The Content-Md5 you specified did not match what we received.\")",
+            ),
+            (
+                "ordinary",
+                "200 bucket=Some(\"e1-complete\") key_len=Some(7) etag=Some(true) crc32=Some(\"AAAAAA==\") sha256=None type=None",
+            ),
+            (
+                "upload id with a dot-dot segment",
+                "Some(404) Some(\"NoSuchUpload\") Some(\"The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.\")",
+            ),
+            (
+                "upload id never minted",
+                "Some(404) Some(\"NoSuchUpload\") Some(\"The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.\")",
+            ),
+            (
+                "no body",
+                "Some(400) Some(\"MalformedXML\") Some(\"This happens when the user sends malformed XML (XML that doesn't conform to the published XSD) for the configuration. The error message is, \\\"The XML you provided was not well-formed or did not validate against our published schema.\\\"\")",
+            ),
+        ];
+        let expected: Vec<(String, String)> = expected
+            .into_iter()
+            .map(|(label, answer)| (label.to_owned(), answer.to_owned()))
+            .collect();
+        assert_eq!(answered, expected);
+        server.shutdown().await;
+    });
+}
+
+#[test]
+fn legacy_absent_upload_refusal_order() {
+    with_stack(LEGACY, |server| async move {
+        use rustfs_ecstore::api::bucket::{
+            metadata_sys,
+            quota::{BucketQuota, checker::QuotaChecker},
+        };
+        use rustfs_ecstore::api::object::{ObjectOptions, PutObjReader};
+        use rustfs_storage_api::{MultipartOperations, ObjectIO};
+        let context = rustfs::app::context::get_global_app_context().expect("embedded context");
+        let store = context.object_store();
+        let client = s3_client(&server.endpoint(), server.access_key(), server.secret_key());
+        for (shape, raw_id) in [("absent", None), ("dot", Some("a/../b")), ("absolute", Some("/upload"))] {
+            for (label, full_quota, locked) in [
+                ("wire-absent-quota", true, false),
+                ("wire-absent", false, false),
+                ("wire-absent-lock", false, true),
+                ("wire-absent-both", true, true),
+            ] {
+                let bucket = format!("{shape}-{label}");
+                let label = bucket.as_str();
+                create_bucket_in_the_store(label).await;
+                let upload = store
+                    .new_multipart_upload(label, "object", &ObjectOptions::default())
+                    .await
+                    .expect("mint");
+                store
+                    .abort_multipart_upload(label, "object", &upload.upload_id, &ObjectOptions::default())
+                    .await
+                    .expect("abort");
+                assert!(
+                    store
+                        .lookup_multipart_completion_binding(label, "object", &upload.upload_id, &ObjectOptions::default())
+                        .await
+                        .expect("binding")
+                        .is_none()
+                );
+                if locked {
+                    metadata_sys::update(
+                        label,
+                        rustfs_ecstore::api::bucket::metadata::OBJECT_LOCK_CONFIG,
+                        b"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>"
+                            .to_vec(),
+                    )
+                    .await
+                    .expect("lock config");
+                    let mut opts = ObjectOptions::default();
+                    opts.user_defined
+                        .insert("x-amz-object-lock-legal-hold".to_owned(), "ON".to_owned());
+                    let mut reader = PutObjReader::from_vec(b"existing".to_vec());
+                    store
+                        .clone()
+                        .put_object(label, "object", &mut reader, &opts)
+                        .await
+                        .expect("held object");
+                }
+                rustfs_ecstore::api::data_usage::seed_bucket_usage_memory_for_test(label, 2).await;
+                if full_quota {
+                    QuotaChecker::new(store.bucket_metadata_sys().expect("metadata context"))
+                        .set_quota_config(
+                            label,
+                            BucketQuota {
+                                quota: Some(1),
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("quota");
+                }
+                let before = super::s3_request_counts();
+                let answer =
+                    complete(&client, label, "object", raw_id.unwrap_or(&upload.upload_id), part(Some(1), "part"), &[]).await;
+                assert_eq!(
+                    super::counted_since(&before),
+                    std::collections::BTreeMap::from([(("POST", "s3:CompleteMultipartUpload", "4xx"), 1)]),
+                    "{shape}/{label}"
+                );
+                let (status, code, message) = if raw_id.is_some() {
+                    (
+                        404,
+                        "NoSuchUpload",
+                        "The specified multipart upload does not exist. The upload ID might be invalid, or the multipart upload might have been aborted or completed.",
+                    )
+                } else if full_quota {
+                    (400, "InvalidRequest", "Bucket quota exceeded. Current usage: 2 bytes, limit: 1 bytes")
+                } else if locked {
+                    (
+                        403,
+                        "AccessDenied",
+                        "Object has a legal hold and cannot be overwritten. Remove the legal hold first.",
+                    )
+                } else {
+                    (
+                        404,
+                        "NoSuchUpload",
+                        "The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
+                    )
+                };
+                assert_eq!(answer, format!("{:?} {:?} {:?}", Some(status), Some(code), Some(message)), "{label}");
+            }
+        }
+        server.shutdown().await;
+    });
+}
+
+#[test]
+fn gateway_completion_authorization_is_scoped_to_the_allowed_bucket() {
+    super::with_stack(super::GATEWAY, |server| async move {
+        const ALLOWED: &str = "e1-complete-allowed";
+        const DENIED: &str = "e1-complete-denied";
+        const USER: &str = "e1completescoped";
+        for bucket in [ALLOWED, DENIED] {
+            create_bucket_in_the_store(bucket).await;
+        }
+        super::create_user_allowed_on_one_bucket(USER, "s3:PutObject", &format!("{ALLOWED}/*")).await;
+        let client = s3_client(&server.endpoint(), USER, super::USER_SECRET_KEY);
+        assert_eq!((MATRIX[0].call)(&client, ALLOWED).await, Ok(()));
+        let refusal = (MATRIX[0].call)(&client, DENIED)
+            .await
+            .expect_err("another bucket is not authorized");
+        assert_eq!(refusal.code.as_deref(), Some("AccessDenied"));
+        assert_eq!(refusal.status, Some(403));
+        server.shutdown().await;
+    });
+}