From 23a5db401206d0d8d8f5ac5f194012d5be5ec31c Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 01:42:02 +0800 Subject: [PATCH] fix(ilm): preserve lifecycle version groups (#5239) * fix(ilm): evaluate complete version groups Co-Authored-By: heihutu * feat(ilm): log replication expiry blocks Co-Authored-By: heihutu * fix(ilm): diagnose incomplete noncurrent chains Co-Authored-By: heihutu * test(ilm): cover purge-pending version groups Add regression coverage for lifecycle-only version listing so purge-pending versions stay present for ILM evaluation while the public ListObjectVersions projection remains filtered. Refs rustfs/backlog#1500 Co-Authored-By: heihutu * feat(ilm): log lifecycle evaluation failures Emit structured lifecycle_evaluation_failed events when version group loading or evaluation fails during immediate and existing-object expiry scans. Refs rustfs/backlog#1503 Co-Authored-By: heihutu * test(ilm): cover incomplete noncurrent chains Pin the fail-closed behavior for noncurrent expiration when successor_mod_time is missing and reuse a stable structured event name for that skip path. Refs rustfs/backlog#1502 Co-Authored-By: heihutu * test(ilm): cover one-day noncurrent expiry boundary Add a runtime regression test proving NoncurrentVersionExpiration Days=1 stays inactive before expected_expiry_time and deletes exactly at the computed due boundary. Refs rustfs/backlog#1504 Co-Authored-By: heihutu * fix(ilm): keep transition checks after incomplete expiry Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 72 +++++++- crates/ecstore/src/object_api/types.rs | 86 +++++++++- crates/ecstore/src/store/list.rs | 13 ++ crates/ecstore/src/store/list_objects.rs | 78 ++++++++- crates/lifecycle/src/core.rs | 154 ++++++++++++++++-- 5 files changed, 374 insertions(+), 29 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 6f23b663d..c89d171f4 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -112,6 +112,7 @@ const EVENT_LIFECYCLE_WORKER_STATE: &str = "lifecycle_worker_state"; const EVENT_LIFECYCLE_TRANSITION_COMPENSATION: &str = "lifecycle_transition_compensation"; const EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP: &str = "lifecycle_stale_multipart_cleanup"; const EVENT_LIFECYCLE_SCAN_SKIPPED: &str = "lifecycle_scan_skipped"; +const EVENT_LIFECYCLE_EVALUATION_FAILED: &str = "lifecycle_evaluation_failed"; const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit"; const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed"; const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed"; @@ -2647,12 +2648,25 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { let mut object_infos = Vec::new(); loop { - let Ok(page) = api + let page = match api .clone() - .list_object_versions(&oi.bucket, &oi.name, marker.clone(), version_marker.clone(), None, 1000) + .list_object_versions_for_lifecycle(&oi.bucket, &oi.name, marker.clone(), version_marker.clone(), None, 1000) .await - else { - return; + { + Ok(page) => page, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_EVALUATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + error = %err, + reason = "list_versions_failed", + "Failed to load lifecycle version group" + ); + return; + } }; object_infos.extend(page.objects.into_iter().filter(|object| object.name == oi.name)); @@ -2677,12 +2691,27 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { .iter() .map(lifecycle::object_opts_from_object_info) .collect::>(); - let Ok(events) = Evaluator::new(Arc::new(lifecycle)) + let events = match Evaluator::new(Arc::new(lifecycle)) .with_lock_retention(lock_config) .eval(&object_opts) .await - else { - return; + { + Ok(events) => events, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_EVALUATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + expected_version_count = oi.num_versions, + observed_version_count = object_infos.len(), + error = %err, + reason = "version_group_evaluation_failed", + "Failed to evaluate lifecycle version group" + ); + return; + } }; let mut to_delete_objs = Vec::new(); @@ -3099,10 +3128,16 @@ async fn enqueue_expiry_for_existing_object_group( Ok(events) => events, Err(err) => { warn!( + event = EVENT_LIFECYCLE_EVALUATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket = context.bucket, object = %object_infos[0].name, + expected_version_count = object_infos[0].num_versions, + observed_version_count = object_infos.len(), error = %err, - "failed to evaluate lifecycle events for existing object versions" + reason = "version_group_evaluation_failed", + "Failed to evaluate lifecycle version group" ); return; } @@ -3210,7 +3245,7 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, bucket: &str loop { let page = api .clone() - .list_object_versions(bucket, "", marker.clone(), version_marker.clone(), None, 1000) + .list_object_versions_for_lifecycle(bucket, "", marker.clone(), version_marker.clone(), None, 1000) .await?; for object in page.objects { @@ -3835,6 +3870,25 @@ pub async fn eval_action_from_lifecycle( } if lifecycle_action_blocked_by_replication(event.action, oi) { + let reason = if oi.version_purge_status.is_pending() { + "version_purge_pending" + } else if oi.replication_status == ReplicationStatusType::Failed { + "replication_failed" + } else { + "replication_pending" + }; + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %oi.name, + version_id = ?oi.version_id, + action = ?event.action, + replication_status = ?oi.replication_status, + version_purge_status = ?oi.version_purge_status, + reason, + "Skipped lifecycle action because replication is not terminal" + ); return lifecycle::Event::default(); } diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 8fd6dcadd..c9a0e560b 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -556,6 +556,29 @@ impl ObjectInfo { prefix: &str, delimiter: Option, after_version_marker: Option, + ) -> Vec { + Self::from_meta_cache_entries_sorted_versions_with_purge(entries, bucket, prefix, delimiter, after_version_marker, false) + .await + } + + pub(crate) async fn from_meta_cache_entries_sorted_versions_for_lifecycle( + entries: &MetaCacheEntriesSorted, + bucket: &str, + prefix: &str, + delimiter: Option, + after_version_marker: Option, + ) -> Vec { + Self::from_meta_cache_entries_sorted_versions_with_purge(entries, bucket, prefix, delimiter, after_version_marker, true) + .await + } + + async fn from_meta_cache_entries_sorted_versions_with_purge( + entries: &MetaCacheEntriesSorted, + bucket: &str, + prefix: &str, + delimiter: Option, + after_version_marker: Option, + include_version_purge: bool, ) -> Vec { let vcfg = get_versioning_config(bucket).await.ok(); let mut objects = Vec::with_capacity(entries.entries().len()); @@ -604,7 +627,7 @@ impl ObjectInfo { }; for fi in versions.iter() { - if !fi.version_purge_status().is_empty() { + if !include_version_purge && !fi.version_purge_status().is_empty() { continue; } @@ -1064,6 +1087,67 @@ mod tests { assert_eq!(objects[0].num_versions, 1); } + #[tokio::test] + async fn lifecycle_versions_listing_preserves_purge_pending_versions() { + let visible_version_id = Uuid::new_v4(); + let purge_version_id = Uuid::new_v4(); + let base_time = OffsetDateTime::now_utc(); + let mut fm = FileMeta::new(); + + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(purge_version_id), + mod_time: Some(base_time), + ..Default::default() + }) + .expect("version pending purge should be added"); + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(visible_version_id), + mod_time: Some(base_time + time::Duration::seconds(1)), + ..Default::default() + }) + .expect("visible version should be added"); + fm.delete_version(&FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(purge_version_id), + replication_state_internal: Some(crate::bucket::replication::replication_state_to_filemeta(&ReplicationState { + version_purge_status_internal: Some("arn:target-a=PENDING;".to_string()), + purge_targets: version_purge_statuses_map("arn:target-a=PENDING;"), + ..Default::default() + })), + ..Default::default() + }) + .expect("version purge status should be persisted"); + + let entries = MetaCacheEntriesSorted { + o: rustfs_filemeta::MetaCacheEntries(vec![Some(MetaCacheEntry { + name: "object".to_string(), + metadata: fm.marshal_msg().expect("metadata should marshal"), + ..Default::default() + })]), + ..Default::default() + }; + + let public_objects = ObjectInfo::from_meta_cache_entries_sorted_versions(&entries, "bucket", "", None, None).await; + let lifecycle_objects = + ObjectInfo::from_meta_cache_entries_sorted_versions_for_lifecycle(&entries, "bucket", "", None, None).await; + + assert_eq!(public_objects.len(), 1); + assert_eq!(public_objects[0].version_id, Some(visible_version_id)); + assert_eq!(public_objects[0].num_versions, 2); + assert_eq!(lifecycle_objects.len(), 2); + assert!( + lifecycle_objects + .iter() + .any(|object| object.version_purge_status == VersionPurgeStatusType::Pending) + ); + assert!(lifecycle_objects.iter().all(|object| object.num_versions == 2)); + } + #[test] fn get_actual_size_prefers_actual_size_field() { let info = ObjectInfo { diff --git a/crates/ecstore/src/store/list.rs b/crates/ecstore/src/store/list.rs index ec4ccb38e..2651797c6 100644 --- a/crates/ecstore/src/store/list.rs +++ b/crates/ecstore/src/store/list.rs @@ -55,6 +55,19 @@ impl ECStore { .await } + pub(crate) async fn list_object_versions_for_lifecycle( + self: Arc, + bucket: &str, + prefix: &str, + marker: Option, + version_marker: Option, + delimiter: Option, + max_keys: i32, + ) -> Result { + self.inner_list_object_versions_for_lifecycle(bucket, prefix, marker, version_marker, delimiter, max_keys) + .await + } + pub(super) async fn handle_walk( self: Arc, rx: CancellationToken, diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 8a1697d8f..17ce6dae8 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -81,6 +81,17 @@ type ListObjectsV2Info = StorageListObjectsV2Info; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ObjectInfoOrErr = StorageObjectInfoOrErr; type WalkOptions = StorageWalkOptions bool>; + +struct ListObjectVersionsInput<'a> { + bucket: &'a str, + prefix: &'a str, + marker: Option, + version_marker: Option, + delimiter: Option, + max_keys: i32, + include_version_purge: bool, +} + const LIST_MERGED_INPUT_BUFFER: usize = 1; fn list_merged_entry_channel() -> (Sender, Receiver) { @@ -3888,6 +3899,52 @@ impl ECStore { delimiter: Option, max_keys: i32, ) -> Result { + self.inner_list_object_versions_with_projection(ListObjectVersionsInput { + bucket, + prefix, + marker, + version_marker, + delimiter, + max_keys, + include_version_purge: false, + }) + .await + } + + pub(crate) async fn inner_list_object_versions_for_lifecycle( + self: Arc, + bucket: &str, + prefix: &str, + marker: Option, + version_marker: Option, + delimiter: Option, + max_keys: i32, + ) -> Result { + self.inner_list_object_versions_with_projection(ListObjectVersionsInput { + bucket, + prefix, + marker, + version_marker, + delimiter, + max_keys, + include_version_purge: true, + }) + .await + } + + async fn inner_list_object_versions_with_projection( + self: Arc, + input: ListObjectVersionsInput<'_>, + ) -> Result { + let ListObjectVersionsInput { + bucket, + prefix, + marker, + version_marker, + delimiter, + max_keys, + include_version_purge, + } = input; let max_keys = normalize_max_keys(max_keys); if marker.is_none() && version_marker.is_some() { return Err(StorageError::NotImplemented); @@ -3947,14 +4004,19 @@ impl ECStore { // Last RAW scanned key, captured before folding (ECA-03 / #944). let last_scanned_key = last_scanned_entry_name(list_result.entries.as_ref()); - let get_objects = ObjectInfo::from_meta_cache_entries_sorted_versions( - &list_result.entries.unwrap_or_default(), - bucket, - prefix, - delimiter.clone(), - version_marker, - ) - .await; + let entries = list_result.entries.unwrap_or_default(); + let get_objects = if include_version_purge { + ObjectInfo::from_meta_cache_entries_sorted_versions_for_lifecycle( + &entries, + bucket, + prefix, + delimiter.clone(), + version_marker, + ) + .await + } else { + ObjectInfo::from_meta_cache_entries_sorted_versions(&entries, bucket, prefix, delimiter.clone(), version_marker).await + }; let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( get_objects, diff --git a/crates/lifecycle/src/core.rs b/crates/lifecycle/src/core.rs index 5ceac58ff..bfed5b270 100644 --- a/crates/lifecycle/src/core.rs +++ b/crates/lifecycle/src/core.rs @@ -33,6 +33,7 @@ const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; const EVENT_LIFECYCLE_EXPIRY_COMPUTED: &str = "lifecycle_expiry_computed"; const EVENT_LIFECYCLE_DEBUG_DAY_SECS: &str = "lifecycle_debug_day_secs"; +const EVENT_LIFECYCLE_NONCURRENT_EXPIRY_SKIPPED: &str = "lifecycle_noncurrent_expiry_skipped"; const ERR_LIFECYCLE_NO_RULE: &str = "Lifecycle configuration should have at least one rule"; const ERR_LIFECYCLE_DUPLICATE_ID: &str = "Rule ID must be unique. Found same ID for more than one rule"; const _ERR_XML_NOT_WELL_FORMED: &str = @@ -594,18 +595,28 @@ impl Lifecycle for BucketLifecycleConfiguration { if !obj.is_latest && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration && let Some(noncurrent_days) = noncurrent_version_expiration.noncurrent_days - && let Some(successor_mod_time) = obj.successor_mod_time { - let expected_expiry = expected_expiry_time(successor_mod_time, noncurrent_days); - if now.unix_timestamp() >= expected_expiry.unix_timestamp() { - events.push(Event { - action: IlmAction::DeleteVersionAction, - rule_id: rule.id.clone().unwrap_or_default(), - due: Some(expected_expiry), - noncurrent_days: 0, - newer_noncurrent_versions: 0, - storage_class: "".into(), - }); + if let Some(successor_mod_time) = obj.successor_mod_time { + let expected_expiry = expected_expiry_time(successor_mod_time, noncurrent_days); + if now.unix_timestamp() >= expected_expiry.unix_timestamp() { + events.push(Event { + action: IlmAction::DeleteVersionAction, + rule_id: rule.id.clone().unwrap_or_default(), + due: Some(expected_expiry), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: "".into(), + }); + } + } else { + debug!( + event = EVENT_LIFECYCLE_NONCURRENT_EXPIRY_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %obj.name, + reason = "missing_successor_mod_time", + "Skipped noncurrent expiration for incomplete version chain" + ); } } @@ -2053,6 +2064,127 @@ mod tests { assert_eq!(event.due, Some(expected_expiry_time(base_time, 1))); } + #[tokio::test] + #[serial] + async fn eval_inner_skips_noncurrent_expiration_without_successor() { + let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp"); + 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-expire".to_string()), + noncurrent_version_expiration: Some(s3s::dto::NoncurrentVersionExpiration { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + }), + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }; + + let opts = ObjectOpts { + name: "obj".to_string(), + mod_time: Some(base_time), + successor_mod_time: None, + is_latest: false, + version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + let event = lc.eval_inner(&opts, base_time + Duration::days(2), 0).await; + + assert_eq!(event.action, IlmAction::NoneAction); + } + + #[tokio::test] + #[serial] + async fn eval_inner_missing_successor_does_not_skip_noncurrent_transition() { + let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp"); + 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-expire-and-transition".to_string()), + noncurrent_version_expiration: Some(s3s::dto::NoncurrentVersionExpiration { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + }), + noncurrent_version_transitions: Some(vec![NoncurrentVersionTransition { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + storage_class: Some(TransitionStorageClass::from_static("COLDTIER44")), + }]), + prefix: None, + transitions: None, + }], + }; + + let opts = ObjectOpts { + name: "obj".to_string(), + mod_time: Some(base_time), + successor_mod_time: None, + is_latest: false, + transition_status: "".to_string(), + version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + let event = lc.eval_inner(&opts, datetime!(2100-01-01 00:00:00 UTC), 0).await; + + assert_eq!(event.action, IlmAction::TransitionVersionAction); + assert_eq!(event.rule_id, "noncurrent-expire-and-transition"); + assert_eq!(event.storage_class, "COLDTIER44"); + } + + #[tokio::test] + #[serial] + async fn eval_inner_noncurrent_expiration_one_day_respects_due_boundary() { + let successor_time = datetime!(2025-06-15 12:00:00 UTC); + let due = expected_expiry_time(successor_time, 1); + 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-one-day".to_string()), + noncurrent_version_expiration: Some(s3s::dto::NoncurrentVersionExpiration { + noncurrent_days: Some(1), + newer_noncurrent_versions: None, + }), + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }; + + let opts = ObjectOpts { + name: "obj".to_string(), + mod_time: Some(successor_time - Duration::seconds(1)), + successor_mod_time: Some(successor_time), + is_latest: false, + version_id: Some(Uuid::new_v4()), + ..Default::default() + }; + + let before_due = lc.eval_inner(&opts, due - Duration::seconds(1), 0).await; + let at_due = lc.eval_inner(&opts, due, 0).await; + + assert_eq!(before_due.action, IlmAction::NoneAction); + assert_eq!(at_due.action, IlmAction::DeleteVersionAction); + assert_eq!(at_due.rule_id, "noncurrent-one-day"); + assert_eq!(at_due.due, Some(due)); + } + #[tokio::test] #[serial] async fn eval_inner_expires_noncurrent_version_immediately_when_zero_days() {