From 67a19021b55f879aaf38c616c8662eba19b00e76 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 14 Aug 2026 21:00:01 +0800 Subject: [PATCH] fix(ecstore): allow migrated unknown part sizes (#6112) --- crates/ecstore/src/set_disk/ops/multipart.rs | 238 +++++++++++++++---- 1 file changed, 187 insertions(+), 51 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 48c6d3076..9684cc1f6 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1906,6 +1906,23 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { return Err(Error::PartMissingOrCorrupt); } let transformed_object = fi.is_compressed() || should_persist_encryption_original_size(&fi.metadata); + let data_movement_actual_size = if opts.data_movement { + rustfs_utils::http::get_consistent_str(&opts.user_defined, SUFFIX_ACTUAL_SIZE) + .map(|value| { + value + .parse::() + .ok() + .filter(|value| *value >= 0) + .ok_or(Error::PartMissingOrCorrupt) + }) + .transpose()? + } else { + None + }; + let data_movement_actual_size_u64 = data_movement_actual_size + .map(u64::try_from) + .transpose() + .map_err(|_| Error::PartMissingOrCorrupt)?; let mut object_size: usize = 0; let mut object_actual_size: i64 = 0; @@ -2034,7 +2051,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } object_size = object_size.checked_add(ext_part.size).ok_or(Error::PartMissingOrCorrupt)?; - if ext_part.actual_size < 0 && (quota_context.is_enforced() || (!opts.replication_request && !opts.data_movement)) { + let unknown_actual_size_allowed = opts.data_movement && transformed_object && data_movement_actual_size.is_some() + || opts.replication_request && !quota_context.is_enforced(); + if ext_part.actual_size < 0 && !unknown_actual_size_allowed { return Err(Error::PartMissingOrCorrupt); } let normalized_actual_size = if ext_part.actual_size >= 0 && !transformed_object { @@ -2051,7 +2070,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { completed_part.actual_size = normalized_actual_size; fi.parts.push(completed_part); } - + if !transformed_object && data_movement_actual_size.is_some_and(|actual_size| actual_size < object_actual_size) { + return Err(Error::PartMissingOrCorrupt); + } if let Some(wtcs) = opts.want_checksum.as_ref() { if checksum_type.full_object_requested() { if wtcs.encoded != checksum.encoded { @@ -2097,7 +2118,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } else { None }; - let quota_new_size = match replication_actual_size { + let data_movement_quota_size = data_movement_actual_size_u64.filter(|_| quota_context.is_enforced()); + let quota_new_size = match data_movement_quota_size.or(replication_actual_size) { Some(size) => size.max(u64::try_from(object_size).map_err(|_| Error::PartMissingOrCorrupt)?), None if quota_context.is_enforced() => u64::try_from(object_actual_size) .map_err(|_| Error::PartMissingOrCorrupt)? @@ -2155,20 +2177,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ); } - let data_movement_actual_size = if opts.data_movement { - rustfs_utils::http::get_consistent_str(&opts.user_defined, SUFFIX_ACTUAL_SIZE) - .map(|value| { - value - .parse::() - .ok() - .filter(|value| *value >= 0) - .ok_or_else(|| Error::other("data movement actual size metadata is invalid")) - }) - .transpose()? - } else { - None - }; - if let Some(actual_size) = data_movement_actual_size { insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.to_string()); if persist_encryption_original_size { @@ -3109,12 +3117,34 @@ mod tests { content: &[u8], actual_size: i64, ) -> CompletePart { + put_test_part_with_opts( + set_disks, + bucket, + object, + upload_id, + part_number, + (content, actual_size), + &ObjectOptions::default(), + ) + .await + } + + async fn put_test_part_with_opts( + set_disks: &Arc, + bucket: &str, + object: &str, + upload_id: &str, + part_number: usize, + part: (&[u8], i64), + opts: &ObjectOptions, + ) -> CompletePart { + let (content, actual_size) = part; let mut reader = PutObjReader::new( HashReader::from_stream(Cursor::new(content.to_vec()), content.len() as i64, actual_size, None, None, false) .expect("hash reader should be constructed"), ); let part = set_disks - .put_object_part(bucket, object, upload_id, part_number, &mut reader, &ObjectOptions::default()) + .put_object_part(bucket, object, upload_id, part_number, &mut reader, opts) .await .expect("uploading the part should succeed"); CompletePart { @@ -3850,7 +3880,6 @@ mod tests { async fn data_movement_complete_accepts_unknown_compressed_part_actual_size() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "data-movement-unknown-actual-size-bucket"; - let object = "object"; make_bucket_on_all(&disk_stores, bucket).await; let mut metadata = HashMap::new(); @@ -3864,6 +3893,63 @@ mod tests { rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD, "source-generation".to_string(), ); + + for (object, quota_limit) in [("without-quota", None), ("with-quota", Some(u64::MAX))] { + let create_opts = ObjectOptions { + data_movement: true, + user_defined: metadata.clone(), + ..Default::default() + }; + let upload = set_disks + .new_multipart_upload(bucket, object, &create_opts) + .await + .expect("data movement upload should be created"); + + let mut completed_parts = Vec::new(); + for (number, actual_size) in [(1, -1), (2, 1)] { + let mut reader = PutObjReader::new( + HashReader::from_stream(Cursor::new(vec![number as u8]), 1, actual_size, None, None, false) + .expect("part reader should be constructed"), + ); + let part = set_disks + .put_object_part(bucket, object, &upload.upload_id, number, &mut reader, &create_opts) + .await + .expect("data movement part should be written"); + completed_parts.push(CompletePart { + part_num: number, + etag: part.etag, + ..Default::default() + }); + } + + let missing_size_err = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, completed_parts.clone(), &create_opts) + .await + .expect_err("data movement completion must require an authoritative total size"); + assert!(matches!(missing_size_err, StorageError::PartMissingOrCorrupt)); + + let mut complete_opts = create_opts.clone(); + rustfs_utils::http::insert_str( + &mut complete_opts.user_defined, + rustfs_utils::http::SUFFIX_ACTUAL_SIZE, + "2".to_string(), + ); + if let Some(quota_limit) = quota_limit { + assert!(complete_opts.set_quota_admission(0, quota_limit)); + } + let completion = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, completed_parts, &complete_opts) + .await; + + let completed = completion.expect("data movement completion should accept the persisted unknown-size sentinel"); + + assert_eq!(completed.parts[0].actual_size, -1); + assert_eq!(completed.get_actual_size().expect("completed object actual size"), 2); + } + + let object = "all-unknown-with-quota"; let create_opts = ObjectOptions { data_movement: true, user_defined: metadata.clone(), @@ -3873,43 +3959,93 @@ mod tests { .new_multipart_upload(bucket, object, &create_opts) .await .expect("data movement upload should be created"); - - let mut completed_parts = Vec::new(); - for (number, actual_size) in [(1, -1), (2, 1)] { - let mut reader = PutObjReader::new( - HashReader::from_stream(Cursor::new(vec![number as u8]), 1, actual_size, None, None, false) - .expect("part reader should be constructed"), - ); - let part = set_disks - .put_object_part(bucket, object, &upload.upload_id, number, &mut reader, &create_opts) - .await - .expect("data movement part should be written"); - completed_parts.push(CompletePart { - part_num: number, - etag: part.etag, - ..Default::default() - }); - } - - rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "2".to_string()); + let part = put_test_part_with_opts(&set_disks, bucket, object, &upload.upload_id, 1, (&[0x40], -1), &create_opts).await; + let mut complete_opts = create_opts; + rustfs_utils::http::insert_str(&mut complete_opts.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "2".to_string()); + assert!(complete_opts.set_quota_admission(0, u64::MAX)); let completed = set_disks .clone() - .complete_multipart_upload( - bucket, - object, - &upload.upload_id, - completed_parts, - &ObjectOptions { - data_movement: true, - user_defined: metadata, - ..Default::default() - }, - ) + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts) .await - .expect("data movement completion should accept the persisted unknown-size sentinel"); - + .expect("quota must accept an all-unknown data movement upload with an authoritative total"); assert_eq!(completed.parts[0].actual_size, -1); assert_eq!(completed.get_actual_size().expect("completed object actual size"), 2); + + let object = "legacy-zero-fallback"; + let create_opts = ObjectOptions { + data_movement: true, + user_defined: metadata.clone(), + ..Default::default() + }; + let upload = set_disks + .new_multipart_upload(bucket, object, &create_opts) + .await + .expect("data movement upload should be created"); + let part = + put_test_part_with_opts(&set_disks, bucket, object, &upload.upload_id, 1, (&[0x41; 128], 128), &create_opts).await; + let mut complete_opts = create_opts; + rustfs_utils::http::insert_str( + &mut complete_opts.user_defined, + rustfs_utils::http::SUFFIX_ACTUAL_SIZE, + "100".to_string(), + ); + assert!(complete_opts.set_quota_admission(0, u64::MAX)); + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts) + .await + .expect("legacy zero-to-physical part fallback must remain migratable"); + assert_eq!(completed.parts[0].actual_size, 128); + assert_eq!(completed.get_actual_size().expect("completed object actual size"), 100); + + let object = "inconsistent-untransformed-total"; + let mut untransformed_metadata = metadata.clone(); + rustfs_utils::http::remove_str(&mut untransformed_metadata, rustfs_utils::http::SUFFIX_COMPRESSION); + let create_opts = ObjectOptions { + data_movement: true, + user_defined: untransformed_metadata, + ..Default::default() + }; + let upload = set_disks + .new_multipart_upload(bucket, object, &create_opts) + .await + .expect("data movement upload should be created"); + let part = put_test_part_with_opts(&set_disks, bucket, object, &upload.upload_id, 1, (&[0x42; 2], 2), &create_opts).await; + let mut complete_opts = create_opts; + rustfs_utils::http::insert_str(&mut complete_opts.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "1".to_string()); + let err = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts) + .await + .expect_err("authoritative total below known logical part sizes must fail closed"); + assert!(matches!(err, StorageError::PartMissingOrCorrupt)); + + for (object, declared_size) in [("invalid-total", "invalid"), ("negative-total", "-1")] { + let create_opts = ObjectOptions { + data_movement: true, + user_defined: metadata.clone(), + ..Default::default() + }; + let upload = set_disks + .new_multipart_upload(bucket, object, &create_opts) + .await + .expect("data movement upload should be created"); + let part = + put_test_part_with_opts(&set_disks, bucket, object, &upload.upload_id, 1, (&[0x41], 1), &create_opts).await; + let mut complete_opts = create_opts.clone(); + rustfs_utils::http::insert_str( + &mut complete_opts.user_defined, + rustfs_utils::http::SUFFIX_ACTUAL_SIZE, + declared_size.to_string(), + ); + + let err = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts) + .await + .expect_err("invalid authoritative total size must fail closed"); + assert!(matches!(err, StorageError::PartMissingOrCorrupt)); + } } async fn assert_complete_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {