fix(s3): return InvalidPart and streamline multipart completion (#8780)

* fix(s3): classify empty multipart completion as InvalidPart Treat absent per-drive multipart transaction directories as empty listings so completion reaches the existing InvalidPart validation. Preserve quorum and transaction recovery, and cover the HTTP and degraded-drive boundaries. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * perf(ecstore): move multipart completion metadata in one pass Reuse quorum-validated part metadata instead of repeatedly inserting, sorting, finding and cloning parts. Keep legacy cleanup identities and ordering while moving the original part allocation into the completion manifest. Compare ETags by borrowing their normalized values. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-27 06:18:43 +00:00 · 2026-10-22 02:45:05 +08:00
parent 65f0599c70
commit 5f5485e332
5 changed files with 309 additions and 78 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=f801018455712d8767ff6d467f3883036f762936f75435a22de74317274bf90d
-sha256-linux=1621c50e25d63f124b411babbf613c04474c5e68864d2e810e50ec892379e91e
+sha256-darwin=359b16f0bba8baa6e995fd3701062a61088fd4c0196ade3bac9e8fda7a050833
+sha256-linux=137098f8bc31a5ca38cac35a5b628346c1082b0277ec3982e2e0ffb69d0ea2e4
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-//! Harbor / Docker Distribution multipart staging to CopyObject regressions.
+//! Multipart completion validation, staging preservation, and CopyObject readiness.

 use crate::common::{RustFSTestEnvironment, init_logging};
 use aws_sdk_s3::error::ProvideErrorMetadata;
@@ -39,6 +39,132 @@ fn list_contains_key(output: &aws_sdk_s3::operation::list_objects_v2::ListObject
        .any(|object| object.key().is_some_and(|candidate| candidate == key))
 }

+#[tokio::test]
+async fn multipart_complete_invalid_etag_returns_400_and_preserves_retry() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
+    let bucket = "multipart-invalid-etag";
+    let key = "object";
+    env.create_test_bucket(bucket).await?;
+    let original = b"original object";
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(original))
+        .send()
+        .await?;
+    let upload = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
+    let upload_id = upload.upload_id().ok_or("missing upload id")?;
+    let payload = b"completed multipart payload";
+    let mut uploaded_etag = None;
+
+    // The first case reproduces #8771's Create -> Complete sequence without UploadPart.
+    for (upload_first, part_number, etag) in [
+        (false, 1, Some("unknown-etag")),
+        (true, 1, Some("unknown-etag")),
+        (true, 1, Some("\"unknown-etag\"")),
+        (true, 1, Some("00000000000000000000000000000000")),
+        (true, 1, None),
+        (true, 2, Some("unknown-etag")),
+    ] {
+        if upload_first && uploaded_etag.is_none() {
+            let part = client
+                .upload_part()
+                .bucket(bucket)
+                .key(key)
+                .upload_id(upload_id)
+                .part_number(1)
+                .body(ByteStream::from_static(payload))
+                .send()
+                .await?;
+            uploaded_etag = Some(part.e_tag().ok_or("missing part ETag")?.to_owned());
+        }
+        let error = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .multipart_upload(
+                CompletedMultipartUpload::builder()
+                    .parts(
+                        CompletedPart::builder()
+                            .part_number(part_number)
+                            .set_e_tag(etag.map(str::to_owned))
+                            .build(),
+                    )
+                    .build(),
+            )
+            .send()
+            .await
+            .expect_err("missing or mismatched parts must not complete the upload");
+        assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(400), "{error:?}");
+        assert_eq!(
+            error.as_service_error().and_then(ProvideErrorMetadata::code),
+            Some("InvalidPart"),
+            "{error:?}"
+        );
+
+        let staged = client
+            .list_parts()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .send()
+            .await?;
+        assert_eq!(staged.parts().len(), usize::from(uploaded_etag.is_some()));
+        if let Some(etag) = uploaded_etag.as_deref() {
+            assert_eq!(staged.parts()[0].part_number(), Some(1));
+            assert_eq!(staged.parts()[0].e_tag(), Some(etag));
+        }
+        let current = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await?
+            .body
+            .collect()
+            .await?
+            .into_bytes();
+        assert_eq!(current.as_ref(), original, "rejected completion must preserve the original object");
+    }
+
+    let etag = uploaded_etag.ok_or("part was not uploaded")?;
+    // Accept the unquoted value, then replay with the quoted UploadPart response.
+    for etag in [etag.trim_matches('"'), etag.as_str()] {
+        client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .multipart_upload(
+                CompletedMultipartUpload::builder()
+                    .parts(CompletedPart::builder().part_number(1).e_tag(etag).build())
+                    .build(),
+            )
+            .send()
+            .await?;
+    }
+    let current = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await?
+        .body
+        .collect()
+        .await?
+        .into_bytes();
+    assert_eq!(current.as_ref(), payload, "a valid retry must publish the exact uploaded bytes");
+    client.delete_object().bucket(bucket).key(key).send().await?;
+    env.delete_test_bucket(bucket).await?;
+    env.stop_server();
+    Ok(())
+}
+
 async fn upload_one_part_mpu(
    client: &aws_sdk_s3::Client,
    bucket: &str,
@@ -5238,6 +5238,8 @@ impl SetDisks {
                        transaction_parts.insert(part_number);
                    }
                }
+                // UploadPart creates this directory; an empty upload has no transactions to recover.
+                Err(DiskError::FileNotFound) => errs.push(None),
                Err(err) => errs.push(Some(err)),
            }
        }
@@ -7689,18 +7689,25 @@ fn get_complete_multipart_md5(parts: &[CompletePart]) -> String {
    format!("{}-{}", etag_hex, parts.len())
 }

-fn completed_multipart_object_part(part_num: usize, ext_part: &ObjectPartInfo) -> ObjectPartInfo {
-    ObjectPartInfo {
-        etag: ext_part.etag.clone(),
-        number: part_num,
-        size: ext_part.size,
-        mod_time: ext_part.mod_time,
-        actual_size: ext_part.actual_size,
-        index: ext_part.index.clone(),
-        checksums: ext_part.checksums.clone(),
-        integrity: ext_part.integrity.clone(),
-        ..Default::default()
+fn prepare_complete_multipart_parts(fi: &mut FileInfo, parts: Vec<ObjectPartInfo>) -> Vec<usize> {
+    // Preserve legacy cleanup ordering, including replacement-only manifests.
+    let capacity = fi.parts.len() + parts.len();
+    let mut staged_part_numbers = Vec::with_capacity(capacity);
+    staged_part_numbers.extend(fi.parts.iter().map(|part| part.number));
+    let mut seen = HashSet::with_capacity(capacity);
+    seen.extend(staged_part_numbers.iter().copied());
+    let mut inserted = false;
+    for part in &parts {
+        if seen.insert(part.number) {
+            staged_part_numbers.push(part.number);
+            inserted = true;
+        }
    }
+    if inserted {
+        staged_part_numbers.sort_unstable();
+    }
+    fi.parts = parts;
+    staged_part_numbers
 }

 fn complete_part_checksum(part: &CompletePart, checksum_type: rustfs_rio::ChecksumType) -> Option<Option<String>> {
@@ -9336,30 +9343,61 @@ mod tests {
    }

    #[test]
-    fn test_completed_multipart_object_part_preserves_checksums() {
+    fn prepare_complete_multipart_parts_preserves_metadata_and_cleanup_order() {
        let checksums = HashMap::from([
-            (rustfs_rio::ChecksumType::CRC32.to_string(), "crc32-value".to_string()),
-            (rustfs_rio::ChecksumType::CRC32C.to_string(), "crc32c-value".to_string()),
+            (rustfs_rio::ChecksumType::CRC32.to_string(), "crc32-value".to_owned()),
+            (rustfs_rio::ChecksumType::CRC32C.to_string(), "crc32c-value".to_owned()),
        ]);
-        let ext_part = ObjectPartInfo {
+        let part = ObjectPartInfo {
            number: 7,
-            etag: "etag-7".to_string(),
+            etag: "etag-7".to_owned(),
            size: 123,
            actual_size: 456,
            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
            index: Some(Bytes::from_static(&[1, 2, 3])),
-            checksums: Some(checksums.clone()),
+            checksums: Some(checksums),
+            integrity: Some(
+                rustfs_filemeta::shard_integrity::PartIntegrity::new(
+                    rustfs_filemeta::shard_integrity::IntegrityLayout::new(2, 2, 1024 * 1024, false).expect("valid layout"),
+                    7,
+                )
+                .expect("valid integrity"),
+            ),
            ..Default::default()
        };
-
-        let completed = completed_multipart_object_part(7, &ext_part);
-
-        assert_eq!(completed.number, 7);
-        assert_eq!(completed.etag, ext_part.etag);
-        assert_eq!(completed.size, ext_part.size);
-        assert_eq!(completed.actual_size, ext_part.actual_size);
-        assert_eq!(completed.index, ext_part.index);
-        assert_eq!(completed.checksums, Some(checksums));
+        let incoming = vec![part.clone()];
+        let pointer = incoming.as_ptr();
+        let mut fi = FileInfo {
+            parts: vec![
+                ObjectPartInfo {
+                    number: 9,
+                    ..Default::default()
+                },
+                ObjectPartInfo {
+                    number: 7,
+                    etag: "stale".to_owned(),
+                    ..Default::default()
+                },
+            ],
+            ..Default::default()
+        };
+        assert_eq!(prepare_complete_multipart_parts(&mut fi, incoming), vec![9, 7]);
+        assert_eq!(fi.parts, vec![part.clone()]);
+        assert_eq!(fi.parts.as_ptr(), pointer, "move the existing allocation");
+        let mut fi = FileInfo {
+            parts: vec![
+                ObjectPartInfo {
+                    number: 9,
+                    ..Default::default()
+                },
+                ObjectPartInfo {
+                    number: 1,
+                    ..Default::default()
+                },
+            ],
+            ..Default::default()
+        };
+        assert_eq!(prepare_complete_multipart_parts(&mut fi, vec![part]), vec![1, 7, 9]);
    }

    #[test]
@@ -40,12 +40,13 @@ use super::super::{
    SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID, SUFFIX_COMPRESSION_SIZE,
    SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SetDisks, SmallWritePath, StorageError,
    Uuid, WriteLayout, check_object_lock_for_deletion_with_state, classify_multipart_part_write_path, coding,
-    complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, completed_multipart_object_part,
-    contains_key_str, create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str,
+    complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, contains_key_str,
+    create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str,
    is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size, log_multipart_write_quorum_failure,
-    parts_after_marker, path_join_buf, record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs,
-    remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size,
-    strip_internal_multipart_metadata, to_object_err, warn,
+    parts_after_marker, path_join_buf, prepare_complete_multipart_parts, record_compression_total_memory,
+    reduce_read_quorum_errs, reduce_write_quorum_errs, remove_header_map, resolve_write_layout,
+    restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, strip_internal_multipart_metadata,
+    to_object_err, warn,
 };
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
 #[cfg(test)]
@@ -2961,29 +2962,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                );
                return Err(Error::InvalidPart(uploaded_parts[i].part_num, bucket.to_owned(), object.to_owned()));
            }
-
-            fi.add_object_part(
-                part.number,
-                part.etag.clone(),
-                part.size,
-                part.mod_time,
-                part.actual_size,
-                part.index.clone(),
-                part.checksums.clone(),
-            );
-            let inserted = fi
-                .parts
-                .iter_mut()
-                .find(|entry| entry.number == part.number)
-                .ok_or(Error::FileCorrupt)?;
-            inserted.integrity.clone_from(&part.integrity);
        }

        let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files_metas, &fi);

-        let curr_fi = fi.clone();
-
-        fi.parts = Vec::with_capacity(uploaded_parts.len());
+        let staged_part_numbers = prepare_complete_multipart_parts(&mut fi, object_parts);

        let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?;
        let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
@@ -3020,19 +3003,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            ..Default::default()
        };

-        // Build a lookup map for O(1) part resolution instead of O(n) find() in the loop
-        // This optimizes from O(n^2) to O(n) when processing many parts
-        use std::collections::{HashMap, HashSet};
-        let part_lookup: HashMap<usize, &ObjectPartInfo> = curr_fi.parts.iter().map(|part| (part.number, part)).collect();
-
-        for (i, p) in uploaded_parts.iter().enumerate() {
-            let Some(ext_part) = part_lookup.get(&p.part_num) else {
-                error!(
-                    "complete_multipart_upload part not found: part_id={}, bucket={}, object={}",
-                    p.part_num, bucket, object
-                );
-                return Err(Error::InvalidPart(p.part_num, "".to_owned(), p.etag.clone().unwrap_or_default()));
-            };
+        for (i, (p, ext_part)) in uploaded_parts.iter().zip(fi.parts.iter_mut()).enumerate() {
            debug!(
                target:"rustfs_ecstore::set_disk",
                event = EVENT_SET_DISK_MULTIPART,
@@ -3045,9 +3016,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                "Set disk multipart part validated"
            );

-            // Normalize ETags by removing quotes before comparison (PR #592 compatibility)
-            let client_etag = p.etag.as_ref().map(|e| rustfs_utils::path::trim_etag(e));
-            let stored_etag = Some(rustfs_utils::path::trim_etag(&ext_part.etag));
+            let client_etag = p.etag.as_deref().map(|etag| etag.trim_matches('"'));
+            let stored_etag = Some(ext_part.etag.trim_matches('"'));
            if client_etag != stored_etag {
                error!(
                    "complete_multipart_upload etag err client={:?}, stored={:?}, part_id={}, bucket={}, object={}",
@@ -3153,9 +3123,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            object_actual_size = object_actual_size
                .checked_add(normalized_actual_size)
                .ok_or(Error::PartMissingOrCorrupt)?;
-            let mut completed_part = completed_multipart_object_part(p.part_num, ext_part);
-            completed_part.actual_size = normalized_actual_size;
-            fi.parts.push(completed_part);
+            ext_part.actual_size = normalized_actual_size;
        }
        if !transformed_object && data_movement_actual_size.is_some_and(|actual_size| actual_size < object_actual_size) {
            return Err(Error::PartMissingOrCorrupt);
@@ -3376,21 +3344,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
        }

        let completed_part_numbers = fi.parts.iter().map(|part| part.number).collect::<HashSet<_>>();
-        let upload_data_dir = curr_fi.data_dir.unwrap_or(Uuid::nil()).to_string();
-        let mut parts = Vec::with_capacity(curr_fi.parts.len());
+        let upload_data_dir = fi.data_dir.unwrap_or(Uuid::nil()).to_string();
+        let mut parts = Vec::with_capacity(staged_part_numbers.len());

-        for p in curr_fi.parts.iter() {
+        for part_number in staged_part_numbers {
            parts.push(path_join_buf(&[
                &upload_id_path,
                upload_data_dir.as_str(),
-                format!("part.{}.meta", p.number).as_str(),
+                format!("part.{part_number}.meta").as_str(),
            ]));

-            if !completed_part_numbers.contains(&p.number) {
+            if !completed_part_numbers.contains(&part_number) {
                parts.push(path_join_buf(&[
                    &upload_id_path,
                    upload_data_dir.as_str(),
-                    format!("part.{}", p.number).as_str(),
+                    format!("part.{part_number}").as_str(),
                ]));
            }
        }
@@ -6223,6 +6191,103 @@ mod tests {
        assert_quorum_minus_one_retry_preserves_completable_part(6, 2, &[3, 4, 5]).await;
    }

+    #[tokio::test]
+    async fn multipart_transaction_recovery_accepts_missing_directories_but_requires_read_quorum() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let part_path = "empty-upload/data/";
+        set_disks
+            .recover_part_transactions(part_path, 2, 3)
+            .await
+            .expect("an upload without parts has no transactions to recover");
+
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[2] = None;
+            disks[3] = None;
+        }
+        set_disks
+            .recover_part_transactions(part_path, 2, 3)
+            .await
+            .expect("missing directories at the exact read quorum prove there are no transactions");
+
+        set_disks.disks.write().await[1] = None;
+        assert_eq!(
+            set_disks.recover_part_transactions(part_path, 2, 3).await,
+            Err(DiskError::ErasureReadQuorum),
+            "one missing directory must not hide three offline disks"
+        );
+
+        disk_stores[1]
+            .write_all(RUSTFS_META_MULTIPART_BUCKET, "empty-upload/data", Bytes::from_static(b"not a directory"))
+            .await
+            .expect("place a file where the part directory should be");
+        set_disks.disks.write().await[1] = Some(disk_stores[1].clone());
+        assert_eq!(
+            set_disks.recover_part_transactions(part_path, 2, 3).await,
+            Err(DiskError::ErasureReadQuorum),
+            "a corrupt directory path must not count as an empty-directory observation"
+        );
+    }
+
+    #[tokio::test]
+    async fn complete_multipart_owned_parts_preserve_legacy_cleanup_and_normalize_size() {
+        use tokio::io::AsyncReadExt as _;
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-owned-parts";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create upload");
+        let payload = b"current sparse part";
+        let selected = put_test_part(&set_disks, bucket, object, &upload.upload_id, 7, payload, 0).await;
+        put_test_part(&set_disks, bucket, object, &upload.upload_id, 2, b"unselected", 10).await;
+        let (fi, mut metadata) = set_disks
+            .check_upload_id_exists(bucket, object, &upload.upload_id, true)
+            .await
+            .expect("read upload metadata");
+        let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
+        let part_path = format!("{upload_path}/{}/part.7.meta", fi.data_dir.expect("upload data directory"));
+        let mut expected = disk_stores[0]
+            .read_parts(RUSTFS_META_MULTIPART_BUCKET, &[part_path])
+            .await
+            .expect("read selected part metadata")
+            .remove(0);
+        expected.actual_size = i64::try_from(payload.len()).expect("payload size");
+        for meta in &mut metadata {
+            meta.parts = vec![
+                ObjectPartInfo {
+                    number: 2,
+                    ..Default::default()
+                },
+                ObjectPartInfo {
+                    number: 7,
+                    etag: "stale".to_owned(),
+                    ..Default::default()
+                },
+            ];
+        }
+        let disks = set_disks.get_disks_internal().await;
+        SetDisks::write_unique_file_info(&disks, "", RUSTFS_META_MULTIPART_BUCKET, &upload_path, &metadata, 3)
+            .await
+            .expect("persist legacy manifest");
+        let completed = set_disks
+            .clone()
+            .complete_multipart_upload(bucket, object, &upload.upload_id, vec![selected], &ObjectOptions::default())
+            .await
+            .expect("complete with current part metadata instead of stale manifest entries");
+        assert_eq!(completed.parts.as_slice(), &[expected]);
+        let mut reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("read completed object");
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("read all bytes");
+        assert_eq!(body, payload);
+    }
+
    #[tokio::test]
    async fn complete_multipart_upload_recovers_interrupted_part_retry() {
        use tokio::io::AsyncReadExt as _;