From b972e25c51d0e8d42409c0abc5f2508984e794cf Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 17:14:55 +0800 Subject: [PATCH] refactor(replication): move app decisions into crate (#4191) --- crates/replication/src/config.rs | 162 ++++++++++++++++++++++++++++ crates/replication/src/lib.rs | 10 +- crates/replication/src/operation.rs | 162 +++++++++++++++++++++++++++- rustfs/src/app/bucket_usecase.rs | 71 ++++-------- rustfs/src/app/storage_api.rs | 42 +------- 5 files changed, 351 insertions(+), 96 deletions(-) diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 4a3171717..2a72caeaa 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -45,6 +45,67 @@ pub trait ReplicationConfigurationExt { fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec; } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicationTargetValidationError { + RoleWithMultipleDestinations, + StaleTarget, +} + +pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet { + let mut arns = HashSet::new(); + + for rule in &config.rules { + if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { + continue; + } + + let arn = rule.destination.bucket.trim(); + if !arn.is_empty() { + arns.insert(arn.to_string()); + } + } + + arns +} + +pub fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { + let role = config.role.trim(); + if !role.is_empty() { + return HashSet::from([role.to_string()]); + } + + active_replication_rule_destination_arns(config) +} + +pub fn validate_replication_config_target_arns<'a>( + configured_arns: impl IntoIterator, + config: &ReplicationConfiguration, +) -> std::result::Result<(), ReplicationTargetValidationError> { + let configured_arns = configured_arns.into_iter().collect::>(); + + let role = config.role.trim(); + let destination_arns = active_replication_rule_destination_arns(config); + if !role.is_empty() && destination_arns.len() > 1 { + return Err(ReplicationTargetValidationError::RoleWithMultipleDestinations); + } + + for configured_arn in replication_target_arns(config) { + if !configured_arns.contains(configured_arn.as_str()) { + return Err(ReplicationTargetValidationError::StaleTarget); + } + } + + Ok(()) +} + +pub fn should_remove_replication_target( + target_arn: &str, + is_replication_target: bool, + config_target_arns: &HashSet, +) -> bool { + is_replication_target && config_target_arns.contains(target_arn) +} + impl ReplicationConfigurationExt for ReplicationConfiguration { /// Check whether any object-replication rules exist fn has_existing_object_replication(&self, arn: &str) -> (bool, bool) { @@ -389,4 +450,105 @@ mod tests { assert!(arns.contains(&"arn:target:enabled".to_string())); assert!(arns.contains(&"arn:target:disabled".to_string())); } + + #[test] + fn replication_target_arns_use_role_when_present() { + let role = "arn:rustfs:replication:us-east-1:source:bucket"; + let destination = "arn:rustfs:replication:us-east-1:target:bucket"; + let config = ReplicationConfiguration { + role: format!(" {role} "), + rules: vec![replication_rule("rule-1", destination)], + }; + + let arns = replication_target_arns(&config); + + assert!(arns.contains(role)); + assert!(!arns.contains(destination)); + } + + #[test] + fn replication_target_arns_use_rule_destinations_without_role() { + let destination = "arn:rustfs:replication:us-east-1:target:bucket"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule("rule-1", destination)], + }; + + let arns = replication_target_arns(&config); + + assert!(arns.contains(destination)); + } + + #[test] + fn validate_replication_config_target_arns_accepts_matching_destination_arns() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule("rule-1", arn)], + }; + + validate_replication_config_target_arns([arn], &config).expect("matching target should pass validation"); + } + + #[test] + fn validate_replication_config_target_arns_rejects_stale_destination_arns() { + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule("rule-1", "arn:rustfs:replication:us-east-1:target-b:bucket")], + }; + + let err = validate_replication_config_target_arns(["arn:rustfs:replication:us-east-1:target-a:bucket"], &config) + .expect_err("stale target should fail validation"); + assert_eq!(err, ReplicationTargetValidationError::StaleTarget); + } + + #[test] + fn validate_replication_config_target_arns_rejects_role_with_multiple_destinations() { + let role = "arn:rustfs:replication:us-east-1:role-target:bucket"; + let config = ReplicationConfiguration { + role: role.to_string(), + rules: vec![ + replication_rule("rule-a", "arn:rustfs:replication:us-east-1:target-a:bucket"), + replication_rule("rule-b", "arn:rustfs:replication:us-east-1:target-b:bucket"), + ], + }; + + let err = validate_replication_config_target_arns([role], &config) + .expect_err("role plus multiple destinations should be rejected"); + assert_eq!(err, ReplicationTargetValidationError::RoleWithMultipleDestinations); + } + + #[test] + fn validate_replication_config_target_arns_ignores_disabled_rules() { + let mut rule = replication_rule("rule-1", "arn:rustfs:replication:us-east-1:stale:bucket"); + rule.status = ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![rule], + }; + + validate_replication_config_target_arns(std::iter::empty::<&str>(), &config) + .expect("disabled rules should not require live targets"); + } + + #[test] + fn should_remove_replication_target_only_matches_replication_target_arns() { + let target_arns = HashSet::from(["arn:rustfs:replication:us-east-1:removed:bucket".to_string()]); + + assert!(should_remove_replication_target( + "arn:rustfs:replication:us-east-1:removed:bucket", + true, + &target_arns + )); + assert!(!should_remove_replication_target( + "arn:rustfs:replication:us-east-1:kept:bucket", + true, + &target_arns + )); + assert!(!should_remove_replication_target( + "arn:rustfs:replication:us-east-1:removed:bucket", + false, + &target_arns + )); + } } diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 1c1cfe10b..840b6cf1a 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -24,7 +24,10 @@ pub mod runtime; pub mod stats; pub mod tagging; -pub use config::{ObjectOpts, ReplicationConfigurationExt}; +pub use config::{ + ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, + replication_target_arns, should_remove_replication_target, validate_replication_config_target_arns, +}; pub use delete::{ DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication, should_retry_delete_marker_purge, @@ -35,8 +38,9 @@ pub use object::{ replication_etags_match, target_is_newer_than_source_null_version, }; pub use operation::{ - MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision, - delete_replication_object_opts, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, + MustReplicateOptions, ReplicationDeleteSource, ReplicationDeleteStateSource, ReplicationResyncTargetObject, + delete_replication_missing_source_decision, delete_replication_object_opts, delete_replication_state_from_config, + heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, }; pub use queue::{ ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation, diff --git a/crates/replication/src/operation.rs b/crates/replication/src/operation.rs index a603559b2..c95bbcb8b 100644 --- a/crates/replication/src/operation.rs +++ b/crates/replication/src/operation.rs @@ -19,10 +19,14 @@ use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, SUFFIX_REPLICATION_RESET_STATUS, get_header_map, }; +use s3s::dto::ReplicationConfiguration; use time::OffsetDateTime; +use uuid::Uuid; use crate::{ - ObjectOpts, ReplicationStatusType, ReplicationType, ResyncTargetDecision, VersionPurgeStatusType, target_reset_header, + ObjectOpts, ReplicateDecision, ReplicateTargetDecision, ReplicationConfigurationExt as _, ReplicationState, + ReplicationStatusType, ReplicationType, ResyncTargetDecision, VersionPurgeStatusType, replication_statuses_map, + target_reset_header, version_purge_statuses_map, }; #[derive(Debug, Clone, Default)] @@ -97,6 +101,59 @@ pub fn delete_replication_object_opts(dobj: &ObjectToDelete, source: &Replicatio } } +#[derive(Debug, Clone)] +pub struct ReplicationDeleteStateSource { + pub name: String, + pub user_tags: String, + pub version_id: Option, + pub delete_marker: bool, + pub replica: bool, +} + +pub fn delete_replication_state_from_config( + config: &ReplicationConfiguration, + source: &ReplicationDeleteStateSource, +) -> Option { + let opts = ObjectOpts { + name: source.name.clone(), + user_tags: source.user_tags.clone(), + version_id: source.version_id, + delete_marker: source.delete_marker, + op_type: ReplicationType::Delete, + replica: source.replica, + ..Default::default() + }; + let target_arns = config.filter_target_arns(&opts); + if target_arns.is_empty() { + return None; + } + + let mut decision = ReplicateDecision::new(); + for target_arn in target_arns { + let mut target_opts = opts.clone(); + target_opts.target_arn = target_arn.clone(); + let replicate = config.replicate(&target_opts); + decision.set(ReplicateTargetDecision::new(target_arn, replicate, false)); + } + if !decision.replicate_any() { + return None; + } + + let pending_status = decision.pending_status(); + let mut state = ReplicationState { + replicate_decision_str: decision.to_string(), + ..Default::default() + }; + if source.version_id.is_some() { + state.version_purge_status_internal = pending_status.clone(); + state.purge_targets = version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); + } else { + state.replication_status_internal = pending_status.clone(); + state.targets = replication_statuses_map(pending_status.as_deref().unwrap_or_default()); + } + Some(state) +} + pub fn heal_uses_delete_replication_path(delete_marker: bool, version_purge_status: &VersionPurgeStatusType) -> bool { delete_marker || !version_purge_status.is_empty() } @@ -189,12 +246,18 @@ pub fn resync_target_for_object( #[cfg(test)] mod tests { use super::{ - MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision, - delete_replication_object_opts, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, + MustReplicateOptions, ReplicationDeleteSource, ReplicationDeleteStateSource, ReplicationResyncTargetObject, + delete_replication_missing_source_decision, delete_replication_object_opts, delete_replication_state_from_config, + heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object, }; use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header}; use rustfs_storage_api::ObjectToDelete; use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER}; + use s3s::dto::{ + DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication, + ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, + ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria, + }; use std::collections::HashMap; use time::{Duration, OffsetDateTime}; use uuid::Uuid; @@ -288,6 +351,99 @@ mod tests { assert_eq!(opts.op_type, ReplicationType::Delete); } + fn delete_replication_rule(arn: &str, replica_modifications: bool) -> ReplicationRule { + ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }), + delete_replication: None, + destination: Destination { + bucket: arn.to_string(), + ..Default::default() + }, + existing_object_replication: Some(ExistingObjectReplication { + status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED), + }), + filter: None, + id: Some("rule-1".to_string()), + prefix: Some("test/".to_string()), + priority: Some(1), + source_selection_criteria: replica_modifications.then_some(SourceSelectionCriteria { + replica_modifications: Some(ReplicaModifications { + status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED), + }), + sse_kms_encrypted_objects: None, + }), + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + } + } + + #[test] + fn delete_replication_state_tracks_downstream_delete_marker_targets() { + let arn = "arn:aws:s3:::target-bucket"; + let config = ReplicationConfiguration { + role: arn.to_string(), + rules: vec![delete_replication_rule(arn, true)], + }; + let source = ReplicationDeleteStateSource { + name: "test/object.txt".to_string(), + user_tags: String::new(), + version_id: None, + delete_marker: true, + replica: true, + }; + + let state = delete_replication_state_from_config(&config, &source) + .expect("replica delete marker should be forwarded to downstream targets"); + let pending = format!("{arn}=PENDING;"); + + assert_eq!(state.replication_status_internal.as_deref(), Some(pending.as_str())); + assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};")); + assert!(state.targets.contains_key(arn)); + } + + #[test] + fn delete_replication_state_skips_replica_delete_without_replica_modifications() { + let arn = "arn:aws:s3:::target-bucket"; + let config = ReplicationConfiguration { + role: arn.to_string(), + rules: vec![delete_replication_rule(arn, false)], + }; + let source = ReplicationDeleteStateSource { + name: "test/object.txt".to_string(), + user_tags: String::new(), + version_id: None, + delete_marker: true, + replica: true, + }; + + assert!(delete_replication_state_from_config(&config, &source).is_none()); + } + + #[test] + fn delete_replication_state_tracks_delete_marker_version_purges() { + let arn = "arn:aws:s3:::target-bucket"; + let config = ReplicationConfiguration { + role: arn.to_string(), + rules: vec![delete_replication_rule(arn, false)], + }; + let source = ReplicationDeleteStateSource { + name: "test/object.txt".to_string(), + user_tags: String::new(), + version_id: Some(Uuid::new_v4()), + delete_marker: true, + replica: false, + }; + + let state = delete_replication_state_from_config(&config, &source) + .expect("delete-marker version purge should honor delete-marker replication rules"); + let pending = format!("{arn}=PENDING;"); + + assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str())); + assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};")); + assert!(state.purge_targets.contains_key(arn)); + } + #[test] fn resync_target_includes_object_at_reset_before_boundary() { let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index d1f331965..0ccf3f32a 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -105,8 +105,7 @@ use s3s::dto::{ PutBucketNotificationConfigurationOutput, PutBucketPolicyInput, PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput, PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput, ReplicationConfiguration, - ReplicationRuleStatus, ServerSideEncryption, Tagging, Timestamp, UserMetadataCollection, UserMetadataEntry, - VersioningConfiguration, + ServerSideEncryption, Tagging, Timestamp, UserMetadataCollection, UserMetadataEntry, VersioningConfiguration, }; use s3s::region::Region; use s3s::xml; @@ -246,58 +245,25 @@ fn notify_bucket_metadata_reload( }); } -fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet { - let mut arns = HashSet::new(); - - for rule in &config.rules { - if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { - continue; - } - - let arn = rule.destination.bucket.trim(); - if !arn.is_empty() { - arns.insert(arn.to_string()); - } - } - - arns -} - -fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { - let role = config.role.trim(); - if !role.is_empty() { - let mut arns = HashSet::new(); - arns.insert(role.to_string()); - return arns; - } - - active_replication_rule_destination_arns(config) -} - fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> { let configured_arns = targets .targets .iter() .filter(|target| target.target_type == BucketTargetType::ReplicationService) - .map(|target| target.arn.as_str()) - .collect::>(); + .map(|target| target.arn.as_str()); - let role = config.role.trim(); - let destination_arns = active_replication_rule_destination_arns(config); - if !role.is_empty() && destination_arns.len() > 1 { - return Err(s3_error!( - InvalidRequest, - "replication config with Role cannot define multiple destination targets" - )); - } - - for configured_arn in replication_target_arns(config) { - if !configured_arns.contains(configured_arn.as_str()) { - return Err(s3_error!(InvalidRequest, "replication config has a stale target")); + match rustfs_replication::validate_replication_config_target_arns(configured_arns, config) { + Ok(()) => Ok(()), + Err(err) => { + let message = match err { + rustfs_replication::ReplicationTargetValidationError::RoleWithMultipleDestinations => { + "replication config with Role cannot define multiple destination targets" + } + rustfs_replication::ReplicationTargetValidationError::StaleTarget => "replication config has a stale target", + }; + Err(S3Error::with_message(S3ErrorCode::InvalidRequest, message.to_string())) } } - - Ok(()) } async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> { @@ -324,7 +290,7 @@ async fn replication_targets_without_config_targets( bucket: &str, config: &ReplicationConfiguration, ) -> S3Result> { - let target_arns = replication_target_arns(config); + let target_arns = rustfs_replication::replication_target_arns(config); if target_arns.is_empty() { return Ok(None); } @@ -349,7 +315,11 @@ async fn replication_targets_without_config_targets( fn remove_replication_targets_from_config_targets(targets: &mut BucketTargets, target_arns: &HashSet) -> usize { let original_len = targets.targets.len(); targets.targets.retain(|target| { - target.target_type != BucketTargetType::ReplicationService || !target_arns.contains(target.arn.as_str()) + !rustfs_replication::should_remove_replication_target( + target.arn.as_str(), + target.target_type == BucketTargetType::ReplicationService, + target_arns, + ) }); original_len - targets.targets.len() @@ -2291,6 +2261,7 @@ impl DefaultBucketUsecase { mod tests { use super::*; use http::{Extensions, HeaderMap, Method, Uri}; + use s3s::dto::ReplicationRuleStatus; use s3s::dto::{ BucketVersioningStatus, CORSConfiguration, Destination, ExcludedPrefix, FilterRule, FilterRuleName, LifecycleExpiration, NoncurrentVersionTransition, PublicAccessBlockConfiguration, QueueConfiguration, ReplicationRule, S3KeyFilter, @@ -2374,7 +2345,7 @@ mod tests { rules: vec![replication_rule_for_target(destination)], }; - let arns = replication_target_arns(&config); + let arns = rustfs_replication::replication_target_arns(&config); assert!(arns.contains(role)); assert!(!arns.contains(destination)); @@ -2388,7 +2359,7 @@ mod tests { rules: vec![replication_rule_for_target(destination)], }; - let arns = replication_target_arns(&config); + let arns = rustfs_replication::replication_target_arns(&config); assert!(arns.contains(destination)); } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 84049e758..dab24be9b 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -610,7 +610,6 @@ pub(crate) mod bucket { pub(crate) type DeletedObjectReplicationInfo = crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo; - type ObjectOpts = crate::storage::storage_api::ecstore_bucket::replication::ObjectOpts; type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge; pub(crate) type ReplicateDecision = rustfs_replication::ReplicateDecision; @@ -660,51 +659,14 @@ pub(crate) mod bucket { version_id: Option, replica: bool, ) -> Option { - let opts = ObjectOpts { + let source = rustfs_replication::ReplicationDeleteStateSource { name: obj_info.name.clone(), user_tags: (*obj_info.user_tags).clone(), version_id, delete_marker: obj_info.delete_marker, - op_type: rustfs_replication::ReplicationType::Delete, replica, - ..Default::default() }; - let target_arns = - crate::storage::storage_api::ecstore_bucket::replication::ReplicationConfigurationExt::filter_target_arns( - config, &opts, - ); - if target_arns.is_empty() { - return None; - } - - let mut decision = ReplicateDecision::new(); - for target_arn in target_arns { - let mut target_opts = opts.clone(); - target_opts.target_arn = target_arn.clone(); - let replicate = crate::storage::storage_api::ecstore_bucket::replication::ReplicationConfigurationExt::replicate( - config, - &target_opts, - ); - decision.set(rustfs_replication::ReplicateTargetDecision::new(target_arn, replicate, false)); - } - if !decision.replicate_any() { - return None; - } - - let pending_status = decision.pending_status(); - let mut state = rustfs_replication::ReplicationState { - replicate_decision_str: decision.to_string(), - ..Default::default() - }; - if version_id.is_some() { - state.version_purge_status_internal = pending_status.clone(); - state.purge_targets = - rustfs_replication::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default()); - } else { - state.replication_status_internal = pending_status.clone(); - state.targets = rustfs_replication::replication_statuses_map(pending_status.as_deref().unwrap_or_default()); - } - Some(state) + rustfs_replication::delete_replication_state_from_config(config, &source) } }