fix: handle duplicate part numbers in CompleteMultipartUpload (#1584)

Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/set_disk.rs

@@ -5718,17 +5718,20 @@ impl StorageAPI 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;
+        let part_lookup: HashMap<usize, &rustfs_filemeta::ObjectPartInfo> =
+            curr_fi.parts.iter().map(|part| (part.number, part)).collect();
+
         for (i, p) in uploaded_parts.iter().enumerate() {
-            let has_part = curr_fi.parts.iter().find(|v| v.number == p.part_num);
+            let Some(ext_part) = part_lookup.get(&p.part_num) else {
-            if has_part.is_none() {
                 error!(
-                    "complete_multipart_upload has_part.is_none() {:?}, part_id={}, bucket={}, object={}",
+                    "complete_multipart_upload part not found: part_id={}, bucket={}, object={}",
-                    has_part, p.part_num, bucket, object
+                    p.part_num, bucket, object
                 );
                 return Err(Error::InvalidPart(p.part_num, "".to_owned(), p.etag.clone().unwrap_or_default()));
-            }
+            };
-
-            let ext_part = &curr_fi.parts[i];
             info!(target:"rustfs_ecstore::set_disk", part_number = p.part_num, part_size = ext_part.size, part_actual_size = ext_part.actual_size, "Completing multipart part");
 
             // Normalize ETags by removing quotes before comparison (PR #592 compatibility)

rustfs/src/storage/ecfs.rs

@@ -4985,7 +4985,7 @@ impl S3 for FS {
 
         let opts = &get_complete_multipart_upload_opts(&req.headers).map_err(ApiError::from)?;
 
-        let uploaded_parts = multipart_upload
+        let uploaded_parts_vec = multipart_upload
             .parts
             .unwrap_or_default()
             .into_iter()
@@ -4993,10 +4993,24 @@ impl S3 for FS {
             .collect::<Vec<_>>();
 
         // is part number sorted?
-        if !uploaded_parts.is_sorted_by_key(|p| p.part_num) {
+        if !uploaded_parts_vec.is_sorted_by_key(|p| p.part_num) {
             return Err(s3_error!(InvalidPart, "Part numbers must be sorted"));
         }
 
+        // Handle duplicate part numbers: according to S3 specification, when the same part number
+        // is uploaded multiple times, the last uploaded part (in the list order) should be used.
+        // This can happen in concurrent upload scenarios where a part is re-uploaded before completion.
+        // We deduplicate by keeping the last occurrence of each part number using a HashMap.
+        use std::collections::HashMap;
+        let mut part_map: HashMap<usize, CompletePart> = HashMap::new();
+        for part in uploaded_parts_vec {
+            part_map.insert(part.part_num, part);
+        }
+
+        // Reconstruct the parts list in sorted order, keeping only the last occurrence of each part number
+        let mut uploaded_parts: Vec<CompletePart> = part_map.into_values().collect();
+        uploaded_parts.sort_by_key(|p| p.part_num);
+
         // TODO: check object lock
 
         let Some(store) = new_object_layer_fn() else {