From 75381d4ffe90a43cfb6bb2abd67cef700e2a3934 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 21 Jul 2026 13:18:45 +0800 Subject: [PATCH] fix(ilm): accept null version transition sources (#5069) * fix(ilm): accept null version transition sources Treat nil/null source version IDs as null-version transition sources when building transition transaction source identity, while preserving fail-closed validation for truly missing IDs on versioned sources. Co-Authored-By: heihutu * fix(ilm): preserve multipart source versions Propagate bucket versioning into CompleteMultipartUpload, assign a concrete version ID for versioned multipart completions, and classify unversioned FileInfo sources as null-version transition sources. Co-Authored-By: heihutu * fix(ilm): normalize multipart completion versions Treat nil staged version IDs as missing for versioned multipart completion and clear staged version IDs when completion publishes a null-version object under suspended or unversioned semantics. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/ecstore/src/set_disk/ops/multipart.rs | 166 +++++++++++++++++- crates/ecstore/src/set_disk/ops/object.rs | 121 ++++++++++++- .../src/app/lifecycle_transition_api_test.rs | 6 +- rustfs/src/app/multipart_usecase.rs | 13 +- 4 files changed, 288 insertions(+), 18 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 08b19bf11..0a84df2be 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1043,6 +1043,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { if expected_restore_operation_id.is_some() { rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_RESTORE_OPERATION_ID); } + if opts.versioned { + fi.version_id = Some( + fi.version_id + .filter(|version_id| !version_id.is_nil()) + .unwrap_or_else(Uuid::new_v4), + ); + } else { + fi.version_id = None; + } let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); let write_quorum = fi.write_quorum(self.default_write_quorum()); @@ -1380,6 +1389,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { meta.parts.clone_from(&fi.parts); meta.metadata = fi.metadata.clone(); meta.versioned = opts.versioned || opts.version_suspended; + meta.version_id = fi.version_id; meta.checksum = fi.checksum.clone(); } } @@ -1629,7 +1639,7 @@ mod tests { #[serial(metadata_cache_invalidation_probe)] async fn complete_multipart_generation_retires_cached_snapshot() { use crate::storage_api_contracts::multipart::MultipartOperations as _; - use crate::storage_api_contracts::object::ObjectIO as _; + use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "multipart-metadata-generation-bucket"; @@ -2358,14 +2368,15 @@ mod tests { /// Stage a single-part multipart upload without completing it. A single /// part is the last part, so the 5 MiB minimum-part-size gate does not /// apply. Returns the upload id and the `CompletePart` list. - async fn stage_upload( + async fn stage_upload_with_create_opts( set_disks: &Arc, bucket: &str, object: &str, content: &[u8], + create_opts: &ObjectOptions, ) -> (String, Vec) { let upload = set_disks - .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .new_multipart_upload(bucket, object, create_opts) .await .expect("multipart upload should be created"); let mut reader = PutObjReader::new( @@ -2393,6 +2404,43 @@ mod tests { ) } + async fn stage_upload( + set_disks: &Arc, + bucket: &str, + object: &str, + content: &[u8], + ) -> (String, Vec) { + stage_upload_with_create_opts(set_disks, bucket, object, content, &ObjectOptions::default()).await + } + + async fn rewrite_staged_upload_version_id( + set_disks: &Arc, + bucket: &str, + object: &str, + upload_id: &str, + version_id: Option, + ) { + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, upload_id); + let disks = set_disks.disks.read().await.clone(); + let (fi, mut files_metas) = set_disks + .check_upload_id_exists(bucket, object, upload_id, true) + .await + .expect("staged upload metadata should be readable"); + for meta in files_metas.iter_mut() { + meta.version_id = version_id; + } + SetDisks::write_unique_file_info( + &disks, + bucket, + RUSTFS_META_MULTIPART_BUCKET, + &upload_id_path, + &files_metas, + fi.write_quorum(set_disks.default_write_quorum()), + ) + .await + .expect("staged upload metadata should be rewritten"); + } + async fn complete( set_disks: &Arc, bucket: &str, @@ -2406,6 +2454,118 @@ mod tests { .await } + #[tokio::test] + #[serial] + async fn complete_multipart_upload_assigns_completion_version_id() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-versioned-complete-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + let (upload_id, parts) = stage_upload(&set_disks, bucket, object, b"versioned multipart body").await; + let complete_opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &complete_opts) + .await + .expect("versioned multipart completion should succeed"); + let version_id = completed.version_id.expect("versioned completion must return a version id"); + + assert!(!version_id.is_nil(), "versioned completion must not use a null version id"); + + let lookup_opts = ObjectOptions { + versioned: true, + version_id: Some(version_id.to_string()), + ..Default::default() + }; + let current = set_disks + .get_object_info(bucket, object, &lookup_opts) + .await + .expect("completed version should be addressable by exact version id"); + + assert_eq!(current.version_id, Some(version_id)); + } + + #[tokio::test] + #[serial] + async fn complete_multipart_upload_replaces_staged_nil_version_id() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-nil-versioned-complete-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + let (upload_id, parts) = stage_upload(&set_disks, bucket, object, b"versioned multipart body").await; + rewrite_staged_upload_version_id(&set_disks, bucket, object, &upload_id, Some(Uuid::nil())).await; + let complete_opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &complete_opts) + .await + .expect("versioned multipart completion should succeed"); + let version_id = completed.version_id.expect("versioned completion must return a version id"); + + assert!(!version_id.is_nil(), "versioned completion must replace a staged null version id"); + + let lookup_opts = ObjectOptions { + versioned: true, + version_id: Some(version_id.to_string()), + ..Default::default() + }; + let current = set_disks + .get_object_info(bucket, object, &lookup_opts) + .await + .expect("completed version should be addressable by exact version id"); + + assert_eq!(current.version_id, Some(version_id)); + } + + #[tokio::test] + #[serial] + async fn complete_multipart_upload_suspended_clears_staged_version_id() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-suspended-complete-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let create_opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, b"suspended multipart body", &create_opts).await; + let (staged_fi, _) = set_disks + .check_upload_id_exists(bucket, object, &upload_id, true) + .await + .expect("staged upload metadata should be readable"); + assert!( + staged_fi.version_id.is_some_and(|version_id| !version_id.is_nil()), + "enabled multipart create should stage a concrete version id" + ); + let complete_opts = ObjectOptions { + version_suspended: true, + ..Default::default() + }; + + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &complete_opts) + .await + .expect("suspended multipart completion should succeed"); + + assert!( + completed.version_id.is_some_and(|version_id| version_id.is_nil()), + "suspended completion must publish a null-version object internally" + ); + } + /// Read the whole committed object back; returns `(body, etag)`. The full /// body read also proves the served length matches (never a torn/short body). async fn read_object(set_disks: &Arc, bucket: &str, object: &str) -> (Vec, Option) { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index d23ee4d7d..84391a611 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1698,10 +1698,16 @@ fn transition_transaction_not_after_unix_nanos() -> Result { i64::try_from(not_after).map_err(|_| Error::other("transition transaction deadline timestamp overflow")) } -fn transition_source_version_mode(opts: &ObjectOptions) -> TransitionSourceVersionMode { - if opts.versioned { +fn transition_source_version_mode(opts: &ObjectOptions, fi: &FileInfo) -> TransitionSourceVersionMode { + let requested_null_version = opts + .version_id + .as_deref() + .and_then(|version_id| Uuid::parse_str(version_id).ok()) + .is_some_and(|version_id| version_id.is_nil()); + let source_version_id = fi.version_id.filter(|version_id| !version_id.is_nil()); + if opts.versioned && (source_version_id.is_some() || (fi.versioned && !requested_null_version)) { TransitionSourceVersionMode::Versioned - } else if opts.version_suspended { + } else if opts.version_suspended || opts.versioned { TransitionSourceVersionMode::VersionSuspended } else { TransitionSourceVersionMode::Unversioned @@ -1715,16 +1721,18 @@ fn transition_source_identity( opts: &ObjectOptions, stored_etag: &str, ) -> Result { - let version_mode = transition_source_version_mode(opts); + let version_mode = transition_source_version_mode(opts, fi); let mod_time = fi .mod_time .ok_or_else(|| Error::other("transition source identity requires mod_time"))? .unix_timestamp_nanos(); let mod_time_unix_nanos = i64::try_from(mod_time).map_err(|_| Error::other("transition source mod_time timestamp overflow"))?; - let version_id = (version_mode == TransitionSourceVersionMode::Versioned) - .then_some(fi.version_id) - .flatten(); + let version_id = if version_mode == TransitionSourceVersionMode::Versioned { + fi.version_id.filter(|version_id| !version_id.is_nil()) + } else { + None + }; let data_dir = fi .data_dir .ok_or_else(|| Error::other("transition source identity requires data_dir"))?; @@ -6166,6 +6174,105 @@ mod transition_source_identity_matrix_tests { use crate::services::tier::test_util::register_mock_tier; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; + #[test] + fn transition_source_identity_treats_nil_version_as_null_source() { + let fi = FileInfo { + version_id: Some(Uuid::nil()), + data_dir: Some(Uuid::new_v4()), + mod_time: Some(OffsetDateTime::now_utc()), + size: 1, + ..Default::default() + }; + let opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") + .expect("nil source version should build a null-version identity"); + + assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); + assert_eq!(source.version_id, None); + source.validate().expect("null-version source identity should validate"); + } + + #[test] + fn transition_source_identity_treats_requested_nil_version_as_null_source() { + let fi = FileInfo { + version_id: None, + data_dir: Some(Uuid::new_v4()), + mod_time: Some(OffsetDateTime::now_utc()), + size: 1, + ..Default::default() + }; + let opts = ObjectOptions { + version_id: Some(Uuid::nil().to_string()), + versioned: true, + ..Default::default() + }; + + let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") + .expect("requested nil source version should build a null-version identity"); + + assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); + assert_eq!(source.version_id, None); + source + .validate() + .expect("requested null-version source identity should validate"); + } + + #[test] + fn transition_source_identity_treats_unversioned_fileinfo_as_null_source() { + let fi = FileInfo { + version_id: None, + versioned: false, + data_dir: Some(Uuid::new_v4()), + mod_time: Some(OffsetDateTime::now_utc()), + size: 1, + ..Default::default() + }; + let opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") + .expect("unversioned fileinfo should build a null-version identity"); + + assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); + assert_eq!(source.version_id, None); + source + .validate() + .expect("unversioned fileinfo null-version source identity should validate"); + } + + #[test] + fn transition_source_identity_still_rejects_missing_versioned_source_id() { + let fi = FileInfo { + version_id: None, + versioned: true, + data_dir: Some(Uuid::new_v4()), + mod_time: Some(OffsetDateTime::now_utc()), + size: 1, + ..Default::default() + }; + let opts = ObjectOptions { + versioned: true, + ..Default::default() + }; + + let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") + .expect("source identity construction should preserve missing version evidence"); + + assert_eq!(source.version_mode, TransitionSourceVersionMode::Versioned); + assert!(matches!( + source.validate(), + Err(crate::bucket::lifecycle::transition_transaction::TransitionTransactionError::Corrupt( + "versioned source is missing version_id" + )) + )); + } + #[tokio::test] #[serial_test::serial] async fn transition_source_identity_field_matrix_rejects_single_field_drift() { diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 3cad45679..1a4dc96e7 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -836,9 +836,13 @@ async fn complete_multipart_upload_transitions_immediately_via_usecase() { .build() .unwrap(); - Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST))) + let complete_output = Box::pin(usecase.execute_complete_multipart_upload(build_request(complete_input, Method::POST))) .await .expect("Failed to complete multipart upload through usecase"); + assert!( + complete_output.output.version_id.is_some(), + "versioned CompleteMultipartUpload should return the created version ID" + ); let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) .await diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index a31580fd1..d740acf65 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -420,6 +420,10 @@ impl DefaultMultipartUsecase { let Some(multipart_upload) = multipart_upload else { return Err(s3_error!(InvalidPart)) }; let mut opts = get_complete_multipart_upload_opts(&req.headers).map_err(ApiError::from)?; + let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; + let version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await; + opts.versioned = versioned; + opts.version_suspended = version_suspended; let capacity_scope_token = Uuid::new_v4(); opts.capacity_scope_token = Some(capacity_scope_token); @@ -512,8 +516,7 @@ impl DefaultMultipartUsecase { )); } // Update quota tracking after successful multipart upload - let mpu_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; - if mpu_versioned { + if versioned { record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64) .await; } else { @@ -529,11 +532,7 @@ impl DefaultMultipartUsecase { enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; let raw_mpu_version = obj_info.version_id.map(|v| v.to_string()); - let mpu_version = if BucketVersioningSys::prefix_enabled(&bucket, &key).await { - raw_mpu_version.clone() - } else { - None - }; + let mpu_version = if versioned { raw_mpu_version.clone() } else { None }; let mpu_version_for_event = mpu_version.clone(); // checksum: stored (decrypted) values take precedence over the request input; // additional algorithms (XXHash3/64/128, SHA-512, MD5), which have no typed