refactor(replication): move app decisions into crate (#4191)

crates/replication/src/config.rs

@@ -45,6 +45,67 @@ pub trait ReplicationConfigurationExt {
     fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String>;
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReplicationTargetValidationError {
+    RoleWithMultipleDestinations,
+    StaleTarget,
+}
+
+pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet<String> {
+    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<String> {
+    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<Item = &'a str>,
+    config: &ReplicationConfiguration,
+) -> std::result::Result<(), ReplicationTargetValidationError> {
+    let configured_arns = configured_arns.into_iter().collect::<HashSet<_>>();
+
+    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<String>,
+) -> 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
+        ));
+    }
 }

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,

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<Uuid>,
+    pub delete_marker: bool,
+    pub replica: bool,
+}
+
+pub fn delete_replication_state_from_config(
+    config: &ReplicationConfiguration,
+    source: &ReplicationDeleteStateSource,
+) -> Option<ReplicationState> {
+    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);

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<String> {
-    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<String> {
-    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::<HashSet<_>>();
+        .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<Option<(BucketTargets, usize)>> {
-    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<String>) -> 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));
     }

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<Uuid>,
             replica: bool,
         ) -> Option<rustfs_replication::ReplicationState> {
-            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)
         }
     }