From 17dab96a89e6e35ab34d8a9bab953bffbf2f9b30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sun, 2 Aug 2026 18:51:29 +0800 Subject: [PATCH] fix(replication): make sync delivery target-granular --- .../bucket/replication/replication_pool.rs | 18 +++-- .../replication/replication_resyncer.rs | 71 ++++--------------- crates/replication/src/filemeta.rs | 47 ++++++++++++ 3 files changed, 74 insertions(+), 62 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index d54a168ad..d0f0c5afd 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -1823,12 +1823,20 @@ pub(crate) async fn schedule_replication( dsc: ReplicateDecision, op_type: ReplicationType, ) { - let synchronous = dsc.is_synchronous(); - let ri = replicate_object_info_from_object_info(oi, dsc, op_type); + let (synchronous, asynchronous) = dsc.partition_by_sync(); + let mut async_oi = oi; - if synchronous { - replicate_object(ri, o).await - } else if let Some(pool) = runtime_sources::replication_pool() { + if synchronous.replicate_any() { + let ri = replicate_object_info_from_object_info(async_oi.clone(), synchronous, op_type); + let state = replicate_object(ri, o.clone()).await; + async_oi.replication_status_internal = state.replication_status_internal; + async_oi.version_purge_status_internal = state.version_purge_status_internal; + } + + if asynchronous.replicate_any() + && let Some(pool) = runtime_sources::replication_pool() + { + let ri = replicate_object_info_from_object_info(async_oi, asynchronous, op_type); let _ = pool.queue_replica_task(ri).await; } } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index f28c18cb6..edd54ff06 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -19,7 +19,7 @@ use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_ use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, - ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, + ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; use super::replication_lock_boundary::ReplicationLockTiming; @@ -2001,61 +2001,14 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli rinfo } -pub async fn replicate_object(roi: ReplicateObjectInfo, storage: Arc) { +pub async fn replicate_object(roi: ReplicateObjectInfo, storage: Arc) -> ReplicationState { let bucket = roi.bucket.clone(); let object = roi.name.clone(); - let cfg = match get_replication_config(&bucket).await { - Ok(Some(config)) => config, - Ok(None) => { - debug!( - event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - reason = "replication_config_missing", - "Skipping replication object because replication config is missing" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: roi.to_object_info(), - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - return; - } - Err(err) => { - error!( - event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - reason = "replication_config_lookup_failed", - error = %err, - "Failed to look up replication config for object replication" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: roi.to_object_info(), - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - return; - } - }; - - let tgt_arns = cfg.filter_target_arns(&ObjectOpts { - name: object.clone(), - user_tags: roi.user_tags.clone(), - ssec: roi.ssec, - op_type: roi.op_type, - // ExistingObject ops must respect per-rule ExistingObjectReplicationStatus. - // Heal ops intentionally bypass it (repairing a past failure is not an initial sync). - existing_object: roi.op_type == ReplicationType::ExistingObject, - ..Default::default() - }); + // 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(); // 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. @@ -2080,7 +2033,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return roi.replication_state.unwrap_or_default(); } }; let _obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { @@ -2103,7 +2056,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); - return; + return roi.replication_state.unwrap_or_default(); } }; @@ -2179,8 +2132,10 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s } } - let replication_status = rinfos.replication_status(); - let new_replication_internal = rinfos.replication_status_internal(); + let previous_state = roi.replication_state.clone().unwrap_or_default(); + let merged_state = get_replication_state(&rinfos, &previous_state, roi.version_id.map(|v| v.to_string())); + 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(); if roi.replication_status_internal != new_replication_internal || rinfos.replication_resynced() { @@ -2249,6 +2204,8 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s } } } + + merged_state } trait ReplicateObjectInfoExt { diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index 667e9a3d1..7d6aae4a0 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -660,6 +660,27 @@ impl ReplicateDecision { self.targets_map.values().any(|t| t.synchronous) } + /// Split admitted targets by their configured delivery mode. + /// + /// Non-replicating entries are intentionally omitted from both decisions. + /// Callers must not promote an async target merely because another target is + /// synchronous, and unsupported operation paths can keep both partitions + /// empty or explicitly async. + pub fn partition_by_sync(&self) -> (Self, Self) { + let mut synchronous = Self::new(); + let mut asynchronous = Self::new(); + + for target in self.targets_map.values().filter(|target| target.replicate) { + if target.synchronous { + synchronous.set(target.clone()); + } else { + asynchronous.set(target.clone()); + } + } + + (synchronous, asynchronous) + } + /// Updates ReplicateDecision with target's replication decision pub fn set(&mut self, target: ReplicateTargetDecision) { self.targets_map.insert(target.arn.clone(), target); @@ -1067,6 +1088,32 @@ mod tests { assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]); } + #[test] + fn partition_by_sync_keeps_mixed_targets_independent() { + let mut decision = ReplicateDecision::new(); + decision.set(ReplicateTargetDecision::new("arn:sync".to_string(), true, true)); + decision.set(ReplicateTargetDecision::new("arn:async".to_string(), true, false)); + decision.set(ReplicateTargetDecision::new("arn:disabled".to_string(), false, true)); + + let (synchronous, asynchronous) = decision.partition_by_sync(); + + assert_eq!(synchronous.replicate_target_arns(), vec!["arn:sync".to_string()]); + assert_eq!(asynchronous.replicate_target_arns(), vec!["arn:async".to_string()]); + assert!(synchronous.is_synchronous()); + assert!(!asynchronous.is_synchronous()); + } + + #[test] + fn partition_by_sync_does_not_promote_async_targets() { + let mut decision = ReplicateDecision::new(); + decision.set(ReplicateTargetDecision::new("arn:async".to_string(), true, false)); + + let (synchronous, asynchronous) = decision.partition_by_sync(); + + assert!(!synchronous.replicate_any()); + assert_eq!(asynchronous.replicate_target_arns(), vec!["arn:async".to_string()]); + } + #[test] fn target_state_reads_resync_timestamp_from_target_reset_header_key() { let arn = "arn:rustfs:replication:us-east-1:target:bucket";