From 3bc8d79fe5559fde1fc1c6af087f3a75b3121440 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 7 Jul 2026 04:28:39 +0800 Subject: [PATCH] fix(multipart): serialize put_object_part per uploadId (#4329) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 106 ++++++++++++++++++ crates/ecstore/src/set_disk/ops/multipart.rs | 26 +++++ 2 files changed, 132 insertions(+) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 77bb98048..2badf75f9 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/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> = (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() { diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 1d9ceaf00..b97d2958e 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/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,