fix(ecstore): set expiration header for put object via lifecycle prediction (#2003)

crates/ecstore/src/bucket/lifecycle/lifecycle.rs

@@ -46,6 +46,9 @@ const ERR_LIFECYCLE_BUCKET_LOCKED: &str =
     "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an retention bucket";
 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 be greater than 0";
+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";
 
 pub use rustfs_common::metrics::IlmAction;
 
@@ -139,6 +142,7 @@ pub trait Lifecycle {
     async fn validate(&self, lr: &ObjectLockConfiguration) -> Result<(), std::io::Error>;
     async fn filter_rules(&self, obj: &ObjectOpts) -> Option<Vec<LifecycleRule>>;
     async fn eval(&self, obj: &ObjectOpts) -> Event;
+    async fn predict_expiration(&self, obj: &ObjectOpts) -> Event;
     async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event;
     //fn set_prediction_headers(&self, w: http.ResponseWriter, obj: ObjectOpts);
     async fn noncurrent_versions_expiration_limit(self: Arc<Self>, obj: &ObjectOpts) -> Event;
@@ -233,13 +237,30 @@ impl Lifecycle for BucketLifecycleConfiguration {
         }
 
         for r in &self.rules {
+            if r.status != ExpirationStatus::from_static(ExpirationStatus::ENABLED)
+                && r.status != ExpirationStatus::from_static(ExpirationStatus::DISABLED)
+            {
+                return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_STATUS));
+            }
+
             if let Some(expiration) = &r.expiration {
+                if let Some(expiration_date) = &expiration.date {
+                    let date = OffsetDateTime::from(expiration_date.clone());
+                    if date.hour() != 0 || date.minute() != 0 || date.second() != 0 || date.nanosecond() != 0 {
+                        return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT));
+                    }
+                }
                 if let Some(days) = expiration.days {
                     if days <= 0 {
                         return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS));
                     }
                 }
             }
+            if let Some(id) = &r.id {
+                if id.len() > 255 {
+                    return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
+                }
+            }
             r.validate()?;
             /*if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref() {
                 if let Some(expiration) = r.expiration.as_ref() {
@@ -257,8 +278,10 @@ impl Lifecycle for BucketLifecycleConfiguration {
             }
             let other_rules = &self.rules[i + 1..];
             for other_rule in other_rules {
-                if self.rules[i].id == other_rule.id {
-                    return Err(std::io::Error::other(ERR_LIFECYCLE_DUPLICATE_ID));
+                if let (Some(id1), Some(id2)) = (&self.rules[i].id, &other_rule.id) {
+                    if id1 == id2 {
+                        return Err(std::io::Error::other(ERR_LIFECYCLE_DUPLICATE_ID));
+                    }
                 }
             }
         }
@@ -295,6 +318,61 @@ impl Lifecycle for BucketLifecycleConfiguration {
         self.eval_inner(obj, OffsetDateTime::now_utc(), 0).await
     }
 
+    async fn predict_expiration(&self, obj: &ObjectOpts) -> Event {
+        let mod_time = match obj.mod_time {
+            Some(time) => {
+                if time.unix_timestamp() == 0 {
+                    return Event::default();
+                }
+                time
+            }
+            None => return Event::default(),
+        };
+
+        if obj.delete_marker || !(obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) {
+            return Event::default();
+        }
+
+        let Some(lc_rules) = self.filter_rules(obj).await else {
+            return Event::default();
+        };
+
+        let mut event: Option<Event> = None;
+        for rule in lc_rules {
+            let Some(expiration) = rule.expiration else {
+                continue;
+            };
+            if expiration.expired_object_delete_marker.is_some_and(|v| v) {
+                continue;
+            }
+            let Some(due) = (match (&expiration.date, expiration.days) {
+                (Some(date), _) => Some(OffsetDateTime::from(date.clone())),
+                (None, Some(days)) => Some(expected_expiry_time(mod_time, days)),
+                _ => None,
+            }) else {
+                continue;
+            };
+
+            let predicted = Event {
+                action: IlmAction::DeleteAction,
+                rule_id: rule.id.clone().unwrap_or_default(),
+                due: Some(due),
+                noncurrent_days: 0,
+                newer_noncurrent_versions: 0,
+                storage_class: "".into(),
+            };
+            let predicted_due = predicted.due.unwrap_or_else(|| OffsetDateTime::UNIX_EPOCH).unix_timestamp();
+            let should_replace = event
+                .as_ref()
+                .is_none_or(|e| predicted_due < e.due.unwrap_or_else(|| OffsetDateTime::UNIX_EPOCH).unix_timestamp());
+            if should_replace {
+                event = Some(predicted);
+            }
+        }
+
+        event.unwrap_or_default()
+    }
+
     async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event {
         let mut events = Vec::<Event>::new();
         info!(
@@ -833,4 +911,167 @@ mod tests {
             .await
             .expect("expected validation to pass");
     }
+
+    #[tokio::test]
+    async fn validate_rejects_non_midnight_expiration_date() {
+        let lc = BucketLifecycleConfiguration {
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    date: Some(time::OffsetDateTime::from_unix_timestamp(20_000_101).unwrap().into()),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT);
+    }
+
+    #[tokio::test]
+    async fn predict_expiration_selects_closest_expiry_for_put_object() {
+        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
+        let lc = BucketLifecycleConfiguration {
+            rules: vec![
+                LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: Some(LifecycleExpiration {
+                        days: Some(30),
+                        ..Default::default()
+                    }),
+                    abort_incomplete_multipart_upload: None,
+                    filter: None,
+                    id: Some("rule-days".to_string()),
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                },
+                LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: Some(LifecycleExpiration {
+                        date: Some((base_time + Duration::days(1)).into()),
+                        ..Default::default()
+                    }),
+                    abort_incomplete_multipart_upload: None,
+                    filter: None,
+                    id: Some("rule-date".to_string()),
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                },
+            ],
+        };
+        let opts = ObjectOpts {
+            name: "obj".to_string(),
+            mod_time: Some(base_time),
+            is_latest: true,
+            version_id: None,
+            ..Default::default()
+        };
+
+        let event = lc.predict_expiration(&opts).await;
+        let expected = base_time + Duration::days(1);
+
+        assert_eq!(event.action, IlmAction::DeleteAction);
+        assert_eq!(event.rule_id, "rule-date");
+        assert_eq!(event.due, Some(expected));
+    }
+
+    #[tokio::test]
+    async fn validate_accepts_multiple_rules_without_ids() {
+        let lc = BucketLifecycleConfiguration {
+            rules: vec![
+                LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: Some(LifecycleExpiration {
+                        days: Some(30),
+                        ..Default::default()
+                    }),
+                    abort_incomplete_multipart_upload: None,
+                    filter: None,
+                    id: None,
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                },
+                LifecycleRule {
+                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                    expiration: Some(LifecycleExpiration {
+                        days: Some(31),
+                        ..Default::default()
+                    }),
+                    abort_incomplete_multipart_upload: None,
+                    filter: None,
+                    id: None,
+                    noncurrent_version_expiration: None,
+                    noncurrent_version_transitions: None,
+                    prefix: None,
+                    transitions: None,
+                },
+            ],
+        };
+
+        lc.validate(&ObjectLockConfiguration::default())
+            .await
+            .expect("expected validation to pass");
+    }
+
+    #[tokio::test]
+    async fn validate_rejects_rule_id_too_long() {
+        let lc = BucketLifecycleConfiguration {
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(30),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: Some("a".repeat(256)),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG);
+    }
+
+    #[tokio::test]
+    async fn validate_rejects_invalid_status_case_sensitive() {
+        let lc = BucketLifecycleConfiguration {
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static("enabled"),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(30),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+
+        assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_RULE_STATUS);
+    }
 }

rustfs/src/app/bucket_usecase.rs

@@ -60,7 +60,7 @@ use s3s::dto::*;
 use s3s::region::Region;
 use s3s::xml;
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
-use std::{fmt::Display, sync::Arc};
+use std::{collections::HashSet, fmt::Display, sync::Arc};
 use tracing::{debug, error, info, instrument, warn};
 use urlencoding::encode;
 
@@ -81,6 +81,45 @@ fn resolve_notification_region(global_region: Option<Region>, request_region: Op
     global_region.or(request_region).unwrap_or_else(default_region)
 }
 
+const ERR_LIFECYCLE_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
+
+fn assign_lifecycle_rule_ids(rules: &mut [LifecycleRule]) {
+    let mut rule_ids: HashSet<String> = HashSet::new();
+
+    for rule in rules.iter() {
+        if let Some(id) = rule.id.as_ref() {
+            rule_ids.insert(id.to_string());
+        }
+    }
+
+    for (idx, rule) in rules.iter_mut().enumerate() {
+        if rule.id.is_none() {
+            let mut suffix = 0usize;
+            let mut generated_id = format!("rule-{}", idx);
+
+            while rule_ids.contains(&generated_id) {
+                suffix += 1;
+                generated_id = format!("rule-{idx}-{suffix}");
+            }
+
+            rule_ids.insert(generated_id.clone());
+            rule.id = Some(generated_id);
+        }
+    }
+}
+
+fn validate_lifecycle_rule_status(rules: &[LifecycleRule]) -> Result<(), &'static str> {
+    for rule in rules {
+        if rule.status != ExpirationStatus::from_static(ExpirationStatus::ENABLED)
+            && rule.status != ExpirationStatus::from_static(ExpirationStatus::DISABLED)
+        {
+            return Err(ERR_LIFECYCLE_RULE_STATUS);
+        }
+    }
+
+    Ok(())
+}
+
 #[derive(Clone, Default)]
 pub struct DefaultBucketUsecase {
     context: Option<Arc<AppContext>>,
@@ -1080,7 +1119,11 @@ impl DefaultBucketUsecase {
             ..
         } = req.input;
 
-        let Some(input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) };
+        let Some(mut input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) };
+        assign_lifecycle_rule_ids(&mut input_cfg.rules);
+        if let Err(err) = validate_lifecycle_rule_status(&input_cfg.rules) {
+            return Err(S3Error::with_message(S3ErrorCode::MalformedXML, format!("Malformed XML: {err}")));
+        }
 
         let rcfg = match metadata_sys::get_object_lock_config(&bucket).await {
             Ok((cfg, _)) => cfg,
@@ -1912,6 +1955,80 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
+    #[test]
+    fn normalize_lifecycle_rules_generate_rule_ids_for_missing_values() {
+        let mut rules = vec![
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(30),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(60),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: Some("rule-1".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+            LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(90),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                filter: None,
+                id: None,
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            },
+        ];
+
+        assign_lifecycle_rule_ids(&mut rules);
+
+        assert_eq!(rules[0].id.as_deref(), Some("rule-0"));
+        assert_eq!(rules[1].id.as_deref(), Some("rule-1"));
+        assert_eq!(rules[2].id.as_deref(), Some("rule-2"));
+    }
+
+    #[test]
+    fn validate_lifecycle_rule_status_rejects_invalid_status() {
+        let rules = vec![LifecycleRule {
+            status: ExpirationStatus::from_static("enabled"),
+            expiration: Some(LifecycleExpiration {
+                days: Some(30),
+                ..Default::default()
+            }),
+            abort_incomplete_multipart_upload: None,
+            filter: None,
+            id: None,
+            noncurrent_version_expiration: None,
+            noncurrent_version_transitions: None,
+            prefix: None,
+            transitions: None,
+        }];
+
+        assert_eq!(validate_lifecycle_rule_status(&rules).unwrap_err(), ERR_LIFECYCLE_RULE_STATUS);
+    }
+
     #[tokio::test]
     async fn execute_list_buckets_returns_internal_error_when_store_uninitialized() {
         let input = ListBucketsInput::builder().build().unwrap();

rustfs/src/app/object_usecase.rs

@@ -42,7 +42,7 @@ use rustfs_ecstore::bucket::quota::checker::QuotaChecker;
 use rustfs_ecstore::bucket::{
     lifecycle::{
         bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts},
-        lifecycle::{self, TransitionOptions},
+        lifecycle::{self, Lifecycle, TransitionOptions},
     },
     metadata::{BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG},
     metadata_sys,
@@ -172,6 +172,36 @@ fn normalize_delete_objects_version_id(version_id: Option<String>) -> Result<(Op
     }
 }
 
+fn build_put_object_expiration_header(event: &lifecycle::Event) -> Option<String> {
+    if !event.action.delete() {
+        return None;
+    }
+
+    let expire_time = event.due?;
+
+    if event.rule_id.is_empty() || expire_time == OffsetDateTime::UNIX_EPOCH {
+        return None;
+    }
+
+    let expiry_date = expire_time.format(&Rfc3339).ok()?;
+    Some(format!("expiry-date=\"{}\", rule-id=\"{}\"", expiry_date, event.rule_id))
+}
+
+async fn resolve_put_object_expiration(bucket: &str, obj_info: &ObjectInfo) -> Option<String> {
+    let Ok((lifecycle_config, _)) = metadata_sys::get_lifecycle_config(bucket).await else {
+        debug!("resolve_put_object_expiration: lifecycle config not found for bucket {bucket}");
+        return None;
+    };
+
+    let obj_opts = lifecycle::ObjectOpts::from_object_info(obj_info);
+    let event = lifecycle_config.predict_expiration(&obj_opts).await;
+    debug!(
+        "resolve_put_object_expiration: bucket={bucket}, action={:?}, rule_id={}, due={:?}",
+        event.action, event.rule_id, event.due
+    );
+    build_put_object_expiration_header(&event)
+}
+
 #[derive(Clone, Default)]
 pub struct DefaultObjectUsecase {
     context: Option<Arc<AppContext>>,
@@ -516,9 +546,10 @@ impl DefaultObjectUsecase {
             get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts);
 
         let dsc = must_replicate(&bucket, &key, repoptions).await;
+        let expiration = resolve_put_object_expiration(&bucket, &obj_info).await;
 
         if dsc.replicate_any() {
-            schedule_replication(obj_info, store, dsc, ReplicationType::Object).await;
+            schedule_replication(obj_info.clone(), store, dsc, ReplicationType::Object).await;
         }
 
         let mut checksums = PutObjectChecksums {
@@ -540,6 +571,7 @@ impl DefaultObjectUsecase {
             sse_customer_algorithm: sse_customer_algorithm.clone(),
             sse_customer_key_md5: sse_customer_key_md5.clone(),
             ssekms_key_id: effective_kms_key_id,
+            expiration,
             checksum_crc32: checksums.crc32,
             checksum_crc32c: checksums.crc32c,
             checksum_sha1: checksums.sha1,
@@ -3832,6 +3864,62 @@ mod tests {
         assert!(!object_attributes_requested(&object_attributes, ObjectAttributes::OBJECT_SIZE));
     }
 
+    #[test]
+    fn build_put_object_expiration_header_returns_none_for_non_delete_events() {
+        let event = lifecycle::Event {
+            action: lifecycle::IlmAction::TransitionAction,
+            rule_id: "rule-1".to_string(),
+            due: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()),
+            noncurrent_days: 0,
+            newer_noncurrent_versions: 0,
+            storage_class: String::new(),
+        };
+
+        assert!(build_put_object_expiration_header(&event).is_none());
+    }
+
+    #[test]
+    fn build_put_object_expiration_header_formats_expected_value() {
+        let expire_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
+        let event = lifecycle::Event {
+            action: lifecycle::IlmAction::DeleteAction,
+            rule_id: "rule-1".to_string(),
+            due: Some(expire_time),
+            noncurrent_days: 0,
+            newer_noncurrent_versions: 0,
+            storage_class: String::new(),
+        };
+
+        let expiry_date = expire_time.format(&Rfc3339).unwrap();
+        let expected = format!("expiry-date=\"{}\", rule-id=\"rule-1\"", expiry_date);
+        assert_eq!(build_put_object_expiration_header(&event), Some(expected));
+    }
+
+    #[test]
+    fn build_put_object_expiration_header_requires_rule_id_and_due_time() {
+        let event = lifecycle::Event {
+            action: lifecycle::IlmAction::DeleteAction,
+            rule_id: String::new(),
+            due: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()),
+            noncurrent_days: 0,
+            newer_noncurrent_versions: 0,
+            storage_class: String::new(),
+        };
+
+        assert!(build_put_object_expiration_header(&event).is_none());
+
+        let event = lifecycle::Event {
+            action: lifecycle::IlmAction::DeleteAction,
+            rule_id: "rule-1".to_string(),
+            due: Some(OffsetDateTime::UNIX_EPOCH),
+            noncurrent_days: 0,
+            newer_noncurrent_versions: 0,
+            storage_class: String::new(),
+        };
+
+        assert!(build_put_object_expiration_header(&event).is_none());
+    }
+
     #[tokio::test]
     async fn execute_get_object_legal_hold_returns_internal_error_when_store_uninitialized() {
         let input = GetObjectLegalHoldInput::builder()