From a918f1a48a3af8e6f2dc0d953815e5fc800b19d6 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 3 Aug 2026 00:13:32 +0800 Subject: [PATCH] fix(replication): snapshot existing object admission targets (#5634) --- .../replication_filemeta_boundary.rs | 2 +- .../bucket/replication/replication_pool.rs | 53 ++++++---- .../replication/replication_resyncer.rs | 7 +- crates/replication/src/filemeta.rs | 97 +++++++++++++++++-- crates/replication/src/lib.rs | 4 +- 5 files changed, 125 insertions(+), 38 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index 1aae83f3d..86171b5fd 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/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, diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 472f3f9b5..f16d1eb32 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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 ReplicationPool { 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 ReplicationPool { 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 ReplicationPool { 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 => { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index fc5eea204..240c356ad 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/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(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. diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index eabb843a1..cd4843a2a 100644 --- a/crates/replication/src/filemeta.rs +++ b/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 = 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 { + 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::>(); + 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 { 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"; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index b60c6aab7..525325db8 100644 --- a/crates/replication/src/lib.rs +++ b/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::{