fix(replication): persist force-delete handoff state (#5641)

* fix(replication): persist force-delete handoff state

* fix(arch): route force-delete config access through boundary

* style: format force-delete imports

---------

Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/replication/src/config.rs

@@ -23,7 +23,7 @@ use s3s::dto::{
     ReplicationRuleStatus, ReplicationRules,
 };
 use serde::{Deserialize, Serialize};
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use uuid::Uuid;
 
 pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1;
@@ -93,6 +93,7 @@ pub trait ReplicationConfigurationExt {
     fn get_destination(&self) -> Destination;
     fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
     fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String>;
+    fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec<String>;
     fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> {
         self.filter_target_arns(obj)
             .into_iter()
@@ -434,6 +435,51 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
         arns
     }
 
+    fn filter_force_delete_target_arns(&self, prefix: &str) -> Vec<String> {
+        let role = self.role.trim();
+        let mut selected = BTreeMap::<String, (&ReplicationRule, bool)>::new();
+
+        for rule in &self.rules {
+            if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
+                continue;
+            }
+
+            let rule_prefix = rule.prefix();
+            if !prefix.starts_with(rule_prefix) && !rule_prefix.starts_with(prefix) {
+                continue;
+            }
+
+            let target = if role.is_empty() {
+                rule.destination.bucket.trim()
+            } else {
+                role
+            };
+            if target.is_empty() {
+                continue;
+            }
+
+            let delete_enabled =
+                rule.delete_replication.as_ref().is_some_and(|delete| {
+                    delete.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED)
+                }) || rule.delete_marker_replication.as_ref().is_some_and(|delete_marker| {
+                    delete_marker.status
+                        == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
+                });
+
+            if selected
+                .get(target)
+                .is_none_or(|(current, _)| rule.priority > current.priority)
+            {
+                selected.insert(target.to_string(), (rule, delete_enabled));
+            }
+        }
+
+        selected
+            .into_iter()
+            .filter_map(|(target, (_, enabled))| enabled.then_some(target))
+            .collect()
+    }
+
     fn filter_target_replication_decisions(&self, obj: &ObjectOpts) -> Vec<(String, bool)> {
         let rules = self.filter_actionable_rules(obj);
         let role = self.role.trim();
@@ -1069,4 +1115,26 @@ mod tests {
 
         assert_eq!(decisions, vec![(target_a.to_string(), false), (target_b.to_string(), true)]);
     }
+
+    #[test]
+    fn force_delete_targets_use_overlapping_rules_and_highest_priority_switch() {
+        let target_a = "arn:target:a";
+        let target_b = "arn:target:b";
+        let mut a_parent = delete_marker_rule("a-parent", target_a, "logs/", 1, true);
+        a_parent.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        let a_child_disabled = delete_marker_rule("a-child", target_a, "logs/2026/", 5, false);
+        let b_child = delete_marker_rule("b-child", target_b, "logs/2026/", 2, true);
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![a_parent, a_child_disabled, b_child],
+        };
+
+        assert_eq!(
+            config.filter_force_delete_target_arns("logs/2026/app.log"),
+            vec![target_b.to_string()],
+            "the child rule must win for target A while the overlapping child target B remains eligible"
+        );
+    }
 }

crates/replication/src/delete.rs

@@ -33,12 +33,15 @@ impl DeletedObjectReplicationInfo {
             return vec![self.target_arn.clone()];
         }
 
-        let mut target_arns = self
-            .delete_object
-            .replication_state
-            .as_ref()
-            .map(admitted_target_arns_from_replication_state)
-            .unwrap_or_default();
+        let mut target_arns = if !self.delete_object.force_delete_target_arns.is_empty() {
+            self.delete_object.force_delete_target_arns.clone()
+        } else {
+            self.delete_object
+                .replication_state
+                .as_ref()
+                .map(admitted_target_arns_from_replication_state)
+                .unwrap_or_default()
+        };
         target_arns.sort();
         target_arns.dedup();
         target_arns
@@ -69,6 +72,9 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
                 .delete_marker_mtime
                 .and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()),
             target_arns: self.admitted_target_arns(),
+            force_delete_id: self.delete_object.force_delete_id,
+            force_delete_generation: self.delete_object.force_delete_generation,
+            force_delete_local_commit: self.delete_object.force_delete,
         }
     }
 
@@ -231,6 +237,35 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deleted_object_replication_info_preserves_force_delete_handoff() {
+        let operation_id = Uuid::new_v4();
+        let info = DeletedObjectReplicationInfo {
+            bucket: "bucket".to_string(),
+            delete_object: DeletedObject {
+                object_name: "prefix/".to_string(),
+                force_delete: true,
+                force_delete_id: Some(operation_id),
+                force_delete_target_arns: vec![
+                    "arn:target-b".to_string(),
+                    "arn:target-a".to_string(),
+                    "arn:target-b".to_string(),
+                ],
+                force_delete_generation: Some(17),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let entry = info.to_mrf_entry();
+
+        assert!(entry.force_delete);
+        assert_eq!(entry.force_delete_id, Some(operation_id));
+        assert_eq!(entry.force_delete_generation, Some(17));
+        assert!(entry.force_delete_local_commit);
+        assert_eq!(entry.target_arns, vec!["arn:target-a", "arn:target-b"]);
+    }
+
     #[test]
     fn version_delete_replication_tracks_delete_marker_version_purge() {
         let dobj = DeletedObject {

crates/replication/src/filemeta.rs

@@ -572,7 +572,7 @@ impl MrfOpKind {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
 pub struct MrfReplicateEntry {
     #[serde(rename = "bucket")]
     pub bucket: String,
@@ -619,6 +619,17 @@ pub struct MrfReplicateEntry {
 
     #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)]
     pub target_arns: Vec<String>,
+
+    // Force-delete entries use the target ARN list above as their immutable target set.
+    // The id distinguishes a durable intent from legacy MRF delete entries.
+    #[serde(rename = "forceDeleteID", skip_serializing_if = "Option::is_none", default)]
+    pub force_delete_id: Option<Uuid>,
+    #[serde(rename = "forceDeleteGeneration", skip_serializing_if = "Option::is_none", default)]
+    pub force_delete_generation: Option<i64>,
+
+    // Replay is allowed only after the source-side recursive delete has committed.
+    #[serde(rename = "forceDeleteLocalCommit", default)]
+    pub force_delete_local_commit: bool,
 }
 
 fn retry_count_to_mrf(retry_count: u32) -> i32 {
@@ -865,6 +876,7 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo {
             delete_marker: false,
             delete_marker_mtime: None,
             target_arns: self.admitted_target_arns(),
+            ..Default::default()
         }
     }
 
@@ -945,6 +957,7 @@ impl ReplicateObjectInfo {
             delete_marker: false,
             delete_marker_mtime: None,
             target_arns: self.admitted_target_arns(),
+            ..Default::default()
         }
     }
 }

crates/replication/src/mrf.rs

@@ -68,6 +68,9 @@ mod tests {
                 retry_count: 1,
                 size: 1024,
                 op: MrfOpKind::Metadata,
+                force_delete_id: None,
+                force_delete_generation: None,
+                force_delete_local_commit: false,
                 force_delete: false,
                 delete_marker_version_id: None,
                 delete_marker: false,
@@ -81,6 +84,9 @@ mod tests {
                 retry_count: 2,
                 size: 1024,
                 op: MrfOpKind::Object,
+                force_delete_id: None,
+                force_delete_generation: None,
+                force_delete_local_commit: false,
                 force_delete: false,
                 delete_marker_version_id: None,
                 delete_marker: false,
@@ -94,6 +100,9 @@ mod tests {
                 retry_count: 0,
                 size: 0,
                 op: MrfOpKind::Delete,
+                force_delete_id: None,
+                force_delete_generation: None,
+                force_delete_local_commit: false,
                 force_delete: true,
                 delete_marker_version_id: Some(del_vid),
                 delete_marker: true,

crates/replication/src/storage_api.rs

@@ -49,6 +49,9 @@ pub struct DeletedObject {
     pub replication_state: Option<ReplicationState>,
     pub found: bool,
     pub force_delete: bool,
+    pub force_delete_id: Option<Uuid>,
+    pub force_delete_target_arns: Vec<String>,
+    pub force_delete_generation: Option<i64>,
 }
 
 impl DeletedObject {