diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 9d9ebfb55..48ec92965 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -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::>(); 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, object: &ObjectInfo, ev } } +struct ExistingObjectExpiryContext<'a> { + api: Arc, + bucket: &'a str, + lc: Arc, + lock_config: Option>, + replication: Option>, + 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::>(); + 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, 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::; loop { let page = api @@ -2076,34 +2185,17 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, 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, 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) -> 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() { diff --git a/crates/ecstore/src/bucket/lifecycle/core.rs b/crates/ecstore/src/bucket/lifecycle/core.rs index 1771b9199..8287b7ef0 100644 --- a/crates/ecstore/src/bucket/lifecycle/core.rs +++ b/crates/ecstore/src/bucket/lifecycle/core.rs @@ -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::::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() { diff --git a/crates/ecstore/src/bucket/lifecycle/rule.rs b/crates/ecstore/src/bucket/lifecycle/rule.rs index ad4670656..e137c5413 100644 --- a/crates/ecstore/src/bucket/lifecycle/rule.rs +++ b/crates/ecstore/src/bucket/lifecycle/rule.rs @@ -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(()) } } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index c248337d8..4754586b3 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -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>, @@ -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::().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(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 5dbfa70e4..9852a623b 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -336,6 +336,13 @@ pub(crate) mod bucket { .await } + pub(crate) async fn run_stale_multipart_upload_cleanup_once(api: Arc) -> 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,