fix(lifecycle): harden S3 lifecycle rule handling (#4115)

* fix(lifecycle): honor noncurrent version retention * fix(lifecycle): validate transition rules * fix(lifecycle): support immediate multipart abort * fix(lifecycle): harden select restore metadata * fix(lifecycle): expose stale multipart cleanup to app * fix(lifecycle): reduce expiry helper arguments
2026-07-27 00:38:16 +00:00 · 2026-06-30 19:06:34 +08:00
parent 7484a61fa3
commit f9340da74a
5 changed files with 683 additions and 100 deletions
@@ -64,8 +64,8 @@ use rustfs_filemeta::{
 };
 use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::strings_has_prefix_fold};
 use s3s::dto::{
-    BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, ReplicationConfiguration, RestoreRequest,
-    RestoreRequestType, RestoreStatus, Timestamp,
+    BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, ObjectLockConfiguration, ReplicationConfiguration,
+    RestoreRequest, RestoreRequestType, RestoreStatus, Timestamp,
 };
 use s3s::header::{X_AMZ_RESTORE, X_AMZ_SERVER_SIDE_ENCRYPTION};
 use sha2::{Digest, Sha256};
@@ -1952,7 +1952,7 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) {

    let object_opts = object_infos
        .iter()
-        .map(|object| object.to_lifecycle_opts())
+        .map(ObjectOpts::from_object_info)
        .collect::<Vec<ObjectOpts>>();
    let Ok(events) = Evaluator::new(Arc::new(lifecycle))
        .with_lock_retention(lock_config)
@@ -2055,20 +2055,129 @@ async fn apply_existing_object_expiry(api: Arc<ECStore>, object: &ObjectInfo, ev
    }
 }

+struct ExistingObjectExpiryContext<'a> {
+    api: Arc<ECStore>,
+    bucket: &'a str,
+    lc: Arc<BucketLifecycleConfiguration>,
+    lock_config: Option<Arc<ObjectLockConfiguration>>,
+    replication: Option<Arc<replication_sink::LifecycleReplicationConfig>>,
+    src: &'a LcEventSrc,
+    defer_date_expiry_once: bool,
+}
+
+async fn enqueue_expiry_for_existing_object_group(
+    context: &ExistingObjectExpiryContext<'_>,
+    object_infos: &[ObjectInfo],
+    date_expiry_deferred_once: &mut bool,
+) {
+    if object_infos.is_empty() {
+        return;
+    }
+
+    let object_opts = object_infos
+        .iter()
+        .map(ObjectOpts::from_object_info)
+        .collect::<Vec<ObjectOpts>>();
+    let events = match Evaluator::new(context.lc.clone())
+        .with_lock_retention(context.lock_config.clone())
+        .with_replication_config(context.replication.clone())
+        .eval(&object_opts)
+        .await
+    {
+        Ok(events) => events,
+        Err(err) => {
+            warn!(
+                bucket = context.bucket,
+                object = %object_infos[0].name,
+                error = %err,
+                "failed to evaluate lifecycle events for existing object versions"
+            );
+            return;
+        }
+    };
+
+    let mut to_delete_objs = Vec::new();
+    let mut noncurrent_event = None;
+
+    for (object, event) in object_infos.iter().zip(events.iter()) {
+        match event.action {
+            IlmAction::DeleteAction
+            | IlmAction::DeleteVersionAction
+            | IlmAction::DeleteRestoredAction
+            | IlmAction::DeleteRestoredVersionAction
+            | IlmAction::DeleteAllVersionsAction
+            | IlmAction::DelMarkerDeleteAllVersionsAction => {
+                let now = OffsetDateTime::now_utc();
+                if event.due.is_some_and(|due| due.unix_timestamp() <= now.unix_timestamp()) {
+                    if context.defer_date_expiry_once
+                        && !*date_expiry_deferred_once
+                        && lifecycle_rule_has_date_expiration(&context.lc, &event.rule_id)
+                    {
+                        tokio::time::sleep(StdDuration::from_secs(DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS as u64)).await;
+                        *date_expiry_deferred_once = true;
+                    }
+
+                    if event.action == IlmAction::DeleteVersionAction {
+                        to_delete_objs.push(ObjectToDelete {
+                            object_name: object.name.clone(),
+                            version_id: object.version_id,
+                            ..Default::default()
+                        });
+                        if noncurrent_event.is_none() {
+                            noncurrent_event = Some(event.clone());
+                        }
+                    } else {
+                        apply_existing_object_expiry(context.api.clone(), object, event, context.src).await;
+                    }
+                } else {
+                    apply_expiry_rule(event, context.src, object).await;
+                }
+            }
+            _ => {}
+        }
+    }
+
+    if !to_delete_objs.is_empty()
+        && let Some(event) = noncurrent_event
+    {
+        let expiry_state = runtime_sources::expiry_state_handle();
+        expiry_state
+            .write()
+            .await
+            .enqueue_by_newer_noncurrent(context.bucket, to_delete_objs, event, context.src)
+            .await;
+    }
+}
+
 pub async fn enqueue_expiry_for_existing_objects(api: Arc<ECStore>, bucket: &str) -> Result<(), Error> {
    let Ok((lc, _)) = metadata_boundary::get_lifecycle_config(bucket).await else {
        return Ok(());
    };
-    let lock_retention = metadata_boundary::get_object_lock_config(bucket)
+    let lc = Arc::new(lc);
+    let lock_config = metadata_boundary::get_object_lock_config(bucket)
        .await
        .ok()
-        .and_then(|(cfg, _)| cfg.rule.and_then(|rule| rule.default_retention));
-    let replication_config = metadata_boundary::get_replication_config(bucket).await.ok();
+        .map(|(cfg, _)| Arc::new(cfg));
+    let replication = match metadata_boundary::get_replication_config(bucket).await {
+        Ok((cfg, _)) if !cfg.rules.is_empty() => Some(Arc::new(replication_sink::new_replication_config(cfg))),
+        _ => None,
+    };
    let mut marker = None;
    let mut version_marker = None;
    let src = LcEventSrc::Scanner;
    let defer_date_expiry_once = should_defer_date_expiry_for_recent_config_update(&lc, OffsetDateTime::now_utc());
+    let expiry_context = ExistingObjectExpiryContext {
+        api: api.clone(),
+        bucket,
+        lc: lc.clone(),
+        lock_config: lock_config.clone(),
+        replication: replication.clone(),
+        src: &src,
+        defer_date_expiry_once,
+    };
    let mut date_expiry_deferred_once = false;
+    let mut pending_group = Vec::new();
+    let mut pending_object = None::<String>;

    loop {
        let page = api
@@ -2076,34 +2185,17 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc<ECStore>, bucket: &str
            .list_object_versions(bucket, "", marker.clone(), version_marker.clone(), None, 1000)
            .await?;

-        for object in &page.objects {
-            let event = eval_action_from_lifecycle(&lc, lock_retention.clone(), replication_config.clone(), object).await;
-            match event.action {
-                IlmAction::DeleteAction
-                | IlmAction::DeleteVersionAction
-                | IlmAction::DeleteRestoredAction
-                | IlmAction::DeleteRestoredVersionAction
-                | IlmAction::DeleteAllVersionsAction
-                | IlmAction::DelMarkerDeleteAllVersionsAction => {
-                    let now = OffsetDateTime::now_utc();
-                    if event.due.is_some_and(|due| due.unix_timestamp() <= now.unix_timestamp()) {
-                        if defer_date_expiry_once
-                            && !date_expiry_deferred_once
-                            && lifecycle_rule_has_date_expiration(&lc, &event.rule_id)
-                        {
-                            tokio::time::sleep(StdDuration::from_secs(DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS as u64)).await;
-                            date_expiry_deferred_once = true;
-                        }
-                        apply_existing_object_expiry(api.clone(), object, &event, &src).await;
-                    } else {
-                        apply_expiry_rule(&event, &src, object).await;
-                    }
-                }
-                _ => {}
+        for object in page.objects {
+            if pending_object.as_ref().is_some_and(|name| name != &object.name) {
+                enqueue_expiry_for_existing_object_group(&expiry_context, &pending_group, &mut date_expiry_deferred_once).await;
+                pending_group.clear();
            }
+            pending_object = Some(object.name.clone());
+            pending_group.push(object);
        }

        if !page.is_truncated {
+            enqueue_expiry_for_existing_object_group(&expiry_context, &pending_group, &mut date_expiry_deferred_once).await;
            return Ok(());
        }

@@ -2326,6 +2418,32 @@ pub async fn post_restore_opts(version_id: &str, bucket: &str, object: &str) ->
    })
 }

+fn select_restore_s3_location(rreq: &RestoreRequest) -> Result<Option<&s3s::dto::S3Location>, std::io::Error> {
+    if rreq
+        .type_
+        .as_ref()
+        .is_none_or(|type_| type_.as_str() != RestoreRequestType::SELECT)
+    {
+        return Ok(None);
+    }
+    let output_location = rreq
+        .output_location
+        .as_ref()
+        .ok_or_else(|| std::io::Error::other("OutputLocation required for SELECT requests"))?;
+    let s3 = output_location
+        .s3
+        .as_ref()
+        .ok_or_else(|| std::io::Error::other("OutputLocation.S3 required for SELECT requests"))?;
+    if let Some(user_metadata) = s3.user_metadata.as_ref() {
+        for metadata in user_metadata {
+            if metadata.name.as_deref().is_none_or(|name| name.is_empty()) {
+                return Err(std::io::Error::other("SELECT restore metadata name is required"));
+            }
+        }
+    }
+    Ok(Some(s3))
+}
+
 pub async fn put_restore_opts(
    bucket: &str,
    object: &str,
@@ -2344,38 +2462,30 @@ pub async fn put_restore_opts(
    if let Some(type_) = &rreq.type_
        && type_.as_str() == RestoreRequestType::SELECT
    {
-        for v in rreq
-            .output_location
-            .as_ref()
-            .unwrap()
-            .s3
-            .as_ref()
-            .unwrap()
-            .user_metadata
-            .as_ref()
-            .unwrap()
-        {
-            if !strings_has_prefix_fold(&v.name.clone().unwrap(), "x-amz-meta") {
-                meta.insert(
-                    format!("x-amz-meta-{}", v.name.as_ref().unwrap()),
-                    v.value.clone().unwrap_or_else(|| "".to_string()),
-                );
-                continue;
+        let Some(s3) = select_restore_s3_location(rreq)? else {
+            return Err(std::io::Error::other("OutputLocation.S3 required for SELECT requests"));
+        };
+        if let Some(user_metadata) = s3.user_metadata.as_ref() {
+            for metadata in user_metadata {
+                let name = metadata
+                    .name
+                    .as_deref()
+                    .ok_or_else(|| std::io::Error::other("SELECT restore metadata name is required"))?;
+                let value = metadata.value.clone().unwrap_or_default();
+                if strings_has_prefix_fold(name, "x-amz-meta") {
+                    meta.insert(name.to_string(), value);
+                } else {
+                    meta.insert(format!("x-amz-meta-{name}"), value);
+                }
            }
-            meta.insert(v.name.clone().unwrap(), v.value.clone().unwrap_or_else(|| "".to_string()));
        }
-        if let Some(output_location) = rreq.output_location.as_ref()
-            && let Some(s3) = &output_location.s3
-            && let Some(tags) = &s3.tagging
-        {
+        if let Some(tags) = &s3.tagging {
            meta.insert(
                AMZ_OBJECT_TAGGING.to_string(),
                serde_urlencoded::to_string(tags.tag_set.clone()).unwrap_or_else(|_| "".to_string()),
            );
        }
-        if let Some(output_location) = rreq.output_location.as_ref()
-            && let Some(s3) = &output_location.s3
-            && let Some(encryption) = &s3.encryption
+        if let Some(encryption) = &s3.encryption
            && encryption.encryption_type.as_str() != ""
        {
            meta.insert(X_AMZ_SERVER_SIDE_ENCRYPTION.as_str().to_string(), AMZ_ENCRYPTION_AES.to_string());
@@ -2466,12 +2576,7 @@ impl RestoreRequestOps for RestoreRequest {
        if self.type_.as_ref().is_none_or(|t| t.as_str() != RestoreRequestType::SELECT) && self.output_location.is_some() {
            return Err(std::io::Error::other("OutputLocation can only be specified with SELECT request type"));
        }
-        if let Some(type_) = &self.type_
-            && type_.as_str() == RestoreRequestType::SELECT
-            && self.output_location.is_none()
-        {
-            return Err(std::io::Error::other("OutputLocation required for SELECT requests"));
-        }
+        select_restore_s3_location(self)?;

        // Days must not be specified with SELECT requests
        if let Some(type_) = &self.type_
@@ -2811,8 +2916,8 @@ mod tests {
        lifecycle_version_purge_state_from_completed_targets, mark_delete_opts_skip_decommissioned_on_remote_success,
        merge_stale_multipart_candidate, replication_state_for_delete, resolve_transition_queue_capacity,
        resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max,
-        should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state,
-        transitioned_cleanup_tuple,
+        select_restore_s3_location, should_defer_date_expiry_for_recent_config_update,
+        should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple,
    };
    use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
    use crate::bucket::lifecycle::runtime_boundary as runtime_sources;
@@ -2835,7 +2940,10 @@ mod tests {
    use rustfs_common::metrics::{IlmAction, global_metrics};
    use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX;
    use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType};
-    use s3s::dto::{BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, Timestamp};
+    use s3s::dto::{
+        BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry, OutputLocation,
+        RestoreRequest, RestoreRequestType, S3Location, Timestamp,
+    };
    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::collections::HashMap;
@@ -3125,6 +3233,57 @@ mod tests {
        assert!(dobj.transitioned_object.name.is_empty());
    }

+    fn select_restore_request(output_location: Option<OutputLocation>) -> RestoreRequest {
+        RestoreRequest {
+            days: None,
+            description: None,
+            glacier_job_parameters: None,
+            output_location,
+            select_parameters: None,
+            tier: None,
+            type_: Some(RestoreRequestType::from_static(RestoreRequestType::SELECT)),
+        }
+    }
+
+    #[test]
+    fn select_restore_s3_location_rejects_missing_s3_output_location() {
+        let request = select_restore_request(Some(OutputLocation { s3: None }));
+
+        let err = select_restore_s3_location(&request).expect_err("missing S3 location should be rejected");
+
+        assert!(err.to_string().contains("OutputLocation.S3 required"));
+    }
+
+    #[test]
+    fn select_restore_s3_location_rejects_missing_metadata_name() {
+        let request = select_restore_request(Some(OutputLocation {
+            s3: Some(S3Location {
+                user_metadata: Some(vec![MetadataEntry {
+                    name: None,
+                    value: Some("value".to_string()),
+                }]),
+                ..Default::default()
+            }),
+        }));
+
+        let err = select_restore_s3_location(&request).expect_err("metadata without name should be rejected");
+
+        assert!(err.to_string().contains("metadata name is required"));
+    }
+
+    #[test]
+    fn select_restore_s3_location_allows_missing_user_metadata() {
+        let request = select_restore_request(Some(OutputLocation {
+            s3: Some(S3Location::default()),
+        }));
+
+        let s3 = select_restore_s3_location(&request)
+            .expect("missing user metadata should be allowed")
+            .expect("S3 location should be present");
+
+        assert!(s3.user_metadata.is_none());
+    }
+
    // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
    // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
    // process environment while `env::set_var`/`env::remove_var` is active.
@@ -3980,6 +4139,43 @@ mod tests {
        assert!(is_err_invalid_upload_id(&err));
    }

+    #[tokio::test]
+    #[serial]
+    async fn stale_multipart_cleanup_applies_zero_day_abort_lifecycle_immediately() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let bucket = format!("stale-zero-lifecycle-{}", Uuid::new_v4().simple());
+        let object = "logs/immediate/object.txt";
+        create_test_bucket(&ecstore, &bucket).await;
+        set_abort_incomplete_lifecycle(&bucket, "logs/", 0).await;
+
+        let initiated = OffsetDateTime::now_utc() - time::Duration::minutes(5);
+        let upload = ecstore
+            .new_multipart_upload(
+                &bucket,
+                object,
+                &ObjectOptions {
+                    mod_time: Some(initiated),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("multipart upload should be created");
+
+        let deleted = cleanup_stale_multipart_uploads_once_at(
+            ecstore.clone(),
+            OffsetDateTime::now_utc(),
+            StdDuration::from_secs(7 * 24 * 60 * 60),
+        )
+        .await;
+        assert!(deleted >= 1, "expected zero-day lifecycle abort cleanup to run immediately");
+
+        let err = ecstore
+            .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect_err("multipart upload should be removed by zero-day lifecycle abort rule");
+        assert!(is_err_invalid_upload_id(&err));
+    }
+
    #[tokio::test]
    #[serial]
    async fn stale_multipart_cleanup_applies_abort_lifecycle_with_size_filter() {
@@ -16,7 +16,7 @@ use rustfs_config::{DEFAULT_ILM_PROCESS_TIME_SECS, ENV_ILM_PROCESS_TIME, ENV_ILM
 use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType};
 use s3s::dto::{
    BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter,
-    NoncurrentVersionTransition, ObjectLockConfiguration, ObjectLockEnabled, RestoreRequest, Transition, TransitionStorageClass,
+    NoncurrentVersionTransition, ObjectLockConfiguration, ObjectLockEnabled, RestoreRequest, Transition,
 };
 use std::cmp::Ordering;
 use std::collections::HashMap;
@@ -26,6 +26,7 @@ use time::{self, Duration, OffsetDateTime};
 use tracing::debug;
 use uuid::Uuid;

+use crate::bucket::lifecycle::rule::{NoncurrentVersionTransitionOps, TransitionOps};
 use crate::object_api::ObjectInfo;

 pub const TRANSITION_COMPLETE: &str = "complete";
@@ -42,6 +43,8 @@ const ERR_LIFECYCLE_BUCKET_LOCKED: &str =
 const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration should have at most 1000 rules";
 const ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS: &str = "Lifecycle expiration days must not be negative";
 const ERR_LIFECYCLE_INVALID_NONCURRENT_EXPIRATION_DAYS: &str = "Lifecycle noncurrent expiration days must not be negative";
+const ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS: &str =
+    "DaysAfterInitiation must be 0 or greater when used with AbortIncompleteMultipartUpload";
 const ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT: &str = "Expiration.Date must be at midnight UTC";
 const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
 const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
@@ -200,7 +203,7 @@ pub trait Lifecycle {
 impl Lifecycle for BucketLifecycleConfiguration {
    async fn has_transition(&self) -> bool {
        for rule in self.rules.iter() {
-            if rule.transitions.is_some() {
+            if rule.transitions.as_ref().is_some_and(|transitions| !transitions.is_empty()) {
                return true;
            }
        }
@@ -246,7 +249,11 @@ impl Lifecycle for BucketLifecycleConfiguration {
                    return true;
                }
            }
-            if rule.noncurrent_version_transitions.is_some() {
+            if rule
+                .noncurrent_version_transitions
+                .as_ref()
+                .is_some_and(|transitions| !transitions.is_empty())
+            {
                return true;
            }
            if let Some(rule_expiration) = &rule.expiration {
@@ -267,15 +274,12 @@ impl Lifecycle for BucketLifecycleConfiguration {
                    return true;
                }
            }
-            if let Some(rule_transitions) = &rule.transitions {
-                let rule_transitions_0 = rule_transitions[0].clone();
-                if let Some(date1) = rule_transitions_0.date
+            if let Some(rule_transition) = rule.transitions.as_ref().and_then(|transitions| transitions.first()) {
+                if let Some(date1) = rule_transition.date.clone()
                    && OffsetDateTime::from(date1).unix_timestamp() < OffsetDateTime::now_utc().unix_timestamp()
                {
                    return true;
                }
-            }
-            if rule.transitions.is_some() {
                return true;
            }
        }
@@ -316,6 +320,22 @@ impl Lifecycle for BucketLifecycleConfiguration {
            {
                return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_NONCURRENT_EXPIRATION_DAYS));
            }
+            if let Some(abort_incomplete_multipart_upload) = &r.abort_incomplete_multipart_upload {
+                match abort_incomplete_multipart_upload.days_after_initiation {
+                    Some(days) if days >= 0 => {}
+                    _ => return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS)),
+                }
+            }
+            if let Some(transitions) = &r.transitions {
+                for transition in transitions {
+                    TransitionOps::validate(transition)?;
+                }
+            }
+            if let Some(noncurrent_transitions) = &r.noncurrent_version_transitions {
+                for transition in noncurrent_transitions {
+                    NoncurrentVersionTransitionOps::validate(transition)?;
+                }
+            }
            if let Some(id) = &r.id
                && id.len() > 255
            {
@@ -441,7 +461,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
        event.unwrap_or_default()
    }

-    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, _newer_noncurrent_versions: usize) -> Event {
+    async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event {
        let mut events = Vec::<Event>::new();
        debug!(
            "eval_inner: object={}, mod_time={:?}, successor_mod_time={:?}, now={:?}, is_latest={}, delete_marker={}",
@@ -526,8 +546,8 @@ impl Lifecycle for BucketLifecycleConfiguration {

                if !obj.is_latest
                    && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
-                    && let Some(newer_noncurrent_versions) = noncurrent_version_expiration.newer_noncurrent_versions
-                    && newer_noncurrent_versions > 0
+                    && let Some(retain_newer_noncurrent_versions) = noncurrent_version_expiration.newer_noncurrent_versions
+                    && newer_noncurrent_versions < retain_newer_noncurrent_versions as usize
                {
                    continue;
                }
@@ -551,13 +571,16 @@ impl Lifecycle for BucketLifecycleConfiguration {
                }

                if !obj.is_latest
-                    && let Some(ref noncurrent_version_transitions) = rule.noncurrent_version_transitions
-                    && let Some(ref storage_class) = noncurrent_version_transitions[0].storage_class
-                    && storage_class.as_str() != ""
+                    && let Some(noncurrent_version_transition) = rule
+                        .noncurrent_version_transitions
+                        .as_ref()
+                        .and_then(|transitions| transitions.first())
+                    && let Some(storage_class) = noncurrent_version_transition.storage_class.as_ref()
+                    && !storage_class.as_str().is_empty()
                    && !obj.delete_marker
                    && obj.transition_status != TRANSITION_COMPLETE
                {
-                    let due = rule.noncurrent_version_transitions.as_ref().unwrap()[0].next_due(obj);
+                    let due = noncurrent_version_transition.next_due(obj);
                    if let Some(due0) = due
                        && (now.unix_timestamp() == 0 || now.unix_timestamp() > due0.unix_timestamp())
                    {
@@ -565,12 +588,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
                            action: IlmAction::TransitionVersionAction,
                            rule_id: rule.id.clone().unwrap_or_default(),
                            due,
-                            storage_class: rule.noncurrent_version_transitions.as_ref().unwrap()[0]
-                                .storage_class
-                                .clone()
-                                .unwrap()
-                                .as_str()
-                                .to_string(),
+                            storage_class: storage_class.as_str().to_string(),
                            ..Default::default()
                        });
                    }
@@ -639,9 +657,11 @@ impl Lifecycle for BucketLifecycleConfiguration {
                    }

                    if obj.transition_status != TRANSITION_COMPLETE
-                        && let Some(ref transitions) = rule.transitions
+                        && let Some(transition) = rule.transitions.as_ref().and_then(|transitions| transitions.first())
+                        && let Some(storage_class) = transition.storage_class.as_ref()
+                        && !storage_class.as_str().is_empty()
                    {
-                        let due = transitions[0].next_due(obj);
+                        let due = transition.next_due(obj);
                        if let Some(due0) = due
                            && (now.unix_timestamp() == 0 || now.unix_timestamp() > due0.unix_timestamp())
                        {
@@ -649,12 +669,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
                                action: IlmAction::TransitionAction,
                                rule_id: rule.id.clone().unwrap_or_default(),
                                due,
-                                storage_class: transitions[0]
-                                    .storage_class
-                                    .clone()
-                                    .unwrap_or_else(|| TransitionStorageClass::from_static(""))
-                                    .as_str()
-                                    .to_string(),
+                                storage_class: storage_class.as_str().to_string(),
                                noncurrent_days: 0,
                                newer_noncurrent_versions: 0,
                            });
@@ -856,7 +871,7 @@ pub async fn abort_incomplete_multipart_upload_due(
                .abort_incomplete_multipart_upload
                .as_ref()?
                .days_after_initiation
-                .filter(|days| *days > 0)?;
+                .filter(|days| *days >= 0)?;
            Some((expected_expiry_time(initiated, days), rule.id.unwrap_or_default()))
        })
        .min_by_key(|(due, _)| due.unix_timestamp_nanos())
@@ -962,7 +977,7 @@ impl Default for TransitionOptions {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use s3s::dto::LifecycleRuleFilter;
+    use s3s::dto::{LifecycleRuleFilter, TransitionStorageClass};
    use serial_test::serial;
    use std::sync::Arc;
    use time::macros::datetime;
@@ -1165,6 +1180,119 @@ mod tests {
            .expect("expected validation to pass");
    }

+    #[tokio::test]
+    #[serial]
+    async fn validate_accepts_zero_abort_incomplete_multipart_upload_days() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: Some(0),
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("abort-zero".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: Some("test/".to_string()),
+                transitions: None,
+            }],
+        };
+
+        lc.validate(&ObjectLockConfiguration::default())
+            .await
+            .expect("zero-day abort incomplete multipart upload should be accepted");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn validate_rejects_missing_abort_incomplete_multipart_upload_days() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: None,
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("abort-missing".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: Some("test/".to_string()),
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn validate_rejects_negative_abort_incomplete_multipart_upload_days() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: Some(-1),
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("abort-negative".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: Some("test/".to_string()),
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn abort_incomplete_multipart_upload_due_accepts_zero_days() {
+        let initiated = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: Some(0),
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("abort-zero".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: Some("test/".to_string()),
+                transitions: None,
+            }],
+        };
+        let opts = ObjectOpts {
+            name: "test/object".to_string(),
+            mod_time: Some(initiated),
+            ..Default::default()
+        };
+
+        let (due, rule_id) = abort_incomplete_multipart_upload_due(&lc, &opts)
+            .await
+            .expect("zero-day abort rule should be due");
+
+        assert_eq!(rule_id, "abort-zero");
+        assert_eq!(due, OffsetDateTime::UNIX_EPOCH);
+    }
+
    #[tokio::test]
    #[serial]
    async fn validate_rejects_non_midnight_expiration_date() {
@@ -1361,6 +1489,120 @@ mod tests {
        assert_eq!(err.to_string(), ERR_LIFECYCLE_DUPLICATE_ID);
    }

+    #[tokio::test]
+    #[serial]
+    async fn validate_rejects_transition_without_storage_class() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("transition-no-storage".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: Some(vec![Transition {
+                    days: Some(1),
+                    date: None,
+                    storage_class: None,
+                }]),
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), "ERR_XML_NOT_WELL_FORMED");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn validate_rejects_transition_without_date_or_days() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("transition-no-schedule".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: Some(vec![Transition {
+                    days: None,
+                    date: None,
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]),
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert!(err.to_string().contains("Exactly one of Days"));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn validate_rejects_noncurrent_transition_without_days() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("noncurrent-transition-no-days".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: None,
+                    newer_noncurrent_versions: None,
+                    storage_class: Some(TransitionStorageClass::from_static("COLD")),
+                }]),
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert!(err.to_string().contains("Exactly one of Days"));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn empty_transition_vectors_are_not_active_or_due() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("empty-transition".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: Some(vec![]),
+                prefix: None,
+                transitions: Some(vec![]),
+            }],
+        };
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(1_000_000).unwrap()),
+            is_latest: true,
+            ..Default::default()
+        };
+
+        assert!(!lc.has_transition().await);
+        assert!(!lc.has_active_rules("obj"));
+        assert_eq!(lc.eval_inner(&opts, OffsetDateTime::now_utc(), 0).await.action, IlmAction::NoneAction);
+    }
+
    #[tokio::test]
    #[serial]
    async fn eval_inner_expires_latest_object_after_days_due() {
@@ -1662,6 +1904,60 @@ mod tests {
        assert_eq!(event.newer_noncurrent_versions, 3);
    }

+    #[tokio::test]
+    #[serial]
+    async fn evaluator_honors_newer_noncurrent_versions_retention_count() {
+        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
+        let lc = Arc::new(BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("retain-two-noncurrent".to_string()),
+                noncurrent_version_expiration: Some(s3s::dto::NoncurrentVersionExpiration {
+                    noncurrent_days: Some(1),
+                    newer_noncurrent_versions: Some(2),
+                }),
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        });
+        let mut objs = vec![ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(base_time + Duration::days(4)),
+            successor_mod_time: None,
+            is_latest: true,
+            version_id: Some(Uuid::new_v4()),
+            num_versions: 5,
+            ..Default::default()
+        }];
+        for days_ago in (0..4).rev() {
+            objs.push(ObjectOpts {
+                name: "obj".to_string(),
+                mod_time: Some(base_time + Duration::days(days_ago)),
+                successor_mod_time: Some(base_time + Duration::days(days_ago + 1)),
+                is_latest: false,
+                version_id: Some(Uuid::new_v4()),
+                num_versions: 5,
+                ..Default::default()
+            });
+        }
+
+        let events = crate::bucket::lifecycle::evaluator::Evaluator::new(lc)
+            .eval(&objs)
+            .await
+            .expect("version group should evaluate");
+
+        assert_eq!(events[1].action, IlmAction::NoneAction);
+        assert_eq!(events[2].action, IlmAction::NoneAction);
+        assert_eq!(events[3].action, IlmAction::DeleteVersionAction);
+        assert_eq!(events[4].action, IlmAction::DeleteVersionAction);
+    }
+
    #[tokio::test]
    #[serial]
    async fn validate_rejects_invalid_status_case_sensitive() {
@@ -12,7 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 use super::tagging_boundary;
-use s3s::dto::{LifecycleRuleAndOperator, LifecycleRuleFilter, Tag, Transition};
+use s3s::dto::{LifecycleRuleAndOperator, LifecycleRuleFilter, NoncurrentVersionTransition, Tag, Transition};
+use time::OffsetDateTime;

 const _ERR_TRANSITION_INVALID_DAYS: &str = "Days must be 0 or greater when used with Transition";
 const _ERR_TRANSITION_INVALID_DATE: &str = "Date must be provided in ISO 8601 format";
@@ -80,15 +81,56 @@ pub trait TransitionOps {
    fn validate(&self) -> Result<(), std::io::Error>;
 }

+pub trait NoncurrentVersionTransitionOps {
+    fn validate(&self) -> Result<(), std::io::Error>;
+}
+
+fn is_midnight(date: OffsetDateTime) -> bool {
+    date.hour() == 0 && date.minute() == 0 && date.second() == 0 && date.nanosecond() == 0
+}
+
 impl TransitionOps for Transition {
    fn validate(&self) -> Result<(), std::io::Error> {
-        if self.date.is_some() && self.days.is_some_and(|d| d > 0) {
+        if self
+            .storage_class
+            .as_ref()
+            .is_none_or(|storage_class| storage_class.as_str().is_empty())
+        {
+            return Err(std::io::Error::other("ERR_XML_NOT_WELL_FORMED"));
+        }
+        if self.date.is_some() == self.days.is_some() {
            return Err(std::io::Error::other(ERR_TRANSITION_INVALID));
        }
+        if self.days.is_some_and(|days| days < 0) {
+            return Err(std::io::Error::other(ERR_TRANSITION_INVALID));
+        }
+        if let Some(date) = self.date.clone() {
+            let date = OffsetDateTime::from(date);
+            if date.unix_timestamp() <= 0 || !is_midnight(date) {
+                return Err(std::io::Error::other(ERR_TRANSITION_INVALID));
+            }
+        }
+        Ok(())
+    }
+}

-        if self.storage_class.is_none() {
+impl NoncurrentVersionTransitionOps for NoncurrentVersionTransition {
+    fn validate(&self) -> Result<(), std::io::Error> {
+        if self
+            .storage_class
+            .as_ref()
+            .is_none_or(|storage_class| storage_class.as_str().is_empty())
+        {
            return Err(std::io::Error::other("ERR_XML_NOT_WELL_FORMED"));
        }
+        if self.noncurrent_days.is_none() {
+            return Err(std::io::Error::other(ERR_TRANSITION_INVALID));
+        }
+        if self.noncurrent_days.is_some_and(|days| days < 0)
+            || self.newer_noncurrent_versions.is_some_and(|versions| versions < 0)
+        {
+            return Err(std::io::Error::other(ERR_TRANSITION_INVALID));
+        }
        Ok(())
    }
 }
@@ -23,8 +23,8 @@ use super::storage_api::bucket_usecase::bucket::{
    ObjectLockConfigExt as _, VersioningConfigExt as _,
    bucket_target_sys::BucketTargetSys,
    lifecycle::bucket_lifecycle_ops::{
-        enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, validate_lifecycle_config,
-        validate_transition_tier,
+        enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, run_stale_multipart_upload_cleanup_once,
+        validate_lifecycle_config, validate_transition_tier,
    },
    metadata::{
        BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG,
@@ -756,6 +756,13 @@ fn lifecycle_has_expiry_rules(config: &BucketLifecycleConfiguration) -> bool {
    })
 }

+fn lifecycle_has_abort_multipart_rules(config: &BucketLifecycleConfiguration) -> bool {
+    config.rules.iter().any(|rule| {
+        rule.status == ExpirationStatus::from_static(ExpirationStatus::ENABLED)
+            && rule.abort_incomplete_multipart_upload.is_some()
+    })
+}
+
 #[derive(Clone, Default)]
 pub struct DefaultBucketUsecase {
    context: Option<Arc<AppContext>>,
@@ -1721,6 +1728,17 @@ impl DefaultBucketUsecase {
            });
        }

+        if lifecycle_has_abort_multipart_rules(&input_cfg)
+            && let Some(store) = self.object_store()
+        {
+            let bucket_name = bucket.clone();
+            let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+            spawn_background_with_context(request_context, async move {
+                let deleted = run_stale_multipart_upload_cleanup_once(store).await;
+                debug!(bucket = %bucket_name, deleted, "completed lifecycle abort multipart cleanup trigger");
+            });
+        }
+
        rustfs_scanner::record_dirty_usage_bucket(&bucket);
        Ok(S3Response::new(PutBucketLifecycleConfigurationOutput::default()))
    }
@@ -2789,6 +2807,30 @@ mod tests {
        assert!(lifecycle_has_transition_rules(&config));
    }

+    #[test]
+    fn lifecycle_has_abort_multipart_rules_accepts_enabled_abort_rule() {
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                    days_after_initiation: Some(0),
+                }),
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("enabled-abort".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        assert!(lifecycle_has_abort_multipart_rules(&config));
+        assert!(!lifecycle_has_expiry_rules(&config));
+    }
+
    #[tokio::test]
    async fn execute_list_buckets_returns_internal_error_when_store_uninitialized() {
        let input = ListBucketsInput::builder().build().unwrap();
@@ -336,6 +336,13 @@ pub(crate) mod bucket {
                .await
            }

+            pub(crate) async fn run_stale_multipart_upload_cleanup_once(api: Arc<crate::storage::storage_api::ECStore>) -> usize {
+                crate::storage::storage_api::ecstore_bucket::lifecycle::bucket_lifecycle_ops::run_stale_multipart_upload_cleanup_once(
+                    api,
+                )
+                .await
+            }
+
            pub(crate) async fn enqueue_transition_immediate(
                oi: &crate::storage::storage_api::StorageObjectInfo,
                src: LcEventSrc,