diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index cdb358bed..59ca61b05 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=f801018455712d8767ff6d467f3883036f762936f75435a22de74317274bf90d -sha256-linux=1621c50e25d63f124b411babbf613c04474c5e68864d2e810e50ec892379e91e +sha256-darwin=359b16f0bba8baa6e995fd3701062a61088fd4c0196ade3bac9e8fda7a050833 +sha256-linux=137098f8bc31a5ca38cac35a5b628346c1082b0277ec3982e2e0ffb69d0ea2e4 diff --git a/crates/e2e_test/src/multipart_copy_readiness_test.rs b/crates/e2e_test/src/multipart_copy_readiness_test.rs index 57be4679f..56bedfa8b 100644 --- a/crates/e2e_test/src/multipart_copy_readiness_test.rs +++ b/crates/e2e_test/src/multipart_copy_readiness_test.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Harbor / Docker Distribution multipart staging to CopyObject regressions. +//! Multipart completion validation, staging preservation, and CopyObject readiness. use crate::common::{RustFSTestEnvironment, init_logging}; use aws_sdk_s3::error::ProvideErrorMetadata; @@ -39,6 +39,132 @@ fn list_contains_key(output: &aws_sdk_s3::operation::list_objects_v2::ListObject .any(|object| object.key().is_some_and(|candidate| candidate == key)) } +#[tokio::test] +async fn multipart_complete_invalid_etag_returns_400_and_preserves_retry() -> Result<(), Box> { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + let client = env.create_s3_client(); + let bucket = "multipart-invalid-etag"; + let key = "object"; + env.create_test_bucket(bucket).await?; + let original = b"original object"; + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(original)) + .send() + .await?; + let upload = client.create_multipart_upload().bucket(bucket).key(key).send().await?; + let upload_id = upload.upload_id().ok_or("missing upload id")?; + let payload = b"completed multipart payload"; + let mut uploaded_etag = None; + + // The first case reproduces #8771's Create -> Complete sequence without UploadPart. + for (upload_first, part_number, etag) in [ + (false, 1, Some("unknown-etag")), + (true, 1, Some("unknown-etag")), + (true, 1, Some("\"unknown-etag\"")), + (true, 1, Some("00000000000000000000000000000000")), + (true, 1, None), + (true, 2, Some("unknown-etag")), + ] { + if upload_first && uploaded_etag.is_none() { + let part = client + .upload_part() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .part_number(1) + .body(ByteStream::from_static(payload)) + .send() + .await?; + uploaded_etag = Some(part.e_tag().ok_or("missing part ETag")?.to_owned()); + } + let error = client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .multipart_upload( + CompletedMultipartUpload::builder() + .parts( + CompletedPart::builder() + .part_number(part_number) + .set_e_tag(etag.map(str::to_owned)) + .build(), + ) + .build(), + ) + .send() + .await + .expect_err("missing or mismatched parts must not complete the upload"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(400), "{error:?}"); + assert_eq!( + error.as_service_error().and_then(ProvideErrorMetadata::code), + Some("InvalidPart"), + "{error:?}" + ); + + let staged = client + .list_parts() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .send() + .await?; + assert_eq!(staged.parts().len(), usize::from(uploaded_etag.is_some())); + if let Some(etag) = uploaded_etag.as_deref() { + assert_eq!(staged.parts()[0].part_number(), Some(1)); + assert_eq!(staged.parts()[0].e_tag(), Some(etag)); + } + let current = client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await? + .body + .collect() + .await? + .into_bytes(); + assert_eq!(current.as_ref(), original, "rejected completion must preserve the original object"); + } + + let etag = uploaded_etag.ok_or("part was not uploaded")?; + // Accept the unquoted value, then replay with the quoted UploadPart response. + for etag in [etag.trim_matches('"'), etag.as_str()] { + client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(upload_id) + .multipart_upload( + CompletedMultipartUpload::builder() + .parts(CompletedPart::builder().part_number(1).e_tag(etag).build()) + .build(), + ) + .send() + .await?; + } + let current = client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await? + .body + .collect() + .await? + .into_bytes(); + assert_eq!(current.as_ref(), payload, "a valid retry must publish the exact uploaded bytes"); + client.delete_object().bucket(bucket).key(key).send().await?; + env.delete_test_bucket(bucket).await?; + env.stop_server(); + Ok(()) +} + async fn upload_one_part_mpu( client: &aws_sdk_s3::Client, bucket: &str, diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 48bcff2a3..c16cee586 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -5238,6 +5238,8 @@ impl SetDisks { transaction_parts.insert(part_number); } } + // UploadPart creates this directory; an empty upload has no transactions to recover. + Err(DiskError::FileNotFound) => errs.push(None), Err(err) => errs.push(Some(err)), } } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 02869de27..2d60c8e2b 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -7689,18 +7689,25 @@ fn get_complete_multipart_md5(parts: &[CompletePart]) -> String { format!("{}-{}", etag_hex, parts.len()) } -fn completed_multipart_object_part(part_num: usize, ext_part: &ObjectPartInfo) -> ObjectPartInfo { - ObjectPartInfo { - etag: ext_part.etag.clone(), - number: part_num, - size: ext_part.size, - mod_time: ext_part.mod_time, - actual_size: ext_part.actual_size, - index: ext_part.index.clone(), - checksums: ext_part.checksums.clone(), - integrity: ext_part.integrity.clone(), - ..Default::default() +fn prepare_complete_multipart_parts(fi: &mut FileInfo, parts: Vec) -> Vec { + // Preserve legacy cleanup ordering, including replacement-only manifests. + let capacity = fi.parts.len() + parts.len(); + let mut staged_part_numbers = Vec::with_capacity(capacity); + staged_part_numbers.extend(fi.parts.iter().map(|part| part.number)); + let mut seen = HashSet::with_capacity(capacity); + seen.extend(staged_part_numbers.iter().copied()); + let mut inserted = false; + for part in &parts { + if seen.insert(part.number) { + staged_part_numbers.push(part.number); + inserted = true; + } } + if inserted { + staged_part_numbers.sort_unstable(); + } + fi.parts = parts; + staged_part_numbers } fn complete_part_checksum(part: &CompletePart, checksum_type: rustfs_rio::ChecksumType) -> Option> { @@ -9336,30 +9343,61 @@ mod tests { } #[test] - fn test_completed_multipart_object_part_preserves_checksums() { + fn prepare_complete_multipart_parts_preserves_metadata_and_cleanup_order() { let checksums = HashMap::from([ - (rustfs_rio::ChecksumType::CRC32.to_string(), "crc32-value".to_string()), - (rustfs_rio::ChecksumType::CRC32C.to_string(), "crc32c-value".to_string()), + (rustfs_rio::ChecksumType::CRC32.to_string(), "crc32-value".to_owned()), + (rustfs_rio::ChecksumType::CRC32C.to_string(), "crc32c-value".to_owned()), ]); - let ext_part = ObjectPartInfo { + let part = ObjectPartInfo { number: 7, - etag: "etag-7".to_string(), + etag: "etag-7".to_owned(), size: 123, actual_size: 456, mod_time: Some(OffsetDateTime::UNIX_EPOCH), index: Some(Bytes::from_static(&[1, 2, 3])), - checksums: Some(checksums.clone()), + checksums: Some(checksums), + integrity: Some( + rustfs_filemeta::shard_integrity::PartIntegrity::new( + rustfs_filemeta::shard_integrity::IntegrityLayout::new(2, 2, 1024 * 1024, false).expect("valid layout"), + 7, + ) + .expect("valid integrity"), + ), ..Default::default() }; - - let completed = completed_multipart_object_part(7, &ext_part); - - assert_eq!(completed.number, 7); - assert_eq!(completed.etag, ext_part.etag); - assert_eq!(completed.size, ext_part.size); - assert_eq!(completed.actual_size, ext_part.actual_size); - assert_eq!(completed.index, ext_part.index); - assert_eq!(completed.checksums, Some(checksums)); + let incoming = vec![part.clone()]; + let pointer = incoming.as_ptr(); + let mut fi = FileInfo { + parts: vec![ + ObjectPartInfo { + number: 9, + ..Default::default() + }, + ObjectPartInfo { + number: 7, + etag: "stale".to_owned(), + ..Default::default() + }, + ], + ..Default::default() + }; + assert_eq!(prepare_complete_multipart_parts(&mut fi, incoming), vec![9, 7]); + assert_eq!(fi.parts, vec![part.clone()]); + assert_eq!(fi.parts.as_ptr(), pointer, "move the existing allocation"); + let mut fi = FileInfo { + parts: vec![ + ObjectPartInfo { + number: 9, + ..Default::default() + }, + ObjectPartInfo { + number: 1, + ..Default::default() + }, + ], + ..Default::default() + }; + assert_eq!(prepare_complete_multipart_parts(&mut fi, vec![part]), vec![1, 7, 9]); } #[test] diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index afd8e25e2..5905349cb 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -40,12 +40,13 @@ use super::super::{ SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SetDisks, SmallWritePath, StorageError, Uuid, WriteLayout, check_object_lock_for_deletion_with_state, classify_multipart_part_write_path, coding, - complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, completed_multipart_object_part, - contains_key_str, create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str, + complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, contains_key_str, + create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str, is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size, log_multipart_write_quorum_failure, - parts_after_marker, path_join_buf, record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs, - remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, - strip_internal_multipart_metadata, to_object_err, warn, + parts_after_marker, path_join_buf, prepare_complete_multipart_parts, record_compression_total_memory, + reduce_read_quorum_errs, reduce_write_quorum_errs, remove_header_map, resolve_write_layout, + restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, strip_internal_multipart_metadata, + to_object_err, warn, }; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; #[cfg(test)] @@ -2961,29 +2962,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ); return Err(Error::InvalidPart(uploaded_parts[i].part_num, bucket.to_owned(), object.to_owned())); } - - fi.add_object_part( - part.number, - part.etag.clone(), - part.size, - part.mod_time, - part.actual_size, - part.index.clone(), - part.checksums.clone(), - ); - let inserted = fi - .parts - .iter_mut() - .find(|entry| entry.number == part.number) - .ok_or(Error::FileCorrupt)?; - inserted.integrity.clone_from(&part.integrity); } let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata_by_index(&disks, &files_metas, &fi); - let curr_fi = fi.clone(); - - fi.parts = Vec::with_capacity(uploaded_parts.len()); + let staged_part_numbers = prepare_complete_multipart_parts(&mut fi, object_parts); let quota_context = reservation::begin(&self.ctx, bucket, object, opts, self.pool_index, self.set_index).await?; let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some(); @@ -3020,19 +3003,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations 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, HashSet}; - let part_lookup: HashMap = curr_fi.parts.iter().map(|part| (part.number, part)).collect(); - - for (i, p) in uploaded_parts.iter().enumerate() { - let Some(ext_part) = part_lookup.get(&p.part_num) else { - error!( - "complete_multipart_upload part not found: part_id={}, bucket={}, object={}", - p.part_num, bucket, object - ); - return Err(Error::InvalidPart(p.part_num, "".to_owned(), p.etag.clone().unwrap_or_default())); - }; + for (i, (p, ext_part)) in uploaded_parts.iter().zip(fi.parts.iter_mut()).enumerate() { debug!( target:"rustfs_ecstore::set_disk", event = EVENT_SET_DISK_MULTIPART, @@ -3045,9 +3016,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { "Set disk multipart part validated" ); - // Normalize ETags by removing quotes before comparison (PR #592 compatibility) - let client_etag = p.etag.as_ref().map(|e| rustfs_utils::path::trim_etag(e)); - let stored_etag = Some(rustfs_utils::path::trim_etag(&ext_part.etag)); + let client_etag = p.etag.as_deref().map(|etag| etag.trim_matches('"')); + let stored_etag = Some(ext_part.etag.trim_matches('"')); if client_etag != stored_etag { error!( "complete_multipart_upload etag err client={:?}, stored={:?}, part_id={}, bucket={}, object={}", @@ -3153,9 +3123,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { object_actual_size = object_actual_size .checked_add(normalized_actual_size) .ok_or(Error::PartMissingOrCorrupt)?; - let mut completed_part = completed_multipart_object_part(p.part_num, ext_part); - completed_part.actual_size = normalized_actual_size; - fi.parts.push(completed_part); + ext_part.actual_size = normalized_actual_size; } if !transformed_object && data_movement_actual_size.is_some_and(|actual_size| actual_size < object_actual_size) { return Err(Error::PartMissingOrCorrupt); @@ -3376,21 +3344,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } let completed_part_numbers = fi.parts.iter().map(|part| part.number).collect::>(); - let upload_data_dir = curr_fi.data_dir.unwrap_or(Uuid::nil()).to_string(); - let mut parts = Vec::with_capacity(curr_fi.parts.len()); + let upload_data_dir = fi.data_dir.unwrap_or(Uuid::nil()).to_string(); + let mut parts = Vec::with_capacity(staged_part_numbers.len()); - for p in curr_fi.parts.iter() { + for part_number in staged_part_numbers { parts.push(path_join_buf(&[ &upload_id_path, upload_data_dir.as_str(), - format!("part.{}.meta", p.number).as_str(), + format!("part.{part_number}.meta").as_str(), ])); - if !completed_part_numbers.contains(&p.number) { + if !completed_part_numbers.contains(&part_number) { parts.push(path_join_buf(&[ &upload_id_path, upload_data_dir.as_str(), - format!("part.{}", p.number).as_str(), + format!("part.{part_number}").as_str(), ])); } } @@ -6223,6 +6191,103 @@ mod tests { assert_quorum_minus_one_retry_preserves_completable_part(6, 2, &[3, 4, 5]).await; } + #[tokio::test] + async fn multipart_transaction_recovery_accepts_missing_directories_but_requires_read_quorum() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let part_path = "empty-upload/data/"; + set_disks + .recover_part_transactions(part_path, 2, 3) + .await + .expect("an upload without parts has no transactions to recover"); + + { + let mut disks = set_disks.disks.write().await; + disks[2] = None; + disks[3] = None; + } + set_disks + .recover_part_transactions(part_path, 2, 3) + .await + .expect("missing directories at the exact read quorum prove there are no transactions"); + + set_disks.disks.write().await[1] = None; + assert_eq!( + set_disks.recover_part_transactions(part_path, 2, 3).await, + Err(DiskError::ErasureReadQuorum), + "one missing directory must not hide three offline disks" + ); + + disk_stores[1] + .write_all(RUSTFS_META_MULTIPART_BUCKET, "empty-upload/data", Bytes::from_static(b"not a directory")) + .await + .expect("place a file where the part directory should be"); + set_disks.disks.write().await[1] = Some(disk_stores[1].clone()); + assert_eq!( + set_disks.recover_part_transactions(part_path, 2, 3).await, + Err(DiskError::ErasureReadQuorum), + "a corrupt directory path must not count as an empty-directory observation" + ); + } + + #[tokio::test] + async fn complete_multipart_owned_parts_preserve_legacy_cleanup_and_normalize_size() { + use tokio::io::AsyncReadExt as _; + + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-owned-parts"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("create upload"); + let payload = b"current sparse part"; + let selected = put_test_part(&set_disks, bucket, object, &upload.upload_id, 7, payload, 0).await; + put_test_part(&set_disks, bucket, object, &upload.upload_id, 2, b"unselected", 10).await; + let (fi, mut metadata) = set_disks + .check_upload_id_exists(bucket, object, &upload.upload_id, true) + .await + .expect("read upload metadata"); + let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id); + let part_path = format!("{upload_path}/{}/part.7.meta", fi.data_dir.expect("upload data directory")); + let mut expected = disk_stores[0] + .read_parts(RUSTFS_META_MULTIPART_BUCKET, &[part_path]) + .await + .expect("read selected part metadata") + .remove(0); + expected.actual_size = i64::try_from(payload.len()).expect("payload size"); + for meta in &mut metadata { + meta.parts = vec![ + ObjectPartInfo { + number: 2, + ..Default::default() + }, + ObjectPartInfo { + number: 7, + etag: "stale".to_owned(), + ..Default::default() + }, + ]; + } + let disks = set_disks.get_disks_internal().await; + SetDisks::write_unique_file_info(&disks, "", RUSTFS_META_MULTIPART_BUCKET, &upload_path, &metadata, 3) + .await + .expect("persist legacy manifest"); + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![selected], &ObjectOptions::default()) + .await + .expect("complete with current part metadata instead of stale manifest entries"); + assert_eq!(completed.parts.as_slice(), &[expected]); + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("read completed object"); + let mut body = Vec::new(); + reader.stream.read_to_end(&mut body).await.expect("read all bytes"); + assert_eq!(body, payload); + } + #[tokio::test] async fn complete_multipart_upload_recovers_interrupted_part_retry() { use tokio::io::AsyncReadExt as _;