fix(replication): snapshot existing object admission targets (#5634)

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

@@ -17,7 +17,7 @@ pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
     REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos,
     ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state,
-    parse_replicate_decision, target_reset_header, version_purge_statuses_map,
+    parse_replicate_decision, replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map,
 };
 pub use rustfs_replication::{
     REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType,

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

@@ -16,8 +16,8 @@ use super::replication_config_store::ReplicationConfigStore;
 use super::replication_error_boundary::Error as EcstoreError;
 use super::replication_filemeta_boundary::{
     MrfOpKind, MrfReplicateEntry, REPLICATE_HEAL_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedTargetInfo,
-    ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, replication_statuses_map,
-    version_purge_statuses_map,
+    ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, replicate_decision_for_admitted_targets,
+    replication_statuses_map, version_purge_statuses_map,
 };
 use super::replication_lock_boundary::ReplicationLockTiming;
 use super::replication_logging::{EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION};
@@ -990,21 +990,25 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                             delete_marker: entry.delete_marker,
                             ..Default::default()
                         };
-                        let dsc = check_replicate_delete(
-                            &entry.bucket,
-                            &ObjectToDelete {
-                                object_name: entry.object.clone(),
-                                version_id: entry.version_id,
-                                ..Default::default()
-                            },
-                            &oi,
-                            &ObjectOptions {
-                                versioned,
-                                ..Default::default()
-                            },
-                            None,
-                        )
-                        .await;
+                        let dsc = if entry.target_arns.is_empty() {
+                            check_replicate_delete(
+                                &entry.bucket,
+                                &ObjectToDelete {
+                                    object_name: entry.object.clone(),
+                                    version_id: entry.version_id,
+                                    ..Default::default()
+                                },
+                                &oi,
+                                &ObjectOptions {
+                                    versioned,
+                                    ..Default::default()
+                                },
+                                None,
+                            )
+                            .await
+                        } else {
+                            replicate_decision_for_admitted_targets(&entry.target_arns)
+                        };
                         let mut rstate = oi.replication_state();
                         rstate.replicate_decision_str = dsc.to_string();
 
@@ -1036,7 +1040,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                         schedule_replication_delete(dv).await;
                         queued_count += 1;
                     }
-                    MrfOpKind::Object => {
+                    MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
                         let opts = ObjectOptions {
                             version_id: entry.version_id.map(|u| u.to_string()),
                             ..Default::default()
@@ -1055,9 +1059,16 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 continue;
                             }
                         };
-                        // Route through queue_replication_heal so the replication decision (dsc)
-                        // is computed from the live config — required for replicate_object.
-                        queue_replication_heal(&entry.bucket, oi, entry.retry_count as u32).await;
+                        if entry.target_arns.is_empty() {
+                            // Legacy entries predate target admission persistence. They cannot
+                            // be safely attributed, so retain the old live-config fallback.
+                            queue_replication_heal(&entry.bucket, oi, entry.retry_count as u32).await;
+                        } else if let Some(pool) = runtime_sources::replication_pool() {
+                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+                            let mut roi = replicate_object_info_from_object_info(oi, dsc, entry.op.replication_type());
+                            roi.retry_count = entry.retry_count.max(0) as u32;
+                            let _ = pool.queue_replica_task(roi).await;
+                        }
                         queued_count += 1;
                     }
                     MrfOpKind::Metadata => {

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

@@ -792,6 +792,7 @@ impl ReplicationResyncer {
             let storage = storage.clone();
             let results_tx = results_tx.clone();
             let bucket_name = opts.bucket.clone();
+            let target_arn = opts.arn.clone();
 
             let f = tokio::spawn(async move {
                 while let Some(mut roi) = rx.recv().await {
@@ -819,6 +820,7 @@ impl ReplicationResyncer {
                             bucket: roi.bucket.clone(),
                             event_type: REPLICATE_EXISTING_DELETE.to_string(),
                             op_type: ReplicationType::ExistingObject,
+                            target_arn: target_arn.clone(),
                             ..Default::default()
                         };
                         replicate_delete(doi, storage.clone()).await;
@@ -2009,10 +2011,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
     let bucket = roi.bucket.clone();
     let object = roi.name.clone();
 
-    // The admission decision is the target-granular contract. Re-evaluating the
-    // live config here could fan a synchronous request out to targets that were
-    // not admitted, or promote an async target after a mixed-mode split.
-    let tgt_arns = roi.dsc.replicate_target_arns();
+    let tgt_arns = roi.admitted_target_arns();
 
     // Acquire a per-object namespace lock so that at most one worker (across all cluster
     // nodes and MRF retry goroutines) replicates this object version at a time.

crates/replication/src/filemeta.rs

@@ -552,10 +552,26 @@ pub enum MrfOpKind {
     Object,
     #[serde(rename = "metadata")]
     Metadata,
+    #[serde(rename = "heal")]
+    Heal,
+    #[serde(rename = "existingObject")]
+    ExistingObject,
     #[serde(rename = "delete")]
     Delete,
 }
 
+impl MrfOpKind {
+    pub fn replication_type(self) -> ReplicationType {
+        match self {
+            Self::Object => ReplicationType::Object,
+            Self::Metadata => ReplicationType::Metadata,
+            Self::Heal => ReplicationType::Heal,
+            Self::ExistingObject => ReplicationType::ExistingObject,
+            Self::Delete => ReplicationType::Delete,
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct MrfReplicateEntry {
     #[serde(rename = "bucket")]
@@ -838,16 +854,17 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo {
             version_id: self.version_id,
             retry_count: retry_count_to_mrf(self.retry_count),
             size: self.size,
-            op: if self.op_type == ReplicationType::Metadata {
-                MrfOpKind::Metadata
-            } else {
-                MrfOpKind::Object
+            op: match self.op_type {
+                ReplicationType::Metadata => MrfOpKind::Metadata,
+                ReplicationType::Heal => MrfOpKind::Heal,
+                ReplicationType::ExistingObject => MrfOpKind::ExistingObject,
+                _ => MrfOpKind::Object,
             },
             force_delete: false,
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: self.dsc.replicate_target_arns(),
+            target_arns: self.admitted_target_arns(),
         }
     }
 
@@ -878,6 +895,25 @@ static REPL_STATUS_REGEX: LazyLock<Regex> = LazyLock::new(|| match Regex::new(r"
 });
 
 impl ReplicateObjectInfo {
+    /// Returns the target set captured when this queued operation was admitted.
+    /// Resync decisions are more specific than the general heal decision and must
+    /// win for ExistingObject work.
+    pub fn admitted_target_arns(&self) -> Vec<String> {
+        if self.op_type == ReplicationType::ExistingObject && !self.existing_obj_resync.is_empty() {
+            let mut arns = self
+                .existing_obj_resync
+                .targets
+                .iter()
+                .filter(|(_, decision)| decision.replicate)
+                .map(|(arn, _)| arn.clone())
+                .collect::<Vec<_>>();
+            arns.sort();
+            arns.dedup();
+            return arns;
+        }
+        self.dsc.replicate_target_arns()
+    }
+
     /// Returns replication status of a target
     pub fn target_replication_status(&self, arn: &str) -> ReplicationStatusType {
         let binding = self.replication_status_internal.clone().unwrap_or_default();
@@ -898,20 +934,31 @@ impl ReplicateObjectInfo {
             version_id: self.version_id,
             retry_count: retry_count_to_mrf(self.retry_count),
             size: self.size,
-            op: if self.op_type == ReplicationType::Metadata {
-                MrfOpKind::Metadata
-            } else {
-                MrfOpKind::Object
+            op: match self.op_type {
+                ReplicationType::Metadata => MrfOpKind::Metadata,
+                ReplicationType::Heal => MrfOpKind::Heal,
+                ReplicationType::ExistingObject => MrfOpKind::ExistingObject,
+                _ => MrfOpKind::Object,
             },
             force_delete: false,
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: self.dsc.replicate_target_arns(),
+            target_arns: self.admitted_target_arns(),
         }
     }
 }
 
+pub fn replicate_decision_for_admitted_targets(target_arns: &[String]) -> ReplicateDecision {
+    let mut decision = ReplicateDecision::new();
+    for arn in target_arns {
+        if !arn.is_empty() {
+            decision.set(ReplicateTargetDecision::new(arn.clone(), true, false));
+        }
+    }
+    decision
+}
+
 // constructs a replication status map from string representation
 pub fn replication_statuses_map(s: &str) -> HashMap<String, ReplicationStatusType> {
     let mut targets = HashMap::new();
@@ -1143,6 +1190,36 @@ mod tests {
         assert_eq!(info.to_mrf_entry().op, MrfOpKind::Metadata);
     }
 
+    #[test]
+    fn admission_snapshot_prefers_resync_targets_for_existing_objects() {
+        let mut decision = ReplicateDecision::new();
+        decision.set(ReplicateTargetDecision::new("arn:live".to_string(), true, false));
+        let mut resync = ResyncDecision::new();
+        resync.targets.insert(
+            "arn:admitted".to_string(),
+            ResyncTargetDecision {
+                replicate: true,
+                reset_id: "reset-1".to_string(),
+                ..Default::default()
+            },
+        );
+        let info = ReplicateObjectInfo {
+            op_type: ReplicationType::ExistingObject,
+            dsc: decision,
+            existing_obj_resync: resync,
+            ..Default::default()
+        };
+
+        assert_eq!(info.admitted_target_arns(), vec!["arn:admitted".to_string()]);
+        assert_eq!(info.to_mrf_entry().op, MrfOpKind::ExistingObject);
+    }
+
+    #[test]
+    fn mrf_operation_kind_round_trips_heal_and_existing_object_intent() {
+        assert_eq!(MrfOpKind::Heal.replication_type(), ReplicationType::Heal);
+        assert_eq!(MrfOpKind::ExistingObject.replication_type(), ReplicationType::ExistingObject);
+    }
+
     #[test]
     fn target_state_reads_resync_timestamp_from_target_reset_header_key() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";

crates/replication/src/lib.rs

@@ -44,8 +44,8 @@ pub use filemeta::{
     REPLICATE_INCOMING_DELETE, REPLICATE_MRF, REPLICATE_QUEUED, REPLICATION_RESET, REPLICATION_STATUS, ReplicateDecision,
     ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState,
     ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision,
-    VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header,
-    version_purge_statuses_map,
+    VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replicate_decision_for_admitted_targets,
+    replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file};
 pub use multipart::{