fix(replication): enforce bucket replication switches (#5449)

* fix(replication): enforce bucket replication switches

* fix(replication): satisfy delete admission clippy lint

* fix(replication): restore MinIO tag filter behavior

---------

Co-authored-by: Zhengchao An <anzhengchao@gmail.com>
Co-authored-by: cxymds <cxymds@gmail.com>

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();
 
@@ -217,11 +239,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 +322,11 @@ 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, ReplicationTime, ReplicationTimeStatus, ReplicationTimeValue, SourceSelectionCriteria,
+        SseKmsEncryptedObjects, SseKmsEncryptedObjectsStatus,
+    };
 
     fn replication_rule(id: &str, arn: &str) -> ReplicationRule {
         ReplicationRule {
@@ -605,4 +626,99 @@ mod tests {
             "highest-priority rule disables delete-marker replication, so the delete marker must not replicate"
         );
     }
+
+    #[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"));
+    }
 }