fix(replication): enforce bucket replication switches

crates/replication/src/config.rs

@@ -51,6 +51,28 @@ pub enum ReplicationTargetValidationError {
     StaleTarget,
 }
 
+pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -> Option<&'static str> {
+    for rule in &config.rules {
+        if rule
+            .source_selection_criteria
+            .as_ref()
+            .is_some_and(|criteria| criteria.sse_kms_encrypted_objects.is_some())
+        {
+            return Some("SourceSelectionCriteria.SseKmsEncryptedObjects");
+        }
+        if rule.destination.encryption_configuration.is_some() {
+            return Some("Destination.EncryptionConfiguration");
+        }
+        if rule.destination.metrics.is_some() {
+            return Some("Destination.Metrics");
+        }
+        if rule.destination.replication_time.is_some() {
+            return Some("Destination.ReplicationTime");
+        }
+    }
+    None
+}
+
 pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet<String> {
     let mut arns = HashSet::new();
 
@@ -164,7 +186,17 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
 
             if let Some(filter) = &rule.filter {
                 let object_tags = ReplicationTagFilter::decode_tags_to_map(&obj.user_tags);
-                if filter.test_tags(&object_tags) {
+                let tags_match = if let Some(and) = &filter.and {
+                    and.tags.as_ref().is_none_or(|tags| {
+                        tags.iter()
+                            .all(|tag| tag.key.as_ref().is_some_and(|key| object_tags.get(key) == tag.value.as_ref()))
+                    })
+                } else if let Some(tag) = &filter.tag {
+                    tag.key.as_ref().is_some_and(|key| object_tags.get(key) == tag.value.as_ref())
+                } else {
+                    true
+                };
+                if tags_match {
                     rules.push(rule.clone());
                 }
             } else {
@@ -217,11 +249,6 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
                 }
 
                 if obj.version_id.is_some() {
-                    if obj.delete_marker {
-                        return rule.delete_marker_replication.clone().is_some_and(|d| {
-                            d.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
-                        });
-                    }
                     return rule
                         .delete_replication
                         .clone()
@@ -305,7 +332,12 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use s3s::dto::{DeleteMarkerReplication, Destination, ExistingObjectReplication, ReplicationRule};
+    use s3s::dto::{
+        DeleteMarkerReplication, DeleteReplication, Destination, EncryptionConfiguration, ExistingObjectReplication, Metrics,
+        MetricsStatus, ReplicationRule, ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationTime,
+        ReplicationTimeStatus, ReplicationTimeValue, SourceSelectionCriteria, SseKmsEncryptedObjects,
+        SseKmsEncryptedObjectsStatus, Tag,
+    };
 
     fn replication_rule(id: &str, arn: &str) -> ReplicationRule {
         ReplicationRule {
@@ -605,4 +637,162 @@ mod tests {
             "highest-priority rule disables delete-marker replication, so the delete marker must not replicate"
         );
     }
+
+    #[test]
+    fn replication_filter_requires_every_and_tag_and_accepts_extra_tags() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule("and-tags", arn);
+        rule.filter = Some(ReplicationRuleFilter {
+            and: Some(ReplicationRuleAndOperator {
+                prefix: Some("logs/".to_string()),
+                tags: Some(vec![
+                    Tag {
+                        key: Some("env".to_string()),
+                        value: Some("prod".to_string()),
+                    },
+                    Tag {
+                        key: Some("team".to_string()),
+                        value: Some("storage/core".to_string()),
+                    },
+                ]),
+            }),
+            ..Default::default()
+        });
+        rule.prefix = None;
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        let matching = ObjectOpts {
+            name: "logs/app.log".to_string(),
+            user_tags: "env=prod&team=storage%2Fcore&extra=value".to_string(),
+            op_type: ReplicationType::Object,
+            ..Default::default()
+        };
+        let partial = ObjectOpts {
+            user_tags: "env=prod".to_string(),
+            ..matching.clone()
+        };
+        let wrong_prefix = ObjectOpts {
+            name: "archive/app.log".to_string(),
+            ..matching.clone()
+        };
+
+        for (op_type, existing_object) in [
+            (ReplicationType::Object, false),
+            (ReplicationType::Delete, false),
+            (ReplicationType::ExistingObject, true),
+            (ReplicationType::Heal, false),
+        ] {
+            let mut matching = matching.clone();
+            matching.op_type = op_type;
+            matching.existing_object = existing_object;
+            let mut partial = partial.clone();
+            partial.op_type = op_type;
+            partial.existing_object = existing_object;
+            let mut wrong_prefix = wrong_prefix.clone();
+            wrong_prefix.op_type = op_type;
+            wrong_prefix.existing_object = existing_object;
+
+            assert_eq!(config.filter_actionable_rules(&matching).len(), 1);
+            assert!(config.filter_actionable_rules(&partial).is_empty());
+            assert!(config.filter_actionable_rules(&wrong_prefix).is_empty());
+        }
+    }
+
+    #[test]
+    fn version_purge_uses_delete_replication_for_object_and_marker_versions() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule("delete", arn);
+        rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        });
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        });
+        let mut config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+        let version_id = Some(Uuid::new_v4());
+
+        for delete_marker in [false, true] {
+            assert!(config.replicate(&ObjectOpts {
+                name: "object".to_string(),
+                version_id,
+                delete_marker,
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            }));
+        }
+        assert!(!config.replicate(&ObjectOpts {
+            name: "object".to_string(),
+            delete_marker: true,
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        }));
+
+        let rule = &mut config.rules[0];
+        rule.delete_marker_replication = Some(DeleteMarkerReplication {
+            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        });
+        rule.delete_replication = Some(DeleteReplication {
+            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+        });
+
+        for delete_marker in [false, true] {
+            assert!(!config.replicate(&ObjectOpts {
+                name: "object".to_string(),
+                version_id,
+                delete_marker,
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            }));
+        }
+        assert!(config.replicate(&ObjectOpts {
+            name: "object".to_string(),
+            delete_marker: true,
+            op_type: ReplicationType::Delete,
+            ..Default::default()
+        }));
+    }
+
+    #[test]
+    fn unsupported_replication_fields_are_reported_before_persistence() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule("unsupported", arn)],
+        };
+
+        config.rules[0].source_selection_criteria = Some(SourceSelectionCriteria {
+            replica_modifications: None,
+            sse_kms_encrypted_objects: Some(SseKmsEncryptedObjects {
+                status: SseKmsEncryptedObjectsStatus::from_static(SseKmsEncryptedObjectsStatus::ENABLED),
+            }),
+        });
+        assert_eq!(
+            unsupported_replication_config_field(&config),
+            Some("SourceSelectionCriteria.SseKmsEncryptedObjects")
+        );
+
+        config.rules[0].source_selection_criteria = None;
+        config.rules[0].destination.encryption_configuration = Some(EncryptionConfiguration::default());
+        assert_eq!(unsupported_replication_config_field(&config), Some("Destination.EncryptionConfiguration"));
+
+        config.rules[0].destination.encryption_configuration = None;
+        config.rules[0].destination.metrics = Some(Metrics {
+            event_threshold: None,
+            status: MetricsStatus::from_static(MetricsStatus::ENABLED),
+        });
+        assert_eq!(unsupported_replication_config_field(&config), Some("Destination.Metrics"));
+
+        config.rules[0].destination.metrics = None;
+        config.rules[0].destination.replication_time = Some(ReplicationTime {
+            status: ReplicationTimeStatus::from_static(ReplicationTimeStatus::ENABLED),
+            time: ReplicationTimeValue { minutes: Some(15) },
+        });
+        assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime"));
+    }
 }

crates/replication/src/lib.rs

@@ -30,7 +30,8 @@ pub mod tagging;
 
 pub use config::{
     ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns,
-    replication_target_arns, should_remove_replication_target, validate_replication_config_target_arns,
+    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
+    validate_replication_config_target_arns,
 };
 pub use delete::{
     DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,

crates/replication/src/operation.rs

@@ -322,9 +322,9 @@ mod tests {
     use crate::storage_api::ObjectToDelete;
     use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header};
     use s3s::dto::{
-        DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication,
-        ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration,
-        ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria,
+        DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
+        ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus,
+        ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria,
     };
     use std::collections::HashMap;
     use time::{Duration, OffsetDateTime};
@@ -433,7 +433,9 @@ mod tests {
             delete_marker_replication: Some(DeleteMarkerReplication {
                 status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
             }),
-            delete_replication: None,
+            delete_replication: Some(DeleteReplication {
+                status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+            }),
             destination: Destination {
                 bucket: arn.to_string(),
                 ..Default::default()
@@ -513,7 +515,7 @@ mod tests {
         };
 
         let state = delete_replication_state_from_config(&config, &source)
-            .expect("delete-marker version purge should honor delete-marker replication rules");
+            .expect("delete-marker version purge should honor delete replication rules");
         let pending = format!("{arn}=PENDING;");
 
         assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str()));