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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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<SetDisks>,
             bucket: &str,
             object: &str,
             content: &[u8],
+            create_opts: &ObjectOptions,
         ) -> (String, Vec<CompletePart>) {
             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<SetDisks>,
+            bucket: &str,
+            object: &str,
+            content: &[u8],
+        ) -> (String, Vec<CompletePart>) {
+            stage_upload_with_create_opts(set_disks, bucket, object, content, &ObjectOptions::default()).await
+        }
+
+        async fn rewrite_staged_upload_version_id(
+            set_disks: &Arc<SetDisks>,
+            bucket: &str,
+            object: &str,
+            upload_id: &str,
+            version_id: Option<Uuid>,
+        ) {
+            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<SetDisks>,
             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<SetDisks>, bucket: &str, object: &str) -> (Vec<u8>, Option<String>) {

crates/ecstore/src/set_disk/ops/object.rs

@@ -1698,10 +1698,16 @@ fn transition_transaction_not_after_unix_nanos() -> Result<i64> {
     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<TransitionSourceIdentity> {
-    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() {

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

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