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/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,