diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 48ec92965..1906594b3 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -48,7 +48,6 @@ use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded}; use futures::Future; use http::HeaderMap; use lazy_static::lazy_static; -use rustfs_common::heal_channel::rep_has_active_rules; use rustfs_common::metrics::{ IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics, }; @@ -64,8 +63,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, ObjectLockConfiguration, ReplicationConfiguration, - RestoreRequest, RestoreRequestType, RestoreStatus, Timestamp, + BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, ObjectLockConfiguration, RestoreRequest, + RestoreRequestType, RestoreStatus, Timestamp, }; use s3s::header::{X_AMZ_RESTORE, X_AMZ_SERVER_SIDE_ENCRYPTION}; use sha2::{Digest, Sha256}; @@ -1945,18 +1944,12 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { Ok((cfg, _)) => Some(Arc::new(cfg)), Err(_) => None, }; - let replication = match metadata_boundary::get_replication_config(&oi.bucket).await { - Ok((cfg, _)) if !cfg.rules.is_empty() => Some(Arc::new(replication_sink::new_replication_config(cfg))), - _ => None, - }; - let object_opts = object_infos .iter() .map(ObjectOpts::from_object_info) .collect::>(); let Ok(events) = Evaluator::new(Arc::new(lifecycle)) .with_lock_retention(lock_config) - .with_replication_config(replication) .eval(&object_opts) .await else { @@ -2060,7 +2053,6 @@ struct ExistingObjectExpiryContext<'a> { bucket: &'a str, lc: Arc, lock_config: Option>, - replication: Option>, src: &'a LcEventSrc, defer_date_expiry_once: bool, } @@ -2080,7 +2072,6 @@ async fn enqueue_expiry_for_existing_object_group( .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 { @@ -2127,6 +2118,28 @@ async fn enqueue_expiry_for_existing_object_group( noncurrent_event = Some(event.clone()); } } else { + let blocked_by_replication = match lifecycle_delete_all_versions_blocked_by_replication( + context.api.clone(), + context.bucket, + &object.name, + event.action, + ) + .await + { + Ok(blocked) => blocked, + Err(err) => { + warn!( + bucket = context.bucket, + object = %object.name, + error = %err, + "failed to check lifecycle delete-all replication state" + ); + true + } + }; + if blocked_by_replication { + continue; + } apply_existing_object_expiry(context.api.clone(), object, event, context.src).await; } } else { @@ -2158,10 +2171,6 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, bucket: &str .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; @@ -2171,7 +2180,6 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, bucket: &str bucket, lc: lc.clone(), lock_config: lock_config.clone(), - replication: replication.clone(), src: &src, defer_date_expiry_once, }; @@ -2204,19 +2212,23 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, bucket: &str } } -async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc) { +async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc) -> bool { let event = lc.eval(&oi.to_lifecycle_opts()).await; match event.action { IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { if oi.delete_marker || oi.is_dir { - return; + return false; } - runtime_sources::transition_state_handle() + if lifecycle_action_blocked_by_replication(event.action, oi) { + return false; + } + return runtime_sources::transition_state_handle() .queue_transition_task(oi, &event, src) .await; } _ => (), } + false } pub async fn expire_transitioned_object( @@ -2530,21 +2542,7 @@ pub trait LifecycleOps { impl LifecycleOps for ObjectInfo { fn to_lifecycle_opts(&self) -> lifecycle::ObjectOpts { - lifecycle::ObjectOpts { - name: self.name.clone(), - user_tags: (*self.user_tags).clone(), - version_id: self.version_id, - mod_time: self.mod_time, - size: self.size as usize, - is_latest: self.is_latest, - num_versions: self.num_versions, - delete_marker: self.delete_marker, - successor_mod_time: self.successor_mod_time, - restore_ongoing: self.restore_ongoing, - restore_expires: self.restore_expires, - transition_status: self.transitioned_object.status.clone(), - ..Default::default() - } + lifecycle::ObjectOpts::from_object_info(self) } fn is_remote(&self) -> bool { @@ -2600,7 +2598,6 @@ const _MAX_RESTORE_OBJECT_REQUEST_SIZE: i64 = 2 << 20; pub async fn eval_action_from_lifecycle( lc: &BucketLifecycleConfiguration, lr: Option, - rcfg: Option<(ReplicationConfiguration, OffsetDateTime)>, oi: &ObjectInfo, ) -> lifecycle::Event { let event = lc.eval(&oi.to_lifecycle_opts()).await; @@ -2652,18 +2649,81 @@ pub async fn eval_action_from_lifecycle( } return lifecycle::Event::default(); } - if let Some(rcfg) = rcfg - && rep_has_active_rules(&rcfg.0, &oi.name, true) - { - return lifecycle::Event::default(); - } } _ => (), } + if lifecycle_action_blocked_by_replication(event.action, oi) { + return lifecycle::Event::default(); + } + event } +pub(crate) async fn lifecycle_delete_all_versions_blocked_by_replication( + api: Arc, + bucket: &str, + object: &str, + action: IlmAction, +) -> Result { + if !action.delete_all() { + return Ok(false); + } + + let mut marker = None; + let mut version_marker = None; + loop { + let page = api + .clone() + .list_object_versions(bucket, object, marker.clone(), version_marker.clone(), None, 1000) + .await?; + + match lifecycle_delete_all_versions_replication_scan(object, &page.objects) { + VersionReplicationScan::Blocked => return Ok(true), + VersionReplicationScan::Done => return Ok(false), + VersionReplicationScan::Continue => {} + } + + if !page.is_truncated { + return Ok(false); + } + + marker = page.next_marker; + version_marker = page.next_version_idmarker; + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum VersionReplicationScan { + Blocked, + Done, + Continue, +} + +fn lifecycle_delete_all_versions_replication_scan(object: &str, versions: &[ObjectInfo]) -> VersionReplicationScan { + for version in versions { + let name = version.name.as_str(); + if name == object { + if lifecycle_replication_blocks_action(version) { + return VersionReplicationScan::Blocked; + } + continue; + } + if name > object { + return VersionReplicationScan::Done; + } + } + VersionReplicationScan::Continue +} + +fn lifecycle_action_blocked_by_replication(action: IlmAction, oi: &ObjectInfo) -> bool { + replication_sink::lifecycle_action_waits_for_replication(action) && lifecycle_replication_blocks_action(oi) +} + +fn lifecycle_replication_blocks_action(oi: &ObjectInfo) -> bool { + replication_sink::replication_status_blocks_lifecycle(&oi.replication_status) || oi.version_purge_status.is_pending() +} + pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { if oi.delete_marker || oi.is_dir { return false; @@ -2911,8 +2971,10 @@ mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, StaleMultipartUploadCandidate, TransitionState, TransitionedObject, - cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, - enqueue_recovered_free_version_with_state, lifecycle_deleted_object, lifecycle_rule_has_date_expiration, + VersionReplicationScan, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + enqueue_recovered_free_version_with_state, enqueue_transition_with_lifecycle, eval_action_from_lifecycle, + lifecycle_action_blocked_by_replication, lifecycle_delete_all_versions_replication_scan, lifecycle_deleted_object, + lifecycle_replication_blocks_action, lifecycle_rule_has_date_expiration, 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, @@ -2939,10 +3001,10 @@ mod tests { use futures::FutureExt; use rustfs_common::metrics::{IlmAction, global_metrics}; use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX; - use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType}; + use rustfs_filemeta::{ReplicateDecision, ReplicationStatusType, VersionPurgeStatusType}; use s3s::dto::{ BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry, OutputLocation, - RestoreRequest, RestoreRequestType, S3Location, Timestamp, + RestoreRequest, RestoreRequestType, S3Location, Timestamp, Transition, TransitionStorageClass, }; use serial_test::serial; use sha2::{Digest, Sha256}; @@ -3861,6 +3923,253 @@ mod tests { assert_eq!(state.replicate_decision_str, oi.replication_decision); } + fn expired_delete_marker_lifecycle() -> BucketLifecycleConfiguration { + BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + expired_object_delete_marker: Some(true), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expired-marker".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + } + } + + fn latest_expiration_lifecycle() -> BucketLifecycleConfiguration { + BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(1), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expire-current".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + } + } + + fn latest_transition_lifecycle() -> BucketLifecycleConfiguration { + 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-current".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: Some(vec![Transition { + days: Some(1), + date: None, + storage_class: Some(TransitionStorageClass::from_static("WARM")), + }]), + }], + } + } + + fn delete_marker_object( + replication_status: ReplicationStatusType, + version_purge_status: VersionPurgeStatusType, + ) -> ObjectInfo { + ObjectInfo { + bucket: "bucket".to_string(), + name: "logs/object".to_string(), + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp")), + version_id: Some(Uuid::new_v4()), + is_latest: true, + delete_marker: true, + num_versions: 1, + replication_status, + version_purge_status, + ..Default::default() + } + } + + fn current_object(replication_status: ReplicationStatusType) -> ObjectInfo { + ObjectInfo { + bucket: "bucket".to_string(), + name: "logs/object".to_string(), + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp")), + version_id: Some(Uuid::new_v4()), + is_latest: true, + num_versions: 1, + replication_status, + ..Default::default() + } + } + + #[test] + fn lifecycle_replication_blocks_only_pending_failed_or_pending_purge() { + assert!(lifecycle_replication_blocks_action(&delete_marker_object( + ReplicationStatusType::Pending, + VersionPurgeStatusType::default(), + ))); + assert!(lifecycle_replication_blocks_action(&delete_marker_object( + ReplicationStatusType::Completed, + VersionPurgeStatusType::Failed, + ))); + assert!(!lifecycle_replication_blocks_action(&delete_marker_object( + ReplicationStatusType::Completed, + VersionPurgeStatusType::Complete, + ))); + assert!(!lifecycle_replication_blocks_action(&delete_marker_object( + ReplicationStatusType::Empty, + VersionPurgeStatusType::Empty, + ))); + } + + #[test] + fn lifecycle_action_replication_guard_requires_waiting_action_and_pending_state() { + let pending = current_object(ReplicationStatusType::Pending); + let completed = current_object(ReplicationStatusType::Completed); + + assert!(lifecycle_action_blocked_by_replication(IlmAction::TransitionAction, &pending)); + assert!(lifecycle_action_blocked_by_replication(IlmAction::DeleteAction, &pending)); + assert!(!lifecycle_action_blocked_by_replication(IlmAction::NoneAction, &pending)); + assert!(!lifecycle_action_blocked_by_replication(IlmAction::TransitionAction, &completed)); + } + + #[test] + fn delete_all_version_replication_scan_stops_after_exact_object_key() { + let completed = ObjectInfo { + name: "a".to_string(), + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }; + let pending_exact = ObjectInfo { + name: "a".to_string(), + replication_status: ReplicationStatusType::Pending, + ..Default::default() + }; + let pending_child = ObjectInfo { + name: "a/child".to_string(), + replication_status: ReplicationStatusType::Pending, + ..Default::default() + }; + let later_key = ObjectInfo { + name: "ab".to_string(), + replication_status: ReplicationStatusType::Pending, + ..Default::default() + }; + + assert_eq!( + lifecycle_delete_all_versions_replication_scan("a", &[pending_exact]), + VersionReplicationScan::Blocked + ); + assert_eq!( + lifecycle_delete_all_versions_replication_scan("a", &[completed.clone(), pending_child]), + VersionReplicationScan::Done + ); + assert_eq!( + lifecycle_delete_all_versions_replication_scan("a", &[completed]), + VersionReplicationScan::Continue + ); + assert_eq!( + lifecycle_delete_all_versions_replication_scan("a", &[later_key]), + VersionReplicationScan::Done + ); + } + + #[tokio::test] + async fn enqueue_transition_with_lifecycle_skips_transition_while_replication_pending() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Pending); + + let queued = enqueue_transition_with_lifecycle(&object, &lc, &LcEventSrc::Scanner).await; + + assert!(!queued); + } + + #[tokio::test] + async fn existing_object_lifecycle_allows_expired_marker_after_replication_completed() { + let lc = expired_delete_marker_lifecycle(); + let object = delete_marker_object(ReplicationStatusType::Completed, VersionPurgeStatusType::Complete); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::DeleteVersionAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_skips_expired_marker_while_replication_pending() { + let lc = expired_delete_marker_lifecycle(); + let object = delete_marker_object(ReplicationStatusType::Pending, VersionPurgeStatusType::default()); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_skips_current_expiration_while_replication_pending() { + let lc = latest_expiration_lifecycle(); + let object = current_object(ReplicationStatusType::Pending); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_skips_current_expiration_while_replication_pending_without_config() { + let lc = latest_expiration_lifecycle(); + let object = current_object(ReplicationStatusType::Pending); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_allows_current_expiration_after_replication_completed() { + let lc = latest_expiration_lifecycle(); + let object = current_object(ReplicationStatusType::Completed); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::DeleteAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_skips_transition_while_replication_pending() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Pending); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn existing_object_lifecycle_allows_transition_after_replication_completed() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Completed); + + let event = eval_action_from_lifecycle(&lc, None, &object).await; + + assert_eq!(event.action, IlmAction::TransitionAction); + } + static STALE_MULTIPART_TEST_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); async fn setup_test_env() -> (Vec, Arc) { diff --git a/crates/ecstore/src/bucket/lifecycle/evaluator.rs b/crates/ecstore/src/bucket/lifecycle/evaluator.rs index 406a674eb..30f8d2315 100644 --- a/crates/ecstore/src/bucket/lifecycle/evaluator.rs +++ b/crates/ecstore/src/bucket/lifecycle/evaluator.rs @@ -28,11 +28,10 @@ const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; const EVENT_LIFECYCLE_VERSION_SCAN_SKIPPED: &str = "lifecycle_version_scan_skipped"; /// Evaluator - evaluates lifecycle policy on objects for the given lifecycle -/// configuration, lock retention configuration and replication configuration. +/// configuration and lock retention configuration. pub struct Evaluator { policy: Arc, lock_retention: Option>, - repl_cfg: Option>, } impl Evaluator { @@ -41,7 +40,6 @@ impl Evaluator { Self { policy, lock_retention: None, - repl_cfg: None, } } @@ -51,17 +49,19 @@ impl Evaluator { self } - /// WithReplicationConfig - sets the replication configuration for the evaluator - pub fn with_replication_config(mut self, rcfg: Option>) -> Self { - self.repl_cfg = rcfg; + /// WithReplicationConfig is retained for caller compatibility. + /// Lifecycle replication guards are evaluated from per-object replication state. + pub fn with_replication_config(self, _rcfg: Option>) -> Self { self } /// IsPendingReplication checks if the object is pending replication. pub fn is_pending_replication(&self, obj: &ObjectOpts) -> bool { - self.repl_cfg - .as_ref() - .is_some_and(|rcfg| replication_sink::has_pending_version_purge(rcfg, obj)) + replication_sink::has_pending_lifecycle_replication(obj) + } + + fn any_version_has_pending_replication(&self, objs: &[ObjectOpts]) -> bool { + objs.iter().any(|obj| self.is_pending_replication(obj)) } /// IsObjectLocked checks if it is appropriate to remove an @@ -89,6 +89,9 @@ impl Evaluator { 'top_loop: { for (i, obj) in objs.iter().enumerate() { let mut event = self.policy.eval_inner(obj, now, newer_noncurrent_versions).await; + if replication_sink::lifecycle_action_waits_for_replication(event.action) && self.is_pending_replication(obj) { + event = Event::default(); + } match event.action { IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => { // Skip if bucket has object locking enabled; To prevent the @@ -98,7 +101,8 @@ impl Evaluator { v.object_lock_enabled .as_ref() .is_some_and(|v| v.as_str() == ObjectLockEnabled::ENABLED) - }) { + }) || self.any_version_has_pending_replication(objs) + { event = Event::default(); } else { // No need to evaluate remaining versions' lifecycle @@ -125,10 +129,6 @@ impl Evaluator { if self.is_object_locked(obj) { event = Event::default(); } - - if self.is_pending_replication(obj) { - event = Event::default(); - } } _ => {} } @@ -164,3 +164,254 @@ impl Evaluator { Ok(self.eval_inner(objs, OffsetDateTime::now_utc()).await) } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use rustfs_common::metrics::IlmAction; + use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType}; + use s3s::dto::{ + BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, Transition, TransitionStorageClass, + }; + use time::OffsetDateTime; + use uuid::Uuid; + + use super::*; + fn expired_marker_lifecycle() -> Arc { + Arc::new(BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + expired_object_delete_marker: Some(true), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expired-marker".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }) + } + + fn latest_expiration_lifecycle() -> Arc { + Arc::new(BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(1), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("expire-current".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }) + } + + fn latest_transition_lifecycle() -> Arc { + 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("transition-current".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: Some(vec![Transition { + days: Some(1), + date: None, + storage_class: Some(TransitionStorageClass::from_static("WARM")), + }]), + }], + }) + } + + fn all_versions_expiration_lifecycle() -> Arc { + Arc::new(BucketLifecycleConfiguration { + expiry_updated_at: None, + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(1), + expired_object_all_versions: Some(true), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + del_marker_expiration: None, + filter: None, + id: Some("delete-all".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }) + } + + fn object_opts(replication_status: ReplicationStatusType, version_purge_status: VersionPurgeStatusType) -> ObjectOpts { + ObjectOpts { + name: "logs/object".to_string(), + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp")), + version_id: Some(Uuid::new_v4()), + is_latest: true, + delete_marker: true, + num_versions: 1, + replication_status, + version_purge_status, + ..Default::default() + } + } + + fn current_object_opts(replication_status: ReplicationStatusType) -> ObjectOpts { + ObjectOpts { + name: "logs/object".to_string(), + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp")), + version_id: Some(Uuid::new_v4()), + is_latest: true, + num_versions: 1, + replication_status, + ..Default::default() + } + } + + fn versioned_object_opts(replication_status: ReplicationStatusType, is_latest: bool) -> ObjectOpts { + ObjectOpts { + num_versions: 2, + is_latest, + ..current_object_opts(replication_status) + } + } + + #[tokio::test] + async fn evaluator_allows_expired_delete_marker_after_replication_completed() { + let evaluator = Evaluator::new(expired_marker_lifecycle()); + + let events = evaluator + .eval(&[object_opts( + ReplicationStatusType::Completed, + VersionPurgeStatusType::Complete, + )]) + .await + .expect("completed replication should allow lifecycle evaluation"); + + assert_eq!(events[0].action, IlmAction::DeleteVersionAction); + } + + #[tokio::test] + async fn evaluator_skips_expired_delete_marker_while_replication_pending() { + let evaluator = Evaluator::new(expired_marker_lifecycle()); + + let events = evaluator + .eval(&[object_opts(ReplicationStatusType::Pending, VersionPurgeStatusType::default())]) + .await + .expect("pending replication should still return a lifecycle decision"); + + assert_eq!(events[0].action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn evaluator_skips_expired_delete_marker_while_version_purge_pending() { + let evaluator = Evaluator::new(expired_marker_lifecycle()); + + let events = evaluator + .eval(&[object_opts(ReplicationStatusType::Completed, VersionPurgeStatusType::Pending)]) + .await + .expect("pending version purge should still return a lifecycle decision"); + + assert_eq!(events[0].action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn evaluator_skips_latest_expiration_while_replication_failed() { + let evaluator = Evaluator::new(latest_expiration_lifecycle()); + + let events = evaluator + .eval(&[current_object_opts(ReplicationStatusType::Failed)]) + .await + .expect("failed replication should still return a lifecycle decision"); + + assert_eq!(events[0].action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn evaluator_allows_latest_expiration_after_replication_completed() { + let evaluator = Evaluator::new(latest_expiration_lifecycle()); + + let events = evaluator + .eval(&[current_object_opts(ReplicationStatusType::Completed)]) + .await + .expect("completed replication should allow latest expiration"); + + assert_eq!(events[0].action, IlmAction::DeleteAction); + } + + #[tokio::test] + async fn evaluator_skips_transition_while_replication_pending() { + let evaluator = Evaluator::new(latest_transition_lifecycle()); + + let events = evaluator + .eval(&[current_object_opts(ReplicationStatusType::Pending)]) + .await + .expect("pending replication should still return a lifecycle decision"); + + assert_eq!(events[0].action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn evaluator_allows_transition_after_replication_completed() { + let evaluator = Evaluator::new(latest_transition_lifecycle()); + + let events = evaluator + .eval(&[current_object_opts(ReplicationStatusType::Completed)]) + .await + .expect("completed replication should allow transition"); + + assert_eq!(events[0].action, IlmAction::TransitionAction); + } + + #[tokio::test] + async fn evaluator_skips_delete_all_versions_when_any_version_replication_pending() { + let evaluator = Evaluator::new(all_versions_expiration_lifecycle()); + let latest = versioned_object_opts(ReplicationStatusType::Completed, true); + let noncurrent = versioned_object_opts(ReplicationStatusType::Pending, false); + + let events = evaluator + .eval(&[latest, noncurrent]) + .await + .expect("pending noncurrent replication should still return lifecycle decisions"); + + assert_eq!(events[0].action, IlmAction::NoneAction); + assert_eq!(events[1].action, IlmAction::NoneAction); + } + + #[tokio::test] + async fn evaluator_allows_delete_all_versions_when_all_versions_replication_completed() { + let evaluator = Evaluator::new(all_versions_expiration_lifecycle()); + let latest = versioned_object_opts(ReplicationStatusType::Completed, true); + let noncurrent = versioned_object_opts(ReplicationStatusType::Completed, false); + + let events = evaluator + .eval(&[latest, noncurrent]) + .await + .expect("completed replication should allow delete-all lifecycle decision"); + + assert_eq!(events[0].action, IlmAction::DeleteAllVersionsAction); + assert_eq!(events[1].action, IlmAction::NoneAction); + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs index 1c5bbea77..24c1d8ab7 100644 --- a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs +++ b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_filemeta::ReplicateDecision; -use s3s::dto::ReplicationConfiguration; +use rustfs_common::metrics::IlmAction; +use rustfs_filemeta::{ReplicateDecision, ReplicationStatusType}; use crate::bucket::lifecycle::lifecycle::ObjectOpts; use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; @@ -22,12 +22,34 @@ use crate::storage_api_contracts::object::{DeletedObject, ObjectToDelete}; pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig; -pub(crate) fn new_replication_config(config: ReplicationConfiguration) -> LifecycleReplicationConfig { - ReplicationLifecycleBridge::new_config(config) +pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool { + obj.version_purge_status.is_pending() } -pub(crate) fn has_pending_version_purge(config: &LifecycleReplicationConfig, obj: &ObjectOpts) -> bool { - ReplicationLifecycleBridge::has_pending_version_purge(config, obj.name.as_str(), !obj.version_purge_status.is_empty()) +pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool { + replication_status_blocks_lifecycle(&obj.replication_status) +} + +pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool { + has_pending_object_replication(obj) || has_pending_version_purge(obj) +} + +pub(crate) fn replication_status_blocks_lifecycle(status: &ReplicationStatusType) -> bool { + matches!(status, ReplicationStatusType::Pending | ReplicationStatusType::Failed) +} + +pub(crate) fn lifecycle_action_waits_for_replication(action: IlmAction) -> bool { + matches!( + action, + IlmAction::DeleteAction + | IlmAction::DeleteVersionAction + | IlmAction::DeleteRestoredAction + | IlmAction::DeleteRestoredVersionAction + | IlmAction::DeleteAllVersionsAction + | IlmAction::DelMarkerDeleteAllVersionsAction + | IlmAction::TransitionAction + | IlmAction::TransitionVersionAction + ) } pub(crate) async fn check_delete_replication( @@ -42,3 +64,75 @@ pub(crate) async fn check_delete_replication( pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject) { ReplicationLifecycleBridge::schedule_delete(bucket, delete_object).await; } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use rustfs_common::metrics::IlmAction; + use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType}; + + use super::*; + + fn object_opts(version_purge_status: VersionPurgeStatusType) -> ObjectOpts { + ObjectOpts { + name: "logs/object".to_string(), + user_tags: String::new(), + mod_time: None, + size: 0, + version_id: None, + is_latest: true, + delete_marker: false, + num_versions: 1, + successor_mod_time: None, + transition_status: String::new(), + restore_ongoing: false, + restore_expires: None, + versioned: true, + version_suspended: false, + user_defined: HashMap::new(), + version_purge_status, + replication_status: ReplicationStatusType::default(), + } + } + + fn object_opts_with_replication_status(replication_status: ReplicationStatusType) -> ObjectOpts { + ObjectOpts { + replication_status, + ..object_opts(VersionPurgeStatusType::default()) + } + } + + #[test] + fn pending_version_purge_blocks_lifecycle_actions_from_object_state() { + assert!(has_pending_version_purge(&object_opts(VersionPurgeStatusType::Pending))); + assert!(has_pending_version_purge(&object_opts(VersionPurgeStatusType::Failed))); + assert!(!has_pending_version_purge(&object_opts(VersionPurgeStatusType::Complete))); + assert!(!has_pending_version_purge(&object_opts(VersionPurgeStatusType::default()))); + } + + #[test] + fn pending_replication_status_blocks_lifecycle_actions_from_object_state() { + assert!(has_pending_object_replication(&object_opts_with_replication_status( + ReplicationStatusType::Pending + ))); + assert!(has_pending_object_replication(&object_opts_with_replication_status( + ReplicationStatusType::Failed + ))); + assert!(!has_pending_object_replication(&object_opts_with_replication_status( + ReplicationStatusType::Completed + ))); + assert!(!has_pending_object_replication(&object_opts_with_replication_status( + ReplicationStatusType::Empty + ))); + } + + #[test] + fn lifecycle_action_waits_for_replication_for_expiration_and_transition() { + assert!(lifecycle_action_waits_for_replication(IlmAction::DeleteAction)); + assert!(lifecycle_action_waits_for_replication(IlmAction::DeleteVersionAction)); + assert!(lifecycle_action_waits_for_replication(IlmAction::TransitionAction)); + assert!(lifecycle_action_waits_for_replication(IlmAction::TransitionVersionAction)); + assert!(!lifecycle_action_waits_for_replication(IlmAction::NoneAction)); + } +} diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index aa867b1b0..23df12641 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -18,6 +18,7 @@ use crate::bucket::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule, eval_action_from_lifecycle, + lifecycle_delete_all_versions_blocked_by_replication, }, lifecycle::IlmAction, }, @@ -2155,7 +2156,6 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( version: &rustfs_filemeta::FileInfo, lifecycle_config: Option<&BucketLifecycleConfiguration>, lock_retention: Option, - replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>, apply_actions: bool, event_source: &LcEventSrc, ) -> Result { @@ -2165,7 +2165,7 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( let versioned = BucketVersioningSys::prefix_enabled(bucket, &version.name).await; let object_info = crate::object_api::ObjectInfo::from_file_info(version, bucket, &version.name, versioned); - let event = eval_action_from_lifecycle(lifecycle_config, lock_retention, replication_config, &object_info).await; + let event = eval_action_from_lifecycle(lifecycle_config, lock_retention, &object_info).await; match event.action { IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => { @@ -2175,6 +2175,9 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( Ok(false) } action if lifecycle_action_removes_data_movement_version(action) => { + if lifecycle_delete_all_versions_blocked_by_replication(store.clone(), bucket, &object_info.name, action).await? { + return Ok(false); + } let applied = !apply_actions || apply_expiry_rule(&event, event_source, &object_info).await; resolve_data_movement_lifecycle_expiry_result(action, apply_actions, applied) } @@ -2661,7 +2664,6 @@ impl ECStore { version, lifecycle_config.as_ref(), lock_retention.clone(), - replication_config.clone(), true, &LcEventSrc::Decom, ) @@ -3920,15 +3922,9 @@ impl ECStore { for bucket_info in &buckets { let mut lifecycle_config = None; let mut lock_retention = None; - let mut replication_config = None; if bucket_info.name != RUSTFS_META_BUCKET { lifecycle_config = runtime_sources::bucket_lifecycle_config(&bucket_info.name).await; lock_retention = BucketObjectLockSys::get(&bucket_info.name).await; - replication_config = resolve_decommission_optional_bucket_config_result( - &bucket_info.name, - "replication", - metadata_sys::get_replication_config(&bucket_info.name).await, - )?; } let versions_found = Arc::new(AtomicUsize::new(0)); @@ -3939,7 +3935,6 @@ impl ECStore { let bucket_name = bucket_info.name.clone(); let lifecycle_config_cb = lifecycle_config.clone(); let lock_retention_cb = lock_retention.clone(); - let replication_config_cb = replication_config.clone(); let store = Arc::clone(self); let callback_rx_cb = callback_rx.clone(); @@ -3949,7 +3944,6 @@ impl ECStore { let bucket_name = bucket_name.clone(); let lifecycle_config = lifecycle_config_cb.clone(); let lock_retention = lock_retention_cb.clone(); - let replication_config = replication_config_cb.clone(); let store = Arc::clone(&store); let callback_rx = callback_rx_cb.clone(); Box::pin(async move { @@ -3992,7 +3986,6 @@ impl ECStore { version, lifecycle_config.as_ref(), lock_retention.clone(), - replication_config.clone(), false, &LcEventSrc::Decom, ) diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index a484a6bbc..f585e52bd 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -119,7 +119,6 @@ impl ECStore { version, bucket_configs.lifecycle_config.as_ref(), bucket_configs.lock_retention.clone(), - bucket_configs.replication_config.clone(), true, &crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Rebal, )