test(replication): move cache compatibility to integration tests

.config/s3s-footprint-baseline.txt

@@ -12,6 +12,6 @@
 # s3_error_lines  lines holding s3_error!, same scope, minus those the script
 #                 header's rule clears (rustfs_s3_types's macro); fails closed
 # ecstore_files   Rust files referencing s3s paths under crates/ecstore/src
-files=194
+files=186
 s3_error_lines=1584
-ecstore_files=36
+ecstore_files=31

.config/test-count-baseline.txt

@@ -10,7 +10,7 @@
 # intentionally removed. Regenerate with:
 #   scripts/check_test_count_ratchet.sh --update
 e2e_test	890	18
-rustfs	5540	62
+rustfs	5551	62
 rustfs-audit	70	0
 rustfs-checksums	26	0
 rustfs-common	23	0
@@ -20,7 +20,7 @@ rustfs-connect-symbol-catalog	6	0
 rustfs-credentials	24	0
 rustfs-crypto	67	0
 rustfs-data-usage	54	0
-rustfs-ecstore	6306	1
+rustfs-ecstore	6307	1
 rustfs-extension-schema	23	0
 rustfs-filemeta	306	0
 rustfs-heal	839	3
@@ -42,7 +42,7 @@ rustfs-obs	365	0
 rustfs-policy	677	0
 rustfs-protocols	736	1
 rustfs-protos	52	0
-rustfs-replication	200	0
+rustfs-replication	207	0
 rustfs-rio	198	0
 rustfs-rio-v2	34	6
 rustfs-s3-client	65	0

Cargo.lock

@@ -8657,6 +8657,15 @@ version = "1.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
 
+[[package]]
+name = "quick-xml"
+version = "0.41.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "quick-xml"
 version = "0.42.0"
@@ -9548,7 +9557,7 @@ dependencies = [
  "proptest",
  "prost 0.14.4",
  "pyroscope",
- "quick-xml",
+ "quick-xml 0.42.0",
  "quinn",
  "rand 0.10.3",
  "rcgen",
@@ -9567,6 +9576,7 @@ dependencies = [
  "rustfs-ecstore",
  "rustfs-extension-schema",
  "rustfs-filemeta",
+ "rustfs-gateway-types",
  "rustfs-heal",
  "rustfs-heal-contracts",
  "rustfs-iam",
@@ -9843,6 +9853,7 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-erasure-codec",
  "rustfs-filemeta",
+ "rustfs-gateway-types",
  "rustfs-heal-contracts",
  "rustfs-io-metrics",
  "rustfs-lifecycle",
@@ -9947,6 +9958,45 @@ dependencies = [
  "xxhash-rust",
 ]
 
+[[package]]
+name = "rustfs-gateway-stream"
+version = "0.2.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=772d3385421626aca5735e5581c5dff13a7aebff#772d3385421626aca5735e5581c5dff13a7aebff"
+dependencies = [
+ "bitflags 2.13.2",
+ "bytes",
+ "http 1.5.0",
+ "http-body",
+]
+
+[[package]]
+name = "rustfs-gateway-types"
+version = "0.32.0+aws.2026-10-02"
+source = "git+https://github.com/rustfs/gateway.git?rev=772d3385421626aca5735e5581c5dff13a7aebff#772d3385421626aca5735e5581c5dff13a7aebff"
+dependencies = [
+ "bytes",
+ "crc-fast",
+ "hex",
+ "http 1.5.0",
+ "md-5 0.11.0",
+ "percent-encoding",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-xml",
+ "sha1 0.11.0",
+ "sha2 0.11.0",
+ "twox-hash",
+ "unicode-normalization",
+ "zeroize",
+]
+
+[[package]]
+name = "rustfs-gateway-xml"
+version = "0.4.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=772d3385421626aca5735e5581c5dff13a7aebff#772d3385421626aca5735e5581c5dff13a7aebff"
+dependencies = [
+ "quick-xml 0.41.0",
+]
+
 [[package]]
 name = "rustfs-heal"
 version = "1.0.1"
@@ -10323,7 +10373,7 @@ dependencies = [
  "jiff",
  "metrics",
  "percent-encoding",
- "quick-xml",
+ "quick-xml 0.42.0",
  "rayon",
  "rustc-hash",
  "rustfs-config",
@@ -10573,7 +10623,7 @@ dependencies = [
  "regex",
  "rmp",
  "rmp-serde",
- "s3s",
+ "rustfs-gateway-types",
  "serde",
  "time",
  "tokio",
@@ -10665,7 +10715,7 @@ dependencies = [
  "hyper-util",
  "lazy_static",
  "md-5 0.11.0",
- "quick-xml",
+ "quick-xml 0.42.0",
  "rand 0.10.3",
  "rustfs-checksums",
  "rustfs-config",
@@ -10798,6 +10848,7 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-filemeta",
+ "rustfs-gateway-types",
  "rustfs-heal",
  "rustfs-heal-contracts",
  "rustfs-lifecycle",
@@ -11304,7 +11355,7 @@ dependencies = [
  "numeric_cast",
  "phf 0.14.0",
  "pin-project-lite",
- "quick-xml",
+ "quick-xml 0.42.0",
  "regex",
  "s3s-multipart",
  "s3s-rfc2047",

Cargo.toml

@@ -142,6 +142,12 @@ rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.1" }
 rustfs-utils = { path = "crates/utils", version = "1.0.1" }
 rustfs-zip = { path = "./crates/zip", version = "1.0.1" }
 
+# The gateway's S3 types: the DTOs and persisted bucket-configuration codecs the
+# engine crates move onto from s3s (rustfs/backlog#2745). The one pin of the
+# gateway in this workspace; no crate may enable a `compat-s3s*` feature except
+# the s3s-facing edge of the rustfs binary.
+rustfs-gateway-types = { git = "https://github.com/rustfs/gateway.git", rev = "772d3385421626aca5735e5581c5dff13a7aebff", version = "0.32.0", default-features = false }
+
 # Async Runtime and Networking
 async-channel = "2.5.0"
 async_zip = { default-features = false, version = "0.0.19" }

crates/ecstore/Cargo.toml

@@ -141,6 +141,7 @@ rustfs-lifecycle.workspace = true
 rustfs-s3-client = { workspace = true }
 rustfs-s3-types = { workspace = true }
 rustfs-data-usage.workspace = true
+rustfs-gateway-types.workspace = true
 rustfs-object-capacity.workspace = true
 rustfs-object-data-cache = { workspace = true, features = ["runtime-memory"] }
 arc-swap.workspace = true

crates/ecstore/src/api/mod.rs

@@ -310,9 +310,9 @@ pub mod bucket {
 
     pub mod utils {
         pub use crate::bucket::utils::{
-            check_bucket_and_object_names, check_list_objs_args, check_object_name_for_length_and_slash,
-            check_valid_bucket_name_strict, deserialize, has_bad_path_component, is_meta_bucketname, is_valid_object_prefix,
-            serialize,
+            BucketConfigXml, BucketConfigXmlError, check_bucket_and_object_names, check_list_objs_args,
+            check_object_name_for_length_and_slash, check_valid_bucket_name_strict, deserialize, has_bad_path_component,
+            is_meta_bucketname, is_valid_object_prefix, serialize,
         };
     }
 

crates/ecstore/src/bucket/lifecycle/replication_sink.rs

@@ -77,12 +77,13 @@ mod tests {
     use crate::bucket::replication::{DeleteReplicationConfigSnapshot, ReplicationObjectBridge};
     use crate::object_api::{ObjectInfo, ObjectOptions};
     use crate::storage_api_contracts::object::ObjectToDelete;
-    use rustfs_scanner_metrics::metrics::IlmAction;
-    use s3s::dto::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication,
-        DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus,
-        VersioningConfiguration,
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+        PersistedReplicationRule, PersistedReplicationStatus,
     };
+    use rustfs_scanner_metrics::metrics::IlmAction;
+    use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
     use uuid::Uuid;
 
     use super::*;
@@ -158,24 +159,24 @@ mod tests {
                         status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                         ..Default::default()
                     },
-                    Some(ReplicationConfiguration {
+                    Some(PersistedReplicationConfiguration {
                         role: String::new(),
-                        rules: vec![ReplicationRule {
-                            delete_marker_replication: Some(DeleteMarkerReplication {
+                        rules: vec![PersistedReplicationRule {
+                            delete_marker_replication: Some(PersistedOptionalReplicationStatus {
                                 status: Some(if marker_enabled {
-                                    DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)
+                                    Status::ENABLED.to_string()
                                 } else {
-                                    DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)
+                                    Status::DISABLED.to_string()
                                 }),
                             }),
-                            delete_replication: Some(DeleteReplication {
+                            delete_replication: Some(PersistedReplicationStatus {
                                 status: if purge_enabled {
-                                    DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED)
+                                    Status::ENABLED.to_string()
                                 } else {
-                                    DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED)
+                                    Status::DISABLED.to_string()
                                 },
                             }),
-                            destination: Destination {
+                            destination: PersistedReplicationDestination {
                                 bucket: "arn:rustfs:replication:target".to_string(),
                                 ..Default::default()
                             },
@@ -185,7 +186,7 @@ mod tests {
                             prefix: Some(String::new()),
                             priority: Some(1),
                             source_selection_criteria: None,
-                            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                            status: Status::ENABLED.to_string(),
                         }],
                     }),
                 );

crates/ecstore/src/bucket/metadata.rs

@@ -25,11 +25,12 @@ use crate::object_api::WriteCompletion;
 use crate::runtime::sources as runtime_sources;
 use crate::store::ECStore;
 use byteorder::{BigEndian, ByteOrder, LittleEndian};
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_policy::policy::BucketPolicy;
 use s3s::dto::{
     AccelerateConfiguration, BucketLifecycleConfiguration, BucketLoggingStatus, BucketVersioningStatus, CORSConfiguration,
-    NotificationConfiguration, ObjectLockConfiguration, PublicAccessBlockConfiguration, ReplicationConfiguration,
-    RequestPaymentConfiguration, ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration, WebsiteConfiguration,
+    NotificationConfiguration, ObjectLockConfiguration, PublicAccessBlockConfiguration, RequestPaymentConfiguration,
+    ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration, WebsiteConfiguration,
 };
 use serde::Serializer;
 use sha2::{Digest, Sha256};
@@ -495,7 +496,7 @@ pub struct BucketMetadata {
     pub sse_config: Option<ServerSideEncryptionConfiguration>,
     pub tagging_config: Option<Tagging>,
     pub quota_config: Option<BucketQuota>,
-    pub replication_config: Option<ReplicationConfiguration>,
+    pub replication_config: Option<PersistedReplicationConfiguration>,
     pub bucket_target_config: Option<BucketTargets>,
     pub bucket_target_config_meta: Option<HashMap<String, String>>,
     pub cors_config: Option<CORSConfiguration>,
@@ -1043,7 +1044,7 @@ impl BucketMetadata {
                 let config = if data.is_empty() {
                     None
                 } else {
-                    let config = deserialize::<ReplicationConfiguration>(&data)?;
+                    let config = deserialize::<PersistedReplicationConfiguration>(&data)?;
                     if let Some(field) = invalid_replication_config_status_field(&config) {
                         return Err(Error::other(format!("replication field {field} has an invalid status")));
                     }
@@ -1324,8 +1325,8 @@ impl BucketMetadata {
         }
         self.replication_config = None;
         if !self.replication_config_xml.is_empty()
-            && let Err(e) =
-                deserialize::<ReplicationConfiguration>(&self.replication_config_xml).map(|c| self.replication_config = Some(c))
+            && let Err(e) = deserialize::<PersistedReplicationConfiguration>(&self.replication_config_xml)
+                .map(|c| self.replication_config = Some(c))
         {
             tracing::warn!(
                 event = "bucket_metadata_parse_failed",

crates/ecstore/src/bucket/metadata_sys.rs

@@ -30,9 +30,9 @@ use crate::storage_api_contracts::heal::HealOperations as _;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use crate::store::{ECStore, await_bucket_namespace_operation};
 use futures::future::join_all;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_heal_contracts::heal_channel::HealOpts;
 use rustfs_policy::policy::BucketPolicy;
-use s3s::dto::ReplicationConfiguration;
 use s3s::dto::{
     AccelerateConfiguration, BucketLifecycleConfiguration, BucketLoggingStatus, CORSConfiguration, NotificationConfiguration,
     ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, PublicAccessBlockConfiguration,
@@ -1417,7 +1417,7 @@ fn quota_config_and_incarnation_from_authority(
     }
 }
 
-pub async fn get_replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
+pub async fn get_replication_config(bucket: &str) -> Result<(PersistedReplicationConfiguration, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
 
@@ -1427,7 +1427,7 @@ pub async fn get_replication_config(bucket: &str) -> Result<(ReplicationConfigur
 pub(crate) async fn get_replication_config_in(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
-) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
+) -> Result<(PersistedReplicationConfiguration, OffsetDateTime)> {
     let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
 
@@ -3040,7 +3040,7 @@ impl BucketMetadataSys {
         }
     }
 
-    pub async fn get_replication_config(&self, bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
+    pub async fn get_replication_config(&self, bucket: &str) -> Result<(PersistedReplicationConfiguration, OffsetDateTime)> {
         let (bm, _) = self.get_config(bucket).await?;
 
         if !bm.replication_config_xml.is_empty() && bm.replication_config.is_none() {

crates/ecstore/src/bucket/replication/replication_config_boundary.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub use rustfs_replication::config_cache_serde;
 pub use rustfs_replication::{
     ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS,
     REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION,

crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use s3s::dto::ReplicationConfiguration;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 
 use super::replication_config_boundary::ReplicationConfigurationExt as _;
 use super::replication_filemeta_boundary::{
@@ -42,7 +42,7 @@ impl ReplicationLifecycleBridge {
         dead_code,
         reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
     )]
-    pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
+    pub(crate) fn new_config(config: PersistedReplicationConfiguration) -> ReplicationLifecycleConfig {
         ReplicationConfig::new(Some(config), None)
     }
 
@@ -118,16 +118,17 @@ impl ReplicationLifecycleBridge {
 
 #[cfg(test)]
 mod tests {
-    use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus};
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{PersistedReplicationDestination, PersistedReplicationRule};
 
     use super::super::replication_filemeta_boundary::{ReplicateTargetDecision, VersionPurgeStatusType};
     use super::*;
 
-    fn replication_rule() -> ReplicationRule {
-        ReplicationRule {
+    fn replication_rule() -> PersistedReplicationRule {
+        PersistedReplicationRule {
             delete_marker_replication: None,
             delete_replication: None,
-            destination: Destination {
+            destination: PersistedReplicationDestination {
                 bucket: "arn:aws:s3:::target-bucket".to_string(),
                 ..Default::default()
             },
@@ -137,13 +138,13 @@ mod tests {
             prefix: Some(String::new()),
             priority: Some(1),
             source_selection_criteria: None,
-            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            status: Status::ENABLED.to_string(),
         }
     }
 
     #[test]
     fn has_pending_version_purge_preserves_replication_active_rule_behavior() {
-        let config = ReplicationLifecycleBridge::new_config(ReplicationConfiguration {
+        let config = ReplicationLifecycleBridge::new_config(PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule()],
         });

crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs

@@ -17,8 +17,8 @@ use std::sync::Arc;
 use crate::bucket::{metadata::BucketMetadata, metadata_sys};
 use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
 use crate::runtime::instance::InstanceContext;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_utils::path::path_join_buf;
-use s3s::dto::ReplicationConfiguration;
 use time::OffsetDateTime;
 
 use super::replication_error_boundary::{Error, Result};
@@ -36,11 +36,11 @@ impl ReplicationMetadataStore {
     pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin";
     pub(crate) const FORCE_DELETE_REPLICATION_TRANSACTION_LOCK: &'static str = "config/replication/force-delete.bin.transaction";
 
-    pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
+    pub(crate) async fn replication_config(bucket: &str) -> Result<(PersistedReplicationConfiguration, OffsetDateTime)> {
         metadata_sys::get_replication_config(bucket).await
     }
 
-    pub(crate) async fn optional_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
+    pub(crate) async fn optional_replication_config(bucket: &str) -> Result<Option<PersistedReplicationConfiguration>> {
         let config = match Self::replication_config(bucket).await {
             Ok((config, _)) => Some(config),
             Err(err) => {

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -16,7 +16,9 @@ use std::{collections::HashMap, fmt, sync::Arc};
 
 use super::replication_filemeta_boundary::metadata_keys;
 use crate::bucket::metadata::BucketMetadata;
-use s3s::dto::{BucketVersioningStatus, ReplicationConfiguration, ReplicationRuleStatus, VersioningConfiguration};
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
+use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
 use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
 use tracing::error;
@@ -39,7 +41,7 @@ use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore};
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
 
-pub(crate) async fn get_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
+pub(crate) async fn get_replication_config(bucket: &str) -> Result<Option<PersistedReplicationConfiguration>> {
     let config = ReplicationMetadataStore::optional_replication_config(bucket).await?;
     validate_delete_replication_config(&VersioningConfiguration::default(), config.as_ref())?;
     Ok(config)
@@ -64,7 +66,7 @@ impl DeleteReplicationConfigSnapshot {
     #[cfg(test)]
     pub(crate) fn from_configs_for_test(
         versioning: VersioningConfiguration,
-        replication: Option<ReplicationConfiguration>,
+        replication: Option<PersistedReplicationConfiguration>,
     ) -> Self {
         let metadata = replication.map(|config| {
             let mut metadata = BucketMetadata::new("test-bucket");
@@ -78,7 +80,7 @@ impl DeleteReplicationConfigSnapshot {
         &self.versioning
     }
 
-    pub fn replication_config(&self) -> Option<&ReplicationConfiguration> {
+    pub fn replication_config(&self) -> Option<&PersistedReplicationConfiguration> {
         self.metadata
             .as_ref()
             .and_then(|metadata| metadata.replication_config.as_ref())
@@ -101,7 +103,7 @@ impl DeleteReplicationConfigSnapshot {
     pub(crate) fn active_delete_marker_rules_require_tags(&self, object: &str) -> bool {
         self.replication_config().is_some_and(|config| {
             config.rules.iter().any(|rule| {
-                if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
+                if rule.status == Status::DISABLED.as_str() {
                     return false;
                 }
                 if !object.starts_with(rule.prefix()) {
@@ -122,7 +124,7 @@ impl DeleteReplicationConfigSnapshot {
 
 fn validate_delete_replication_config(
     versioning: &VersioningConfiguration,
-    config: Option<&ReplicationConfiguration>,
+    config: Option<&PersistedReplicationConfiguration>,
 ) -> Result<()> {
     if versioning
         .status
@@ -144,7 +146,7 @@ fn validate_delete_replication_config(
         let role = config.role.trim();
         let mut role_destination = None;
         for rule in &config.rules {
-            if rule.status.as_str() == ReplicationRuleStatus::ENABLED {
+            if rule.status.as_str() == Status::ENABLED.as_str() {
                 let destination = rule.destination.bucket.trim();
                 if role.is_empty() && destination.is_empty() {
                     return Err(super::replication_error_boundary::Error::other(
@@ -169,7 +171,7 @@ fn validate_delete_replication_config(
     Ok(())
 }
 
-fn replication_config_from_metadata(metadata: &BucketMetadata) -> Result<Option<&ReplicationConfiguration>> {
+fn replication_config_from_metadata(metadata: &BucketMetadata) -> Result<Option<&PersistedReplicationConfiguration>> {
     if !metadata.replication_config_xml.is_empty() && metadata.replication_config.is_none() {
         return Err(super::replication_error_boundary::Error::other(
             "persisted bucket replication configuration is invalid",
@@ -250,12 +252,15 @@ pub(crate) async fn load_delete_replication_config_in(
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct ReplicationConfig {
-    pub config: Option<ReplicationConfiguration>,
+    /// Carried in the scanner's persisted data usage cache, so its serde form
+    /// is the one the s3s DTO wrote there (see `config_cache_serde` in the replication crate).
+    #[serde(default, with = "super::replication_config_boundary::config_cache_serde")]
+    pub config: Option<PersistedReplicationConfiguration>,
     pub remotes: Option<BucketTargets>,
 }
 
 impl ReplicationConfig {
-    pub fn new(config: Option<ReplicationConfiguration>, remotes: Option<BucketTargets>) -> Self {
+    pub fn new(config: Option<PersistedReplicationConfiguration>, remotes: Option<BucketTargets>) -> Self {
         Self { config, remotes }
     }
 
@@ -468,7 +473,7 @@ fn check_replicate_delete_with_config(
     oi: &ObjectInfo,
     del_opts: &ObjectOptions,
     source_error: bool,
-    config: Option<&ReplicationConfiguration>,
+    config: Option<&PersistedReplicationConfiguration>,
     existing_delete_marker: bool,
     trust_persisted_replica_status: bool,
 ) -> ReplicateDecision {
@@ -587,7 +592,7 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic
 }
 
 fn metadata_target_should_replicate(
-    cfg: &ReplicationConfiguration,
+    cfg: &PersistedReplicationConfiguration,
     opts: &ObjectOpts,
     mopts: &MustReplicateOptions,
     arn: &str,
@@ -597,20 +602,21 @@ fn metadata_target_should_replicate(
 
 #[cfg(test)]
 mod tests {
-    use s3s::dto::{
-        DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
-        ReplicaModifications, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, SourceSelectionCriteria, Tag,
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedOptionalReplicationStatus, PersistedReplicationDestination, PersistedReplicationFilter,
+        PersistedReplicationRule, PersistedReplicationStatus, PersistedReplicationTag, PersistedSourceSelectionCriteria,
     };
 
     use super::super::replication_filemeta_boundary::VersionPurgeStatusType;
     use super::super::replication_target_boundary::BucketTarget;
     use super::*;
 
-    fn replication_rule() -> ReplicationRule {
-        ReplicationRule {
+    fn replication_rule() -> PersistedReplicationRule {
+        PersistedReplicationRule {
             delete_marker_replication: None,
             delete_replication: None,
-            destination: Destination {
+            destination: PersistedReplicationDestination {
                 bucket: "arn:aws:s3:::target-bucket".to_string(),
                 ..Default::default()
             },
@@ -620,7 +626,7 @@ mod tests {
             prefix: Some(String::new()),
             priority: Some(1),
             source_selection_criteria: None,
-            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            status: Status::ENABLED.to_string(),
         }
     }
 
@@ -629,11 +635,11 @@ mod tests {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let mut rule = replication_rule();
         rule.destination.bucket = arn.to_string();
-        rule.filter = Some(ReplicationRuleFilter {
+        rule.filter = Some(PersistedReplicationFilter {
             prefix: Some("admitted/".to_string()),
             ..Default::default()
         });
-        let cfg = ReplicationConfiguration {
+        let cfg = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -671,7 +677,7 @@ mod tests {
         assert!(!empty.replicate(&ObjectOpts::default()));
 
         let config = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![replication_rule()],
             }),
@@ -725,7 +731,7 @@ mod tests {
     fn delete_snapshot_rejects_enabled_rules_without_a_destination() {
         let mut rule = replication_rule();
         rule.destination.bucket.clear();
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -741,7 +747,7 @@ mod tests {
         let first = replication_rule();
         let mut second = replication_rule();
         second.destination.bucket = "arn:aws:s3:::other-target".to_string();
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: "arn:aws:s3:::role-target".to_string(),
             rules: vec![first, second],
         };
@@ -758,41 +764,41 @@ mod tests {
         assert!(validate_delete_replication_config(&invalid_versioning, None).is_err());
 
         let mut invalid_rule = replication_rule();
-        invalid_rule.status = "Enabld".to_string().into();
-        let config = ReplicationConfiguration {
+        invalid_rule.status = "Enabld".to_string();
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![invalid_rule],
         };
         assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
 
         let mut invalid_delete = replication_rule();
-        invalid_delete.delete_replication = Some(DeleteReplication {
-            status: "Enabld".to_string().into(),
+        invalid_delete.delete_replication = Some(PersistedReplicationStatus {
+            status: "Enabld".to_string(),
         });
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![invalid_delete],
         };
         assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
 
         let mut invalid_delete_marker = replication_rule();
-        invalid_delete_marker.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some("Enabld".to_string().into()),
+        invalid_delete_marker.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some("Enabld".to_string()),
         });
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![invalid_delete_marker],
         };
         assert!(validate_delete_replication_config(&VersioningConfiguration::default(), Some(&config)).is_err());
 
         let mut invalid_replica_modifications = replication_rule();
-        invalid_replica_modifications.source_selection_criteria = Some(SourceSelectionCriteria {
-            replica_modifications: Some(ReplicaModifications {
-                status: "Enabld".to_string().into(),
+        invalid_replica_modifications.source_selection_criteria = Some(PersistedSourceSelectionCriteria {
+            replica_modifications: Some(PersistedReplicationStatus {
+                status: "Enabld".to_string(),
             }),
             sse_kms_encrypted_objects: None,
         });
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![invalid_replica_modifications],
         };
@@ -808,7 +814,7 @@ mod tests {
             ..Default::default()
         });
         metadata.replication_config_xml = b"configured".to_vec();
-        metadata.replication_config = Some(ReplicationConfiguration {
+        metadata.replication_config = Some(PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule()],
         });
@@ -848,7 +854,7 @@ mod tests {
         );
 
         let mut inconsistent = BucketMetadata::new("bucket");
-        inconsistent.replication_config = Some(ReplicationConfiguration {
+        inconsistent.replication_config = Some(PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule()],
         });
@@ -863,11 +869,11 @@ mod tests {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let mut rule = replication_rule();
         rule.destination.bucket = arn.to_string();
-        rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        rule.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::ENABLED.to_string()),
         });
         let mut metadata = BucketMetadata::new("bucket");
-        metadata.replication_config = Some(ReplicationConfiguration {
+        metadata.replication_config = Some(PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         });
@@ -898,8 +904,8 @@ mod tests {
     fn delete_marker_source_read_is_required_only_for_tag_filtered_rules() {
         let mut prefix_rule = replication_rule();
         prefix_rule.prefix = Some("logs/".to_string());
-        prefix_rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        prefix_rule.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::ENABLED.to_string()),
         });
 
         let prefix_snapshot = DeleteReplicationConfigSnapshot::from_configs_for_test(
@@ -907,18 +913,18 @@ mod tests {
                 status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                 ..Default::default()
             },
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![prefix_rule],
             }),
         );
 
         let mut tag_rule = replication_rule();
-        tag_rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+        tag_rule.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::ENABLED.to_string()),
         });
-        tag_rule.filter = Some(ReplicationRuleFilter {
-            tag: Some(Tag {
+        tag_rule.filter = Some(PersistedReplicationFilter {
+            tag: Some(PersistedReplicationTag {
                 key: Some("class".to_string()),
                 value: Some("audit".to_string()),
             }),
@@ -929,7 +935,7 @@ mod tests {
                 status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                 ..Default::default()
             },
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![tag_rule],
             }),
@@ -944,14 +950,14 @@ mod tests {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let mut rule = replication_rule();
         rule.destination.bucket = arn.to_string();
-        rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        rule.delete_replication = Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         });
-        rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        rule.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::DISABLED.to_string()),
         });
         let config = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![rule],
             }),
@@ -985,16 +991,16 @@ mod tests {
     #[test]
     fn live_delete_does_not_trust_persisted_replica_status() {
         let mut rule = replication_rule();
-        rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        rule.delete_replication = Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         });
-        rule.source_selection_criteria = Some(SourceSelectionCriteria {
-            replica_modifications: Some(ReplicaModifications {
-                status: s3s::dto::ReplicaModificationsStatus::from_static(s3s::dto::ReplicaModificationsStatus::DISABLED),
+        rule.source_selection_criteria = Some(PersistedSourceSelectionCriteria {
+            replica_modifications: Some(PersistedReplicationStatus {
+                status: Status::DISABLED.to_string(),
             }),
             sse_kms_encrypted_objects: None,
         });
-        let replication = ReplicationConfiguration {
+        let replication = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -1036,14 +1042,14 @@ mod tests {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let mut rule = replication_rule();
         rule.destination.bucket = arn.to_string();
-        rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        rule.delete_replication = Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         });
-        rule.delete_marker_replication = Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        rule.delete_marker_replication = Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::DISABLED.to_string()),
         });
         let config = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![rule],
             }),

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -4689,9 +4689,10 @@ mod tests {
     #[tokio::test]
     async fn heal_queue_marks_missing_versioning_state_as_missed() {
         use super::super::replication_target_boundary::BucketTargets;
-        use s3s::dto::{
-            DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule,
-            ReplicationRuleStatus,
+        use rustfs_gateway_types::dto::Status;
+        use rustfs_gateway_types::persistence::{
+            PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationRule,
+            PersistedReplicationStatus,
         };
 
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
@@ -4706,14 +4707,14 @@ mod tests {
                 ..Default::default()
             },
             ReplicationConfig::new(
-                Some(ReplicationConfiguration {
+                Some(PersistedReplicationConfiguration {
                     role: String::new(),
-                    rules: vec![ReplicationRule {
+                    rules: vec![PersistedReplicationRule {
                         delete_marker_replication: None,
-                        delete_replication: Some(DeleteReplication {
-                            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                        delete_replication: Some(PersistedReplicationStatus {
+                            status: Status::ENABLED.to_string(),
                         }),
-                        destination: Destination {
+                        destination: PersistedReplicationDestination {
                             bucket: arn.to_string(),
                             ..Default::default()
                         },
@@ -4723,7 +4724,7 @@ mod tests {
                         prefix: Some(String::new()),
                         priority: Some(1),
                         source_selection_criteria: None,
-                        status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                        status: Status::ENABLED.to_string(),
                     }],
                 }),
                 Some(BucketTargets::default()),

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -84,6 +84,8 @@ use http_body_util::StreamBody;
 use metrics::counter;
 #[cfg(test)]
 use rmp_serde;
+#[cfg(test)]
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_s3_types::EventName;
 use rustfs_utils::http::{
     AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_TAGGING_DIRECTIVE,
@@ -91,8 +93,6 @@ use rustfs_utils::http::{
     insert_str, replication_target_versions,
 };
 use rustfs_utils::{DEFAULT_SIP_HASH_KEY, get_env_usize, sip_hash};
-#[cfg(test)]
-use s3s::dto::ReplicationConfiguration;
 use std::collections::{HashMap, HashSet};
 use std::fmt::Display;
 use std::pin::Pin;
@@ -6256,10 +6256,11 @@ mod tests {
 
     use super::super::replication_target_boundary::{BucketTarget, BucketTargets};
     use super::*;
-    use s3s::dto::{
-        BucketVersioningStatus, DeleteReplication, DeleteReplicationStatus, Destination, ExcludedPrefix, ReplicationRule,
-        ReplicationRuleStatus, VersioningConfiguration,
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedReplicationDestination, PersistedReplicationRule, PersistedReplicationStatus,
     };
+    use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration};
     use std::collections::HashMap;
     use time::OffsetDateTime;
     use uuid::Uuid;
@@ -7816,7 +7817,7 @@ mod tests {
             ..Default::default()
         };
         let rcfg = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: role.to_string(),
                 rules: vec![],
             }),
@@ -7883,14 +7884,14 @@ mod tests {
             },
         );
         let rcfg = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
+                rules: vec![PersistedReplicationRule {
                     delete_marker_replication: None,
-                    delete_replication: Some(DeleteReplication {
-                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                    delete_replication: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: arn.to_string(),
                         ..Default::default()
                     },
@@ -7900,7 +7901,7 @@ mod tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             }),
             Some(BucketTargets {
@@ -7936,14 +7937,14 @@ mod tests {
         let admitted_arn = "arn:rustfs:replication:us-east-1:target:admitted";
         let current_role = "arn:rustfs:replication:us-east-1:target:current";
         let rcfg = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: current_role.to_string(),
-                rules: vec![ReplicationRule {
+                rules: vec![PersistedReplicationRule {
                     delete_marker_replication: None,
-                    delete_replication: Some(DeleteReplication {
-                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+                    delete_replication: Some(PersistedReplicationStatus {
+                        status: Status::DISABLED.to_string(),
                     }),
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: current_role.to_string(),
                         ..Default::default()
                     },
@@ -7953,7 +7954,7 @@ mod tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             }),
             Some(BucketTargets {
@@ -8001,12 +8002,12 @@ mod tests {
     #[tokio::test]
     async fn heal_rejects_semantically_invalid_replication_config() {
         let rcfg = ReplicationConfig::new(
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
+                rules: vec![PersistedReplicationRule {
                     delete_marker_replication: None,
                     delete_replication: None,
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: "arn:rustfs:replication:us-east-1:target:bucket".to_string(),
                         ..Default::default()
                     },
@@ -8016,7 +8017,7 @@ mod tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static("Enabld"),
+                    status: "Enabld".to_owned(),
                 }],
             }),
             Some(BucketTargets::default()),

crates/ecstore/src/bucket/replication/replication_target_config_bridge.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use s3s::dto::ReplicationConfiguration;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 
 use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt};
 use super::replication_filemeta_boundary::ReplicationType;
@@ -20,7 +20,7 @@ use super::replication_filemeta_boundary::ReplicationType;
 pub(crate) struct ReplicationTargetConfigBridge;
 
 impl ReplicationTargetConfigBridge {
-    pub(crate) fn target_is_used_by_rules(config: &ReplicationConfiguration, arn: &str) -> bool {
+    pub(crate) fn target_is_used_by_rules(config: &PersistedReplicationConfiguration, arn: &str) -> bool {
         config
             .filter_target_arns(&ObjectOpts {
                 op_type: ReplicationType::All,
@@ -37,7 +37,7 @@ mod tests {
 
     #[test]
     fn target_config_bridge_matches_role_target() {
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: " arn:target ".to_string(),
             rules: Vec::new(),
         };

crates/ecstore/src/bucket/utils.rs

@@ -15,6 +15,9 @@
 use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
 use crate::error::{Error, Result, StorageError};
 use regex::Regex;
+use rustfs_gateway_types::persistence::{
+    PersistedReplicationConfiguration, PersistenceCodecError, parse_replication, serialize_replication,
+};
 use rustfs_utils::path::SLASH_SEPARATOR;
 use s3s::xml;
 use tracing::instrument;
@@ -73,38 +76,94 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
     check_bucket_name_common(bucket_name, true)
 }
 
+/// A bucket configuration persisted as XML in the bucket metadata, together
+/// with the codec that reads and writes its stored bytes.
+///
+/// The gateway persistence codecs (`rustfs_gateway_types::persistence`)
+/// replace the s3s XML codec one configuration family at a time
+/// (rustfs/backlog#2745): a family that has moved implements this trait with
+/// its gateway codec, and every other family still implements it with the s3s
+/// codec it was persisted with. Both must keep the persisted bytes the
+/// `bucket_config_goldens` pin.
+pub trait BucketConfigXml: Sized {
+    /// Decodes the stored bytes of one configuration document.
+    fn decode_config_xml(input: &[u8]) -> std::result::Result<Self, BucketConfigXmlError>;
+
+    /// Encodes the value as RustFS stores it.
+    fn encode_config_xml(&self) -> std::result::Result<Vec<u8>, BucketConfigXmlError>;
+}
+
+/// Why a persisted bucket configuration could not be decoded or encoded.
+#[derive(Debug, thiserror::Error)]
+pub enum BucketConfigXmlError {
+    /// The s3s XML decoder of a family that has not moved yet refused the bytes.
+    #[error(transparent)]
+    LegacyDecode(#[from] xml::DeError),
+    /// The s3s XML encoder of a family that has not moved yet failed.
+    #[error(transparent)]
+    LegacyEncode(#[from] xml::SerError),
+    /// A gateway persistence codec refused the bytes or the value.
+    #[error(transparent)]
+    Persisted(#[from] PersistenceCodecError),
+}
+
+/// Decodes one persisted bucket configuration document.
+pub fn deserialize<T: BucketConfigXml>(input: &[u8]) -> std::result::Result<T, BucketConfigXmlError> {
+    T::decode_config_xml(input)
+}
+
+/// Encodes one bucket configuration as RustFS persists it.
+pub fn serialize<T: BucketConfigXml>(val: &T) -> std::result::Result<Vec<u8>, BucketConfigXmlError> {
+    val.encode_config_xml()
+}
+
+impl BucketConfigXml for PersistedReplicationConfiguration {
+    fn decode_config_xml(input: &[u8]) -> std::result::Result<Self, BucketConfigXmlError> {
+        Ok(parse_replication(input)?)
+    }
+
+    fn encode_config_xml(&self) -> std::result::Result<Vec<u8>, BucketConfigXmlError> {
+        Ok(serialize_replication(self)?)
+    }
+}
+
 // RUSTFS_COMPAT_TODO(s3gate-metadata-xml): the s3s codec reads persisted XML during migration. Remove after every supported writer uses the gateway codec and every retained metadata object and backup archive is verified or rewritten.
-pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
-where
-    T: for<'xml> xml::Deserialize<'xml>,
-{
-    let mut d = xml::Deserializer::new(input);
-    let ans = T::deserialize(&mut d)?;
-    d.expect_eof()?;
-    Ok(ans)
+macro_rules! legacy_s3s_bucket_config_xml {
+    ($($ty:ty),+ $(,)?) => {$(
+        impl BucketConfigXml for $ty {
+            fn decode_config_xml(input: &[u8]) -> std::result::Result<Self, BucketConfigXmlError> {
+                let mut d = xml::Deserializer::new(input);
+                let ans = <$ty as xml::Deserialize>::deserialize(&mut d)?;
+                d.expect_eof()?;
+                Ok(ans)
+            }
+
+            fn encode_config_xml(&self) -> std::result::Result<Vec<u8>, BucketConfigXmlError> {
+                let mut buf = Vec::with_capacity(256);
+                {
+                    let mut ser = xml::Serializer::new(&mut buf);
+                    <$ty as xml::Serialize>::serialize(self, &mut ser)?;
+                }
+                Ok(buf)
+            }
+        }
+    )+};
 }
 
-#[allow(
-    dead_code,
-    reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
-)]
-pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
-    let mut buf = Vec::with_capacity(256);
-    {
-        let mut ser = xml::Serializer::new(&mut buf);
-        val.serialize_content(&mut ser)?;
-    }
-    Ok(String::from_utf8(buf).unwrap())
-}
-
-pub fn serialize<T: xml::Serialize>(val: &T) -> xml::SerResult<Vec<u8>> {
-    let mut buf = Vec::with_capacity(256);
-    {
-        let mut ser = xml::Serializer::new(&mut buf);
-        val.serialize(&mut ser)?;
-    }
-    Ok(buf)
-}
+legacy_s3s_bucket_config_xml!(
+    s3s::dto::AccelerateConfiguration,
+    s3s::dto::BucketLifecycleConfiguration,
+    s3s::dto::BucketLoggingStatus,
+    s3s::dto::CORSConfiguration,
+    s3s::dto::NotificationConfiguration,
+    s3s::dto::ObjectLockConfiguration,
+    s3s::dto::PublicAccessBlockConfiguration,
+    s3s::dto::RequestPaymentConfiguration,
+    s3s::dto::ServerSideEncryptionConfiguration,
+    s3s::dto::Tagging,
+    s3s::dto::VersioningConfiguration,
+    s3s::dto::WebsiteConfiguration,
+);
 
 pub fn has_bad_path_component(path: &str) -> bool {
     let n = path.len();

crates/ecstore/src/core/pools.rs

@@ -79,12 +79,13 @@ use rand::RngExt as _;
 use rmp_serde::Deserializer;
 use rmp_serde::Serializer;
 use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_heal_contracts::heal_channel::HealOpts;
 use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
 use rustfs_utils::path::{
     decode_dir_object, encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path,
 };
-use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration};
+use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::{HashMap, HashSet};
@@ -13945,7 +13946,7 @@ impl ECStore {
         set: Arc<SetDisks>,
         lifecycle_config: Option<BucketLifecycleConfiguration>,
         object_lock_config: Option<ObjectLockConfiguration>,
-        replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
+        replication_config: Option<(PersistedReplicationConfiguration, OffsetDateTime)>,
         expected_bucket_incarnation_id: Option<uuid::Uuid>,
         source_changed_exhaustions: Arc<AtomicUsize>,
         entry_budget: Arc<Semaphore>,
@@ -14115,7 +14116,7 @@ impl ECStore {
         set: Arc<SetDisks>,
         lifecycle_config: Option<BucketLifecycleConfiguration>,
         object_lock_config: Option<ObjectLockConfiguration>,
-        replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
+        replication_config: Option<(PersistedReplicationConfiguration, OffsetDateTime)>,
         expected_bucket_incarnation_id: Option<uuid::Uuid>,
         source_changed_exhaustions: Arc<AtomicUsize>,
         entry_budget: Arc<Semaphore>,
@@ -14223,7 +14224,7 @@ impl ECStore {
         bi: DecomBucketInfo,
         lifecycle_config: Option<BucketLifecycleConfiguration>,
         object_lock_config: Option<ObjectLockConfiguration>,
-        replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
+        replication_config: Option<(PersistedReplicationConfiguration, OffsetDateTime)>,
         expected_bucket_incarnation_id: Option<uuid::Uuid>,
         source_changed_exhaustions: Arc<AtomicUsize>,
         entry_budget: Arc<Semaphore>,
@@ -14848,7 +14849,7 @@ impl ECStore {
         set: Arc<SetDisks>,
         lifecycle_config: Option<BucketLifecycleConfiguration>,
         object_lock_config: Option<ObjectLockConfiguration>,
-        replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
+        replication_config: Option<(PersistedReplicationConfiguration, OffsetDateTime)>,
         expected_bucket_incarnation_id: Option<uuid::Uuid>,
         source_changed_exhaustions: Arc<AtomicUsize>,
         attempts: &mut usize,
@@ -15005,7 +15006,7 @@ impl ECStore {
         set: Arc<SetDisks>,
         lifecycle_config: Option<BucketLifecycleConfiguration>,
         object_lock_config: Option<ObjectLockConfiguration>,
-        replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>,
+        replication_config: Option<(PersistedReplicationConfiguration, OffsetDateTime)>,
         expected_bucket_incarnation_id: Option<uuid::Uuid>,
         entry_attempt: usize,
         scanner_backlog_handoff_attempt: usize,

crates/ecstore/src/error/mod.rs

@@ -1076,6 +1076,12 @@ impl From<s3s::xml::DeError> for Error {
     }
 }
 
+impl From<crate::bucket::utils::BucketConfigXmlError> for Error {
+    fn from(e: crate::bucket::utils::BucketConfigXmlError) -> Self {
+        Error::other(e)
+    }
+}
+
 impl From<tonic::Status> for Error {
     fn from(e: tonic::Status) -> Self {
         // Keep the typed status as the io::Error payload instead of a

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -67,8 +67,8 @@ use crate::error::{Error, Result};
 use crate::storage_api_contracts::range::HTTPRangeSpec;
 use rustfs_filemeta::TRANSITION_COMPLETE;
 use rustfs_filemeta::{FileInfo, MetaCacheEntry};
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_rio::Index;
-use s3s::dto::ReplicationConfiguration;
 use serde::Serialize;
 use std::io::Cursor;
 use std::sync::Arc;
@@ -2629,7 +2629,7 @@ fn test_resolve_rebalance_optional_bucket_config_result_passthrough() {
     let result = resolve_rebalance_optional_bucket_config_result(
         "bucket-a",
         "replication",
-        Ok((ReplicationConfiguration::default(), OffsetDateTime::UNIX_EPOCH)),
+        Ok((PersistedReplicationConfiguration::default(), OffsetDateTime::UNIX_EPOCH)),
     )
     .expect("bucket config should pass through");
     assert!(result.is_some());

crates/ecstore/src/services/rebalance/types.rs

@@ -36,7 +36,7 @@ pub(super) struct RebalanceBucketConfigs {
     pub(super) bucket_incarnation_id: Option<uuid::Uuid>,
     pub(super) lifecycle_config: Option<s3s::dto::BucketLifecycleConfiguration>,
     pub(super) object_lock_config: Option<s3s::dto::ObjectLockConfiguration>,
-    pub(super) replication_config: Option<(s3s::dto::ReplicationConfiguration, OffsetDateTime)>,
+    pub(super) replication_config: Option<(rustfs_gateway_types::persistence::PersistedReplicationConfiguration, OffsetDateTime)>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]

crates/ecstore/src/set_disk/ops/object.rs

@@ -21882,11 +21882,13 @@ mod delete_objects_lock_gating_tests {
 
     #[tokio::test]
     async fn batch_version_delete_uses_tags_read_under_the_delete_lock() {
-        use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
-        use s3s::dto::{
-            BucketVersioningStatus, DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration,
-            ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, Tag, VersioningConfiguration,
+        use rustfs_gateway_types::dto::Status;
+        use rustfs_gateway_types::persistence::{
+            PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationFilter,
+            PersistedReplicationRule, PersistedReplicationStatus, PersistedReplicationTag,
         };
+        use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
+        use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
 
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
         let bucket = "batch-delete-locked-tags";
@@ -21925,20 +21927,20 @@ mod delete_objects_lock_gating_tests {
                 status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                 ..Default::default()
             },
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
+                rules: vec![PersistedReplicationRule {
                     delete_marker_replication: None,
-                    delete_replication: Some(DeleteReplication {
-                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                    delete_replication: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: arn.to_string(),
                         ..Default::default()
                     },
                     existing_object_replication: None,
-                    filter: Some(ReplicationRuleFilter {
-                        tag: Some(Tag {
+                    filter: Some(PersistedReplicationFilter {
+                        tag: Some(PersistedReplicationTag {
                             key: Some("generation".to_string()),
                             value: Some("locked".to_string()),
                         }),
@@ -21948,7 +21950,7 @@ mod delete_objects_lock_gating_tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             }),
         ));
@@ -21986,10 +21988,12 @@ mod delete_objects_lock_gating_tests {
     #[tokio::test]
     #[serial]
     async fn lifecycle_delete_all_history_records_exact_replication_purge() {
-        use s3s::dto::{
-            BucketVersioningStatus, DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration,
-            ReplicationRule, ReplicationRuleStatus, VersioningConfiguration,
+        use rustfs_gateway_types::dto::Status;
+        use rustfs_gateway_types::persistence::{
+            PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationRule,
+            PersistedReplicationStatus,
         };
+        use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
 
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
         let bucket = "lifecycle-delete-all-replication";
@@ -22033,14 +22037,14 @@ mod delete_objects_lock_gating_tests {
                 status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                 ..Default::default()
             },
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
+                rules: vec![PersistedReplicationRule {
                     delete_marker_replication: None,
-                    delete_replication: Some(DeleteReplication {
-                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                    delete_replication: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: arn.to_string(),
                         ..Default::default()
                     },
@@ -22050,7 +22054,7 @@ mod delete_objects_lock_gating_tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             }),
         ));
@@ -22092,11 +22096,12 @@ mod delete_objects_lock_gating_tests {
 
     #[tokio::test]
     async fn synthetic_directory_delete_uses_decoded_prefix_and_marker_switch() {
-        use s3s::dto::{
-            BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication,
-            DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter,
-            ReplicationRuleStatus, VersioningConfiguration,
+        use rustfs_gateway_types::dto::Status;
+        use rustfs_gateway_types::persistence::{
+            PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+            PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationStatus,
         };
+        use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
 
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
         let bucket = "batch-directory-replication";
@@ -22112,21 +22117,21 @@ mod delete_objects_lock_gating_tests {
                 status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
                 ..Default::default()
             },
-            Some(ReplicationConfiguration {
+            Some(PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
-                    delete_marker_replication: Some(DeleteMarkerReplication {
-                        status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+                rules: vec![PersistedReplicationRule {
+                    delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                        status: Some(Status::ENABLED.to_string()),
                     }),
-                    delete_replication: Some(DeleteReplication {
-                        status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+                    delete_replication: Some(PersistedReplicationStatus {
+                        status: Status::DISABLED.to_string(),
                     }),
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: arn.to_string(),
                         ..Default::default()
                     },
                     existing_object_replication: None,
-                    filter: Some(ReplicationRuleFilter {
+                    filter: Some(PersistedReplicationFilter {
                         prefix: Some("photos/".to_string()),
                         ..Default::default()
                     }),
@@ -22134,7 +22139,7 @@ mod delete_objects_lock_gating_tests {
                     prefix: None,
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             }),
         ));

crates/ecstore/tests/bucket_config_goldens.rs

@@ -26,7 +26,9 @@
 //! metadata store, or choosing a canonical shape.
 //!
 //! Upstream: the bucket-config `deserialize`/`serialize` pair, reached through the
-//! integration-test facade `tests/storage_api.rs`, and the s3s DTOs they are generic over. Downstream: the DTO/codec
+//! integration-test facade `tests/storage_api.rs`, and the configuration types they are
+//! generic over (`BucketConfigXml`: the gateway persistence shapes for the families that
+//! have moved, the s3s DTOs for the rest). Downstream: the DTO/codec
 //! replacement (T1.4) and the persisted-bytes comparison (T3.7), which must
 //! keep every golden green without regenerating it. Regeneration is reserved
 //! to `scripts/gen_bucket_config_goldens.sh`, which sets
@@ -35,23 +37,25 @@
 
 mod storage_api;
 
+use rustfs_gateway_types::dto::{Status, StorageClass};
+use rustfs_gateway_types::persistence::{
+    PersistedOptionalReplicationStatus, PersistedReplicationAnd, PersistedReplicationConfiguration,
+    PersistedReplicationDestination, PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationStatus,
+    PersistedReplicationTag, PersistedSourceSelectionCriteria,
+};
 use s3s::dto::{
     AbortIncompleteMultipartUpload, AccelerateConfiguration, BucketAccelerateStatus, BucketLifecycleConfiguration,
     BucketLoggingStatus, BucketVersioningStatus, CORSConfiguration, CORSRule, Condition, DefaultRetention, DelMarkerExpiration,
-    DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
-    ErrorDocument, Event, ExcludedPrefix, ExistingObjectReplication, ExistingObjectReplicationStatus, ExpirationStatus,
-    FilterRule, FilterRuleName, IndexDocument, LifecycleExpiration, LifecycleRule, LifecycleRuleAndOperator, LifecycleRuleFilter,
-    LoggingEnabled, NoncurrentVersionExpiration, NotificationConfiguration, NotificationConfigurationFilter,
-    ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, Payer, PublicAccessBlockConfiguration,
-    QueueConfiguration, Redirect, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
-    ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationRuleStatus, RequestPaymentConfiguration, RoutingRule,
-    S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
-    ServerSideEncryptionRule, SourceSelectionCriteria, StorageClass, Tag, Tagging, Transition, TransitionStorageClass,
+    ErrorDocument, Event, ExcludedPrefix, ExpirationStatus, FilterRule, FilterRuleName, IndexDocument, LifecycleExpiration,
+    LifecycleRule, LifecycleRuleAndOperator, LifecycleRuleFilter, LoggingEnabled, NoncurrentVersionExpiration,
+    NotificationConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled,
+    ObjectLockRetentionMode, ObjectLockRule, Payer, PublicAccessBlockConfiguration, QueueConfiguration, Redirect,
+    RequestPaymentConfiguration, RoutingRule, S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault,
+    ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging, Transition, TransitionStorageClass,
     VersioningConfiguration, WebsiteConfiguration,
 };
-use s3s::xml;
 use std::path::{Path, PathBuf};
-use storage_api::bucket_config_codec::{deserialize, serialize};
+use storage_api::bucket_config_codec::{BucketConfigXml, deserialize, serialize};
 
 /// Set only by `scripts/gen_bucket_config_goldens.sh`. Every pair test then
 /// writes its `.out.xml` (and, for the `rustfs` shape, its `.in.xml`) and
@@ -83,7 +87,7 @@ fn contains(haystack: &[u8], needle: &[u8]) -> bool {
 
 fn round_trip<T>(input: &[u8], label: &str) -> Vec<u8>
 where
-    T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
+    T: BucketConfigXml,
 {
     let parsed: T = deserialize(input).unwrap_or_else(|err| panic!("{label}: deserialize failed: {err}"));
     serialize(&parsed).unwrap_or_else(|err| panic!("{label}: serialize failed: {err}"))
@@ -106,7 +110,7 @@ fn assert_bytes_eq(actual: &[u8], expected: &[u8], context: &str) {
 /// the `rustfs` shape; the `aws` and `minio` shapes pass `None`.
 fn check_pair<T>(family: &str, shape: &str, typed: Option<&T>)
 where
-    T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
+    T: BucketConfigXml,
 {
     let in_path = pair_path(family, shape, "in");
     let out_path = pair_path(family, shape, "out");
@@ -291,41 +295,44 @@ mod rustfs_values {
         }
     }
 
-    pub(super) fn replication() -> ReplicationConfiguration {
-        ReplicationConfiguration {
+    pub(super) fn replication() -> PersistedReplicationConfiguration {
+        PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+            rules: vec![PersistedReplicationRule {
+                delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                    status: Some(Status::ENABLED.to_string()),
                 }),
-                delete_replication: Some(DeleteReplication {
-                    status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+                delete_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:rustfs:replication::0b6f3c2a-5e6a-4d2b-9c1e-7a8f9d0e1b2c:interop-dr".to_owned(),
-                    storage_class: Some(StorageClass::from_static(StorageClass::STANDARD)),
+                    storage_class: Some(StorageClass::STANDARD.to_string()),
                     ..Default::default()
                 },
-                existing_object_replication: Some(ExistingObjectReplication {
-                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
-                filter: Some(ReplicationRuleFilter {
-                    and: Some(ReplicationRuleAndOperator {
+                filter: Some(PersistedReplicationFilter {
+                    and: Some(PersistedReplicationAnd {
                         prefix: Some("docs/".to_owned()),
-                        tags: Some(vec![tag("replicate", "yes")]),
+                        tags: Some(vec![PersistedReplicationTag {
+                            key: Some("replicate".to_owned()),
+                            value: Some("yes".to_owned()),
+                        }]),
                     }),
                     ..Default::default()
                 }),
                 id: Some("rustfs-dr".to_owned()),
                 prefix: None,
                 priority: Some(1),
-                source_selection_criteria: Some(SourceSelectionCriteria {
-                    replica_modifications: Some(ReplicaModifications {
-                        status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
+                source_selection_criteria: Some(PersistedSourceSelectionCriteria {
+                    replica_modifications: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
                     sse_kms_encrypted_objects: None,
                 }),
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         }
     }
@@ -451,7 +458,7 @@ mod bucket_config_goldens {
     /// element is skipped on read and never re-emitted on write.
     #[test]
     fn replication_aws() {
-        check_pair::<ReplicationConfiguration>("replication", "aws", None);
+        check_pair::<PersistedReplicationConfiguration>("replication", "aws", None);
         let input = read(&pair_path("replication", "aws", "in"));
         let output = read(&pair_path("replication", "aws", "out"));
         assert!(
@@ -464,7 +471,7 @@ mod bucket_config_goldens {
             String::from_utf8_lossy(&output)
         );
     }
-    golden!(replication_minio, "replication", "minio", ReplicationConfiguration);
+    golden!(replication_minio, "replication", "minio", PersistedReplicationConfiguration);
     golden!(replication_rustfs, "replication", rustfs = rustfs_values::replication());
 
     golden!(cors_aws, "cors", "aws", CORSConfiguration);
@@ -495,3 +502,151 @@ mod bucket_config_goldens {
         rustfs = rustfs_values::public_access_block()
     );
 }
+
+mod bucket_config_goldens_carrier_compat {
+    use super::storage_api::replication_compat::{BucketTargets, ReplicationConfig};
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationFilter, PersistedReplicationRule,
+    };
+    use serde::{Deserialize, Serialize};
+
+    fn replication_rule() -> PersistedReplicationRule {
+        PersistedReplicationRule {
+            delete_marker_replication: None,
+            delete_replication: None,
+            destination: PersistedReplicationDestination {
+                bucket: "arn:aws:s3:::target-bucket".to_string(),
+                ..Default::default()
+            },
+            existing_object_replication: None,
+            filter: None,
+            id: Some("rule".to_string()),
+            prefix: Some(String::new()),
+            priority: Some(1),
+            source_selection_criteria: None,
+            status: Status::ENABLED.to_string(),
+        }
+    }
+
+    #[test]
+    fn replication_cache_matches_the_legacy_reader_for_nested_tag_cache_shapes() {
+        #[derive(Serialize, Deserialize)]
+        struct LegacyReplicationConfig {
+            config: Option<s3s::dto::ReplicationConfiguration>,
+            remotes: Option<BucketTargets>,
+        }
+
+        fn encode(value: &impl Serialize, named: bool) -> Vec<u8> {
+            if named {
+                rmp_serde::to_vec_named(value).expect("encode the actual cache carrier as a named map")
+            } else {
+                rmp_serde::to_vec(value).expect("encode the actual cache carrier as an array")
+            }
+        }
+
+        let mut rule = replication_rule();
+        rule.prefix = None;
+        rule.filter = Some(PersistedReplicationFilter {
+            prefix: Some("cache-marker/".to_string()),
+            ..Default::default()
+        });
+        let expected = PersistedReplicationConfiguration {
+            role: "arn:aws:iam::123456789012:role/replication".to_string(),
+            rules: vec![rule],
+        };
+        let legacy = LegacyReplicationConfig {
+            config: Some(s3s::dto::ReplicationConfiguration {
+                role: expected.role.clone(),
+                rules: vec![s3s::dto::ReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: None,
+                    destination: s3s::dto::Destination {
+                        bucket: "arn:aws:s3:::target-bucket".to_string(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: Some(s3s::dto::ReplicationRuleFilter {
+                        prefix: Some("cache-marker/".to_string()),
+                        ..Default::default()
+                    }),
+                    id: Some("rule".to_string()),
+                    prefix: None,
+                    priority: Some(1),
+                    source_selection_criteria: None,
+                    status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                }],
+            }),
+            remotes: None,
+        };
+        let mut mismatches = Vec::new();
+        for named in [false, true] {
+            let bytes = encode(&legacy, named);
+            let marker = if named {
+                let mut marker = rmp_serde::to_vec("cached_tags").expect("encode the legacy cache field name");
+                marker.push(0x80);
+                marker
+            } else {
+                let mut marker = vec![0x80];
+                marker.extend(rmp_serde::to_vec("cache-marker/").expect("encode the unique filter prefix"));
+                marker
+            };
+            let position = bytes
+                .windows(marker.len())
+                .position(|window| window == marker)
+                .expect("the real legacy serializer must write the nested empty tag cache");
+            let cache_position = position + if named { marker.len() - 1 } else { 0 };
+            for (name, cache, valid) in [
+                ("empty-map", &[0x80][..], true),
+                ("string-map", &[0x81, 0xa1, b'k', 0xa1, b'v'][..], true),
+                ("nil", &[0xc0][..], false),
+                ("scalar", &[0x01][..], false),
+                ("array", &[0x90][..], false),
+                ("nonstring-key", &[0x81, 0x01, 0xa1, b'v'][..], false),
+                ("nonstring-value", &[0x81, 0xa1, b'k', 0x01][..], false),
+            ] {
+                let mut input = bytes.clone();
+                input.splice(cache_position..=cache_position, cache.iter().copied());
+                let old = rmp_serde::from_slice::<LegacyReplicationConfig>(&input);
+                let new = rmp_serde::from_slice::<ReplicationConfig>(&input);
+                assert_eq!(old.is_ok(), valid, "the real old reader's contract for {name}, named={named}");
+                if new.is_ok() != old.is_ok() {
+                    mismatches.push(format!("{name}, named={named}: old={:?}, new={}", old.as_ref().err(), new.is_ok()));
+                }
+                if let (Ok(old), Ok(new)) = (old, new) {
+                    assert_eq!(old.config, legacy.config, "the old reader must discard only cached tags");
+                    assert_eq!(
+                        new.config.as_ref(),
+                        Some(&expected),
+                        "the new reader must preserve every configuration field"
+                    );
+                    let reread = rmp_serde::from_slice::<LegacyReplicationConfig>(&encode(&new, named))
+                        .expect("the real old reader must read the newly serialized carrier");
+                    assert_eq!(reread.config, legacy.config, "the new writer must preserve the old configuration");
+                    assert!(reread.remotes.is_none(), "the absent remote configuration must remain absent");
+                }
+            }
+            let absent = LegacyReplicationConfig {
+                config: None,
+                remotes: None,
+            };
+            let input = encode(&absent, named);
+            let new = rmp_serde::from_slice::<ReplicationConfig>(&input).expect("read an absent optional configuration");
+            assert!(new.config.is_none());
+            assert_eq!(
+                encode(&new, named),
+                input,
+                "an absent configuration must keep its actual carrier encoding"
+            );
+        }
+        let missing =
+            rmp_serde::to_vec_named(&serde_json::json!({"remotes": null})).expect("encode the named map without config");
+        let old = rmp_serde::from_slice::<LegacyReplicationConfig>(&missing).expect("the old reader permits an omitted config");
+        assert!(old.config.is_none());
+        let new = rmp_serde::from_slice::<ReplicationConfig>(&missing);
+        if !matches!(&new, Ok(value) if value.config.is_none()) {
+            mismatches.push(format!("omitted config, named=true: old=None, new={new:?}"));
+        }
+        assert!(mismatches.is_empty(), "cache acceptance must match the real old reader: {mismatches:?}");
+    }
+}

crates/ecstore/tests/replication_facade_compat_test.rs

@@ -14,7 +14,7 @@
 
 mod storage_api;
 
-use s3s::dto::ReplicationConfiguration;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use std::future::Future;
 use storage_api::contract_compat::{Error, ObjectInfo, ObjectOptions, ObjectToDelete};
 use storage_api::replication_compat::{
@@ -40,7 +40,7 @@ fn assert_strict_delete_future<F: Future<Output = Result<ReplicateDecision, Erro
 
 #[test]
 fn replication_facade_exports_config_extension_contract() {
-    assert_replication_config_ext::<ReplicationConfiguration>();
+    assert_replication_config_ext::<PersistedReplicationConfiguration>();
 }
 
 #[test]
@@ -73,7 +73,7 @@ fn replication_facade_exports_runtime_and_dto_types() {
 
 #[test]
 fn replication_facade_exports_app_storage_helper_contracts() {
-    let config = ReplicationConfiguration::default();
+    let config = PersistedReplicationConfiguration::default();
     let target_arns = replication_target_arns(&config);
     assert!(target_arns.is_empty());
 

crates/ecstore/tests/storage_api.rs

@@ -43,12 +43,13 @@ pub(crate) mod manual_transition {
 pub(crate) mod replication_compat {
     pub(crate) use rustfs_ecstore::api::bucket::replication::{
         BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
-        ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationObjectBridge,
-        ReplicationState, ReplicationStats, ReplicationTargetValidationError, ResyncStatusType,
+        ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource,
+        ReplicationObjectBridge, ReplicationState, ReplicationStats, ReplicationTargetValidationError, ResyncStatusType,
         delete_replication_state_from_config, delete_replication_version_id, replication_statuses_map, replication_target_arns,
         should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info,
         should_use_existing_delete_replication_source, validate_replication_config_target_arns,
     };
+    pub(crate) use rustfs_ecstore::api::bucket::target::BucketTargets;
 }
 
 pub(crate) mod legacy_bitrot_read {
@@ -74,5 +75,5 @@ pub(crate) mod pinned_erasure_fixtures {
 }
 
 pub(crate) mod bucket_config_codec {
-    pub(crate) use rustfs_ecstore::api::bucket::utils::{deserialize, serialize};
+    pub(crate) use rustfs_ecstore::api::bucket::utils::{BucketConfigXml, deserialize, serialize};
 }

crates/replication/Cargo.toml

@@ -38,7 +38,7 @@ byteorder.workspace = true
 regex.workspace = true
 rmp.workspace = true
 rmp-serde.workspace = true
-s3s = { workspace = true, features = ["minio"] }
+rustfs-gateway-types.workspace = true
 serde = { workspace = true, features = ["derive"] }
 time = { workspace = true, features = ["parsing", "formatting", "macros", "serde"] }
 tokio = { workspace = true, features = ["sync"] }

crates/replication/src/config_cache_serde.rs

@@ -0,0 +1,576 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The serde form of a replication configuration inside the scanner's data
+//! usage cache.
+//!
+//! Responsible for: encoding `PersistedReplicationConfiguration` with serde in
+//! exactly the layout the s3s DTO's derived serde produced, because the
+//! ECStore `ReplicationConfig` carrying it is a member of the scanner's
+//! `DataUsageCacheInfo`, which RustFS persists with MessagePack: a cache written
+//! before the gateway types must still decode, and an older reader must still
+//! decode a cache written now.
+//!
+//! Not responsible for: the persisted bucket-metadata XML (that is the gateway
+//! persistence codec) or deciding whether the cache should carry the
+//! configuration at all.
+//!
+//! The layout is the derived one of the s3s `minio` DTOs: every struct is its
+//! fields in declaration order (alphabetical), every string enum is its wire
+//! text, and the rule filter carries the s3s tag cache as an empty map.
+//!
+//! Upstream: the serde derive of ECStore's `ReplicationConfig`, through
+//! `#[serde(with = ...)]`. Downstream: none.
+
+use rustfs_gateway_types::persistence::{
+    PersistedAccessControlTranslation, PersistedEncryptionConfiguration, PersistedOptionalReplicationStatus,
+    PersistedReplicationAnd, PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationFilter,
+    PersistedReplicationMetrics, PersistedReplicationRule, PersistedReplicationStatus, PersistedReplicationTag,
+    PersistedReplicationTime, PersistedReplicationTimeValue, PersistedSourceSelectionCriteria,
+};
+use serde::de::{MapAccess, Visitor};
+use serde::ser::SerializeMap;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+pub fn serialize<S: Serializer>(value: &Option<PersistedReplicationConfiguration>, serializer: S) -> Result<S::Ok, S::Error> {
+    value.as_ref().map(CacheConfiguration::from).serialize(serializer)
+}
+
+pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Option<PersistedReplicationConfiguration>, D::Error> {
+    Ok(Option::<CacheConfiguration>::deserialize(deserializer)?.map(Into::into))
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheConfiguration {
+    role: String,
+    rules: Vec<CacheRule>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheRule {
+    delete_marker_replication: Option<CacheOptionalStatus>,
+    delete_replication: Option<CacheStatus>,
+    destination: CacheDestination,
+    existing_object_replication: Option<CacheStatus>,
+    filter: Option<CacheFilter>,
+    id: Option<String>,
+    prefix: Option<String>,
+    priority: Option<i32>,
+    source_selection_criteria: Option<CacheSourceSelection>,
+    status: String,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheOptionalStatus {
+    status: Option<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheStatus {
+    status: String,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheDestination {
+    access_control_translation: Option<CacheAccessControlTranslation>,
+    account: Option<String>,
+    bucket: String,
+    encryption_configuration: Option<CacheEncryptionConfiguration>,
+    metrics: Option<CacheMetrics>,
+    replication_time: Option<CacheReplicationTime>,
+    storage_class: Option<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheAccessControlTranslation {
+    owner: String,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheEncryptionConfiguration {
+    replica_kms_key_id: Option<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheMetrics {
+    event_threshold: Option<CacheTimeValue>,
+    status: String,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheReplicationTime {
+    status: String,
+    time: CacheTimeValue,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheTimeValue {
+    minutes: Option<i32>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheFilter {
+    and: Option<CacheAnd>,
+    cached_tags: EmptyTagCache,
+    prefix: Option<String>,
+    tag: Option<CacheTag>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheAnd {
+    prefix: Option<String>,
+    tags: Option<Vec<CacheTag>>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheTag {
+    key: Option<String>,
+    value: Option<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+struct CacheSourceSelection {
+    replica_modifications: Option<CacheStatus>,
+    sse_kms_encrypted_objects: Option<CacheStatus>,
+}
+
+/// The s3s rule filter's parse cache: written as an empty map, and whatever
+/// an older writer stored there is read and dropped.
+struct EmptyTagCache;
+
+impl Serialize for EmptyTagCache {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_map(Some(0))?.end()
+    }
+}
+
+impl<'de> Deserialize<'de> for EmptyTagCache {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        struct TagCacheVisitor;
+
+        impl<'de> Visitor<'de> for TagCacheVisitor {
+            type Value = EmptyTagCache;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("a map")
+            }
+
+            fn visit_map<M: MapAccess<'de>>(self, mut access: M) -> Result<Self::Value, M::Error> {
+                while access.next_entry::<String, String>()?.is_some() {}
+                Ok(EmptyTagCache)
+            }
+        }
+
+        deserializer.deserialize_map(TagCacheVisitor)
+    }
+}
+
+impl From<&PersistedReplicationConfiguration> for CacheConfiguration {
+    fn from(value: &PersistedReplicationConfiguration) -> Self {
+        Self {
+            role: value.role.clone(),
+            rules: value.rules.iter().map(CacheRule::from).collect(),
+        }
+    }
+}
+
+impl From<CacheConfiguration> for PersistedReplicationConfiguration {
+    fn from(value: CacheConfiguration) -> Self {
+        Self {
+            role: value.role,
+            rules: value.rules.into_iter().map(Into::into).collect(),
+        }
+    }
+}
+
+impl From<&PersistedReplicationRule> for CacheRule {
+    fn from(value: &PersistedReplicationRule) -> Self {
+        Self {
+            delete_marker_replication: value.delete_marker_replication.as_ref().map(|wrapper| CacheOptionalStatus {
+                status: wrapper.status.clone(),
+            }),
+            delete_replication: value.delete_replication.as_ref().map(CacheStatus::from),
+            destination: CacheDestination::from(&value.destination),
+            existing_object_replication: value.existing_object_replication.as_ref().map(CacheStatus::from),
+            filter: value.filter.as_ref().map(CacheFilter::from),
+            id: value.id.clone(),
+            prefix: value.prefix.clone(),
+            priority: value.priority,
+            source_selection_criteria: value.source_selection_criteria.as_ref().map(|criteria| CacheSourceSelection {
+                replica_modifications: criteria.replica_modifications.as_ref().map(CacheStatus::from),
+                sse_kms_encrypted_objects: criteria.sse_kms_encrypted_objects.as_ref().map(CacheStatus::from),
+            }),
+            status: value.status.clone(),
+        }
+    }
+}
+
+impl From<CacheRule> for PersistedReplicationRule {
+    fn from(value: CacheRule) -> Self {
+        Self {
+            delete_marker_replication: value
+                .delete_marker_replication
+                .map(|wrapper| PersistedOptionalReplicationStatus { status: wrapper.status }),
+            delete_replication: value.delete_replication.map(Into::into),
+            destination: value.destination.into(),
+            existing_object_replication: value.existing_object_replication.map(Into::into),
+            filter: value.filter.map(Into::into),
+            id: value.id,
+            prefix: value.prefix,
+            priority: value.priority,
+            source_selection_criteria: value
+                .source_selection_criteria
+                .map(|criteria| PersistedSourceSelectionCriteria {
+                    replica_modifications: criteria.replica_modifications.map(Into::into),
+                    sse_kms_encrypted_objects: criteria.sse_kms_encrypted_objects.map(Into::into),
+                }),
+            status: value.status,
+        }
+    }
+}
+
+impl From<&PersistedReplicationStatus> for CacheStatus {
+    fn from(value: &PersistedReplicationStatus) -> Self {
+        Self {
+            status: value.status.clone(),
+        }
+    }
+}
+
+impl From<CacheStatus> for PersistedReplicationStatus {
+    fn from(value: CacheStatus) -> Self {
+        Self { status: value.status }
+    }
+}
+
+impl From<&PersistedReplicationDestination> for CacheDestination {
+    fn from(value: &PersistedReplicationDestination) -> Self {
+        Self {
+            access_control_translation: value.access_control_translation.as_ref().map(|translation| {
+                CacheAccessControlTranslation {
+                    owner: translation.owner.clone(),
+                }
+            }),
+            account: value.account.clone(),
+            bucket: value.bucket.clone(),
+            encryption_configuration: value
+                .encryption_configuration
+                .as_ref()
+                .map(|configuration| CacheEncryptionConfiguration {
+                    replica_kms_key_id: configuration.replica_kms_key_id.clone(),
+                }),
+            metrics: value.metrics.as_ref().map(|metrics| CacheMetrics {
+                event_threshold: metrics.event_threshold.as_ref().map(CacheTimeValue::from),
+                status: metrics.status.clone(),
+            }),
+            replication_time: value.replication_time.as_ref().map(|replication_time| CacheReplicationTime {
+                status: replication_time.status.clone(),
+                time: CacheTimeValue::from(&replication_time.time),
+            }),
+            storage_class: value.storage_class.clone(),
+        }
+    }
+}
+
+impl From<CacheDestination> for PersistedReplicationDestination {
+    fn from(value: CacheDestination) -> Self {
+        Self {
+            account: value.account,
+            access_control_translation: value
+                .access_control_translation
+                .map(|translation| PersistedAccessControlTranslation {
+                    owner: translation.owner,
+                }),
+            bucket: value.bucket,
+            encryption_configuration: value
+                .encryption_configuration
+                .map(|configuration| PersistedEncryptionConfiguration {
+                    replica_kms_key_id: configuration.replica_kms_key_id,
+                }),
+            metrics: value.metrics.map(|metrics| PersistedReplicationMetrics {
+                event_threshold: metrics.event_threshold.map(Into::into),
+                status: metrics.status,
+            }),
+            replication_time: value.replication_time.map(|replication_time| PersistedReplicationTime {
+                status: replication_time.status,
+                time: replication_time.time.into(),
+            }),
+            storage_class: value.storage_class,
+        }
+    }
+}
+
+impl From<&PersistedReplicationTimeValue> for CacheTimeValue {
+    fn from(value: &PersistedReplicationTimeValue) -> Self {
+        Self { minutes: value.minutes }
+    }
+}
+
+impl From<CacheTimeValue> for PersistedReplicationTimeValue {
+    fn from(value: CacheTimeValue) -> Self {
+        Self { minutes: value.minutes }
+    }
+}
+
+impl From<&PersistedReplicationFilter> for CacheFilter {
+    fn from(value: &PersistedReplicationFilter) -> Self {
+        Self {
+            and: value.and.as_ref().map(|and| CacheAnd {
+                prefix: and.prefix.clone(),
+                tags: and.tags.as_ref().map(|tags| tags.iter().map(CacheTag::from).collect()),
+            }),
+            cached_tags: EmptyTagCache,
+            prefix: value.prefix.clone(),
+            tag: value.tag.as_ref().map(CacheTag::from),
+        }
+    }
+}
+
+impl From<CacheFilter> for PersistedReplicationFilter {
+    fn from(value: CacheFilter) -> Self {
+        Self {
+            and: value.and.map(|and| PersistedReplicationAnd {
+                prefix: and.prefix,
+                tags: and.tags.map(|tags| tags.into_iter().map(Into::into).collect()),
+            }),
+            prefix: value.prefix,
+            tag: value.tag.map(Into::into),
+        }
+    }
+}
+
+impl From<&PersistedReplicationTag> for CacheTag {
+    fn from(value: &PersistedReplicationTag) -> Self {
+        Self {
+            key: value.key.clone(),
+            value: value.value.clone(),
+        }
+    }
+}
+
+impl From<CacheTag> for PersistedReplicationTag {
+    fn from(value: CacheTag) -> Self {
+        Self {
+            key: value.key,
+            value: value.value,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A newtype carrying the field exactly as `ReplicationConfig` does, so
+    /// the bytes below are the ones the cache holds for that field.
+    #[derive(Debug, PartialEq, Serialize, Deserialize)]
+    struct Field(#[serde(with = "super")] Option<PersistedReplicationConfiguration>);
+
+    /// The derived s3s serde of `Some(every_member())`, as the data usage cache
+    /// writes it (`rmp_serde::Serializer::new`), recorded from s3s rev
+    /// 5761ddfe with the `minio` feature.
+    const S3S_COMPACT: &str = concat!(
+        "92d92a61726e3a6177733a69616d3a3a3132333435363738393031323a726f6c652f7265706c69636174696f6e929a91",
+        "a7456e61626c656491a7456e61626c65649791ab44657374696e6174696f6eac313233343536373839303132b361726e",
+        "3a6177733a73333a3a3a6261636b757091a76b6d732d6b657992910fa7456e61626c656492a7456e61626c6564911ea8",
+        "5354414e4441524491a844697361626c65649492a5646f63732f9292a47465616da773746f7261676592c0c080a76669",
+        "6c7465722f92a3656e76a470726f64ac65766572792d6d656d626572a76c65676163792f079291a7456e61626c656491",
+        "a844697361626c6564a7456e61626c65649ac0c097c0c0b361726e3a6177733a73333a3a3a7365636f6e64c0c0c0c0c0",
+        "c0c0c0c0c0a844697361626c6564",
+    );
+
+    /// The same value written with field names (`rmp_serde::to_vec_named`).
+    const S3S_NAMED: &str = concat!(
+        "82a4726f6c65d92a61726e3a6177733a69616d3a3a3132333435363738393031323a726f6c652f7265706c6963617469",
+        "6f6ea572756c6573928ab964656c6574655f6d61726b65725f7265706c69636174696f6e81a6737461747573a7456e61",
+        "626c6564b264656c6574655f7265706c69636174696f6e81a6737461747573a7456e61626c6564ab64657374696e6174",
+        "696f6e87ba6163636573735f636f6e74726f6c5f7472616e736c6174696f6e81a56f776e6572ab44657374696e617469",
+        "6f6ea76163636f756e74ac313233343536373839303132a66275636b6574b361726e3a6177733a73333a3a3a6261636b",
+        "7570b8656e6372797074696f6e5f636f6e66696775726174696f6e81b27265706c6963615f6b6d735f6b65795f6964a7",
+        "6b6d732d6b6579a76d65747269637382af6576656e745f7468726573686f6c6481a76d696e757465730fa67374617475",
+        "73a7456e61626c6564b07265706c69636174696f6e5f74696d6582a6737461747573a7456e61626c6564a474696d6581",
+        "a76d696e757465731ead73746f726167655f636c617373a85354414e44415244bb6578697374696e675f6f626a656374",
+        "5f7265706c69636174696f6e81a6737461747573a844697361626c6564a666696c74657284a3616e6482a67072656669",
+        "78a5646f63732fa4746167739282a36b6579a47465616da576616c7565a773746f7261676582a36b6579c0a576616c75",
+        "65c0ab6361636865645f7461677380a6707265666978a766696c7465722fa374616782a36b6579a3656e76a576616c75",
+        "65a470726f64a26964ac65766572792d6d656d626572a6707265666978a76c65676163792fa87072696f7269747907b9",
+        "736f757263655f73656c656374696f6e5f637269746572696182b57265706c6963615f6d6f64696669636174696f6e73",
+        "81a6737461747573a7456e61626c6564b97373655f6b6d735f656e637279707465645f6f626a6563747381a673746174",
+        "7573a844697361626c6564a6737461747573a7456e61626c65648ab964656c6574655f6d61726b65725f7265706c6963",
+        "6174696f6ec0b264656c6574655f7265706c69636174696f6ec0ab64657374696e6174696f6e87ba6163636573735f63",
+        "6f6e74726f6c5f7472616e736c6174696f6ec0a76163636f756e74c0a66275636b6574b361726e3a6177733a73333a3a",
+        "3a7365636f6e64b8656e6372797074696f6e5f636f6e66696775726174696f6ec0a76d657472696373c0b07265706c69",
+        "636174696f6e5f74696d65c0ad73746f726167655f636c617373c0bb6578697374696e675f6f626a6563745f7265706c",
+        "69636174696f6ec0a666696c746572c0a26964c0a6707265666978c0a87072696f72697479c0b9736f757263655f7365",
+        "6c656374696f6e5f6372697465726961c0a6737461747573a844697361626c6564",
+    );
+
+    fn bytes(hex: &str) -> Vec<u8> {
+        (0..hex.len())
+            .step_by(2)
+            .map(|index| u8::from_str_radix(&hex[index..index + 2], 16).expect("fixture hex is valid"))
+            .collect()
+    }
+
+    fn encode(value: Option<PersistedReplicationConfiguration>) -> Vec<u8> {
+        let mut buf = Vec::new();
+        Field(value)
+            .serialize(&mut rmp_serde::Serializer::new(&mut buf))
+            .expect("the cache encoding never fails");
+        buf
+    }
+
+    fn decode(input: &[u8]) -> Result<Option<PersistedReplicationConfiguration>, rmp_serde::decode::Error> {
+        rmp_serde::from_slice::<Field>(input).map(|field| field.0)
+    }
+
+    fn status(value: &str) -> PersistedReplicationStatus {
+        PersistedReplicationStatus {
+            status: value.to_owned(),
+        }
+    }
+
+    fn tag(key: Option<&str>, value: Option<&str>) -> PersistedReplicationTag {
+        PersistedReplicationTag {
+            key: key.map(str::to_owned),
+            value: value.map(str::to_owned),
+        }
+    }
+
+    fn every_member() -> PersistedReplicationConfiguration {
+        PersistedReplicationConfiguration {
+            role: "arn:aws:iam::123456789012:role/replication".to_owned(),
+            rules: vec![
+                PersistedReplicationRule {
+                    delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                        status: Some("Enabled".to_owned()),
+                    }),
+                    delete_replication: Some(status("Enabled")),
+                    destination: PersistedReplicationDestination {
+                        account: Some("123456789012".to_owned()),
+                        access_control_translation: Some(PersistedAccessControlTranslation {
+                            owner: "Destination".to_owned(),
+                        }),
+                        bucket: "arn:aws:s3:::backup".to_owned(),
+                        encryption_configuration: Some(PersistedEncryptionConfiguration {
+                            replica_kms_key_id: Some("kms-key".to_owned()),
+                        }),
+                        metrics: Some(PersistedReplicationMetrics {
+                            event_threshold: Some(PersistedReplicationTimeValue { minutes: Some(15) }),
+                            status: "Enabled".to_owned(),
+                        }),
+                        replication_time: Some(PersistedReplicationTime {
+                            status: "Enabled".to_owned(),
+                            time: PersistedReplicationTimeValue { minutes: Some(30) },
+                        }),
+                        storage_class: Some("STANDARD".to_owned()),
+                    },
+                    existing_object_replication: Some(status("Disabled")),
+                    filter: Some(PersistedReplicationFilter {
+                        and: Some(PersistedReplicationAnd {
+                            prefix: Some("docs/".to_owned()),
+                            tags: Some(vec![tag(Some("team"), Some("storage")), tag(None, None)]),
+                        }),
+                        prefix: Some("filter/".to_owned()),
+                        tag: Some(tag(Some("env"), Some("prod"))),
+                    }),
+                    id: Some("every-member".to_owned()),
+                    prefix: Some("legacy/".to_owned()),
+                    priority: Some(7),
+                    source_selection_criteria: Some(PersistedSourceSelectionCriteria {
+                        replica_modifications: Some(status("Enabled")),
+                        sse_kms_encrypted_objects: Some(status("Disabled")),
+                    }),
+                    status: "Enabled".to_owned(),
+                },
+                PersistedReplicationRule {
+                    delete_marker_replication: None,
+                    delete_replication: None,
+                    destination: PersistedReplicationDestination {
+                        bucket: "arn:aws:s3:::second".to_owned(),
+                        ..Default::default()
+                    },
+                    existing_object_replication: None,
+                    filter: None,
+                    id: None,
+                    prefix: None,
+                    priority: None,
+                    source_selection_criteria: None,
+                    status: "Disabled".to_owned(),
+                },
+            ],
+        }
+    }
+
+    #[test]
+    fn writes_the_bytes_the_old_derive_wrote() {
+        assert_eq!(encode(Some(every_member())), bytes(S3S_COMPACT));
+    }
+
+    #[test]
+    fn reads_a_cache_the_old_derive_wrote() {
+        assert_eq!(decode(&bytes(S3S_COMPACT)).expect("the old cache field decodes"), Some(every_member()));
+        assert_eq!(decode(&bytes(S3S_NAMED)).expect("the named form decodes"), Some(every_member()));
+    }
+
+    #[test]
+    fn an_absent_configuration_stays_nil() {
+        assert_eq!(encode(None), vec![0xc0]);
+        assert_eq!(decode(&[0xc0]).expect("nil decodes"), None);
+    }
+
+    #[test]
+    fn a_filled_tag_cache_from_an_older_writer_is_dropped_not_refused() {
+        let mut stored = every_member();
+        stored.rules.truncate(1);
+        let mut buf = encode(Some(stored.clone()));
+        // The filter is the fifth member of the rule; its second member is the
+        // empty tag cache map (0x80). Put one entry in it.
+        let position = buf
+            .windows(2)
+            .position(|pair| pair == [0x80, 0xa7])
+            .expect("the empty tag cache is written");
+        buf.splice(position..=position, [0x81, 0xa1, b'k', 0xa1, b'v']);
+        assert_eq!(decode(&buf).expect("a filled tag cache is ignored"), Some(stored));
+    }
+
+    #[test]
+    fn a_truncated_rule_is_refused_rather_than_defaulted() {
+        let mut stored = every_member();
+        stored.rules.truncate(1);
+        let mut buf = encode(Some(stored));
+        // Drop the rule's last member (its status) and shorten its array header.
+        buf.truncate(buf.len() - "Enabled".len() - 1);
+        let header = buf
+            .iter()
+            .position(|byte| *byte == 0x9a)
+            .expect("the rule is a ten-member array");
+        buf[header] = 0x99;
+        assert!(decode(&buf).is_err(), "a rule without its status must not decode into a default status");
+    }
+
+    #[test]
+    fn a_wrong_shape_is_refused() {
+        let buf = rmp_serde::to_vec(&"not a configuration").expect("encode");
+        assert!(decode(&buf).is_err());
+    }
+
+    #[test]
+    fn a_status_outside_the_vocabulary_keeps_its_text() {
+        let mut stored = every_member();
+        stored.rules[0].status = "enabled".to_owned();
+        let crossed = decode(&encode(Some(stored.clone()))).expect("round trip");
+        assert_eq!(crossed, Some(stored));
+    }
+}

crates/replication/src/lib.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod config;
+pub mod config_cache_serde;
 pub mod delete;
 mod filemeta;
 mod http;

crates/replication/src/operation.rs

@@ -18,7 +18,7 @@ use crate::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER,
     SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_STATUS, get_consistent_internal_metadata, get_header_metadata,
 };
-use s3s::dto::ReplicationConfiguration;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use time::OffsetDateTime;
 use uuid::Uuid;
 
@@ -167,7 +167,7 @@ pub struct ReplicationDeleteStateSource {
 }
 
 pub fn delete_replication_state_from_config(
-    config: &ReplicationConfiguration,
+    config: &PersistedReplicationConfiguration,
     source: &ReplicationDeleteStateSource,
 ) -> Option<ReplicationState> {
     let opts = ObjectOpts {
@@ -359,10 +359,10 @@ mod tests {
     };
     use crate::storage_api::ObjectToDelete;
     use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header};
-    use s3s::dto::{
-        DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
-        ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus,
-        ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria,
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+        PersistedReplicationRule, PersistedReplicationStatus, PersistedSourceSelectionCriteria,
     };
     use std::collections::HashMap;
     use time::{Duration, OffsetDateTime};
@@ -546,37 +546,37 @@ 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)),
+    fn delete_replication_rule(arn: &str, replica_modifications: bool) -> PersistedReplicationRule {
+        PersistedReplicationRule {
+            delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                status: Some(Status::ENABLED.to_string()),
             }),
             delete_replication: None,
-            destination: Destination {
+            destination: PersistedReplicationDestination {
                 bucket: arn.to_string(),
                 ..Default::default()
             },
-            existing_object_replication: Some(ExistingObjectReplication {
-                status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+            existing_object_replication: Some(PersistedReplicationStatus {
+                status: Status::ENABLED.to_string(),
             }),
             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),
+            source_selection_criteria: replica_modifications.then_some(PersistedSourceSelectionCriteria {
+                replica_modifications: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
                 sse_kms_encrypted_objects: None,
             }),
-            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            status: Status::ENABLED.to_string(),
         }
     }
 
     #[test]
     fn delete_replication_state_tracks_downstream_delete_marker_targets() {
         let arn = "arn:aws:s3:::target-bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: arn.to_string(),
             rules: vec![delete_replication_rule(arn, true)],
         };
@@ -600,7 +600,7 @@ mod tests {
     #[test]
     fn delete_replication_state_skips_replica_delete_without_replica_modifications() {
         let arn = "arn:aws:s3:::target-bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: arn.to_string(),
             rules: vec![delete_replication_rule(arn, false)],
         };
@@ -619,10 +619,10 @@ mod tests {
     fn delete_replication_state_uses_delete_switch_for_marker_version_purges() {
         let arn = "arn:aws:s3:::target-bucket";
         let mut rule = delete_replication_rule(arn, false);
-        rule.delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
+        rule.delete_replication = Some(PersistedReplicationStatus {
+            status: Status::DISABLED.to_string(),
         });
-        let mut config = ReplicationConfiguration {
+        let mut config = PersistedReplicationConfiguration {
             role: arn.to_string(),
             rules: vec![rule],
         };
@@ -639,8 +639,8 @@ mod tests {
             "delete-marker version purge must not use the enabled marker-creation switch"
         );
 
-        config.rules[0].delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        config.rules[0].delete_replication = Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         });
         let state = delete_replication_state_from_config(&config, &source)
             .expect("delete-marker version purge should honor the enabled permanent-delete switch");

crates/replication/src/rule.rs

@@ -13,15 +13,15 @@
 // limitations under the License.
 
 use crate::config::ObjectOpts;
-use s3s::dto::ReplicaModificationsStatus;
-use s3s::dto::ReplicationRule;
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::PersistedReplicationRule;
 
 pub trait ReplicationRuleExt {
     fn prefix(&self) -> &str;
     fn metadata_replicate(&self, obj: &ObjectOpts) -> bool;
 }
 
-impl ReplicationRuleExt for ReplicationRule {
+impl ReplicationRuleExt for PersistedReplicationRule {
     /// The rule's key prefix: `Filter.Prefix`, else `Filter.And.Prefix`, else
     /// the deprecated top-level `Prefix` of a V1 rule written without a
     /// `<Filter>` (backlog#2367 C-2). A rule that carries both keeps AWS's
@@ -47,8 +47,8 @@ impl ReplicationRuleExt for ReplicationRule {
 
         self.source_selection_criteria.as_ref().is_some_and(|s| {
             s.replica_modifications
-                .clone()
-                .is_some_and(|r| r.status == ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED))
+                .as_ref()
+                .is_some_and(|r| r.status == Status::ENABLED.as_str())
         })
     }
 }

crates/scanner/Cargo.toml

@@ -98,6 +98,7 @@ rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }
 http = { workspace = true }
 rand = { workspace = true, features = ["serde"] }
+rustfs-gateway-types.workspace = true
 s3s = { workspace = true, features = ["minio"] }
 metrics = { workspace = true }
 rustfs-data-usage = { workspace = true }

crates/scanner/src/lib.rs

@@ -342,7 +342,7 @@ pub(crate) async fn get_object_lock_config(
 
 pub(crate) async fn get_replication_config(
     bucket: &str,
-) -> EcstoreResult<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
+) -> EcstoreResult<(rustfs_gateway_types::persistence::PersistedReplicationConfiguration, time::OffsetDateTime)> {
     ecstore_get_replication_config(bucket).await
 }
 
@@ -360,9 +360,9 @@ pub(crate) trait ScannerReplicationConfigExt {
     fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
 }
 
-impl ScannerReplicationConfigExt for s3s::dto::ReplicationConfiguration {
+impl ScannerReplicationConfigExt for rustfs_gateway_types::persistence::PersistedReplicationConfiguration {
     fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
-        <s3s::dto::ReplicationConfiguration as EcstoreReplicationConfigurationExt>::has_active_rules(self, prefix, recursive)
+        <rustfs_gateway_types::persistence::PersistedReplicationConfiguration as EcstoreReplicationConfigurationExt>::has_active_rules(self, prefix, recursive)
     }
 }
 

crates/scanner/src/scanner_io.rs

@@ -29,15 +29,14 @@ use rand::seq::SliceRandom as _;
 use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
 use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo, observed_data_usage_is_newer};
 use rustfs_filemeta::FileMeta;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_heal_contracts::heal_channel::HealScanMode;
 use rustfs_lock::{LockError, NamespaceLockGuard};
 use rustfs_scanner_metrics::metrics::{
     Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics,
 };
 use rustfs_utils::path::path_join_buf;
-use s3s::dto::{
-    BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration,
-};
+use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, VersioningConfiguration};
 use sha2::{Digest as _, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::future::Future;

crates/scanner/src/scanner_io/io_disk.rs

@@ -188,7 +188,7 @@ impl ScannerIODisk for Disk {
             crate::storage_api::ecstore_with_background_disk_io(ctx.clone(), get_replication_config(&cache.info.name))
                 .await
                 .unwrap_or((
-                    ReplicationConfiguration {
+                    PersistedReplicationConfiguration {
                         role: "".to_string(),
                         rules: vec![],
                     },

crates/scanner/src/storage_api.rs

@@ -197,7 +197,10 @@ pub(crate) mod owner {
 pub struct ScannerReplicationConfig(EcstoreReplicationConfig);
 
 impl ScannerReplicationConfig {
-    pub(crate) fn new(config: Option<s3s::dto::ReplicationConfiguration>, remotes: Option<EcstoreBucketTargets>) -> Self {
+    pub(crate) fn new(
+        config: Option<rustfs_gateway_types::persistence::PersistedReplicationConfiguration>,
+        remotes: Option<EcstoreBucketTargets>,
+    ) -> Self {
         Self(EcstoreReplicationConfig::new(config, remotes))
     }
 

deny.toml

@@ -39,6 +39,10 @@ allow-git = [
     # Official s3s repository pinned for Iceberg SigV4 path encoding support.
     # owner: GatewayJ  review: 2026-10
     "https://github.com/s3s-project/s3s",
+    # rustfs-gateway-types, pinned to one gateway commit in the workspace
+    # dependencies while the engine crates move off s3s (rustfs/backlog#2745).
+    # owner: rustfs-maintainers  review: 2027-01
+    "https://github.com/rustfs/gateway",
     # RustFS io_uring backend is now consumed from crates.io at 0.3.0.
     # owner: rustfs-maintainers  review: 2026-10
     # "https://github.com/rustfs/uring",

rustfs/Cargo.toml

@@ -246,6 +246,7 @@ rustfs-notify = { workspace = true }
 rustfs-obs = { workspace = true }
 rustfs-policy = { workspace = true }
 rustfs-protocols = { workspace = true }
+rustfs-gateway-types = { workspace = true }
 rustfs-protos = { workspace = true }
 rustfs-rio = { workspace = true }
 rustfs-s3-client = { workspace = true }

rustfs/src/admin/handlers/account_info.rs

@@ -27,10 +27,12 @@ use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_data_usage::{BucketUsageInfo, DataUsageInfo};
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_policy::policy::BucketPolicy;
 use rustfs_policy::policy::default::DEFAULT_POLICIES;
 use rustfs_policy::policy::{Args, action::Action, action::S3Action};
-use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, ReplicationRuleStatus};
+use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled};
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::Serialize;
@@ -97,11 +99,11 @@ fn object_lock_config_enabled(config: &ObjectLockConfiguration) -> bool {
         .is_some_and(|enabled| enabled.as_str() == ObjectLockEnabled::ENABLED)
 }
 
-fn replication_config_enabled(config: &ReplicationConfiguration) -> bool {
+fn replication_config_enabled(config: &PersistedReplicationConfiguration) -> bool {
     config
         .rules
         .iter()
-        .any(|rule| rule.status.as_str() == ReplicationRuleStatus::ENABLED)
+        .any(|rule| rule.status.as_str() == Status::ENABLED.as_str())
 }
 
 async fn bucket_locking_enabled(bucket: &str) -> bool {
@@ -317,9 +319,9 @@ impl Operation for AccountInfoHandler {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rustfs_gateway_types::persistence::{PersistedReplicationDestination, PersistedReplicationRule};
     use rustfs_madmin::BackendInfo;
     use rustfs_policy::policy::BucketPolicy;
-    use s3s::dto::{Destination, ReplicationRule};
 
     #[test]
     fn accountinfo_response_adds_optional_oidc_display_fields() {
@@ -434,20 +436,20 @@ mod tests {
 
     #[test]
     fn accountinfo_bucket_details_marks_replication_enabled_rules() {
-        let enabled = replication_config_with_status(ReplicationRuleStatus::ENABLED);
-        let disabled = replication_config_with_status(ReplicationRuleStatus::DISABLED);
+        let enabled = replication_config_with_status(Status::ENABLED.as_str());
+        let disabled = replication_config_with_status(Status::DISABLED.as_str());
 
         assert!(replication_config_enabled(&enabled));
         assert!(!replication_config_enabled(&disabled));
     }
 
-    fn replication_config_with_status(status: &'static str) -> ReplicationConfiguration {
-        ReplicationConfiguration {
+    fn replication_config_with_status(status: &str) -> PersistedReplicationConfiguration {
+        PersistedReplicationConfiguration {
             role: "arn:aws:iam::123456789012:role/replication".to_string(),
-            rules: vec![ReplicationRule {
+            rules: vec![PersistedReplicationRule {
                 delete_marker_replication: None,
                 delete_replication: None,
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:aws:s3:::target".to_string(),
                     ..Default::default()
                 },
@@ -457,7 +459,7 @@ mod tests {
                 prefix: None,
                 priority: None,
                 source_selection_criteria: None,
-                status: ReplicationRuleStatus::from_static(status),
+                status: status.to_owned(),
             }],
         }
     }

rustfs/src/admin/handlers/bucket_meta.rs

@@ -39,6 +39,7 @@ use http::{HeaderMap, StatusCode};
 use hyper::Method;
 use matchit::Params;
 use rustfs_config::MAX_BUCKET_METADATA_IMPORT_SIZE;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
 use rustfs_policy::policy::{
     BucketPolicy,
@@ -48,8 +49,8 @@ use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use s3s::{
     Body, S3Request, S3Response, S3Result,
     dto::{
-        BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration, ServerSideEncryptionConfiguration,
-        Tagging, VersioningConfiguration,
+        BucketLifecycleConfiguration, ObjectLockConfiguration, ServerSideEncryptionConfiguration, Tagging,
+        VersioningConfiguration,
     },
     header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE},
     s3_error,
@@ -288,7 +289,7 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> Result<Option<Vec<u
                 .map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
                 .replication_config_xml
                 .clone();
-            let config_xml = checked_raw_xml(&config, raw_config, deserialize::<ReplicationConfiguration>)?;
+            let config_xml = checked_raw_xml(&config, raw_config, deserialize::<PersistedReplicationConfiguration>)?;
 
             Ok(Some(config_xml))
         }
@@ -895,7 +896,7 @@ fn apply_imported_bucket_config(
         }
         BUCKET_REPLICATION_CONFIG => {
             validated_config!(
-                deserialize::<ReplicationConfiguration>,
+                deserialize::<PersistedReplicationConfiguration>,
                 replication_config_xml,
                 replication_config_updated_at
             )
@@ -1940,15 +1941,19 @@ mod backup_zip_compatibility_tests {
         checked_raw_xml(
             &validated_replication,
             DIFFERENT_REPLICATION_XML.to_vec(),
-            deserialize::<ReplicationConfiguration>,
+            deserialize::<PersistedReplicationConfiguration>,
         )
         .expect_err("a different valid revision must not be exported after validating the old revision");
-        checked_raw_xml(&validated_replication, b"not xml".to_vec(), deserialize::<ReplicationConfiguration>)
-            .expect_err("an invalid raw revision must not be exported after validating the old revision");
+        checked_raw_xml(
+            &validated_replication,
+            b"not xml".to_vec(),
+            deserialize::<PersistedReplicationConfiguration>,
+        )
+        .expect_err("an invalid raw revision must not be exported after validating the old revision");
         let matching_raw = checked_raw_xml(
             &validated_replication,
             OLD_REPLICATION_XML.to_vec(),
-            deserialize::<ReplicationConfiguration>,
+            deserialize::<PersistedReplicationConfiguration>,
         )
         .expect("the exact validated raw revision must remain exportable");
         assert_eq!(matching_raw, OLD_REPLICATION_XML);
@@ -2010,8 +2015,12 @@ mod backup_zip_compatibility_tests {
                 "g-zip-002 rollback import did not restore {config_file}"
             );
         }
-        let _: ReplicationConfiguration =
-            deserialize(&restored.replication_config_xml).expect("old parser must read the newly exported archive payload");
+        let mut old_reader = s3s::xml::Deserializer::new(&restored.replication_config_xml);
+        let _: s3s::dto::ReplicationConfiguration =
+            s3s::xml::Deserialize::deserialize(&mut old_reader).expect("old parser must read the newly exported archive payload");
+        old_reader
+            .expect_eof()
+            .expect("old parser must consume the newly exported archive payload completely");
 
         // The array-shaped compatibility fixture is unreadable as targets.
         // A backup must not claim success after silently omitting that setting.

rustfs/src/admin/handlers/site_replication.rs

@@ -54,6 +54,8 @@ use http::Uri;
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_config::{DEFAULT_DELIMITER, MAX_ADMIN_REQUEST_BODY_SIZE};
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::{PersistedReplicationConfiguration, PersistedReplicationRule};
 use rustfs_iam::error::is_err_no_such_service_account;
 use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
 use rustfs_iam::store::object::ObjectStore;
@@ -75,10 +77,7 @@ use rustfs_policy::policy::{
     Policy,
     action::{Action, AdminAction},
 };
-use s3s::dto::{
-    DeleteMarkerReplicationStatus, DeleteReplicationStatus, ExistingObjectReplicationStatus, ObjectLockConfiguration,
-    ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, VersioningConfiguration,
-};
+use s3s::dto::{ObjectLockConfiguration, VersioningConfiguration};
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::Deserialize;
 use serde::Serialize;
@@ -2419,26 +2418,25 @@ fn canonical_status_json(value: &Value) -> Value {
     }
 }
 
-fn site_replication_rule_complete(rule: &ReplicationRule, owner_deployment_id: &str) -> bool {
+fn site_replication_rule_complete(rule: &PersistedReplicationRule, owner_deployment_id: &str) -> bool {
     let delete_marker_enabled = rule
         .delete_marker_replication
         .as_ref()
         .and_then(|delete_marker| delete_marker.status.as_ref())
-        .is_some_and(|status| status == &DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED));
+        .is_some_and(|status| status == Status::ENABLED.as_str());
     let delete_enabled = rule
         .delete_replication
         .as_ref()
-        .is_some_and(|delete| delete.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED));
-    let existing_object_enabled = rule.existing_object_replication.as_ref().is_some_and(|existing| {
-        existing.status == ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED)
-    });
+        .is_some_and(|delete| delete.status == Status::ENABLED.as_str());
+    let existing_object_enabled = rule
+        .existing_object_replication
+        .as_ref()
+        .is_some_and(|existing| existing.status == Status::ENABLED.as_str());
     let replica_modifications_enabled = rule
         .source_selection_criteria
         .as_ref()
         .and_then(|criteria| criteria.replica_modifications.as_ref())
-        .is_some_and(|replica_modifications| {
-            replica_modifications.status == ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED)
-        });
+        .is_some_and(|replica_modifications| replica_modifications.status == Status::ENABLED.as_str());
 
     // A rule whose destination ARN names the site that holds it can never replicate:
     // `reconcile_site_replication_bucket_targets` skips the local peer, so no bucket
@@ -2448,7 +2446,7 @@ fn site_replication_rule_complete(rule: &ReplicationRule, owner_deployment_id: &
         .is_some_and(|deployment_id| deployment_id != owner_deployment_id);
 
     rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-"))
-        && rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED)
+        && rule.status == Status::ENABLED.as_str()
         && points_at_remote_site
         && delete_marker_enabled
         && delete_enabled
@@ -2479,7 +2477,7 @@ fn site_replication_config_mismatch<'a>(
         // out-of-sync ("0/N Buckets in sync"). decode_bucket_meta_wire_value falls back to
         // the raw bytes when the value is not base64, so plain-XML callers still work.
         let xml = decode_bucket_meta_wire_value(raw);
-        deserialize::<ReplicationConfiguration>(&xml).is_ok_and(|config| {
+        deserialize::<PersistedReplicationConfiguration>(&xml).is_ok_and(|config| {
             config.rules.len() == expected_rules
                 && config
                     .rules
@@ -4752,7 +4750,7 @@ fn bucket_target_deployment_id(target: &BucketTarget) -> Option<String> {
 /// will drop for `removed_deployment_ids`, plus a site-replication `Role`
 /// naming one of them: the targets those ARNs back are the site's own.
 fn removed_site_replication_rule_arns(
-    config: Option<&ReplicationConfiguration>,
+    config: Option<&PersistedReplicationConfiguration>,
     removed_deployment_ids: &HashSet<String>,
 ) -> HashSet<String> {
     let Some(config) = config else {
@@ -4998,7 +4996,7 @@ fn is_zero_rule_lifecycle_tombstone(raw: &[u8]) -> bool {
     well_formed_document && quick_xml::de::from_reader::<_, Tombstone>(raw).is_ok()
 }
 
-fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option<String> {
+fn replication_rule_deployment_id(rule: &PersistedReplicationRule) -> Option<String> {
     if let Some(rule_id) = rule.id.as_deref() {
         if let Some(deployment_id) = rule_id.strip_prefix("site-repl-")
             && !deployment_id.is_empty()
@@ -5012,9 +5010,9 @@ fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option<String> {
 }
 
 fn prune_removed_site_replication_rules(
-    mut config: ReplicationConfiguration,
+    mut config: PersistedReplicationConfiguration,
     removed_deployment_ids: &HashSet<String>,
-) -> (Option<ReplicationConfiguration>, usize) {
+) -> (Option<PersistedReplicationConfiguration>, usize) {
     if removed_deployment_ids.is_empty() {
         return (Some(config), 0);
     }
@@ -5554,7 +5552,7 @@ async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: O
 /// `getRemoteARNForPeer` does: the same-name target bucket on the peer.
 fn site_bucket_resync_target_for_peer<'a>(
     bucket: &str,
-    config: &ReplicationConfiguration,
+    config: &PersistedReplicationConfiguration,
     targets: &'a BucketTargets,
     peer: &PeerInfo,
 ) -> Option<&'a BucketTarget> {
@@ -5578,7 +5576,9 @@ fn site_bucket_resync_target_for_peer<'a>(
 /// The persisted replication configuration and bucket targets, bypassing the
 /// node-local metadata cache. `ConfigNotFound` surfaces for a bucket without
 /// a replication configuration, matching the cached read's error.
-async fn site_bucket_resync_persisted_wiring(bucket: &str) -> Result<(ReplicationConfiguration, BucketTargets), StorageError> {
+async fn site_bucket_resync_persisted_wiring(
+    bucket: &str,
+) -> Result<(PersistedReplicationConfiguration, BucketTargets), StorageError> {
     let metadata = metadata_sys::get_config_from_disk(bucket).await?;
     let config = metadata.replication_config.ok_or(StorageError::ConfigNotFound)?;
     Ok((config, metadata.bucket_target_config.unwrap_or_default()))
@@ -5999,7 +5999,7 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
                 .as_ref()
                 .map(|raw| {
                     let data = decode_bucket_meta_wire_value(raw);
-                    deserialize::<ReplicationConfiguration>(&data)
+                    deserialize::<PersistedReplicationConfiguration>(&data)
                 })
                 .transpose()
                 .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))?;
@@ -10779,7 +10779,7 @@ mod tests {
 
     #[test]
     fn test_site_replication_config_status_accepts_peer_specific_targets() {
-        let site_a_config = ReplicationConfiguration {
+        let site_a_config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(
                 "arn:rustfs:replication::site-b:test-replication",
@@ -10787,7 +10787,7 @@ mod tests {
                 "site-repl-site-b",
             )],
         };
-        let site_b_config = ReplicationConfiguration {
+        let site_b_config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(
                 "arn:rustfs:replication::site-a:test-replication",
@@ -10890,7 +10890,7 @@ mod tests {
     // reported "in sync" — the operator's single health signal agreed with the broken state.
     #[test]
     fn test_site_replication_config_mismatch_rejects_rule_pointing_at_owning_site() {
-        let shared_config = ReplicationConfiguration {
+        let shared_config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(
                 "arn:rustfs:replication::site-b:test-replication",
@@ -13271,8 +13271,8 @@ mod tests {
         assert_eq!(deployment_id.as_deref(), Some("remote-dep"));
     }
 
-    fn site_repl_config(peer: &str) -> ReplicationConfiguration {
-        ReplicationConfiguration {
+    fn site_repl_config(peer: &str) -> PersistedReplicationConfiguration {
+        PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(
                 &format!("arn:rustfs:replication::{peer}:photos"),
@@ -13781,7 +13781,7 @@ mod tests {
         assert_eq!(credentials.secret_key, "runtime-iam-secret");
 
         let regional_arn = "arn:rustfs:replication:eu-west-1:remote:photos";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(regional_arn, 1, "site-repl-remote")],
         };
@@ -14027,7 +14027,7 @@ mod tests {
     #[test]
     fn test_prune_removed_site_replication_bucket_targets_keeps_operator_target_to_removed_peer() {
         let removed_deployment_ids = HashSet::from(["removed-dep".to_string()]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 build_site_replication_rule("arn:rustfs:replication:eu-west-1:removed-dep:photos", 1, "site-repl-removed-dep"),
@@ -14099,7 +14099,7 @@ mod tests {
                 },
             ],
         };
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 build_site_replication_rule("arn:minio:replication::7c0c5a1e-operator:photos-dst", 9, "operator-rule"),
@@ -14109,7 +14109,7 @@ mod tests {
         let picked = site_bucket_resync_target_for_peer("photos", &config, &targets, &peer).expect("site target");
         assert_eq!(picked.arn, "arn:minio:replication::remote:photos");
 
-        let without_rule = ReplicationConfiguration {
+        let without_rule = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![build_site_replication_rule(
                 "arn:minio:replication::7c0c5a1e-operator:photos-dst",
@@ -14127,7 +14127,7 @@ mod tests {
         let kept_rule = build_site_replication_rule("arn:rustfs:replication::kept-dep:photos", 3, "site-repl-kept-dep");
         let removed_rule = build_site_replication_rule("arn:rustfs:replication::removed-dep:photos", 1, "site-repl-removed-dep");
         let user_rule = build_site_replication_rule("arn:rustfs:replication::removed-dep:photos", 9, "user-managed-rule");
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: "arn:rustfs:replication::removed-dep:photos".to_string(),
             rules: vec![removed_rule, user_rule, kept_rule],
         };
@@ -14157,7 +14157,7 @@ mod tests {
             "site-repl-gone-dep"
         )));
 
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 build_site_replication_rule("arn:rustfs:replication::removed-dep:photos", 1, "site-repl-removed-dep"),
@@ -14910,7 +14910,7 @@ mod tests {
         use rustfs_madmin::{SRBucketInfo, SRInfo};
 
         let repl_xml = {
-            let config = ReplicationConfiguration {
+            let config = PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![build_site_replication_rule(
                     "arn:rustfs:replication::site-b:photos",
@@ -15039,7 +15039,7 @@ mod tests {
         // A peer that is reachable and has complete replication rules for all other peers
         // should be Enable; one that is reachable but has an incomplete config should be Disable.
         let site_config_xml = |peer: &str| {
-            let config = ReplicationConfiguration {
+            let config = PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![build_site_replication_rule(
                     &format!("arn:rustfs:replication::{peer}:bucket"),
@@ -15076,7 +15076,7 @@ mod tests {
     #[test]
     fn test_site_replication_config_mismatch_accepts_base64_wire_form() {
         let site_config_xml = |peer: &str| {
-            let config = ReplicationConfiguration {
+            let config = PersistedReplicationConfiguration {
                 role: String::new(),
                 rules: vec![build_site_replication_rule(
                     &format!("arn:rustfs:replication::{peer}:bucket"),

rustfs/src/admin/handlers/sts.rs

@@ -18,7 +18,6 @@ use crate::admin::handlers::mfa::verify_for_session as mfa_verify_for_session;
 use crate::admin::runtime_sources::object_store_from_req;
 use crate::admin::service::federated_identity::DefaultFederatedSessionBinding;
 use crate::admin::service::session_policy::populate_session_policy;
-use crate::admin::storage_api::bucket::utils::serialize;
 use crate::{
     admin::runtime_sources::{current_action_credentials, current_federated_identity_service, current_token_signing_key},
     admin::{
@@ -335,12 +334,20 @@ async fn handle_assume_role(
     };
 
     // getAssumeRoleCredentials
-    let output = serialize::<AssumeRoleOutput>(&resp)
+    let output = serialize_assume_role_output(&resp)
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize assume role output failed: {e}")))?;
 
     Ok(S3Response::new((StatusCode::OK, Body::from(output))))
 }
 
+/// The STS answer is an s3s response body, so the s3s encoder writes it; it
+/// is not a bucket configuration and never goes through the bucket-config codec.
+fn serialize_assume_role_output(output: &AssumeRoleOutput) -> s3s::xml::SerResult<Vec<u8>> {
+    let mut buf = Vec::with_capacity(256);
+    s3s::xml::Serialize::serialize(output, &mut s3s::xml::Serializer::new(&mut buf))?;
+    Ok(buf)
+}
+
 /// Session claim marking a session that presented a second factor.
 ///
 /// Namespaced with the `x-rustfs-` prefix so it cannot collide with an OIDC

rustfs/src/admin/router.rs

@@ -62,6 +62,8 @@ use rustfs_config::{
     ENABLE_KEY, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT,
     WEBHOOK_SKIP_TLS_VERIFY,
 };
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_madmin::utils::parse_duration;
 use rustfs_notify::{Event as NotificationEvent, notification_system};
 use rustfs_policy::policy::action::{Action, S3Action};
@@ -1720,19 +1722,19 @@ fn parse_reset_start_target(uri: &Uri) -> S3Result<ReplicationResetStartRequest>
     })
 }
 
-fn collect_resettable_replication_target_arns(config: &s3s::dto::ReplicationConfiguration) -> Vec<String> {
+fn collect_resettable_replication_target_arns(config: &PersistedReplicationConfiguration) -> Vec<String> {
     let mut arns = Vec::new();
     let mut seen = HashSet::new();
 
     for rule in &config.rules {
-        if rule.status == s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::DISABLED) {
+        if rule.status == Status::DISABLED.as_str() {
             continue;
         }
 
-        let existing_object_enabled = rule.existing_object_replication.as_ref().is_some_and(|status| {
-            status.status
-                == s3s::dto::ExistingObjectReplicationStatus::from_static(s3s::dto::ExistingObjectReplicationStatus::ENABLED)
-        });
+        let existing_object_enabled = rule
+            .existing_object_replication
+            .as_ref()
+            .is_some_and(|status| status.status == Status::ENABLED.as_str());
         if !existing_object_enabled {
             continue;
         }
@@ -1755,7 +1757,7 @@ fn collect_resettable_replication_target_arns(config: &s3s::dto::ReplicationConf
     arns
 }
 
-fn resolve_replication_reset_target_arn(config: &s3s::dto::ReplicationConfiguration, requested_arn: &str) -> S3Result<String> {
+fn resolve_replication_reset_target_arn(config: &PersistedReplicationConfiguration, requested_arn: &str) -> S3Result<String> {
     let resettable_arns = collect_resettable_replication_target_arns(config);
 
     if requested_arn.is_empty() {
@@ -1943,7 +1945,7 @@ fn is_object_lock_not_enabled_error(err: &S3ClientError) -> bool {
 
 fn validate_replication_check_config_targets(
     targets: &BucketTargets,
-    config: &s3s::dto::ReplicationConfiguration,
+    config: &PersistedReplicationConfiguration,
 ) -> S3Result<()> {
     let configured_arns = targets
         .targets
@@ -1961,7 +1963,7 @@ fn validate_replication_check_config_targets(
     }
 
     for rule in &config.rules {
-        if rule.status == s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::DISABLED) {
+        if rule.status == Status::DISABLED.as_str() {
             continue;
         }
 
@@ -1978,7 +1980,7 @@ fn validate_replication_check_config_targets(
     Ok(())
 }
 
-fn filter_replication_check_targets(targets: BucketTargets, config: &s3s::dto::ReplicationConfiguration) -> Vec<BucketTarget> {
+fn filter_replication_check_targets(targets: BucketTargets, config: &PersistedReplicationConfiguration) -> Vec<BucketTarget> {
     let referenced_arns = config
         .filter_all_replication_target_arns()
         .into_iter()
@@ -3406,6 +3408,9 @@ mod tests {
     use http::HeaderMap;
     use http::Method;
     use http::Uri;
+    use rustfs_gateway_types::persistence::{
+        PersistedReplicationDestination, PersistedReplicationRule, PersistedReplicationStatus,
+    };
     use s3s::S3Request;
     use std::net::{IpAddr, SocketAddr};
     use time::macros::datetime;
@@ -3750,26 +3755,24 @@ mod tests {
 
     #[test]
     fn resolve_replication_reset_target_arn_uses_single_existing_object_target_by_default() {
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![s3s::dto::ReplicationRule {
+            rules: vec![PersistedReplicationRule {
                 delete_marker_replication: None,
                 delete_replication: None,
-                destination: s3s::dto::Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:replication:a".to_string(),
                     ..Default::default()
                 },
-                existing_object_replication: Some(s3s::dto::ExistingObjectReplication {
-                    status: s3s::dto::ExistingObjectReplicationStatus::from_static(
-                        s3s::dto::ExistingObjectReplicationStatus::ENABLED,
-                    ),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
                 filter: None,
                 id: Some("rule-a".to_string()),
                 prefix: Some(String::new()),
                 priority: None,
                 source_selection_criteria: None,
-                status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
 
@@ -3779,46 +3782,42 @@ mod tests {
 
     #[test]
     fn resolve_replication_reset_target_arn_requires_arn_for_multiple_targets() {
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
-                s3s::dto::ReplicationRule {
+                PersistedReplicationRule {
                     delete_marker_replication: None,
                     delete_replication: None,
-                    destination: s3s::dto::Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: "arn:replication:a".to_string(),
                         ..Default::default()
                     },
-                    existing_object_replication: Some(s3s::dto::ExistingObjectReplication {
-                        status: s3s::dto::ExistingObjectReplicationStatus::from_static(
-                            s3s::dto::ExistingObjectReplicationStatus::ENABLED,
-                        ),
+                    existing_object_replication: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
                     filter: None,
                     id: Some("rule-a".to_string()),
                     prefix: Some(String::new()),
                     priority: None,
                     source_selection_criteria: None,
-                    status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 },
-                s3s::dto::ReplicationRule {
+                PersistedReplicationRule {
                     delete_marker_replication: None,
                     delete_replication: None,
-                    destination: s3s::dto::Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: "arn:replication:b".to_string(),
                         ..Default::default()
                     },
-                    existing_object_replication: Some(s3s::dto::ExistingObjectReplication {
-                        status: s3s::dto::ExistingObjectReplicationStatus::from_static(
-                            s3s::dto::ExistingObjectReplicationStatus::ENABLED,
-                        ),
+                    existing_object_replication: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
                     filter: None,
                     id: Some("rule-b".to_string()),
                     prefix: Some(String::new()),
                     priority: None,
                     source_selection_criteria: None,
-                    status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 },
             ],
         };
@@ -3830,26 +3829,24 @@ mod tests {
 
     #[test]
     fn resolve_replication_reset_target_arn_rejects_target_without_existing_object_replication() {
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![s3s::dto::ReplicationRule {
+            rules: vec![PersistedReplicationRule {
                 delete_marker_replication: None,
                 delete_replication: None,
-                destination: s3s::dto::Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:replication:a".to_string(),
                     ..Default::default()
                 },
-                existing_object_replication: Some(s3s::dto::ExistingObjectReplication {
-                    status: s3s::dto::ExistingObjectReplicationStatus::from_static(
-                        s3s::dto::ExistingObjectReplicationStatus::DISABLED,
-                    ),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::DISABLED.to_string(),
                 }),
                 filter: None,
                 id: Some("rule-a".to_string()),
                 prefix: Some(String::new()),
                 priority: None,
                 source_selection_criteria: None,
-                status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
 
@@ -4373,7 +4370,7 @@ mod tests {
 
     #[test]
     fn build_replication_check_response_rejects_empty_target_list_at_runtime_boundary() {
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![],
         };
@@ -4521,12 +4518,12 @@ mod tests {
                 },
             ],
         };
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![s3s::dto::ReplicationRule {
+            rules: vec![PersistedReplicationRule {
                 delete_marker_replication: None,
                 delete_replication: None,
-                destination: s3s::dto::Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:replication:b".to_string(),
                     ..Default::default()
                 },
@@ -4536,7 +4533,7 @@ mod tests {
                 prefix: Some(String::new()),
                 priority: None,
                 source_selection_criteria: None,
-                status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
 
@@ -4555,12 +4552,12 @@ mod tests {
                 ..Default::default()
             }],
         };
-        let config = s3s::dto::ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![s3s::dto::ReplicationRule {
+            rules: vec![PersistedReplicationRule {
                 delete_marker_replication: None,
                 delete_replication: None,
-                destination: s3s::dto::Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:replication:missing".to_string(),
                     ..Default::default()
                 },
@@ -4570,7 +4567,7 @@ mod tests {
                 prefix: Some(String::new()),
                 priority: None,
                 source_selection_criteria: None,
-                status: s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
 

rustfs/src/admin/storage_api.rs

@@ -173,19 +173,19 @@ pub(crate) trait AdminReplicationConfigExt {
     fn has_existing_object_replication(&self, arn: &str) -> (bool, bool);
 }
 
-impl AdminReplicationConfigExt for s3s::dto::ReplicationConfiguration {
+impl AdminReplicationConfigExt for rustfs_gateway_types::persistence::PersistedReplicationConfiguration {
     fn filter_all_replication_target_arns(&self) -> Vec<String> {
         let obj = ecstore_bucket::replication::ObjectOpts {
             op_type: ecstore_bucket::replication::ReplicationType::All,
             ..Default::default()
         };
-        <s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::filter_target_arns(
+        <rustfs_gateway_types::persistence::PersistedReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::filter_target_arns(
             self, &obj,
         )
     }
 
     fn has_existing_object_replication(&self, arn: &str) -> (bool, bool) {
-        <s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_existing_object_replication(
+        <rustfs_gateway_types::persistence::PersistedReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_existing_object_replication(
             self, arn,
         )
     }
@@ -512,7 +512,9 @@ pub(crate) mod metadata_sys {
         super::ecstore_bucket::metadata_sys::get_quota_config(bucket).await
     }
 
-    pub(crate) async fn get_replication_config(bucket: &str) -> Result<(s3s::dto::ReplicationConfiguration, OffsetDateTime)> {
+    pub(crate) async fn get_replication_config(
+        bucket: &str,
+    ) -> Result<(rustfs_gateway_types::persistence::PersistedReplicationConfiguration, OffsetDateTime)> {
         super::ecstore_bucket::metadata_sys::get_replication_config(bucket).await
     }
 
@@ -760,20 +762,11 @@ pub(crate) mod target {
 }
 
 pub(crate) mod ecstore_utils {
-    pub(crate) fn deserialize<T>(input: &[u8]) -> s3s::xml::DeResult<T>
-    where
-        T: for<'xml> s3s::xml::Deserialize<'xml>,
-    {
-        super::ecstore_bucket::utils::deserialize(input)
-    }
+    pub(crate) use super::ecstore_bucket::utils::{deserialize, serialize};
 
     pub(crate) fn is_valid_object_prefix(object: &str) -> bool {
         super::ecstore_bucket::utils::is_valid_object_prefix(object)
     }
-
-    pub(crate) fn serialize<T: s3s::xml::Serialize>(val: &T) -> s3s::xml::SerResult<Vec<u8>> {
-        super::ecstore_bucket::utils::serialize(val)
-    }
 }
 
 pub(crate) mod versioning {}

rustfs/src/app/bucket_usecase.rs

@@ -43,7 +43,7 @@ use super::storage_api::bucket_usecase::bucket::{
         unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
     },
     target::{BucketTargetType, BucketTargets},
-    utils::serialize,
+    utils::{BucketConfigXml, serialize},
     versioning_sys::BucketVersioningSys,
 };
 use super::storage_api::bucket_usecase::contract::bucket::{
@@ -83,6 +83,7 @@ use crate::storage::storage_api::lock_bucket_targets_metadata;
 use http::StatusCode;
 use metrics::counter;
 use rustfs_config::RUSTFS_REGION;
+use rustfs_gateway_types::persistence::PersistedReplicationConfiguration;
 use rustfs_io_metrics::record_s3_op;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
 use rustfs_policy::policy::{
@@ -107,17 +108,16 @@ use s3s::dto::{
     GetBucketLifecycleConfigurationInput, GetBucketLifecycleConfigurationOutput, GetBucketLocationInput, GetBucketLocationOutput,
     GetBucketNotificationConfigurationInput, GetBucketNotificationConfigurationOutput, GetBucketPolicyInput,
     GetBucketPolicyOutput, GetBucketPolicyStatusInput, GetBucketPolicyStatusOutput, GetBucketReplicationInput,
-    GetBucketReplicationOutput, GetBucketTaggingInput, GetBucketTaggingOutput, GetBucketVersioningInput,
-    GetBucketVersioningOutput, GetPublicAccessBlockInput, GetPublicAccessBlockOutput, HeadBucketInput, HeadBucketOutput,
-    LifecycleRule, ListBucketsInput, ListBucketsOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsInput,
-    ListObjectsOutput, ListObjectsV2Input, ListObjectsV2Output, MetadataEntry, NotificationConfiguration,
-    NotificationConfigurationFilter, Object, ObjectLockConfiguration, ObjectStorageClass, ObjectVersion,
-    ObjectVersionStorageClass, PolicyStatus, PutBucketCorsInput, PutBucketCorsOutput, PutBucketEncryptionInput,
-    PutBucketEncryptionOutput, PutBucketLifecycleConfigurationInput, PutBucketLifecycleConfigurationOutput,
-    PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput, PutBucketPolicyInput,
-    PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput,
-    PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput,
-    ReplicationConfiguration, ServerSideEncryption, ServerSideEncryptionConfiguration, Tagging, Timestamp, UserMetadata,
+    GetBucketTaggingInput, GetBucketTaggingOutput, GetBucketVersioningInput, GetBucketVersioningOutput,
+    GetPublicAccessBlockInput, GetPublicAccessBlockOutput, HeadBucketInput, HeadBucketOutput, LifecycleRule, ListBucketsInput,
+    ListBucketsOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsInput, ListObjectsOutput,
+    ListObjectsV2Input, ListObjectsV2Output, MetadataEntry, NotificationConfiguration, NotificationConfigurationFilter, Object,
+    ObjectLockConfiguration, ObjectStorageClass, ObjectVersion, ObjectVersionStorageClass, PolicyStatus, PutBucketCorsInput,
+    PutBucketCorsOutput, PutBucketEncryptionInput, PutBucketEncryptionOutput, PutBucketLifecycleConfigurationInput,
+    PutBucketLifecycleConfigurationOutput, PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput,
+    PutBucketPolicyInput, PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput,
+    PutBucketTaggingOutput, PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput,
+    PutPublicAccessBlockOutput, ServerSideEncryption, ServerSideEncryptionConfiguration, Tagging, Timestamp, UserMetadata,
     VersioningConfiguration,
 };
 use s3s::region::Region;
@@ -439,7 +439,7 @@ impl SerializeContent for ObjectInternalInfo {
     }
 }
 
-fn serialize_config<T: xml::Serialize>(value: &T) -> S3Result<Vec<u8>> {
+fn serialize_config<T: BucketConfigXml>(value: &T) -> S3Result<Vec<u8>> {
     serialize(value).map_err(to_internal_error)
 }
 
@@ -664,7 +664,7 @@ fn notify_bucket_metadata_delete(bucket: String, request_context: Option<request
     });
 }
 
-fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> {
+fn validate_replication_config_targets(targets: &BucketTargets, config: &PersistedReplicationConfiguration) -> S3Result<()> {
     let configured_arns = targets
         .targets
         .iter()
@@ -685,7 +685,7 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica
     }
 }
 
-fn validate_replication_config_capabilities(config: &ReplicationConfiguration) -> S3Result<()> {
+fn validate_replication_config_capabilities(config: &PersistedReplicationConfiguration) -> S3Result<()> {
     if let Err(err) = validate_replication_config_structure(config) {
         return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, err.message()));
     }
@@ -704,7 +704,7 @@ fn validate_replication_config_capabilities(config: &ReplicationConfiguration) -
     Ok(())
 }
 
-async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> {
+async fn validate_bucket_replication_update(bucket: &str, config: &PersistedReplicationConfiguration) -> S3Result<()> {
     if !BucketVersioningSys::enabled(bucket).await {
         return Err(s3_error!(
             InvalidRequest,
@@ -735,11 +735,11 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo
 /// `contract` is what the peers were probed to support; the merged config is
 /// what gets broadcast, so it is built the way every peer will merge it.
 fn merge_user_replication_config_update(
-    incoming: ReplicationConfiguration,
-    existing: Option<ReplicationConfiguration>,
+    incoming: PersistedReplicationConfiguration,
+    existing: Option<PersistedReplicationConfiguration>,
     site_peer_deployment_ids: &HashSet<String>,
     contract: OperatorRuleContract,
-) -> ReplicationConfiguration {
+) -> PersistedReplicationConfiguration {
     if site_peer_deployment_ids.is_empty() {
         return incoming;
     }
@@ -756,10 +756,10 @@ fn merge_user_replication_config_update(
 /// targets may be garbage-collected — never an ARN a surviving reconciler
 /// rule still points at.
 fn split_replication_config_for_user_delete(
-    config: ReplicationConfiguration,
+    config: PersistedReplicationConfiguration,
     site_peer_deployment_ids: &HashSet<String>,
     contract: OperatorRuleContract,
-) -> (Option<ReplicationConfiguration>, HashSet<String>) {
+) -> (Option<PersistedReplicationConfiguration>, HashSet<String>) {
     let mut removable_arns = replication_target_arns(&config);
     let remaining = merge_user_replication_config(None, Some(config), site_peer_deployment_ids, contract);
     if let Some(remaining) = remaining.as_ref() {
@@ -822,7 +822,7 @@ async fn write_replication_targets_after_config_delete(
 
 async fn restore_replication_config_after_target_cleanup_failure(
     bucket: &str,
-    config: &ReplicationConfiguration,
+    config: &PersistedReplicationConfiguration,
     cleanup_err: S3Error,
     expected_incarnation_id: Option<uuid::Uuid>,
 ) -> S3Error {
@@ -2182,10 +2182,12 @@ impl DefaultBucketUsecase {
         Ok(S3Response::new(output))
     }
 
+    /// Answers the stored configuration in the engine's shape; the s3s edge
+    /// (`storage::ecfs`) converts it into the `GetBucketReplication` output.
     pub async fn execute_get_bucket_replication(
         &self,
         req: S3Request<GetBucketReplicationInput>,
-    ) -> S3Result<S3Response<GetBucketReplicationOutput>> {
+    ) -> S3Result<PersistedReplicationConfiguration> {
         let GetBucketReplicationInput { bucket, .. } = req.input;
 
         let Some(store) = self.object_store() else {
@@ -2218,9 +2220,7 @@ impl DefaultBucketUsecase {
             }
         };
 
-        Ok(S3Response::new(GetBucketReplicationOutput {
-            replication_configuration: Some(replication_configuration),
-        }))
+        Ok(replication_configuration)
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -2664,20 +2664,19 @@ impl DefaultBucketUsecase {
     }
 
     /// See [`Self::execute_delete_bucket_replication`] for `site_peers` and
-    /// `contract`.
+    /// `contract`. `replication_configuration` is the request body in the
+    /// engine's shape: the s3s edge (`storage::ecfs`) converts it, and this use
+    /// case never reads `req.input.replication_configuration`.
     pub async fn execute_put_bucket_replication(
         &self,
         req: S3Request<PutBucketReplicationInput>,
+        replication_configuration: PersistedReplicationConfiguration,
         site_peers: HashSet<String>,
         contract: OperatorRuleContract,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
         let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
         let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
-        let PutBucketReplicationInput {
-            bucket,
-            replication_configuration,
-            ..
-        } = req.input;
+        let PutBucketReplicationInput { bucket, .. } = req.input;
         info!(bucket = %bucket, "updating bucket replication config");
 
         validate_replication_config_capabilities(&replication_configuration)?;
@@ -3211,10 +3210,13 @@ fn validate_bucket_encryption_configuration(config: &ServerSideEncryptionConfigu
 mod tests {
     use super::*;
     use http::{Extensions, HeaderMap, Method, Uri};
-    use s3s::dto::ReplicationRuleStatus;
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedEncryptionConfiguration, PersistedReplicationDestination, PersistedReplicationRule,
+    };
     use s3s::dto::{
-        BucketVersioningStatus, CORSConfiguration, Destination, ExcludedPrefix, FilterRule, FilterRuleName, LifecycleExpiration,
-        NoncurrentVersionTransition, PublicAccessBlockConfiguration, QueueConfiguration, ReplicationRule, S3KeyFilter,
+        BucketVersioningStatus, CORSConfiguration, ExcludedPrefix, FilterRule, FilterRuleName, LifecycleExpiration,
+        NoncurrentVersionTransition, PublicAccessBlockConfiguration, QueueConfiguration, S3KeyFilter,
         ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Transition,
         TransitionStorageClass,
     };
@@ -3223,6 +3225,16 @@ mod tests {
     use std::time::Duration;
     use tokio::sync::Notify;
 
+    /// Response bodies are not bucket configurations: they are written by the
+    /// s3s encoder itself, as the legacy stack does.
+    fn serialize_s3s_response<T: xml::Serialize>(value: &T) -> Vec<u8> {
+        let mut buf = Vec::new();
+        value
+            .serialize(&mut xml::Serializer::new(&mut buf))
+            .expect("metadata output should serialize");
+        buf
+    }
+
     fn s3_op_total(op: S3Operation) -> u64 {
         rustfs_io_metrics::s3_op_metrics_snapshot()
             .into_iter()
@@ -3248,6 +3260,7 @@ mod tests {
 
     /// The stored configuration must be one the write path honours as written:
     /// only AES256 and aws:kms exist, and a key id belongs to aws:kms alone.
+
     #[test]
     fn put_bucket_encryption_refuses_configurations_the_write_path_cannot_honour() {
         validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", None)])).expect("AES256 is valid");
@@ -3555,11 +3568,11 @@ mod tests {
         rustfs_scanner::clear_dirty_usage_bucket(BUCKET);
     }
 
-    fn replication_rule_for_target(arn: &str) -> ReplicationRule {
-        ReplicationRule {
+    fn replication_rule_for_target(arn: &str) -> PersistedReplicationRule {
+        PersistedReplicationRule {
             delete_marker_replication: None,
             delete_replication: None,
-            destination: Destination {
+            destination: PersistedReplicationDestination {
                 bucket: arn.to_string(),
                 ..Default::default()
             },
@@ -3569,7 +3582,7 @@ mod tests {
             prefix: None,
             priority: Some(1),
             source_selection_criteria: None,
-            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            status: Status::ENABLED.to_string(),
         }
     }
 
@@ -3577,7 +3590,7 @@ mod tests {
     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 {
+        let config = PersistedReplicationConfiguration {
             role: format!(" {role} "),
             rules: vec![replication_rule_for_target(destination)],
         };
@@ -3591,7 +3604,7 @@ mod tests {
     #[test]
     fn replication_target_arns_use_rule_destinations_without_role() {
         let destination = "arn:rustfs:replication:us-east-1:target:bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_for_target(destination)],
         };
@@ -3601,7 +3614,7 @@ mod tests {
         assert!(arns.contains(destination));
     }
 
-    fn replication_rule_with_id(arn: &str, id: &str, priority: i32) -> ReplicationRule {
+    fn replication_rule_with_id(arn: &str, id: &str, priority: i32) -> PersistedReplicationRule {
         let mut rule = replication_rule_for_target(arn);
         rule.id = Some(id.to_string());
         rule.priority = Some(priority);
@@ -3614,14 +3627,14 @@ mod tests {
 
     #[test]
     fn put_replication_merge_preserves_site_replication_rules() {
-        let existing = ReplicationConfiguration {
+        let existing = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 replication_rule_with_id("arn:rustfs:replication::peer-dep:bucket", "site-repl-peer-dep", 1),
                 replication_rule_with_id("arn:rustfs:replication:us-east-1:old:bucket", "old-user-rule", 2),
             ],
         };
-        let incoming = ReplicationConfiguration {
+        let incoming = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 replication_rule_with_id("arn:rustfs:replication:us-east-1:new:bucket", "new-user-rule", 1),
@@ -3660,7 +3673,7 @@ mod tests {
     #[test]
     fn put_then_delete_replication_without_site_replication_treats_site_repl_id_as_user_rule() {
         let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
-        let incoming = ReplicationConfiguration {
+        let incoming = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_with_id(user_arn, "site-repl-user", 1)],
         };
@@ -3681,7 +3694,7 @@ mod tests {
     fn delete_replication_split_keeps_only_reconciler_derived_rules() {
         let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
         let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 replication_rule_with_id(user_arn, "site-repl-user", 1),
@@ -3706,7 +3719,7 @@ mod tests {
 
     #[test]
     fn put_replication_merge_returns_incoming_verbatim_without_site_rules() {
-        let existing = ReplicationConfiguration {
+        let existing = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_with_id(
                 "arn:rustfs:replication:us-east-1:old:bucket",
@@ -3714,7 +3727,7 @@ mod tests {
                 7,
             )],
         };
-        let incoming = ReplicationConfiguration {
+        let incoming = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_with_id(
                 "arn:rustfs:replication:us-east-1:new:bucket",
@@ -3738,7 +3751,7 @@ mod tests {
     fn delete_replication_split_keeps_site_rules_and_their_targets() {
         let sr_arn = "arn:rustfs:replication::peer-dep:bucket";
         let user_arn = "arn:rustfs:replication:us-east-1:user:bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 replication_rule_with_id(user_arn, "user-rule", 1),
@@ -3762,7 +3775,7 @@ mod tests {
     #[test]
     fn delete_replication_split_protects_targets_shared_with_site_rules() {
         let sr_arn = "arn:rustfs:replication::peer-dep:bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![
                 replication_rule_with_id(sr_arn, "user-rule-on-sr-target", 1),
@@ -3783,7 +3796,7 @@ mod tests {
     #[test]
     fn delete_replication_split_removes_everything_without_site_rules() {
         let user_arn = "arn:rustfs:replication:us-east-1:user:bucket";
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_with_id(user_arn, "user-rule", 1)],
         };
@@ -3812,7 +3825,7 @@ mod tests {
     fn validate_replication_config_targets_accepts_matching_destination_arns() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let targets = replication_targets_with_arn(&[arn]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_for_target(arn)],
         };
@@ -3823,7 +3836,7 @@ mod tests {
     #[test]
     fn validate_replication_config_targets_rejects_stale_destination_arns() {
         let targets = replication_targets_with_arn(&["arn:rustfs:replication:us-east-1:target:bucket-a"]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_for_target(
                 "arn:rustfs:replication:us-east-1:target:bucket-b",
@@ -3838,7 +3851,7 @@ mod tests {
     fn validate_replication_config_targets_accepts_matching_role_arn() {
         let arn = "arn:rustfs:replication:us-east-1:role-target:bucket";
         let targets = replication_targets_with_arn(&[arn]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: format!(" {arn} "),
             rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:ignored:bucket")],
         };
@@ -3850,7 +3863,7 @@ mod tests {
     fn validate_replication_config_targets_rejects_role_with_multiple_destinations() {
         let role = "arn:rustfs:replication:us-east-1:role-target:bucket";
         let targets = replication_targets_with_arn(&[role]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: role.to_string(),
             rules: vec![
                 replication_rule_for_target("arn:rustfs:replication:us-east-1:target-a:bucket"),
@@ -3867,7 +3880,7 @@ mod tests {
     fn validate_replication_config_targets_trims_destination_arns() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";
         let targets = replication_targets_with_arn(&[arn]);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![replication_rule_for_target(
                 " arn:rustfs:replication:us-east-1:target:bucket ",
@@ -3881,8 +3894,8 @@ mod tests {
     fn validate_replication_config_targets_ignores_disabled_rules() {
         let targets = replication_targets_with_arn(&[]);
         let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:stale:bucket");
-        rule.status = ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED);
-        let config = ReplicationConfiguration {
+        rule.status = Status::DISABLED.to_string();
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -3894,10 +3907,10 @@ mod tests {
     fn validate_replication_config_capabilities_names_unsupported_field() {
         let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
         let destination_key_id = "arn:aws:kms:us-east-1:123456789012:key/opaque-key-id";
-        rule.destination.encryption_configuration = Some(s3s::dto::EncryptionConfiguration {
+        rule.destination.encryption_configuration = Some(PersistedEncryptionConfiguration {
             replica_kms_key_id: Some(destination_key_id.to_string()),
         });
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -3919,7 +3932,7 @@ mod tests {
         first.priority = Some(1);
         let mut second = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
         second.priority = Some(1);
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![first, second],
         };
@@ -3934,8 +3947,8 @@ mod tests {
     #[test]
     fn validate_replication_config_capabilities_rejects_invalid_status_before_write() {
         let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
-        rule.status = ReplicationRuleStatus::from_static("Invalid");
-        let config = ReplicationConfiguration {
+        rule.status = "Invalid".to_owned();
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
             rules: vec![rule],
         };
@@ -4855,8 +4868,7 @@ mod tests {
         };
 
         let output = build_list_objects_v2_metadata_output(object_infos, "demo-bucket", &params, None, false, &permissions);
-        let xml = String::from_utf8(serialize_config(&output).expect("metadata output should serialize"))
-            .expect("metadata output should be UTF-8");
+        let xml = String::from_utf8(serialize_s3s_response(&output)).expect("metadata output should be UTF-8");
 
         assert!(xml.contains("<ListBucketResult"));
         assert!(xml.contains("<Contents>"));
@@ -4928,8 +4940,7 @@ mod tests {
         };
 
         let output = build_list_objects_v2_metadata_output(object_infos, "demo-bucket", &params, None, false, &permissions);
-        let xml = String::from_utf8(serialize_config(&output).expect("metadata output should serialize"))
-            .expect("metadata output should be UTF-8");
+        let xml = String::from_utf8(serialize_s3s_response(&output)).expect("metadata output should be UTF-8");
 
         // Good sibling metadata survives.
         assert!(xml.contains("<project>alpha</project>"), "well-formed key must remain: {xml}");
@@ -5274,12 +5285,13 @@ mod tests {
     async fn execute_put_bucket_replication_returns_internal_error_when_store_uninitialized() {
         // The config must clear the structural/capability validators so the
         // request actually reaches the store lookup this test pins.
+        let configuration = PersistedReplicationConfiguration {
+            role: "arn:aws:iam::123456789012:role/test".to_string(),
+            rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket")],
+        };
         let input = PutBucketReplicationInput::builder()
             .bucket("test-bucket".to_string())
-            .replication_configuration(ReplicationConfiguration {
-                role: "arn:aws:iam::123456789012:role/test".to_string(),
-                rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket")],
-            })
+            .replication_configuration(Default::default())
             .build()
             .unwrap();
 
@@ -5287,7 +5299,7 @@ mod tests {
         let usecase = DefaultBucketUsecase::without_context();
 
         let err = usecase
-            .execute_put_bucket_replication(req, HashSet::new(), OperatorRuleContract::Derived)
+            .execute_put_bucket_replication(req, configuration, HashSet::new(), OperatorRuleContract::Derived)
             .await
             .unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
@@ -5297,17 +5309,23 @@ mod tests {
     async fn execute_put_bucket_replication_rejects_unsupported_fields_before_store_or_metadata_write() {
         let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
         rule.destination.account = Some("123456789012".to_string());
+        let configuration = PersistedReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
         let input = PutBucketReplicationInput::builder()
             .bucket("test-bucket".to_string())
-            .replication_configuration(ReplicationConfiguration {
-                role: String::new(),
-                rules: vec![rule],
-            })
+            .replication_configuration(Default::default())
             .build()
             .unwrap();
 
         let err = DefaultBucketUsecase::without_context()
-            .execute_put_bucket_replication(build_request(input, Method::PUT), HashSet::new(), OperatorRuleContract::Derived)
+            .execute_put_bucket_replication(
+                build_request(input, Method::PUT),
+                configuration,
+                HashSet::new(),
+                OperatorRuleContract::Derived,
+            )
             .await
             .expect_err("unsupported fields must be rejected before store access");
 

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -41,6 +41,12 @@ use futures::FutureExt;
 use futures::stream;
 use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATCH};
 use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT};
+use rustfs_gateway_types::dto::Status;
+use rustfs_gateway_types::persistence::{
+    PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+    PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationStatus, PersistedReplicationTag,
+    PersistedSourceSelectionCriteria,
+};
 use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager};
 use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, insert_header};
 use rustfs_utils::path::encode_dir_object;
@@ -3104,9 +3110,9 @@ async fn install_delete_replication_config(
     let mut metadata = (*metadata).clone();
     let target = "arn:aws:s3:::target-bucket";
     let delete_status = if delete_enabled {
-        DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED)
+        Status::ENABLED.to_string()
     } else {
-        DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED)
+        Status::DISABLED.to_string()
     };
 
     metadata.versioning_config_xml = b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>".to_vec();
@@ -3114,27 +3120,27 @@ async fn install_delete_replication_config(
         status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
         ..Default::default()
     });
-    let filter = required_tag.map(|(key, value)| ReplicationRuleFilter {
-        tag: Some(Tag {
+    let filter = required_tag.map(|(key, value)| PersistedReplicationFilter {
+        tag: Some(PersistedReplicationTag {
             key: Some(key.to_string()),
             value: Some(value.to_string()),
         }),
         ..Default::default()
     });
-    let source_selection_criteria = replica_modifications.then(|| SourceSelectionCriteria {
-        replica_modifications: Some(ReplicaModifications {
-            status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
+    let source_selection_criteria = replica_modifications.then(|| PersistedSourceSelectionCriteria {
+        replica_modifications: Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         }),
         sse_kms_encrypted_objects: None,
     });
-    let replication_config = ReplicationConfiguration {
+    let replication_config = PersistedReplicationConfiguration {
         role: String::new(),
-        rules: vec![ReplicationRule {
-            delete_marker_replication: Some(DeleteMarkerReplication {
-                status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+        rules: vec![PersistedReplicationRule {
+            delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                status: Some(Status::DISABLED.to_string()),
             }),
-            delete_replication: Some(DeleteReplication { status: delete_status }),
-            destination: Destination {
+            delete_replication: Some(PersistedReplicationStatus { status: delete_status }),
+            destination: PersistedReplicationDestination {
                 bucket: target.to_string(),
                 ..Default::default()
             },
@@ -3144,7 +3150,7 @@ async fn install_delete_replication_config(
             prefix: Some(String::new()),
             priority: Some(1),
             source_selection_criteria,
-            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+            status: Status::ENABLED.to_string(),
         }],
     };
     metadata.replication_config_xml = serialize(&replication_config).expect("replication test config should serialize");

rustfs/src/app/object/delete.rs

@@ -1298,11 +1298,12 @@ impl DefaultObjectUsecase {
 mod tests {
     use super::*;
     use http::{HeaderMap, HeaderValue, Method};
-    use s3s::dto::{
-        Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
-        ExistingObjectReplication, ExistingObjectReplicationStatus, ObjectIdentifier, ReplicaModifications,
-        ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria,
+    use rustfs_gateway_types::dto::Status;
+    use rustfs_gateway_types::persistence::{
+        PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+        PersistedReplicationRule, PersistedReplicationStatus, PersistedSourceSelectionCriteria,
     };
+    use s3s::dto::{Delete, ObjectIdentifier};
     use std::sync::Arc;
 
     #[test]
@@ -2684,31 +2685,31 @@ mod tests {
     #[test]
     fn delete_replication_state_from_config_tracks_downstream_delete_marker_targets() {
         let arn = "arn:aws:s3:::target-bucket".to_string();
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: arn.clone(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+            rules: vec![PersistedReplicationRule {
+                delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                    status: Some(Status::ENABLED.to_string()),
                 }),
                 delete_replication: None,
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: arn.clone(),
                     ..Default::default()
                 },
-                existing_object_replication: Some(ExistingObjectReplication {
-                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
                 filter: None,
                 id: Some("rule-1".to_string()),
                 prefix: Some("test/".to_string()),
                 priority: Some(1),
-                source_selection_criteria: Some(SourceSelectionCriteria {
-                    replica_modifications: Some(ReplicaModifications {
-                        status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
+                source_selection_criteria: Some(PersistedSourceSelectionCriteria {
+                    replica_modifications: Some(PersistedReplicationStatus {
+                        status: Status::ENABLED.to_string(),
                     }),
                     sse_kms_encrypted_objects: None,
                 }),
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
         let obj_info = ObjectInfo {
@@ -2731,26 +2732,26 @@ mod tests {
     #[test]
     fn delete_replication_state_from_config_skips_replica_delete_without_replica_modifications() {
         let arn = "arn:aws:s3:::target-bucket".to_string();
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: arn.clone(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+            rules: vec![PersistedReplicationRule {
+                delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                    status: Some(Status::ENABLED.to_string()),
                 }),
                 delete_replication: None,
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: arn,
                     ..Default::default()
                 },
-                existing_object_replication: Some(ExistingObjectReplication {
-                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
                 filter: None,
                 id: Some("rule-1".to_string()),
                 prefix: Some("test/".to_string()),
                 priority: Some(1),
                 source_selection_criteria: None,
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
         let obj_info = ObjectInfo {
@@ -2770,26 +2771,26 @@ mod tests {
     #[test]
     fn delete_replication_state_from_config_requires_delete_switch_for_marker_version_purges() {
         let arn = "arn:aws:s3:::target-bucket".to_string();
-        let mut config = ReplicationConfiguration {
+        let mut config = PersistedReplicationConfiguration {
             role: arn.clone(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+            rules: vec![PersistedReplicationRule {
+                delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                    status: Some(Status::ENABLED.to_string()),
                 }),
                 delete_replication: None,
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: arn.clone(),
                     ..Default::default()
                 },
-                existing_object_replication: Some(ExistingObjectReplication {
-                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+                existing_object_replication: Some(PersistedReplicationStatus {
+                    status: Status::ENABLED.to_string(),
                 }),
                 filter: None,
                 id: Some("rule-1".to_string()),
                 prefix: Some("test/".to_string()),
                 priority: Some(1),
                 source_selection_criteria: None,
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
         let obj_info = ObjectInfo {
@@ -2806,8 +2807,8 @@ mod tests {
             "delete-marker version purge must not use DeleteMarkerReplication"
         );
 
-        config.rules[0].delete_replication = Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        config.rules[0].delete_replication = Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         });
         let state = delete_replication_state_from_config(&config, &obj_info, version_id, false)
             .expect("delete-marker version purge should honor DeleteReplication");

rustfs/src/app/object/internal_put.rs

@@ -797,8 +797,9 @@ impl DefaultObjectUsecase {
 mod tests {
     use super::*;
     use crate::app::storage_api::s3::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration,
-        ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, Tag, VersioningConfiguration,
+        BucketVersioningStatus, PersistedOptionalReplicationStatus, PersistedReplicationConfiguration,
+        PersistedReplicationDestination, PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationTag, Status,
+        VersioningConfiguration,
     };
     use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
@@ -895,21 +896,21 @@ mod tests {
         });
 
         if let Some(target) = target {
-            let filter = required_tag.map(|(key, value)| ReplicationRuleFilter {
-                tag: Some(Tag {
+            let filter = required_tag.map(|(key, value)| PersistedReplicationFilter {
+                tag: Some(PersistedReplicationTag {
                     key: Some(key.to_string()),
                     value: Some(value.to_string()),
                 }),
                 ..Default::default()
             });
-            let config = ReplicationConfiguration {
+            let config = PersistedReplicationConfiguration {
                 role: String::new(),
-                rules: vec![ReplicationRule {
-                    delete_marker_replication: Some(DeleteMarkerReplication {
-                        status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+                rules: vec![PersistedReplicationRule {
+                    delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                        status: Some(Status::DISABLED.to_string()),
                     }),
                     delete_replication: None,
-                    destination: Destination {
+                    destination: PersistedReplicationDestination {
                         bucket: target.to_string(),
                         ..Default::default()
                     },
@@ -919,7 +920,7 @@ mod tests {
                     prefix: Some(String::new()),
                     priority: Some(1),
                     source_selection_criteria: None,
-                    status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                    status: Status::ENABLED.to_string(),
                 }],
             };
             metadata.replication_config_xml = serialize(&config).expect("replication test config should serialize");

rustfs/src/app/object/on_demand_migration_put.rs

@@ -299,9 +299,10 @@ mod tests {
     use super::*;
     use crate::app::storage_api::multipart_usecase::contract::multipart::MultipartOperations as _;
     use crate::app::storage_api::s3::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration,
-        ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault,
-        ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
+        BucketVersioningStatus, PersistedOptionalReplicationStatus, PersistedReplicationConfiguration,
+        PersistedReplicationDestination, PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationTag,
+        ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Status,
+        VersioningConfiguration,
     };
     use crate::app::storage_api::test::bucket::utils::serialize;
     use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
@@ -1030,20 +1031,20 @@ mod tests {
             status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
             ..Default::default()
         });
-        let config = ReplicationConfiguration {
+        let config = PersistedReplicationConfiguration {
             role: String::new(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
+            rules: vec![PersistedReplicationRule {
+                delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+                    status: Some(Status::DISABLED.to_string()),
                 }),
                 delete_replication: None,
-                destination: Destination {
+                destination: PersistedReplicationDestination {
                     bucket: "arn:aws:s3:::target-bucket".to_string(),
                     ..Default::default()
                 },
                 existing_object_replication: None,
-                filter: required_tag.map(|(key, value)| ReplicationRuleFilter {
-                    tag: Some(Tag {
+                filter: required_tag.map(|(key, value)| PersistedReplicationFilter {
+                    tag: Some(PersistedReplicationTag {
                         key: Some(key.to_string()),
                         value: Some(value.to_string()),
                     }),
@@ -1053,7 +1054,7 @@ mod tests {
                 prefix: Some(String::new()),
                 priority: Some(1),
                 source_selection_criteria: None,
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+                status: Status::ENABLED.to_string(),
             }],
         };
         metadata.replication_config_xml = serialize(&config).expect("replication config serializes");

rustfs/src/app/storage_api.rs

@@ -27,16 +27,21 @@ pub(crate) fn EndpointServerPools(
 /// S3 wire types for app-layer modules, funneled here so new files stay off
 /// the direct s3s surface (s3s footprint ratchet, `scripts/check_s3s_footprint.sh`).
 pub(crate) mod s3 {
+    #[cfg(test)]
+    pub(crate) use rustfs_gateway_types::dto::Status;
+    #[cfg(test)]
+    pub(crate) use rustfs_gateway_types::persistence::{
+        PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+        PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationTag,
+    };
     #[cfg(test)]
     pub(crate) use s3s::auth::{Credentials, SimpleAuth};
     #[cfg(test)]
     pub(crate) use s3s::config::{S3Config, StaticConfigProvider};
     #[cfg(test)]
     pub(crate) use s3s::dto::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, GetObjectInput,
-        HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output, ReplicationConfiguration, ReplicationRule,
-        ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
-        ServerSideEncryptionRule, Tag, VersioningConfiguration,
+        BucketVersioningStatus, GetObjectInput, HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output,
+        ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, VersioningConfiguration,
     };
     #[cfg(test)]
     pub(crate) use s3s::dto::{ListObjectsInput, StreamingBlob, UploadPartInput, UploadPartOutput};
@@ -553,7 +558,10 @@ pub(crate) mod bucket {
 
         pub(crate) async fn get_replication_config(
             bucket: &str,
-        ) -> Result<(s3s::dto::ReplicationConfiguration, OffsetDateTime), crate::storage::storage_api::StorageError> {
+        ) -> Result<
+            (rustfs_gateway_types::persistence::PersistedReplicationConfiguration, OffsetDateTime),
+            crate::storage::storage_api::StorageError,
+        > {
             crate::storage::storage_api::ecstore_bucket::metadata_sys::get_replication_config(bucket).await
         }
 
@@ -914,7 +922,7 @@ pub(crate) mod bucket {
         }
 
         pub(crate) fn delete_replication_state_from_config(
-            config: &s3s::dto::ReplicationConfiguration,
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
             obj_info: &crate::storage::storage_api::StorageObjectInfo,
             version_id: Option<Uuid>,
             replica: bool,
@@ -929,7 +937,9 @@ pub(crate) mod bucket {
             replication_contracts::delete_replication_state_from_config(config, &source)
         }
 
-        pub(crate) fn replication_target_arns(config: &s3s::dto::ReplicationConfiguration) -> HashSet<String> {
+        pub(crate) fn replication_target_arns(
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
+        ) -> HashSet<String> {
             replication_contracts::replication_target_arns(config)
         }
 
@@ -966,17 +976,19 @@ pub(crate) mod bucket {
 
         pub(crate) fn validate_replication_config_target_arns<'a>(
             configured_arns: impl Iterator<Item = &'a str>,
-            config: &s3s::dto::ReplicationConfiguration,
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
         ) -> Result<(), ReplicationTargetValidationError> {
             replication_contracts::validate_replication_config_target_arns(configured_arns, config)
         }
 
-        pub(crate) fn unsupported_replication_config_field(config: &s3s::dto::ReplicationConfiguration) -> Option<&'static str> {
+        pub(crate) fn unsupported_replication_config_field(
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
+        ) -> Option<&'static str> {
             replication_contracts::unsupported_replication_config_field(config)
         }
 
         pub(crate) fn invalid_replication_config_status_field(
-            config: &s3s::dto::ReplicationConfiguration,
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
         ) -> Option<&'static str> {
             replication_contracts::invalid_replication_config_status_field(config)
         }
@@ -984,7 +996,7 @@ pub(crate) mod bucket {
         pub(crate) type ReplicationConfigStructureError = replication_contracts::ReplicationConfigStructureError;
 
         pub(crate) fn validate_replication_config_structure(
-            config: &s3s::dto::ReplicationConfiguration,
+            config: &rustfs_gateway_types::persistence::PersistedReplicationConfiguration,
         ) -> Result<(), ReplicationConfigStructureError> {
             replication_contracts::validate_replication_config_structure(config)
         }
@@ -1004,9 +1016,7 @@ pub(crate) mod bucket {
     }
 
     pub(crate) mod utils {
-        pub(crate) fn serialize<T: s3s::xml::Serialize>(val: &T) -> s3s::xml::SerResult<Vec<u8>> {
-            crate::storage::storage_api::ecstore_bucket::utils::serialize(val)
-        }
+        pub(crate) use crate::storage::storage_api::ecstore_bucket::utils::{BucketConfigXml, serialize};
     }
 
     pub(crate) mod versioning_sys {

rustfs/src/site_replication/hooks.rs

@@ -619,7 +619,7 @@ fn retry_bucket_metadata_is_redundant(item: &SRBucketMeta) -> bool {
             })
         }),
         "replication-config" => item.replication_config.as_deref().is_some_and(|raw| {
-            deserialize::<ReplicationConfiguration>(&decode_bucket_meta_wire_value(raw))
+            deserialize::<PersistedReplicationConfiguration>(&decode_bucket_meta_wire_value(raw))
                 .is_ok_and(|config| config.role.trim().is_empty() && config.rules.iter().all(is_derived_site_replication_rule))
         }),
         // `Some("")` is the in-memory sentinel used when the bucket is lock
@@ -1277,7 +1277,9 @@ pub(crate) fn bucket_target_matches_peer(target: &BucketTarget, peer: &PeerInfo)
     bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint)
 }
 
-pub(crate) fn site_replication_target_arns_by_peer(config: Option<&ReplicationConfiguration>) -> HashMap<String, String> {
+pub(crate) fn site_replication_target_arns_by_peer(
+    config: Option<&PersistedReplicationConfiguration>,
+) -> HashMap<String, String> {
     let mut arns_by_peer = HashMap::new();
     let Some(config) = config else {
         return arns_by_peer;
@@ -1364,7 +1366,7 @@ pub(crate) fn reconcile_site_replication_bucket_targets(
     bucket: &str,
     state: &SiteReplicationState,
     local_peer: &PeerInfo,
-    config: Option<&ReplicationConfiguration>,
+    config: Option<&PersistedReplicationConfiguration>,
     service_account_secret_key: &str,
 ) -> S3Result<BucketTargets> {
     if !state.enabled() || state.service_account_access_key.is_empty() || service_account_secret_key.is_empty() {
@@ -1454,7 +1456,7 @@ fn is_site_replication_owned_target(
 /// already-resolved client map rather than rebuilding clients, so it stays cheap enough for
 /// the status path.
 pub(crate) async fn site_replication_targets_online(bucket: &str, replication_config_xml: &[u8]) -> bool {
-    let Ok(config) = deserialize::<ReplicationConfiguration>(replication_config_xml) else {
+    let Ok(config) = deserialize::<PersistedReplicationConfiguration>(replication_config_xml) else {
         return true;
     };
 
@@ -1585,36 +1587,36 @@ pub(crate) fn lifecycle_expiry_statement(
 /// keep only the current peers' rules and treat a leftover as operator state
 /// the edit replaces. An operator-authored `site-repl-*` id on an operator
 /// ARN is outside the shape and survives every pass.
-pub(crate) fn is_derived_site_replication_rule(rule: &ReplicationRule) -> bool {
+pub(crate) fn is_derived_site_replication_rule(rule: &PersistedReplicationRule) -> bool {
     site_replication_rule_deployment_id(rule).is_some()
 }
 
-pub(crate) fn build_site_replication_rule(arn: &str, priority: i32, rule_id: &str) -> ReplicationRule {
-    ReplicationRule {
-        delete_marker_replication: Some(DeleteMarkerReplication {
-            status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
+pub(crate) fn build_site_replication_rule(arn: &str, priority: i32, rule_id: &str) -> PersistedReplicationRule {
+    PersistedReplicationRule {
+        delete_marker_replication: Some(PersistedOptionalReplicationStatus {
+            status: Some(Status::ENABLED.to_string()),
         }),
-        delete_replication: Some(DeleteReplication {
-            status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
+        delete_replication: Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         }),
-        destination: Destination {
+        destination: PersistedReplicationDestination {
             bucket: arn.to_string(),
             ..Default::default()
         },
-        existing_object_replication: Some(ExistingObjectReplication {
-            status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
+        existing_object_replication: Some(PersistedReplicationStatus {
+            status: Status::ENABLED.to_string(),
         }),
         filter: None,
         id: Some(rule_id.to_string()),
         prefix: None,
         priority: Some(priority),
-        source_selection_criteria: Some(SourceSelectionCriteria {
-            replica_modifications: Some(ReplicaModifications {
-                status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
+        source_selection_criteria: Some(PersistedSourceSelectionCriteria {
+            replica_modifications: Some(PersistedReplicationStatus {
+                status: Status::ENABLED.to_string(),
             }),
             sse_kms_encrypted_objects: None,
         }),
-        status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+        status: Status::ENABLED.to_string(),
     }
 }
 
@@ -1623,8 +1625,8 @@ pub(crate) fn build_site_replication_config(
     state: &SiteReplicationState,
     local_peer: &PeerInfo,
     service_account_secret_key: &str,
-    existing: Option<&ReplicationConfiguration>,
-) -> S3Result<Option<ReplicationConfiguration>> {
+    existing: Option<&PersistedReplicationConfiguration>,
+) -> S3Result<Option<PersistedReplicationConfiguration>> {
     // Reuse the ARN already recorded for a peer so the rule keeps pointing at the same
     // bucket target `reconcile_site_replication_bucket_targets` keys off (a MinIO-era
     // `arn:minio:...` target would otherwise be orphaned by a freshly minted ARN).
@@ -1655,7 +1657,7 @@ pub(crate) fn build_site_replication_config(
     if rules.is_empty() {
         Ok(None)
     } else {
-        Ok(Some(ReplicationConfiguration {
+        Ok(Some(PersistedReplicationConfiguration {
             role: String::new(),
             rules,
         }))
@@ -1704,7 +1706,7 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
     bucket: &str,
     state: &SiteReplicationState,
     local_peer: &PeerInfo,
-    config: Option<&ReplicationConfiguration>,
+    config: Option<&PersistedReplicationConfiguration>,
     service_account_secret_key: &str,
     expected_incarnation_id: Uuid,
 ) -> S3Result<bool> {
@@ -1736,7 +1738,9 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime(
     Ok(true)
 }
 
-pub(crate) async fn bucket_replication_config_for_target_refresh(bucket: &str) -> S3Result<Option<ReplicationConfiguration>> {
+pub(crate) async fn bucket_replication_config_for_target_refresh(
+    bucket: &str,
+) -> S3Result<Option<PersistedReplicationConfiguration>> {
     match metadata_sys::get_replication_config(bucket).await {
         Ok((config, _)) => Ok(Some(config)),
         Err(StorageError::ConfigNotFound) => Ok(None),
@@ -1771,7 +1775,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti
     let (existing_role, existing_rules) = existing
         .map(|config| (config.role, config.rules))
         .unwrap_or_else(|| (String::new(), Vec::new()));
-    let mut rules: Vec<ReplicationRule> = existing_rules
+    let mut rules: Vec<PersistedReplicationRule> = existing_rules
         .iter()
         .filter(|rule| !is_derived_site_replication_rule(rule))
         .cloned()
@@ -1795,7 +1799,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti
         return Ok(false);
     }
 
-    let data = serialize(&ReplicationConfiguration { role, rules })
+    let data = serialize(&PersistedReplicationConfiguration { role, rules })
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}")))?;
     metadata_sys::update_if_incarnation(bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
         .await

rustfs/src/site_replication/mod.rs

@@ -56,11 +56,9 @@ use crate::runtime_sources::{
     current_deployment_id, current_endpoints_handle, current_iam_handle, current_object_store_handle, current_region,
 };
 use crate::storage_api::site_replication::s3::{
-    Body, BucketLifecycleConfiguration, BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus,
-    DeleteReplication, DeleteReplicationStatus, Destination, ExistingObjectReplication, ExistingObjectReplicationStatus,
-    LifecycleRule, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
-    ReplicationRuleStatus, S3Error, S3ErrorCode, S3Response, S3Result, SourceSelectionCriteria, VersioningConfiguration,
-    s3_error,
+    Body, BucketLifecycleConfiguration, BucketVersioningStatus, LifecycleRule, PersistedOptionalReplicationStatus,
+    PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationRule, PersistedReplicationStatus,
+    PersistedSourceSelectionCriteria, S3Error, S3ErrorCode, S3Response, S3Result, Status, VersioningConfiguration, s3_error,
 };
 #[cfg(test)]
 use crate::storage_api::site_replication::save_config as save_admin_config;

rustfs/src/site_replication/tests.rs

@@ -3166,8 +3166,8 @@ fn home_office() -> HashSet<String> {
     HashSet::from(["home".to_string(), "office".to_string()])
 }
 
-fn site_repl_config(peer: &str) -> ReplicationConfiguration {
-    ReplicationConfiguration {
+fn site_repl_config(peer: &str) -> PersistedReplicationConfiguration {
+    PersistedReplicationConfiguration {
         role: String::new(),
         rules: vec![build_site_replication_rule(
             &format!("arn:rustfs:replication::{peer}:photos"),
@@ -3177,8 +3177,8 @@ fn site_repl_config(peer: &str) -> ReplicationConfiguration {
     }
 }
 
-fn operator_rule(id: &str) -> ReplicationRule {
-    ReplicationRule {
+fn operator_rule(id: &str) -> PersistedReplicationRule {
+    PersistedReplicationRule {
         id: Some(id.to_string()),
         ..build_site_replication_rule("arn:aws:s3:::backup", 1, id)
     }
@@ -3362,7 +3362,7 @@ fn test_build_site_replication_config_reuses_configured_arn() {
             ..peer("remote", "http://remote.example.com:9000")
         },
     );
-    let existing = ReplicationConfiguration {
+    let existing = PersistedReplicationConfiguration {
         role: String::new(),
         rules: vec![build_site_replication_rule(
             "arn:minio:replication::remote:photos",

rustfs/src/storage/ecfs.rs

@@ -27,6 +27,7 @@ use crate::storage::access::{apply_bucket_generation_guard, bucket_config_mutati
 use crate::storage::helper::OperationHelper;
 use crate::storage::options::get_opts;
 use crate::storage::s3_api::{self, acl};
+use crate::storage::s3_compat::replication::{replication_configuration_from_s3s, replication_configuration_to_s3s};
 use crate::storage::storage_api::ecfs_consumer::contract::{
     bucket::{BucketOperations, BucketOptions},
     object::{ObjectLockRetentionOptions, ObjectOperations as _},
@@ -856,7 +857,10 @@ impl S3 for FS {
     ) -> S3Result<S3Response<GetBucketReplicationOutput>> {
         record_s3_op(S3Operation::GetBucketReplication);
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_get_bucket_replication(req).await
+        let replication_configuration = usecase.execute_get_bucket_replication(req).await?;
+        Ok(S3Response::new(GetBucketReplicationOutput {
+            replication_configuration: Some(replication_configuration_to_s3s(replication_configuration)),
+        }))
     }
 
     async fn get_bucket_request_payment(
@@ -1444,12 +1448,16 @@ impl S3 for FS {
 
     async fn put_bucket_replication(
         &self,
-        req: S3Request<PutBucketReplicationInput>,
+        mut req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
         deny_replication_config_edit_for_non_owner(&req).await?;
         let (site_peers, contract) = site_replication_edit_context().await?;
+        let replication_configuration =
+            replication_configuration_from_s3s(std::mem::take(&mut req.input.replication_configuration));
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_replication(req, site_peers, contract).await
+        usecase
+            .execute_put_bucket_replication(req, replication_configuration, site_peers, contract)
+            .await
     }
 
     async fn put_bucket_request_payment(

rustfs/src/storage/s3_compat/mod.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 //! The legacy s3s edge: conversions from s3s request types into the app
-//! layer's own types (rustfs/backlog#2734). Everything here goes away with the
-//! s3s edge itself (task T4.1).
+//! layer's own types (rustfs/backlog#2734), and between the s3s DTOs and the
+//! `rustfs-gateway-types` shapes the engine crates hold (rustfs/backlog#2745),
+//! one module per configuration family; every such conversion is total and
+//! lossless. Everything here goes away with the s3s edge itself (task T4.1).
 
 mod envelope;
+pub(crate) mod replication;

rustfs/src/storage/s3_compat/replication.rs

@@ -0,0 +1,482 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The s3s edge of the bucket replication configuration.
+//!
+//! Responsible for: the one conversion between the s3s DTO the legacy S3 stack
+//! decodes (`PutBucketReplication`) or encodes (`GetBucketReplication`) and the
+//! gateway persistence shape the replication engine, the bucket metadata and
+//! site replication work on (`rustfs_gateway_types::persistence`).
+//!
+//! Not responsible for: validating a configuration (`rustfs-replication`) or
+//! storing it (`rustfs-ecstore`).
+//!
+//! Both directions are total and lossless: every s3s member has a persisted
+//! member of the same optionality, and every s3s string enum carries its wire
+//! text, so an unknown status or storage class crosses unchanged. The one s3s
+//! member without a counterpart, `ReplicationRuleFilter::cached_tags`, is a
+//! parse cache with no wire form.
+//!
+//! Upstream: `storage::ecfs` (`put_bucket_replication`,
+//! `get_bucket_replication`). Downstream: `app::bucket_usecase`, which takes and
+//! answers the persistence shape. The gateway HTTP stack retires this edge
+//! (T1.8, rustfs/backlog#2749).
+
+use rustfs_gateway_types::persistence::{
+    PersistedAccessControlTranslation, PersistedEncryptionConfiguration, PersistedOptionalReplicationStatus,
+    PersistedReplicationAnd, PersistedReplicationConfiguration, PersistedReplicationDestination, PersistedReplicationFilter,
+    PersistedReplicationMetrics, PersistedReplicationRule, PersistedReplicationStatus, PersistedReplicationTag,
+    PersistedReplicationTime, PersistedReplicationTimeValue, PersistedSourceSelectionCriteria,
+};
+use s3s::dto;
+
+/// The decoded `PutBucketReplication` body in the shape the engine stores.
+pub(crate) fn replication_configuration_from_s3s(value: dto::ReplicationConfiguration) -> PersistedReplicationConfiguration {
+    PersistedReplicationConfiguration {
+        role: value.role,
+        rules: value.rules.into_iter().map(rule_from_s3s).collect(),
+    }
+}
+
+/// A stored configuration in the shape the legacy `GetBucketReplication` encoder takes.
+pub(crate) fn replication_configuration_to_s3s(value: PersistedReplicationConfiguration) -> dto::ReplicationConfiguration {
+    dto::ReplicationConfiguration {
+        role: value.role,
+        rules: value.rules.into_iter().map(rule_to_s3s).collect(),
+    }
+}
+
+/// s3s string enums become their wire text; the persisted shape keeps every
+/// value, known or not.
+fn status_text(value: &str) -> String {
+    value.to_owned()
+}
+
+fn rule_from_s3s(value: dto::ReplicationRule) -> PersistedReplicationRule {
+    PersistedReplicationRule {
+        delete_marker_replication: value
+            .delete_marker_replication
+            .map(|wrapper| PersistedOptionalReplicationStatus {
+                status: wrapper.status.map(|status| status_text(status.as_str())),
+            }),
+        delete_replication: value.delete_replication.map(|wrapper| PersistedReplicationStatus {
+            status: status_text(wrapper.status.as_str()),
+        }),
+        destination: destination_from_s3s(value.destination),
+        existing_object_replication: value.existing_object_replication.map(|wrapper| PersistedReplicationStatus {
+            status: status_text(wrapper.status.as_str()),
+        }),
+        filter: value.filter.map(filter_from_s3s),
+        id: value.id,
+        prefix: value.prefix,
+        priority: value.priority,
+        source_selection_criteria: value
+            .source_selection_criteria
+            .map(|criteria| PersistedSourceSelectionCriteria {
+                replica_modifications: criteria.replica_modifications.map(|wrapper| PersistedReplicationStatus {
+                    status: status_text(wrapper.status.as_str()),
+                }),
+                sse_kms_encrypted_objects: criteria.sse_kms_encrypted_objects.map(|wrapper| PersistedReplicationStatus {
+                    status: status_text(wrapper.status.as_str()),
+                }),
+            }),
+        status: status_text(value.status.as_str()),
+    }
+}
+
+fn destination_from_s3s(value: dto::Destination) -> PersistedReplicationDestination {
+    PersistedReplicationDestination {
+        account: value.account,
+        access_control_translation: value
+            .access_control_translation
+            .map(|translation| PersistedAccessControlTranslation {
+                owner: status_text(translation.owner.as_str()),
+            }),
+        bucket: value.bucket,
+        encryption_configuration: value
+            .encryption_configuration
+            .map(|configuration| PersistedEncryptionConfiguration {
+                replica_kms_key_id: configuration.replica_kms_key_id,
+            }),
+        metrics: value.metrics.map(|metrics| PersistedReplicationMetrics {
+            event_threshold: metrics.event_threshold.map(time_value_from_s3s),
+            status: status_text(metrics.status.as_str()),
+        }),
+        replication_time: value.replication_time.map(|replication_time| PersistedReplicationTime {
+            status: status_text(replication_time.status.as_str()),
+            time: time_value_from_s3s(replication_time.time),
+        }),
+        storage_class: value.storage_class.map(|class| status_text(class.as_str())),
+    }
+}
+
+fn time_value_from_s3s(value: dto::ReplicationTimeValue) -> PersistedReplicationTimeValue {
+    PersistedReplicationTimeValue { minutes: value.minutes }
+}
+
+fn filter_from_s3s(value: dto::ReplicationRuleFilter) -> PersistedReplicationFilter {
+    PersistedReplicationFilter {
+        and: value.and.map(|and| PersistedReplicationAnd {
+            prefix: and.prefix,
+            tags: and.tags.map(|tags| tags.into_iter().map(tag_from_s3s).collect()),
+        }),
+        prefix: value.prefix,
+        tag: value.tag.map(tag_from_s3s),
+    }
+}
+
+/// Both members stay optional: a keyless `<Tag/>` (which the console's rule
+/// form sends) is a filter the engine reads as absent, not a malformed body.
+fn tag_from_s3s(value: dto::Tag) -> PersistedReplicationTag {
+    PersistedReplicationTag {
+        key: value.key,
+        value: value.value,
+    }
+}
+
+fn rule_to_s3s(value: PersistedReplicationRule) -> dto::ReplicationRule {
+    dto::ReplicationRule {
+        delete_marker_replication: value.delete_marker_replication.map(|wrapper| dto::DeleteMarkerReplication {
+            status: wrapper.status.map(dto::DeleteMarkerReplicationStatus::from),
+        }),
+        delete_replication: value.delete_replication.map(|wrapper| dto::DeleteReplication {
+            status: dto::DeleteReplicationStatus::from(wrapper.status),
+        }),
+        destination: destination_to_s3s(value.destination),
+        existing_object_replication: value
+            .existing_object_replication
+            .map(|wrapper| dto::ExistingObjectReplication {
+                status: dto::ExistingObjectReplicationStatus::from(wrapper.status),
+            }),
+        filter: value.filter.map(filter_to_s3s),
+        id: value.id,
+        prefix: value.prefix,
+        priority: value.priority,
+        source_selection_criteria: value.source_selection_criteria.map(|criteria| dto::SourceSelectionCriteria {
+            replica_modifications: criteria.replica_modifications.map(|wrapper| dto::ReplicaModifications {
+                status: dto::ReplicaModificationsStatus::from(wrapper.status),
+            }),
+            sse_kms_encrypted_objects: criteria.sse_kms_encrypted_objects.map(|wrapper| dto::SseKmsEncryptedObjects {
+                status: dto::SseKmsEncryptedObjectsStatus::from(wrapper.status),
+            }),
+        }),
+        status: dto::ReplicationRuleStatus::from(value.status),
+    }
+}
+
+fn destination_to_s3s(value: PersistedReplicationDestination) -> dto::Destination {
+    dto::Destination {
+        access_control_translation: value
+            .access_control_translation
+            .map(|translation| dto::AccessControlTranslation {
+                owner: dto::OwnerOverride::from(translation.owner),
+            }),
+        account: value.account,
+        bucket: value.bucket,
+        encryption_configuration: value
+            .encryption_configuration
+            .map(|configuration| dto::EncryptionConfiguration {
+                replica_kms_key_id: configuration.replica_kms_key_id,
+            }),
+        metrics: value.metrics.map(|metrics| dto::Metrics {
+            event_threshold: metrics.event_threshold.map(time_value_to_s3s),
+            status: dto::MetricsStatus::from(metrics.status),
+        }),
+        replication_time: value.replication_time.map(|replication_time| dto::ReplicationTime {
+            status: dto::ReplicationTimeStatus::from(replication_time.status),
+            time: time_value_to_s3s(replication_time.time),
+        }),
+        storage_class: value.storage_class.map(dto::StorageClass::from),
+    }
+}
+
+fn time_value_to_s3s(value: PersistedReplicationTimeValue) -> dto::ReplicationTimeValue {
+    dto::ReplicationTimeValue { minutes: value.minutes }
+}
+
+fn filter_to_s3s(value: PersistedReplicationFilter) -> dto::ReplicationRuleFilter {
+    dto::ReplicationRuleFilter {
+        and: value.and.map(|and| dto::ReplicationRuleAndOperator {
+            prefix: and.prefix,
+            tags: and.tags.map(|tags| tags.into_iter().map(tag_to_s3s).collect()),
+        }),
+        prefix: value.prefix,
+        tag: value.tag.map(tag_to_s3s),
+        ..Default::default()
+    }
+}
+
+fn tag_to_s3s(value: PersistedReplicationTag) -> dto::Tag {
+    dto::Tag {
+        key: value.key,
+        value: value.value,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{replication_configuration_from_s3s, replication_configuration_to_s3s};
+    use rustfs_gateway_types::persistence::{PersistedReplicationConfiguration, parse_replication, serialize_replication};
+    use s3s::{dto, xml};
+
+    fn s3s_bytes(value: &dto::ReplicationConfiguration) -> Vec<u8> {
+        let mut buf = Vec::new();
+        xml::Serialize::serialize(value, &mut xml::Serializer::new(&mut buf)).expect("the s3s encoder writes the fixture");
+        buf
+    }
+
+    fn tag(key: &str, value: &str) -> dto::Tag {
+        dto::Tag {
+            key: Some(key.to_owned()),
+            value: Some(value.to_owned()),
+        }
+    }
+
+    /// Every member of the family populated, each string enum carrying a value
+    /// outside its known set, so a conversion that drops or normalizes one
+    /// member cannot pass.
+    fn every_member() -> dto::ReplicationConfiguration {
+        dto::ReplicationConfiguration {
+            role: "arn:aws:iam::123456789012:role/replication".to_owned(),
+            rules: vec![
+                dto::ReplicationRule {
+                    delete_marker_replication: Some(dto::DeleteMarkerReplication {
+                        status: Some(dto::DeleteMarkerReplicationStatus::from("FutureDeleteMarker".to_owned())),
+                    }),
+                    delete_replication: Some(dto::DeleteReplication {
+                        status: dto::DeleteReplicationStatus::from("FutureDelete".to_owned()),
+                    }),
+                    destination: dto::Destination {
+                        access_control_translation: Some(dto::AccessControlTranslation {
+                            owner: dto::OwnerOverride::from("FutureOwner".to_owned()),
+                        }),
+                        account: Some("123456789012".to_owned()),
+                        bucket: "arn:aws:s3:::backup".to_owned(),
+                        encryption_configuration: Some(dto::EncryptionConfiguration {
+                            replica_kms_key_id: Some("kms-key".to_owned()),
+                        }),
+                        metrics: Some(dto::Metrics {
+                            event_threshold: Some(dto::ReplicationTimeValue { minutes: Some(15) }),
+                            status: dto::MetricsStatus::from("FutureMetrics".to_owned()),
+                        }),
+                        replication_time: Some(dto::ReplicationTime {
+                            status: dto::ReplicationTimeStatus::from("FutureRtc".to_owned()),
+                            time: dto::ReplicationTimeValue { minutes: Some(30) },
+                        }),
+                        storage_class: Some(dto::StorageClass::from("FutureClass".to_owned())),
+                    },
+                    existing_object_replication: Some(dto::ExistingObjectReplication {
+                        status: dto::ExistingObjectReplicationStatus::from("FutureExisting".to_owned()),
+                    }),
+                    filter: Some(dto::ReplicationRuleFilter {
+                        and: Some(dto::ReplicationRuleAndOperator {
+                            prefix: Some("docs/".to_owned()),
+                            tags: Some(vec![tag("team", "storage"), tag("tier", "gold")]),
+                        }),
+                        prefix: Some("filter/".to_owned()),
+                        tag: Some(tag("env", "prod")),
+                        ..Default::default()
+                    }),
+                    id: Some("every-member".to_owned()),
+                    prefix: Some("legacy/".to_owned()),
+                    priority: Some(7),
+                    source_selection_criteria: Some(dto::SourceSelectionCriteria {
+                        replica_modifications: Some(dto::ReplicaModifications {
+                            status: dto::ReplicaModificationsStatus::from("FutureReplica".to_owned()),
+                        }),
+                        sse_kms_encrypted_objects: Some(dto::SseKmsEncryptedObjects {
+                            status: dto::SseKmsEncryptedObjectsStatus::from("FutureKms".to_owned()),
+                        }),
+                    }),
+                    status: dto::ReplicationRuleStatus::from("FutureRule".to_owned()),
+                },
+                rule("arn:aws:s3:::second", dto::ReplicationRuleStatus::DISABLED),
+            ],
+        }
+    }
+
+    /// A rule with only its required members.
+    fn rule(bucket: &str, status: &'static str) -> dto::ReplicationRule {
+        dto::ReplicationRule {
+            delete_marker_replication: None,
+            delete_replication: None,
+            destination: dto::Destination {
+                bucket: bucket.to_owned(),
+                ..Default::default()
+            },
+            existing_object_replication: None,
+            filter: None,
+            id: None,
+            prefix: None,
+            priority: None,
+            source_selection_criteria: None,
+            status: dto::ReplicationRuleStatus::from_static(status),
+        }
+    }
+
+    fn one_filter(filter: dto::ReplicationRuleFilter) -> dto::ReplicationConfiguration {
+        let mut rule = rule("arn:aws:s3:::backup", dto::ReplicationRuleStatus::ENABLED);
+        rule.filter = Some(filter);
+        dto::ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        }
+    }
+
+    #[test]
+    fn every_member_crosses_to_the_engine_and_back_unchanged() {
+        let crossed = replication_configuration_to_s3s(replication_configuration_from_s3s(every_member()));
+        assert_eq!(crossed, every_member());
+    }
+
+    #[test]
+    fn the_engine_shape_persists_the_bytes_the_s3s_encoder_wrote() {
+        let persisted = serialize_replication(&replication_configuration_from_s3s(every_member()))
+            .expect("the gateway writer accepts the converted configuration");
+        assert_eq!(String::from_utf8_lossy(&persisted), String::from_utf8_lossy(&s3s_bytes(&every_member())));
+    }
+
+    #[test]
+    fn stored_bytes_answer_the_s3s_get_unchanged() {
+        let stored = parse_replication(&s3s_bytes(&every_member())).expect("the gateway parser reads what s3s wrote");
+        assert_eq!(replication_configuration_to_s3s(stored), every_member());
+    }
+
+    #[test]
+    fn a_keyless_tag_stays_keyless_instead_of_becoming_an_empty_key() {
+        let config = replication_configuration_from_s3s(one_filter(dto::ReplicationRuleFilter {
+            tag: Some(dto::Tag { key: None, value: None }),
+            ..Default::default()
+        }));
+        let tag = config.rules[0]
+            .filter
+            .as_ref()
+            .and_then(|filter| filter.tag.as_ref())
+            .expect("the keyless tag is kept, not dropped");
+        assert_eq!(tag.key, None);
+        assert_eq!(tag.value, None);
+    }
+
+    #[test]
+    fn an_empty_tag_key_stays_an_empty_key_instead_of_becoming_absent() {
+        let config = replication_configuration_from_s3s(one_filter(dto::ReplicationRuleFilter {
+            tag: Some(tag("", "")),
+            ..Default::default()
+        }));
+        let tag = config.rules[0].filter.as_ref().and_then(|filter| filter.tag.as_ref());
+        assert_eq!(tag.and_then(|tag| tag.key.as_deref()), Some(""));
+        assert_eq!(tag.and_then(|tag| tag.value.as_deref()), Some(""));
+    }
+
+    #[test]
+    fn an_empty_and_tag_list_stays_distinct_from_an_absent_one() {
+        let empty = replication_configuration_from_s3s(one_filter(dto::ReplicationRuleFilter {
+            and: Some(dto::ReplicationRuleAndOperator {
+                prefix: None,
+                tags: Some(Vec::new()),
+            }),
+            ..Default::default()
+        }));
+        let absent = replication_configuration_from_s3s(one_filter(dto::ReplicationRuleFilter {
+            and: Some(dto::ReplicationRuleAndOperator {
+                prefix: None,
+                tags: None,
+            }),
+            ..Default::default()
+        }));
+        let tags = |config: &PersistedReplicationConfiguration| {
+            config.rules[0]
+                .filter
+                .as_ref()
+                .and_then(|filter| filter.and.as_ref())
+                .map(|and| and.tags.clone())
+        };
+        assert_eq!(tags(&empty), Some(Some(Vec::new())));
+        assert_eq!(tags(&absent), Some(None));
+    }
+
+    #[test]
+    fn an_unknown_status_keeps_its_text_so_the_capability_gate_still_sees_it() {
+        let mut source = one_filter(dto::ReplicationRuleFilter::default());
+        source.rules[0].status = dto::ReplicationRuleStatus::from("enabled".to_owned());
+        let config = replication_configuration_from_s3s(source);
+        assert_eq!(
+            config.rules[0].status, "enabled",
+            "a lowercase status must reach the capability gate as sent, not normalized into a valid one"
+        );
+        let back = replication_configuration_to_s3s(config);
+        assert_eq!(back.rules[0].status.as_str(), "enabled");
+    }
+
+    #[test]
+    fn a_delete_marker_wrapper_without_status_stays_without_status() {
+        let mut source = one_filter(dto::ReplicationRuleFilter::default());
+        source.rules[0].delete_marker_replication = Some(dto::DeleteMarkerReplication { status: None });
+        let config = replication_configuration_from_s3s(source);
+        let wrapper = config.rules[0]
+            .delete_marker_replication
+            .as_ref()
+            .expect("the empty wrapper is kept");
+        assert_eq!(wrapper.status, None);
+        let back = replication_configuration_to_s3s(config);
+        assert_eq!(
+            back.rules[0].delete_marker_replication,
+            Some(dto::DeleteMarkerReplication { status: None })
+        );
+    }
+
+    #[test]
+    fn an_absent_storage_class_is_not_defaulted() {
+        let config = replication_configuration_from_s3s(one_filter(dto::ReplicationRuleFilter::default()));
+        assert_eq!(config.rules[0].destination.storage_class, None);
+        let back = replication_configuration_to_s3s(config);
+        assert_eq!(back.rules[0].destination.storage_class, None);
+    }
+
+    fn s3s_reads(input: &[u8]) -> bool {
+        let mut deserializer = xml::Deserializer::new(input);
+        <dto::ReplicationConfiguration as xml::Deserialize>::deserialize(&mut deserializer).is_ok()
+            && deserializer.expect_eof().is_ok()
+    }
+
+    /// The persisted reader the engine now uses is no more permissive than the
+    /// s3s reader it replaces: every document s3s refused stays refused.
+    #[test]
+    fn the_gateway_reader_refuses_what_the_s3s_reader_refused() {
+        let refused: [&[u8]; 7] = [
+            b"<ReplicationConfiguration><Role></Role><Rule><Future>x</Future><Status>Enabled</Status><Destination><Bucket>b</Bucket></Destination></Rule></ReplicationConfiguration>",
+            b"<ReplicationConfiguration><Role></Role><Rule><Status>Enabled</Status><Destination><Bucket>b</Bucket><Future>x</Future></Destination></Rule></ReplicationConfiguration>",
+            b"<ReplicationConfiguration><Role></Role><Rule><Status>Enabled</Status><Filter><Tag><Key>k</Key><Future>x</Future></Tag></Filter><Destination><Bucket>b</Bucket></Destination></Rule></ReplicationConfiguration>",
+            b"<ReplicationConfiguration><Role></Role><Rule><Status>Enabled</Status><Priority>high</Priority><Destination><Bucket>b</Bucket></Destination></Rule></ReplicationConfiguration>",
+            b"<ReplicationConfiguration><Role></Role><Rule><Destination><Bucket>b</Bucket></Destination></Rule></ReplicationConfiguration>",
+            b"<ReplicationConfiguration><Role></Role><Rule><Status>Enabled</Status></Rule></ReplicationConfiguration>",
+            b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
+        ];
+        for input in refused {
+            let text = String::from_utf8_lossy(input);
+            assert!(!s3s_reads(input), "fixture is not a document s3s refuses: {text}");
+            assert!(parse_replication(input).is_err(), "the gateway reader accepted what s3s refused: {text}");
+        }
+    }
+
+    /// The other direction: a root-level element neither reader knows is
+    /// skipped by both, so the gateway reader is no stricter either.
+    #[test]
+    fn both_readers_skip_an_unknown_root_level_element() {
+        let input: &[u8] = b"<ReplicationConfiguration><Role></Role><FutureTopLevel>x</FutureTopLevel><Rule><Status>Enabled</Status><Destination><Bucket>b</Bucket></Destination></Rule></ReplicationConfiguration>";
+        assert!(s3s_reads(input));
+        let stored = parse_replication(input).expect("the gateway reader skips the unknown root element");
+        assert_eq!(stored.rules.len(), 1);
+    }
+}

rustfs/src/storage/storage_api.rs

@@ -2021,8 +2021,12 @@ pub(crate) fn encode_tags(tags: Vec<s3s::dto::Tag>) -> String {
     ecstore_bucket::tagging::encode_tags(tags)
 }
 
+/// Writes an s3s response body. Bucket configurations go through
+/// `ecstore_bucket::utils::serialize` instead, which owns their persisted form.
 pub(crate) fn serialize<T: s3s::xml::Serialize>(val: &T) -> s3s::xml::SerResult<Vec<u8>> {
-    ecstore_bucket::utils::serialize(val)
+    let mut buf = Vec::with_capacity(256);
+    val.serialize(&mut s3s::xml::Serializer::new(&mut buf))?;
+    Ok(buf)
 }
 
 pub(crate) fn is_err_bucket_not_found(err: &Error) -> bool {

rustfs/src/storage_api.rs

@@ -338,12 +338,12 @@ pub(crate) mod site_replication {
     /// S3 wire types for the service module, funneled here so the module
     /// itself stays off the direct s3s surface (s3s footprint ratchet).
     pub(crate) mod s3 {
-        pub(crate) use s3s::dto::{
-            BucketLifecycleConfiguration, BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus,
-            DeleteReplication, DeleteReplicationStatus, Destination, ExistingObjectReplication, ExistingObjectReplicationStatus,
-            LifecycleRule, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
-            ReplicationRuleStatus, SourceSelectionCriteria, VersioningConfiguration,
+        pub(crate) use rustfs_gateway_types::dto::Status;
+        pub(crate) use rustfs_gateway_types::persistence::{
+            PersistedOptionalReplicationStatus, PersistedReplicationConfiguration, PersistedReplicationDestination,
+            PersistedReplicationRule, PersistedReplicationStatus, PersistedSourceSelectionCriteria,
         };
+        pub(crate) use s3s::dto::{BucketLifecycleConfiguration, BucketVersioningStatus, LifecycleRule, VersioningConfiguration};
         #[cfg(test)]
         pub(crate) use s3s::dto::{ExpirationStatus, LifecycleExpiration, Timestamp, Transition, TransitionStorageClass};
         pub(crate) use s3s::{Body, S3Error, S3ErrorCode, S3Response, S3Result, s3_error};