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,