fix(lifecycle): block actions on replication state (#4125)

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,
+    BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, ObjectLockConfiguration, RestoreRequest,
-    RestoreRequest, RestoreRequestType, RestoreStatus, Timestamp,
+    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::<Vec<ObjectOpts>>();
     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<BucketLifecycleConfiguration>,
     lock_config: Option<Arc<ObjectLockConfiguration>>,
-    replication: Option<Arc<replication_sink::LifecycleReplicationConfig>>,
     src: &'a LcEventSrc,
     defer_date_expiry_once: bool,
 }
@@ -2080,7 +2072,6 @@ async fn enqueue_expiry_for_existing_object_group(
         .collect::<Vec<ObjectOpts>>();
     let events = match Evaluator::new(context.lc.clone())
         .with_lock_retention(context.lock_config.clone())
-        .with_replication_config(context.replication.clone())
         .eval(&object_opts)
         .await
     {
@@ -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<ECStore>, 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<ECStore>, 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<ECStore>, 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 {
+        lifecycle::ObjectOpts::from_object_info(self)
-            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()
-        }
     }
 
     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<DefaultRetention>,
-    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<ECStore>,
+    bucket: &str,
+    object: &str,
+    action: IlmAction,
+) -> Result<bool, Error> {
+    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,
+        VersionReplicationScan, 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,
+        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<PathBuf>, Arc<ECStore>)> = OnceLock::new();
 
     async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {

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<BucketLifecycleConfiguration>,
     lock_retention: Option<Arc<ObjectLockConfiguration>>,
-    repl_cfg: Option<Arc<LifecycleReplicationConfig>>,
 }
 
 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
+    /// WithReplicationConfig is retained for caller compatibility.
-    pub fn with_replication_config(mut self, rcfg: Option<Arc<LifecycleReplicationConfig>>) -> Self {
+    /// Lifecycle replication guards are evaluated from per-object replication state.
-        self.repl_cfg = rcfg;
+    pub fn with_replication_config(self, _rcfg: Option<Arc<LifecycleReplicationConfig>>) -> Self {
         self
     }
 
     /// IsPendingReplication checks if the object is pending replication.
     pub fn is_pending_replication(&self, obj: &ObjectOpts) -> bool {
-        self.repl_cfg
+        replication_sink::has_pending_lifecycle_replication(obj)
-            .as_ref()
+    }
-            .is_some_and(|rcfg| replication_sink::has_pending_version_purge(rcfg, 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<BucketLifecycleConfiguration> {
+        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<BucketLifecycleConfiguration> {
+        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<BucketLifecycleConfiguration> {
+        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<BucketLifecycleConfiguration> {
+        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);
+    }
+}

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 rustfs_common::metrics::IlmAction;
-use s3s::dto::ReplicationConfiguration;
+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 {
+pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
-    ReplicationLifecycleBridge::new_config(config)
+    obj.version_purge_status.is_pending()
 }
 
-pub(crate) fn has_pending_version_purge(config: &LifecycleReplicationConfig, obj: &ObjectOpts) -> bool {
+pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
-    ReplicationLifecycleBridge::has_pending_version_purge(config, obj.name.as_str(), !obj.version_purge_status.is_empty())
+    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));
+    }
+}

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<DefaultRetention>,
-    replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
     apply_actions: bool,
     event_source: &LcEventSrc,
 ) -> Result<bool> {
@@ -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,
                             )

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,
             )