From 4310850103d6f3736e27dcdbc8be10922deb82be Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 3 Aug 2026 22:09:24 +0800 Subject: [PATCH] fix(replication): retain MRF entries until completion (#5671) --- .../bucket/replication/replication_pool.rs | 42 ++++++++--- .../replication/replication_resyncer.rs | 70 ++++++++++++++----- crates/replication/src/queue.rs | 2 +- 3 files changed, 87 insertions(+), 27 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index af727334a..a6eca120c 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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 ReplicationPool { 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 ReplicationPool { 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 ReplicationPool { 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 ReplicationPool { // 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 - } else { - ReplicationQueueAdmission::Missed + if replicate_object_with_outcome(roi, storage.clone()).await.1 { + ReplicationQueueAdmission::Queued + } else { + ReplicationQueueAdmission::Missed + } } } MrfOpKind::Metadata => { @@ -1281,7 +1292,18 @@ impl ReplicationPool { 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 + } + } } }; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index b2f070de8..929b9b331 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1203,12 +1203,19 @@ pub(crate) async fn save_resync_status( } pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, storage: Arc) { + let _ = replicate_delete_with_outcome(dobj, storage).await; +} + +pub(crate) async fn replicate_delete_with_outcome( + dobj: DeletedObjectReplicationInfo, + storage: Arc, +) -> bool { if dobj.delete_object.force_delete { - replicate_force_delete_to_targets(&dobj, storage).await; - return; + return replicate_force_delete_to_targets(&dobj, storage).await; } 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(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(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(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(dobj: DeletedObjectReplicat user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return false; } }; @@ -1372,7 +1380,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return false; } }; @@ -1386,6 +1394,11 @@ pub async fn replicate_delete(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(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(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(dobj: DeletedObjectReplicat object, ..Default::default() }); + true } Err(e) => { error!( @@ -1583,8 +1598,16 @@ pub async fn replicate_delete(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( @@ -1639,7 +1662,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb } } -async fn replicate_force_delete_to_targets(dobj: &DeletedObjectReplicationInfo, storage: Arc) { +async fn replicate_force_delete_to_targets(dobj: &DeletedObjectReplicationInfo, storage: Arc) -> 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(dobj: &Deleted user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return false; } }; @@ -1755,7 +1778,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return false; } }; @@ -1764,6 +1787,9 @@ async fn replicate_force_delete_to_targets(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(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 { @@ -2034,6 +2063,13 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli } pub async fn replicate_object(roi: ReplicateObjectInfo, storage: Arc) -> ReplicationState { + replicate_object_with_outcome(roi, storage).await.0 +} + +pub(crate) async fn replicate_object_with_outcome( + roi: ReplicateObjectInfo, + storage: Arc, +) -> (ReplicationState, bool) { let bucket = roi.bucket.clone(); let object = roi.name.clone(); @@ -2062,7 +2098,7 @@ pub async fn replicate_object(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(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(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(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(roi: ReplicateObjectInfo, s } } - merged_state + (merged_state, state_persisted) } trait ReplicateObjectInfoExt { diff --git a/crates/replication/src/queue.rs b/crates/replication/src/queue.rs index 569c0200c..b652a6237 100644 --- a/crates/replication/src/queue.rs +++ b/crates/replication/src/queue.rs @@ -232,7 +232,7 @@ impl ReplicationPriority { } } -#[derive(Debug)] +#[derive(Debug, Clone)] pub enum ReplicationOperation { Object(Box), Delete(Box),