fix(replication): retain MRF entries until completion (#5671)

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

@@ -38,7 +38,8 @@ use super::replication_resync_boundary::{
     encode_mrf_file, should_auto_resume_resync,
 };
 use super::replication_resyncer::{
-    ReplicationResyncer, get_heal_replicate_object_info, replicate_delete, replicate_object, save_resync_status,
+    ReplicationResyncer, get_heal_replicate_object_info, replicate_delete, replicate_delete_with_outcome, replicate_object,
+    replicate_object_with_outcome, save_resync_status,
 };
 use super::replication_state::ReplicationStats;
 use super::replication_storage_boundary::{
@@ -1129,7 +1130,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                             let Some(operation_id) = entry.force_delete_id else {
                                 continue;
                             };
-                            schedule_replication_delete(DeletedObjectReplicationInfo {
+                            let delete = DeletedObjectReplicationInfo {
                                 delete_object: ReplicationDeletedObject {
                                     object_name: entry.object.clone(),
                                     force_delete: true,
@@ -1142,8 +1143,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 op_type: ReplicationType::Heal,
                                 event_type: REPLICATE_HEAL_DELETE.to_string(),
                                 ..Default::default()
-                            })
+                            };
-                            .await
+                            if replicate_delete_with_outcome(delete, storage.clone()).await {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
                         } else if entry.force_delete_id.is_some() {
                             ReplicationQueueAdmission::Skipped
                         } else {
@@ -1221,7 +1226,11 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 event_type: REPLICATE_HEAL_DELETE.to_string(),
                                 ..Default::default()
                             };
-                            schedule_replication_delete(dv).await
+                            if replicate_delete_with_outcome(dv, storage.clone()).await {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
                         }
                     }
                     MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
@@ -1250,13 +1259,15 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                             // 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.max(0) as u32).await
-                        } else if let Some(pool) = runtime_sources::replication_pool() {
+                        } else {
                             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;
-                            pool.queue_replica_task(roi).await
+                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-                        } else {
+                                ReplicationQueueAdmission::Queued
-                            ReplicationQueueAdmission::Missed
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
                         }
                     }
                     MrfOpKind::Metadata => {
@@ -1281,7 +1292,18 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 continue;
                             }
                         };
-                        queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
+                        if entry.target_arns.is_empty() {
+                            queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
+                        } else {
+                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+                            let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata);
+                            roi.retry_count = entry.retry_count.max(0) as u32;
+                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
+                        }
                     }
                 };
 

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

@@ -1203,12 +1203,19 @@ pub(crate) async fn save_resync_status<S: ReplicationObjectIO>(
 }
 
 pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
+    let _ = replicate_delete_with_outcome(dobj, storage).await;
+}
+
+pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
+    dobj: DeletedObjectReplicationInfo,
+    storage: Arc<S>,
+) -> bool {
     if dobj.delete_object.force_delete {
-        replicate_force_delete_to_targets(&dobj, storage).await;
+        return replicate_force_delete_to_targets(&dobj, storage).await;
-        return;
     }
 
     let bucket = dobj.bucket.clone();
+    let mut source_state_verified = true;
     let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
         Some(version_id.to_owned())
     } else {
@@ -1245,7 +1252,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                     reason = "source_not_delete_marker",
                     "Skipping stale delete-marker replication"
                 );
-                return;
+                return true;
             }
             Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {
                 debug!(
@@ -1258,9 +1265,10 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                     reason = "source_version_missing",
                     "Skipping stale delete-marker replication"
                 );
-                return;
+                return true;
             }
             Err(err) => {
+                source_state_verified = false;
                 debug!(
                     event = EVENT_REPLICATION_DELETE_SKIPPED,
                     component = LOG_COMPONENT_ECSTORE,
@@ -1310,7 +1318,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return;
+            return false;
         }
     };
     let ns_lock = match storage
@@ -1342,7 +1350,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return;
+            return false;
         }
     };
 
@@ -1372,7 +1380,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return;
+            return false;
         }
     };
 
@@ -1386,6 +1394,11 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
 
     // Process each target
     let target_arns = dobj.admitted_target_arns();
+    let expected_targets = dsc
+        .targets_map
+        .values()
+        .filter(|target| target.replicate && (target_arns.is_empty() || target_arns.iter().any(|arn| arn == &target.arn)))
+        .count();
     for tgt_entry in dsc.targets_map.values() {
         // Skip targets that should not be replicated
         if !tgt_entry.replicate {
@@ -1465,7 +1478,8 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
 
     let is_version_purge = is_version_delete_replication(&dobj.delete_object);
 
-    if should_retry_delete_marker_purge(&dobj.delete_object) {
+    let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object);
+    if requires_delayed_purge {
         let bucket_clone = bucket.clone();
         let dobj_clone = dobj.clone();
         let dsc_clone = dsc.clone();
@@ -1536,7 +1550,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
         EventName::ObjectReplicationFailed.to_string()
     };
 
-    match storage
+    let state_persisted = match storage
         .delete_object(
             &bucket,
             &dobj.delete_object.object_name,
@@ -1558,6 +1572,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                 object,
                 ..Default::default()
             });
+            true
         }
         Err(e) => {
             error!(
@@ -1583,8 +1598,16 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
                 },
                 ..Default::default()
             });
+            false
         }
-    }
+    };
+
+    expected_targets > 0
+        && rinfos.targets.len() == expected_targets
+        && state_persisted
+        && source_state_verified
+        && !requires_delayed_purge
+        && replication_status == ReplicationStatusType::Completed
 }
 
 async fn source_delete_marker_missing<S: EcstoreObjectOperations>(
@@ -1639,7 +1662,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
     }
 }
 
-async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &DeletedObjectReplicationInfo, storage: Arc<S>) {
+async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &DeletedObjectReplicationInfo, storage: Arc<S>) -> bool {
     let bucket = &dobj.bucket;
     let object_name = &dobj.delete_object.object_name;
     let admitted_target_arns = dobj.admitted_target_arns();
@@ -1727,7 +1750,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return;
+            return false;
         }
     };
 
@@ -1755,7 +1778,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return;
+            return false;
         }
     };
 
@@ -1764,6 +1787,9 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     } else {
         admitted_target_arns
     };
+    if tgt_arns.is_empty() {
+        return false;
+    }
 
     let mut join_set = JoinSet::new();
     let mut all_succeeded = true;
@@ -1888,7 +1914,10 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
             error = %error,
             "Force-delete replication completed but durable intent cleanup failed"
         );
+        return false;
     }
+
+    all_succeeded
 }
 
 fn target_delete_version_id(version_id: Uuid, version_purge: bool) -> Option<String> {
@@ -2034,6 +2063,13 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
 }
 
 pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, storage: Arc<S>) -> ReplicationState {
+    replicate_object_with_outcome(roi, storage).await.0
+}
+
+pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
+    roi: ReplicateObjectInfo,
+    storage: Arc<S>,
+) -> (ReplicationState, bool) {
     let bucket = roi.bucket.clone();
     let object = roi.name.clone();
 
@@ -2062,7 +2098,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return roi.replication_state.unwrap_or_default();
+            return (roi.replication_state.unwrap_or_default(), false);
         }
     };
     let _obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
@@ -2085,7 +2121,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return roi.replication_state.unwrap_or_default();
+            return (roi.replication_state.unwrap_or_default(), false);
         }
     };
 
@@ -2166,6 +2202,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
     let replication_status = merged_state.composite_replication_status();
     let new_replication_internal = merged_state.replication_status_internal.clone();
     let mut object_info = roi.to_object_info();
+    let mut state_persisted = true;
 
     if roi.replication_status_internal != new_replication_internal || rinfos.replication_resynced() {
         let mut eval_metadata = HashMap::new();
@@ -2181,6 +2218,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
         match storage.put_object_metadata(&bucket, &object, &popts).await {
             Ok(u) => object_info = u,
             Err(e) => {
+                state_persisted = false;
                 // Persisting the resynced replication status failed. Don't swallow
                 // it silently — the object's on-disk status now disagrees with the
                 // resync result and needs operator visibility (backlog#799 B23).
@@ -2234,7 +2272,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
         }
     }
 
-    merged_state
+    (merged_state, state_persisted)
 }
 
 trait ReplicateObjectInfoExt {

crates/replication/src/queue.rs

@@ -232,7 +232,7 @@ impl ReplicationPriority {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum ReplicationOperation {
     Object(Box<ReplicateObjectInfo>),
     Delete(Box<DeletedObjectReplicationInfo>),