fix(lifecycle): correct delete replication fanout (#2609)

Co-authored-by: loverustfs <hello@rustfs.com>

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

@@ -1354,7 +1354,7 @@ pub async fn expire_transitioned_object(
         }
     };
 
-    schedule_lifecycle_replication_delete_if_needed(oi).await;
+    schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await;
 
     //defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn)
 
@@ -1778,7 +1778,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
             return false;
         }
     };
-    schedule_lifecycle_replication_delete_if_needed(oi).await;
+    schedule_lifecycle_replication_delete_if_needed(oi, &dobj).await;
     //debug!("dobj: {:?}", dobj);
     if dobj.name.is_empty() {
         dobj = oi.clone();
@@ -1819,25 +1819,55 @@ pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &
     true
 }
 
-async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo) {
+fn lifecycle_deleted_object(oi: &ObjectInfo, dobj: &ObjectInfo) -> crate::store_api::DeletedObject {
-    if !oi.delete_marker || oi.version_id.is_none() {
+    if dobj.delete_marker {
-        return;
+        return crate::store_api::DeletedObject {
+            object_name: oi.name.clone(),
+            delete_marker: true,
+            delete_marker_version_id: dobj.version_id,
+            delete_marker_mtime: dobj.mod_time.or(oi.mod_time),
+            ..Default::default()
+        };
     }
 
-    let replication_state = lifecycle_delete_replication_state(oi).await;
+    if oi.delete_marker && oi.version_id.is_some() {
+        return crate::store_api::DeletedObject {
+            object_name: oi.name.clone(),
+            delete_marker: false,
+            delete_marker_version_id: oi.version_id,
+            delete_marker_mtime: oi.mod_time,
+            ..Default::default()
+        };
+    }
+
+    crate::store_api::DeletedObject {
+        object_name: oi.name.clone(),
+        delete_marker: false,
+        version_id: oi.version_id,
+        delete_marker_mtime: oi.mod_time,
+        ..Default::default()
+    }
+}
+
+async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo, dobj: &ObjectInfo) {
+    let mut delete_object = lifecycle_deleted_object(oi, dobj);
+    let version_id = if delete_object.delete_marker {
+        None
+    } else if delete_object.delete_marker_version_id.is_some() {
+        delete_object.delete_marker_version_id
+    } else {
+        delete_object.version_id
+    };
+
+    let replication_state = lifecycle_delete_replication_state(oi, version_id).await;
     if replication_state.is_none() {
         return;
     }
 
+    delete_object.replication_state = replication_state;
+
     schedule_replication_delete(DeletedObjectReplicationInfo {
-        delete_object: crate::store_api::DeletedObject {
+        delete_object,
-            object_name: oi.name.clone(),
-            delete_marker_version_id: oi.version_id,
-            delete_marker: false,
-            delete_marker_mtime: oi.mod_time,
-            replication_state,
-            ..Default::default()
-        },
         bucket: oi.bucket.clone(),
         event_type: REPLICATE_INCOMING_DELETE.to_string(),
         ..Default::default()
@@ -1845,21 +1875,56 @@ async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo) {
     .await;
 }
 
-async fn lifecycle_delete_replication_state(oi: &ObjectInfo) -> Option<ReplicationState> {
+fn should_reuse_lifecycle_delete_replication_state(oi: &ObjectInfo, version_delete: bool) -> bool {
-    if !oi.replication_decision.is_empty() || oi.version_purge_status == VersionPurgeStatusType::Pending {
+    let state = oi.replication_state();
+    if version_delete {
+        oi.version_purge_status == VersionPurgeStatusType::Pending && !state.purge_targets.is_empty()
+    } else {
+        oi.replication_status == rustfs_filemeta::ReplicationStatusType::Pending && !state.targets.is_empty()
+    }
+}
+
+fn lifecycle_version_purge_state_from_completed_targets(oi: &ObjectInfo) -> Option<ReplicationState> {
+    if oi.replication_status != rustfs_filemeta::ReplicationStatusType::Completed {
+        return None;
+    }
+
+    let targets = oi.replication_state().targets;
+    if targets.is_empty() {
+        return None;
+    }
+
+    let pending_status = targets.keys().map(|arn| format!("{arn}=PENDING;")).collect::<String>();
+
+    Some(ReplicationState {
+        replicate_decision_str: oi.replication_decision.clone(),
+        version_purge_status_internal: Some(pending_status.clone()),
+        purge_targets: rustfs_filemeta::version_purge_statuses_map(&pending_status),
+        ..Default::default()
+    })
+}
+
+async fn lifecycle_delete_replication_state(oi: &ObjectInfo, version_id: Option<Uuid>) -> Option<ReplicationState> {
+    if should_reuse_lifecycle_delete_replication_state(oi, version_id.is_some()) {
         return Some(oi.replication_state());
     }
 
+    if version_id.is_some()
+        && let Some(state) = lifecycle_version_purge_state_from_completed_targets(oi)
+    {
+        return Some(state);
+    }
+
     let dsc = check_replicate_delete(
         &oi.bucket,
         &ObjectToDelete {
             object_name: oi.name.clone(),
-            version_id: oi.version_id,
+            version_id,
             ..Default::default()
         },
         oi,
         &ObjectOptions {
-            version_id: oi.version_id.map(|v| v.to_string()),
+            version_id: version_id.map(|v| v.to_string()),
             versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await,
             ..Default::default()
         },
@@ -1870,17 +1935,23 @@ async fn lifecycle_delete_replication_state(oi: &ObjectInfo) -> Option<Replicati
         return None;
     }
 
-    Some(replication_state_for_version_delete(dsc))
+    Some(replication_state_for_delete(dsc, version_id.is_some()))
 }
 
-fn replication_state_for_version_delete(dsc: ReplicateDecision) -> ReplicationState {
+fn replication_state_for_delete(dsc: ReplicateDecision, version_delete: bool) -> ReplicationState {
     let pending_status = dsc.pending_status();
-    ReplicationState {
+    let mut state = ReplicationState {
         replicate_decision_str: dsc.to_string(),
-        version_purge_status_internal: pending_status.clone(),
-        purge_targets: rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()),
         ..Default::default()
+    };
+    if version_delete {
+        state.version_purge_status_internal = pending_status.clone();
+        state.purge_targets = rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default());
+    } else {
+        state.replication_status_internal = pending_status.clone();
+        state.targets = rustfs_filemeta::replication_statuses_map(pending_status.as_deref().unwrap_or_default());
     }
+    state
 }
 
 pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
@@ -1906,7 +1977,9 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc,
 mod tests {
     use super::{
         StaleMultipartUploadCandidate, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at,
-        mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate,
+        lifecycle_deleted_object, lifecycle_version_purge_state_from_completed_targets,
+        mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete,
+        should_reuse_lifecycle_delete_replication_state,
     };
     use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG;
     use crate::bucket::metadata_sys;
@@ -1917,8 +1990,9 @@ mod tests {
     use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY};
     use crate::store::ECStore;
     use crate::store_api::{
-        BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectOptions, PutObjReader,
+        BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectInfo, ObjectOptions, PutObjReader,
     };
+    use rustfs_filemeta::{ReplicateDecision, VersionPurgeStatusType};
     use serial_test::serial;
     use sha2::{Digest, Sha256};
     use std::collections::HashMap;
@@ -1960,6 +2034,136 @@ mod tests {
         assert!(opts.skip_decommissioned);
     }
 
+    #[test]
+    fn lifecycle_deleted_object_uses_delete_marker_created_by_expiry() {
+        let source = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "key".to_string(),
+            ..Default::default()
+        };
+        let delete_result = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "key".to_string(),
+            delete_marker: true,
+            version_id: Some(Uuid::new_v4()),
+            mod_time: Some(OffsetDateTime::now_utc()),
+            ..Default::default()
+        };
+
+        let deleted = lifecycle_deleted_object(&source, &delete_result);
+
+        assert!(deleted.delete_marker);
+        assert_eq!(deleted.delete_marker_version_id, delete_result.version_id);
+        assert_eq!(deleted.version_id, None);
+        assert_eq!(deleted.object_name, "key");
+    }
+
+    #[test]
+    fn lifecycle_deleted_object_uses_version_id_for_noncurrent_version_purge() {
+        let version_id = Uuid::new_v4();
+        let source = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "key".to_string(),
+            version_id: Some(version_id),
+            ..Default::default()
+        };
+
+        let deleted = lifecycle_deleted_object(&source, &ObjectInfo::default());
+
+        assert!(!deleted.delete_marker);
+        assert_eq!(deleted.version_id, Some(version_id));
+        assert_eq!(deleted.delete_marker_version_id, None);
+    }
+
+    #[test]
+    fn lifecycle_deleted_object_uses_delete_marker_version_for_marker_purge() {
+        let version_id = Uuid::new_v4();
+        let source = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "key".to_string(),
+            delete_marker: true,
+            version_id: Some(version_id),
+            ..Default::default()
+        };
+
+        let deleted = lifecycle_deleted_object(&source, &ObjectInfo::default());
+
+        assert!(!deleted.delete_marker);
+        assert_eq!(deleted.delete_marker_version_id, Some(version_id));
+        assert_eq!(deleted.version_id, None);
+    }
+
+    #[test]
+    fn replication_state_for_delete_uses_replication_targets_for_current_delete() {
+        let arn = "arn:aws:s3:::target-bucket";
+        let mut dsc = ReplicateDecision::default();
+        dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false));
+
+        let state = replication_state_for_delete(dsc, false);
+
+        assert_eq!(state.replication_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str()));
+        assert!(state.version_purge_status_internal.is_none());
+        assert!(state.targets.contains_key(arn));
+    }
+
+    #[test]
+    fn replication_state_for_delete_uses_purge_targets_for_version_delete() {
+        let arn = "arn:aws:s3:::target-bucket";
+        let mut dsc = ReplicateDecision::default();
+        dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false));
+
+        let state = replication_state_for_delete(dsc, true);
+
+        assert_eq!(state.version_purge_status_internal.as_deref(), Some(format!("{arn}=PENDING;").as_str()));
+        assert!(state.replication_status_internal.is_none());
+        assert!(state.purge_targets.contains_key(arn));
+    }
+
+    #[test]
+    fn lifecycle_delete_replication_state_reuses_only_pending_version_purge_state() {
+        let oi = ObjectInfo {
+            version_purge_status: VersionPurgeStatusType::Pending,
+            version_purge_status_internal: Some("arn:aws:s3:::target=PENDING;".to_string()),
+            replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
+            ..Default::default()
+        };
+
+        assert!(should_reuse_lifecycle_delete_replication_state(&oi, true));
+        assert!(!should_reuse_lifecycle_delete_replication_state(&oi, false));
+    }
+
+    #[test]
+    fn lifecycle_delete_replication_state_does_not_reuse_put_replication_for_version_delete() {
+        let oi = ObjectInfo {
+            replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
+            replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()),
+            replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
+            ..Default::default()
+        };
+
+        assert!(
+            !should_reuse_lifecycle_delete_replication_state(&oi, true),
+            "version purges must not reuse plain object replication state from prior PUT/delete-marker replication"
+        );
+    }
+
+    #[test]
+    fn lifecycle_version_purge_state_from_completed_targets_derives_pending_purge_targets() {
+        let oi = ObjectInfo {
+            replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
+            replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()),
+            replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
+            ..Default::default()
+        };
+
+        let state = lifecycle_version_purge_state_from_completed_targets(&oi)
+            .expect("completed replication targets should be convertible into version-purge targets");
+
+        assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;"));
+        assert!(state.purge_targets.contains_key("arn:aws:s3:::target"));
+        assert_eq!(state.replicate_decision_str, oi.replication_decision);
+    }
+
     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/replication/config.rs

@@ -152,6 +152,10 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
             }
 
             if obj.op_type == ReplicationType::Delete {
+                if !rule.metadata_replicate(obj) {
+                    return false;
+                }
+
                 if obj.version_id.is_some() {
                     if obj.delete_marker {
                         return rule.delete_marker_replication.clone().is_some_and(|d| {

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -85,7 +85,7 @@ use tokio::task::JoinSet;
 use tokio::time::Duration as TokioDuration;
 use tokio_util::io::ReaderStream;
 use tokio_util::sync::CancellationToken;
-use tracing::{error, info, instrument, warn};
+use tracing::{debug, error, info, instrument, warn};
 use uuid::Uuid;
 
 pub(crate) const REPLICATION_DIR: &str = ".replication";
@@ -1272,15 +1272,7 @@ pub async fn check_replicate_delete(
         return ReplicateDecision::default();
     }
 
-    let opts = ObjectOpts {
+    let opts = delete_replication_object_opts(dobj, oi);
-        name: dobj.object_name.clone(),
-        ssec: is_ssec_encrypted(&oi.user_defined),
-        user_tags: oi.user_tags.clone(),
-        delete_marker: oi.delete_marker,
-        version_id: dobj.version_id,
-        op_type: ReplicationType::Delete,
-        ..Default::default()
-    };
 
     let tgt_arns = rcfg.filter_target_arns(&opts);
     let mut dsc = ReplicateDecision::new();
@@ -1332,6 +1324,19 @@ pub async fn check_replicate_delete(
     dsc
 }
 
+fn delete_replication_object_opts(dobj: &ObjectToDelete, oi: &ObjectInfo) -> ObjectOpts {
+    ObjectOpts {
+        name: dobj.object_name.clone(),
+        ssec: is_ssec_encrypted(&oi.user_defined),
+        user_tags: oi.user_tags.clone(),
+        delete_marker: oi.delete_marker,
+        version_id: dobj.version_id,
+        op_type: ReplicationType::Delete,
+        replica: oi.replication_status == ReplicationStatusType::Replica,
+        ..Default::default()
+    }
+}
+
 /// Check if the user-defined metadata contains SSEC encryption headers
 fn is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
     user_defined.contains_key(SSEC_ALGORITHM_HEADER)
@@ -1703,7 +1708,7 @@ pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo,
 
     let is_version_purge = is_version_delete_replication(&dobj.delete_object);
 
-    if !is_version_purge && dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
+    if should_retry_delete_marker_purge(&dobj.delete_object) {
         let bucket_clone = bucket.clone();
         let dobj_clone = dobj.clone();
         let dsc_clone = dsc.clone();
@@ -2061,6 +2066,10 @@ fn is_version_delete_replication(dobj: &DeletedObject) -> bool {
     dobj.version_id.is_some() || (dobj.delete_marker_version_id.is_some() && !dobj.delete_marker)
 }
 
+fn should_retry_delete_marker_purge(dobj: &DeletedObject) -> bool {
+    dobj.delete_marker_version_id.is_some()
+}
+
 fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Option<&str>) -> bool {
     !is_not_found && !matches!(code, Some("MethodNotAllowed" | "405"))
 }
@@ -2152,6 +2161,14 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
         .await
     {
         Ok(_) => {
+            debug!(
+                bucket = tgt_client.bucket,
+                object = dobj.delete_object.object_name,
+                version_id = ?version_id,
+                delete_marker = dobj.delete_object.delete_marker,
+                is_version_purge,
+                "replicate_delete_to_target succeeded"
+            );
             if !is_version_purge {
                 rinfo.replication_status = ReplicationStatusType::Completed;
             } else {
@@ -2159,6 +2176,15 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
             }
         }
         Err(e) => {
+            warn!(
+                bucket = tgt_client.bucket,
+                object = dobj.delete_object.object_name,
+                version_id = ?version_id,
+                delete_marker = dobj.delete_object.delete_marker,
+                is_version_purge,
+                error = %e,
+                "replicate_delete_to_target failed"
+            );
             rinfo.error = Some(e.to_string());
             if !is_version_purge {
                 rinfo.replication_status = ReplicationStatusType::Failed;
@@ -3603,6 +3629,49 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_delete_replication_object_opts_marks_replica_deletes() {
+        let dobj = ObjectToDelete {
+            object_name: "obj".to_string(),
+            version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+        let oi = ObjectInfo {
+            bucket: "b".to_string(),
+            name: "obj".to_string(),
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+
+        let opts = delete_replication_object_opts(&dobj, &oi);
+
+        assert!(
+            opts.replica,
+            "replica deletes must preserve replica status for downstream ReplicaModifications rules"
+        );
+        assert_eq!(opts.version_id, dobj.version_id);
+        assert_eq!(opts.name, dobj.object_name);
+        assert_eq!(opts.op_type, ReplicationType::Delete);
+    }
+
+    #[test]
+    fn test_delete_replication_object_opts_keeps_non_replica_deletes_local() {
+        let dobj = ObjectToDelete {
+            object_name: "obj".to_string(),
+            ..Default::default()
+        };
+        let oi = ObjectInfo {
+            bucket: "b".to_string(),
+            name: "obj".to_string(),
+            replication_status: ReplicationStatusType::Completed,
+            ..Default::default()
+        };
+
+        let opts = delete_replication_object_opts(&dobj, &oi);
+
+        assert!(!opts.replica, "source-originated deletes should not be treated as replica modifications");
+    }
+
     #[test]
     fn test_is_version_delete_replication_for_delete_marker_version_purge() {
         let dobj = DeletedObject {
@@ -3631,6 +3700,34 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_should_retry_delete_marker_purge_for_version_purge() {
+        let dobj = DeletedObject {
+            delete_marker: false,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(
+            should_retry_delete_marker_purge(&dobj),
+            "delete-marker version purge should schedule delayed target cleanup in case the target marker arrives late"
+        );
+    }
+
+    #[test]
+    fn test_should_retry_delete_marker_purge_for_delete_marker_creation() {
+        let dobj = DeletedObject {
+            delete_marker: true,
+            delete_marker_version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert!(
+            should_retry_delete_marker_purge(&dobj),
+            "delete-marker creation should keep the late-arrival cleanup path so downstream purges can catch up"
+        );
+    }
+
     #[test]
     fn test_is_retryable_delete_replication_head_error_allows_delete_marker_head_responses() {
         assert!(

crates/ecstore/src/client/object_handlers_common.rs

@@ -16,12 +16,26 @@ use std::sync::Arc;
 use tracing::warn;
 
 use crate::bucket::lifecycle::lifecycle;
+use crate::bucket::replication::{DeletedObjectReplicationInfo, check_replicate_delete, schedule_replication_delete};
 use crate::bucket::versioning::VersioningApi;
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::store::ECStore;
 use crate::store_api::{ObjectOperations, ObjectOptions, ObjectToDelete};
+use rustfs_filemeta::{REPLICATE_INCOMING_DELETE, ReplicationState, version_purge_statuses_map};
 use rustfs_lock::MAX_DELETE_LIST;
 
+fn lifecycle_version_delete_replication_state(
+    replicate_decision_str: String,
+    pending_status: Option<String>,
+) -> ReplicationState {
+    ReplicationState {
+        replicate_decision_str,
+        version_purge_status_internal: pending_status.clone(),
+        purge_targets: version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()),
+        ..Default::default()
+    }
+}
+
 pub async fn delete_object_versions(api: &Arc<ECStore>, bucket: &str, to_del: &[ObjectToDelete], _lc_event: lifecycle::Event) {
     let version_suspended = match BucketVersioningSys::get(bucket).await {
         Ok(vc) => vc.suspended(),
@@ -39,7 +53,37 @@ pub async fn delete_object_versions(api: &Arc<ECStore>, bucket: &str, to_del: &[
         } else {
             remaining = &[];
         }
-        let (_deleted_objs, errors) = api
+
+        let mut replication_candidates: Vec<Option<ReplicationState>> = Vec::with_capacity(to_del.len());
+        for object in to_del.iter() {
+            let version_id = object.version_id.map(|vid| vid.to_string());
+            let opts = ObjectOptions {
+                version_id: version_id.clone(),
+                versioned: true,
+                version_suspended,
+                ..Default::default()
+            };
+            let candidate = match api.get_object_info(bucket, &object.object_name, &opts).await {
+                Ok(info) => {
+                    let dsc = check_replicate_delete(bucket, object, &info, &opts, None).await;
+                    dsc.replicate_any()
+                        .then(|| lifecycle_version_delete_replication_state(dsc.to_string(), dsc.pending_status()))
+                }
+                Err(err) => {
+                    warn!(
+                        bucket,
+                        object = %object.object_name,
+                        version_id = ?version_id,
+                        error = ?err,
+                        "failed to get object info during lifecycle noncurrent version cleanup; skipping delete replication scheduling"
+                    );
+                    None
+                }
+            };
+            replication_candidates.push(candidate);
+        }
+
+        let (mut deleted_objs, errors) = api
             .delete_objects(
                 bucket,
                 to_del.to_vec(),
@@ -49,6 +93,24 @@ pub async fn delete_object_versions(api: &Arc<ECStore>, bucket: &str, to_del: &[
                 },
             )
             .await;
+
+        for (i, deleted_obj) in deleted_objs.iter_mut().enumerate() {
+            if errors.get(i).and_then(|err| err.as_ref()).is_some() {
+                continue;
+            }
+            let Some(replication_state) = replication_candidates.get(i).and_then(|c| c.clone()) else {
+                continue;
+            };
+            deleted_obj.replication_state = Some(replication_state);
+            schedule_replication_delete(DeletedObjectReplicationInfo {
+                delete_object: deleted_obj.clone(),
+                bucket: bucket.to_string(),
+                event_type: REPLICATE_INCOMING_DELETE.to_string(),
+                ..Default::default()
+            })
+            .await;
+        }
+
         for (i, err) in errors.iter().enumerate() {
             if let Some(e) = err {
                 let obj_name = to_del.get(i).map(|o| o.object_name.as_str()).unwrap_or("<unknown>");
@@ -65,3 +127,20 @@ pub async fn delete_object_versions(api: &Arc<ECStore>, bucket: &str, to_del: &[
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::lifecycle_version_delete_replication_state;
+
+    #[test]
+    fn lifecycle_version_delete_replication_state_tracks_pending_purge_targets() {
+        let state = lifecycle_version_delete_replication_state(
+            "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
+            Some("arn:aws:s3:::target=PENDING;".to_string()),
+        );
+
+        assert_eq!(state.version_purge_status_internal.as_deref(), Some("arn:aws:s3:::target=PENDING;"));
+        assert!(state.purge_targets.contains_key("arn:aws:s3:::target"));
+        assert_eq!(state.replicate_decision_str, "arn:aws:s3:::target=true;false;arn:aws:s3:::target;");
+    }
+}

rustfs/src/app/object_usecase.rs

@@ -515,7 +515,8 @@ async fn enrich_delete_replication_state_if_needed(
     let Some(replication_state) = delete_object.replication_state.as_ref() else {
         return;
     };
-    if !replication_state.replicate_decision_str.is_empty()
+    if obj_info.replication_status != ReplicationStatusType::Replica
+        && !replication_state.replicate_decision_str.is_empty()
         && (!replication_state.targets.is_empty() || !replication_state.purge_targets.is_empty())
     {
         return;
@@ -550,12 +551,61 @@ fn should_schedule_delete_replication(
         return false;
     }
 
+    if opts.version_id.is_some() && !deleted_delete_marker_version && !replication_source.delete_marker {
+        return matches!(
+            replication_source.replication_status,
+            ReplicationStatusType::Replica
+                | ReplicationStatusType::Pending
+                | ReplicationStatusType::Completed
+                | ReplicationStatusType::Failed
+        );
+    }
+
     replication_source.replication_status == ReplicationStatusType::Replica
         || replication_source.replication_status == ReplicationStatusType::Pending
         || replication_source.version_purge_status == VersionPurgeStatusType::Pending
         || (deleted_delete_marker_version && replication_source.replication_status == ReplicationStatusType::Completed)
 }
 
+async fn should_schedule_replica_delete_replication(
+    bucket: &str,
+    replication_source: &ObjectInfo,
+    version_id: Option<Uuid>,
+) -> bool {
+    let Ok((config, _)) = metadata_sys::get_replication_config(bucket).await else {
+        return false;
+    };
+
+    delete_replication_state_from_config(&config, replication_source, version_id, true).is_some()
+}
+
+fn delete_replication_version_id(replication_source: &ObjectInfo, deleted_delete_marker_version: bool) -> Option<Uuid> {
+    if replication_source.delete_marker && !deleted_delete_marker_version {
+        None
+    } else {
+        replication_source.version_id
+    }
+}
+
+fn should_use_existing_delete_replication_info(opts: &ObjectOptions) -> bool {
+    opts.version_id.is_some() && !opts.delete_marker
+}
+
+fn delete_replication_state_source<'a>(
+    opts: &ObjectOptions,
+    existing_object_info: Option<&'a ObjectInfo>,
+    deleted_object_info: &'a ObjectInfo,
+) -> &'a ObjectInfo {
+    if opts.replication_request
+        && deleted_object_info.delete_marker
+        && let Some(existing) = existing_object_info
+    {
+        return existing;
+    }
+
+    deleted_object_info
+}
+
 const AMZ_SNOWBALL_EXTRACT_COMPAT: &str = "X-Amz-Snowball-Auto-Extract";
 #[cfg(test)]
 const AMZ_SNOWBALL_PREFIX_INTERNAL: &str = "X-Amz-Meta-Rustfs-Snowball-Prefix";
@@ -3242,34 +3292,45 @@ impl DefaultObjectUsecase {
             return result;
         }
 
-        let deleted_replication_info = existing_object_info.as_ref().filter(|_| opts.version_id.is_some());
+        let deleted_replication_info = existing_object_info
-        let replication_source = deleted_replication_info.unwrap_or(&obj_info);
+            .as_ref()
+            .filter(|_| should_use_existing_delete_replication_info(&opts));
+        let deleted_object_source = deleted_replication_info.unwrap_or(&obj_info);
+        let replication_state_source =
+            delete_replication_state_source(&opts, existing_object_info.as_ref(), deleted_object_source);
         let deleted_delete_marker_version = deleted_replication_info.is_some_and(|info| info.delete_marker);
 
-        if should_schedule_delete_replication(&opts, replication_source, deleted_delete_marker_version) {
+        let delete_replication_version_id = delete_replication_version_id(deleted_object_source, deleted_delete_marker_version);
+        let schedule_delete_replication = if opts.replication_request && replica {
+            should_schedule_replica_delete_replication(&bucket, replication_state_source, delete_replication_version_id).await
+        } else {
+            should_schedule_delete_replication(&opts, deleted_object_source, deleted_delete_marker_version)
+        };
+
+        if schedule_delete_replication {
             let mut deleted_object = DeletedObjectReplicationInfo {
                 delete_object: rustfs_ecstore::store_api::DeletedObject {
-                    delete_marker: replication_source.delete_marker && !deleted_delete_marker_version,
+                    delete_marker: deleted_object_source.delete_marker && !deleted_delete_marker_version,
-                    delete_marker_version_id: if replication_source.delete_marker {
+                    delete_marker_version_id: if deleted_object_source.delete_marker {
-                        replication_source.version_id
+                        deleted_object_source.version_id
                     } else {
                         None
                     },
                     object_name: key.clone(),
-                    version_id: if replication_source.delete_marker {
+                    version_id: if deleted_object_source.delete_marker {
                         None
                     } else {
-                        replication_source.version_id
+                        deleted_object_source.version_id
                     },
-                    delete_marker_mtime: replication_source.mod_time,
+                    delete_marker_mtime: deleted_object_source.mod_time,
-                    replication_state: Some(replication_source.replication_state()),
+                    replication_state: Some(replication_state_source.replication_state()),
                     ..Default::default()
                 },
                 bucket: bucket.clone(),
                 event_type: REPLICATE_INCOMING_DELETE.to_string(),
                 ..Default::default()
             };
-            enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object.delete_object, replication_source).await;
+            enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object.delete_object, replication_state_source).await;
             schedule_replication_delete(deleted_object).await;
         }
 
@@ -4802,6 +4863,25 @@ mod tests {
         );
     }
 
+    #[test]
+    fn should_schedule_delete_replication_keeps_object_version_purge_from_completed_source() {
+        let opts = ObjectOptions {
+            replication_request: false,
+            version_id: Some(Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        let replication_source = ObjectInfo {
+            delete_marker: false,
+            replication_status: ReplicationStatusType::Completed,
+            ..Default::default()
+        };
+
+        assert!(
+            should_schedule_delete_replication(&opts, &replication_source, false),
+            "source-side object version purge must still enqueue delete replication after the original PUT completed"
+        );
+    }
+
     #[tokio::test]
     async fn execute_get_object_attributes_returns_internal_error_when_store_uninitialized() {
         let input = GetObjectAttributesInput::builder()
@@ -4990,7 +5070,12 @@ mod tests {
                 id: Some("rule-1".to_string()),
                 prefix: Some("test/".to_string()),
                 priority: Some(1),
-                source_selection_criteria: None,
+                source_selection_criteria: Some(SourceSelectionCriteria {
+                    replica_modifications: Some(ReplicaModifications {
+                        status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
+                    }),
+                    sse_kms_encrypted_objects: None,
+                }),
                 status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
             }],
         };
@@ -5011,6 +5096,45 @@ mod tests {
         assert!(state.targets.contains_key(&arn));
     }
 
+    #[test]
+    fn delete_replication_state_from_config_skips_replica_delete_without_replica_modifications() {
+        let arn = "arn:aws:s3:::target-bucket".to_string();
+        let config = ReplicationConfiguration {
+            role: arn.clone(),
+            rules: vec![ReplicationRule {
+                delete_marker_replication: Some(DeleteMarkerReplication {
+                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+                }),
+                delete_replication: None,
+                destination: Destination {
+                    bucket: arn,
+                    ..Default::default()
+                },
+                existing_object_replication: Some(ExistingObjectReplication {
+                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+                }),
+                filter: None,
+                id: Some("rule-1".to_string()),
+                prefix: Some("test/".to_string()),
+                priority: Some(1),
+                source_selection_criteria: None,
+                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            }],
+        };
+        let obj_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "test/object.txt".to_string(),
+            delete_marker: true,
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+
+        assert!(
+            delete_replication_state_from_config(&config, &obj_info, None, true).is_none(),
+            "replica deletes must only fan out when ReplicaModifications are enabled"
+        );
+    }
+
     #[test]
     fn delete_replication_state_from_config_tracks_delete_marker_version_purges() {
         let arn = "arn:aws:s3:::target-bucket".to_string();
@@ -5053,4 +5177,139 @@ mod tests {
         assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};"));
         assert!(state.purge_targets.contains_key(&arn));
     }
+
+    #[test]
+    fn delete_replication_state_source_prefers_existing_replica_for_replication_delete_marker_creation() {
+        let opts = ObjectOptions {
+            replication_request: true,
+            version_id: Some(Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        let existing = ObjectInfo {
+            name: "test/object.txt".to_string(),
+            replication_status: ReplicationStatusType::Completed,
+            ..Default::default()
+        };
+        let deleted = ObjectInfo {
+            name: "test/object.txt".to_string(),
+            delete_marker: true,
+            ..Default::default()
+        };
+
+        let source = delete_replication_state_source(&opts, Some(&existing), &deleted);
+
+        assert_eq!(source.replication_status, ReplicationStatusType::Completed);
+        assert!(
+            !source.delete_marker,
+            "downstream fanout should inherit replica identity from the pre-delete object"
+        );
+    }
+
+    #[test]
+    fn delete_replication_state_source_keeps_deleted_marker_for_non_replication_requests() {
+        let opts = ObjectOptions::default();
+        let existing = ObjectInfo {
+            name: "test/object.txt".to_string(),
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+        let deleted = ObjectInfo {
+            name: "test/object.txt".to_string(),
+            delete_marker: true,
+            ..Default::default()
+        };
+
+        let source = delete_replication_state_source(&opts, Some(&existing), &deleted);
+
+        assert!(
+            source.delete_marker,
+            "source-originated deletes should keep using the new delete marker state"
+        );
+    }
+
+    #[test]
+    fn replica_delete_enrichment_must_not_reuse_upstream_targets() {
+        let delete_object = rustfs_ecstore::store_api::DeletedObject {
+            replication_state: Some(ReplicationState {
+                replicate_decision_str: "arn:aws:s3:::upstream=true;false;arn:aws:s3:::upstream;".to_string(),
+                replication_status_internal: Some("arn:aws:s3:::upstream=COMPLETED;".to_string()),
+                targets: replication_statuses_map("arn:aws:s3:::upstream=COMPLETED;"),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let obj_info = ObjectInfo {
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+
+        let should_keep_existing = delete_object.replication_state.as_ref().is_some_and(|state| {
+            obj_info.replication_status != ReplicationStatusType::Replica
+                && !state.replicate_decision_str.is_empty()
+                && (!state.targets.is_empty() || !state.purge_targets.is_empty())
+        });
+
+        assert!(
+            !should_keep_existing,
+            "replica fanout deletes must recompute targets from the local bucket config instead of reusing upstream replication state"
+        );
+    }
+
+    #[test]
+    fn delete_replication_version_id_uses_none_for_delete_marker_creation() {
+        let source = ObjectInfo {
+            delete_marker: true,
+            version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+
+        assert_eq!(
+            delete_replication_version_id(&source, false),
+            None,
+            "delete-marker creation must stay on the delete-marker replication path"
+        );
+    }
+
+    #[test]
+    fn delete_replication_version_id_keeps_version_for_marker_purge() {
+        let version_id = Uuid::new_v4();
+        let source = ObjectInfo {
+            delete_marker: true,
+            version_id: Some(version_id),
+            ..Default::default()
+        };
+
+        assert_eq!(
+            delete_replication_version_id(&source, true),
+            Some(version_id),
+            "delete-marker version purge must preserve the concrete version id for downstream purge replication"
+        );
+    }
+
+    #[test]
+    fn should_use_existing_delete_replication_info_ignores_replication_delete_marker_creation() {
+        let opts = ObjectOptions {
+            version_id: Some(Uuid::new_v4().to_string()),
+            delete_marker: true,
+            ..Default::default()
+        };
+
+        assert!(
+            !should_use_existing_delete_replication_info(&opts),
+            "replicated delete-marker creation carries a source version id header but must not be treated as a version purge"
+        );
+    }
+
+    #[test]
+    fn should_use_existing_delete_replication_info_keeps_version_delete_requests() {
+        let opts = ObjectOptions {
+            version_id: Some(Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+
+        assert!(
+            should_use_existing_delete_replication_info(&opts),
+            "true version-delete requests should keep using the pre-delete object info"
+        );
+    }
 }