fix(multipart): serialize put_object_part per uploadId (#4329)

fix(multipart): serialize the part commit per uploadId to prevent mixed-generation shards (backlog#853)

Concurrently re-transmitting the same part could land two shard
generations across different disks: each shard is individually
bitrot-valid, so the corruption surfaces only as a silent mix at read
time.

The hazard is confined to rename_part, where two temp parts are moved
cross-disk onto the SAME final part path — interleaving there can leave
shards from two generations. Each concurrent stream already writes to its
own unique temp dir, so the encode/stream phase never conflicts and must
stay lock-free: holding a lock across it would serialize slow re-transmits
of the same part, break the S3 'last finisher wins' semantics, and cause
UploadPart lock-acquire timeouts (ServiceUnavailable).

Scope a write lock to the uploadId namespace around only the rename_part
commit, so each commit is atomic across disks and the last committer wins
consistently. Mirrors MinIO's per-uploadID NS lock; disjoint from the
object lock held by complete_multipart_upload (no lock-ordering cycle).
Honors opts.no_lock.

The pre-delete-before-rename half of backlog#853 was already fixed in
#4316; this completes the issue.

Adds an end-to-end regression test: six concurrent resends of the same
part must all succeed (no lock-acquire timeout), exactly one generation
stays visible, and the reassembled object equals one intact generation.

Closes rustfs/backlog#853

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -4688,6 +4688,112 @@ mod tests {
         assert_eq!(completed.size, second_part.size as i64);
     }
 
+    // backlog#853: concurrently re-transmitting the same part must not mix two
+    // shard generations. The uploadId commit lock only wraps rename_part, so the
+    // slow streaming phase stays fully concurrent (regression guard: no
+    // ServiceUnavailable / lock-acquire timeout) while the cross-disk commit is
+    // serialized, leaving exactly one self-consistent generation visible.
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn concurrent_resend_same_part_commits_one_generation() {
+        use crate::storage_api_contracts::object::ObjectIO as _;
+
+        let (_paths, ecstore) = setup_test_env().await;
+        let bucket = format!("multipart-resend-{}", Uuid::new_v4().simple());
+        let object = "resend/object.txt";
+        create_test_bucket(&ecstore, &bucket).await;
+
+        let upload = ecstore
+            .new_multipart_upload(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("multipart upload should be created");
+
+        // Distinct payloads with distinct sizes: a mixed-generation reassembly
+        // would produce bytes matching none of them (or fail the read outright).
+        let candidates: Vec<Vec<u8>> = (0..6)
+            .map(|g| {
+                let len = 4096 + g * 512;
+                vec![b'a' + g as u8; len]
+            })
+            .collect();
+
+        let mut tasks = tokio::task::JoinSet::new();
+        for payload in candidates.iter().cloned() {
+            let store = ecstore.clone();
+            let bucket = bucket.clone();
+            let upload_id = upload.upload_id.clone();
+            tasks.spawn(async move {
+                let mut data = PutObjReader::from_vec(payload.clone());
+                store
+                    .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
+                    .await
+                    .map(|info| (info, payload))
+            });
+        }
+
+        // Every concurrent resend must succeed; the fix must not serialize the
+        // streaming phase into lock-acquire timeouts.
+        let mut results = Vec::new();
+        while let Some(joined) = tasks.join_next().await {
+            let outcome = joined.expect("put_object_part task should not panic");
+            results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
+        }
+        assert_eq!(results.len(), candidates.len());
+
+        // Exactly one generation is visible after the serialized commits, and its
+        // recorded size/etag matches one of the payloads we actually wrote.
+        let parts = ecstore
+            .list_object_parts(
+                &bucket,
+                object,
+                &upload.upload_id,
+                None,
+                crate::set_disk::MAX_PARTS_COUNT,
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("multipart parts should be readable after concurrent resends");
+        assert_eq!(parts.parts.len(), 1, "only one generation of part 1 should remain visible");
+        let visible = &parts.parts[0];
+        assert_eq!(visible.part_num, 1);
+
+        let winner = results
+            .iter()
+            .find(|(info, _)| info.etag == visible.etag && info.size == visible.size)
+            .map(|(_, payload)| payload.clone())
+            .expect("the visible part must match exactly one payload that was committed");
+
+        let completed = ecstore
+            .clone()
+            .complete_multipart_upload(
+                &bucket,
+                object,
+                &upload.upload_id,
+                vec![crate::storage_api_contracts::multipart::CompletePart {
+                    part_num: 1,
+                    etag: visible.etag.clone(),
+                    checksum_crc32: None,
+                    checksum_crc32c: None,
+                    checksum_sha1: None,
+                    checksum_sha256: None,
+                    checksum_crc64nvme: None,
+                }],
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("complete multipart upload should succeed with the committed generation");
+        assert_eq!(completed.size, winner.len() as i64);
+
+        // The reassembled object bytes must equal the winning generation exactly
+        // — proof that no shard from a different generation leaked into the read.
+        let mut reader = ecstore
+            .get_object_reader(&bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("completed object should be readable");
+        let bytes = reader.read_all().await.expect("object bytes should be readable");
+        assert_eq!(bytes, winner, "reassembled object must match one intact generation, not a mix of shards");
+    }
+
     #[tokio::test]
     #[serial]
     async fn cleanup_removes_empty_multipart_sha_dirs() {

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -460,6 +460,30 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         drop(writers); // drop writers to close all files
 
         let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix);
+
+        // Serialize only the commit (rename_part), not the whole upload. Each
+        // concurrent stream writes to its own unique temp dir (see `tmp_part`
+        // above), so the encode/stream phase never conflicts and must stay
+        // lock-free — holding a lock across it would serialize slow re-transmits
+        // of the same part and defeat the S3 "last finisher wins" semantics
+        // (it also caused UploadPart lock-acquire timeouts). The mixed-generation
+        // hazard is confined to rename_part, where two temp parts are moved
+        // cross-disk onto the SAME final part_path: interleaving there can leave
+        // shards from two generations, each individually bitrot-valid, that only
+        // surface as silent corruption at read time (backlog#853). A write lock
+        // scoped to the uploadId namespace makes each commit atomic across disks,
+        // so the last committer wins consistently. Mirrors MinIO's per-uploadID
+        // NS lock; distinct from the object lock held by complete_multipart_upload
+        // (disjoint namespaces, no lock-ordering cycle).
+        let _upload_commit_guard = if opts.no_lock {
+            None
+        } else {
+            Some(
+                self.acquire_write_lock_diag("put_object_part_commit", RUSTFS_META_MULTIPART_BUCKET, &upload_id_path)
+                    .await?,
+            )
+        };
+
         let _ = self
             .rename_part(
                 &disks,
@@ -479,6 +503,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             )
             .await?;
 
+        drop(_upload_commit_guard);
+
         let ret: PartInfo = PartInfo {
             etag: Some(etag.clone()),
             part_num: part_id,