Merge branch 'main' into overtrue/wait-bucket-metadata-reload

SSE and compression wrap the payload so its length is unknown and
advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved
that sentinel except create_bitrot_writer, which clamped it to 0 before
calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in
the put_file_stream query, so remote peers were told the body was empty.

Since the authenticated put-file trailer (#5868) the receiver used the
declared size to split body from trailer, turning the clamp into a fatal
"auth trailer has trailing data" failure for every SSE PUT on multi-node
deployments (rc.2). #6320 relaxed the receiver to only trust size > 0;
this change fixes the sender so the sentinel survives end to end and the
wire no longer conflates empty objects with unknown-length streams.

Refs #6331

crates/e2e_test/src/cluster_concurrency_test.rs

@@ -15,12 +15,14 @@
 use crate::common::RustFSTestClusterEnvironment;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
 use bytes::Bytes;
 use std::sync::Arc;
 use tokio::sync::Barrier;
 use tracing::{info, warn};
 
 const BUCKET: &str = "conditional-put-race-bucket";
+const BUCKET_METADATA_RELOAD_BUCKET: &str = "bucket-metadata-reload-barrier";
 
 async fn cleanup_object(client: &Client, key: &str) {
     if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
@@ -28,6 +30,16 @@ async fn cleanup_object(client: &Client, key: &str) {
     }
 }
 
+async fn assert_bucket_cors_missing(client: &Client) {
+    let result = client.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await;
+    match result {
+        Err(SdkError::ServiceError(error)) => {
+            assert_eq!(error.err().meta().code(), Some("NoSuchCORSConfiguration"));
+        }
+        result => panic!("expected the peer to report a missing CORS configuration: {result:?}"),
+    }
+}
+
 async fn conditional_put(
     client: &Client,
     key: &str,
@@ -236,3 +248,48 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
     cleanup_object(&client, test_key).await;
     Ok(())
 }
+
+#[tokio::test]
+async fn test_bucket_cors_write_is_visible_on_peer_before_response() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    crate::common::init_logging();
+
+    let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
+    cluster.start().await?;
+    cluster.create_test_bucket(BUCKET_METADATA_RELOAD_BUCKET).await?;
+
+    let writer = cluster.create_s3_client(0)?;
+    let reader = cluster.create_s3_client(1)?;
+    assert_bucket_cors_missing(&reader).await;
+
+    let rule = CorsRule::builder()
+        .allowed_methods("GET")
+        .allowed_origins("https://example.com")
+        .build()?;
+    let configuration = CorsConfiguration::builder().cors_rules(rule).build()?;
+
+    writer
+        .put_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .cors_configuration(configuration)
+        .send()
+        .await?;
+
+    let response = reader.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
+    let rules = response.cors_rules();
+    assert_eq!(
+        rules.len(),
+        1,
+        "peer should observe the committed CORS rule before the write response returns"
+    );
+    assert_eq!(rules[0].allowed_methods(), ["GET"]);
+    assert_eq!(rules[0].allowed_origins(), ["https://example.com"]);
+
+    writer
+        .delete_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .send()
+        .await?;
+    assert_bucket_cors_missing(&reader).await;
+    writer.delete_bucket().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
+    Ok(())
+}

crates/ecstore/src/api/mod.rs

@@ -196,16 +196,15 @@ pub mod bucket {
             ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
             ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
             ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            VersionPurgeStatusType, XferStats, assign_site_replication_rule_priorities, commit_force_delete_intent,
-            complete_force_delete_intent, delete_replication_state_from_config, delete_replication_version_id,
-            get_global_replication_pool, get_global_replication_stats, get_proxy_targets, init_background_replication,
-            invalid_replication_config_status_field, is_site_replication_rule, merge_incoming_replication_config,
-            merge_user_replication_config, persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta,
-            replication_status_to_filemeta, replication_statuses_map, replication_target_arn_deployment_id,
-            replication_target_arns, resync_start_conflict_id, should_remove_replication_target,
-            should_schedule_delete_replication, should_use_existing_delete_replication_info,
-            should_use_existing_delete_replication_source, unsupported_replication_config_field,
-            validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta,
+            VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
+            delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
+            get_global_replication_stats, get_proxy_targets, init_background_replication,
+            invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog,
+            replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
+            resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
+            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
+            unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+            version_purge_status_to_filemeta,
         };
     }
 

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

@@ -47,10 +47,8 @@ pub use replication_config_boundary::{
     ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
     REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
     ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError,
-    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_rule,
-    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
-    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
-    validate_replication_config_structure, validate_replication_config_target_arns,
+    invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target,
+    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
 };
 pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{

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

@@ -16,8 +16,6 @@ pub use rustfs_replication::{
     ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
     REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
     ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError,
-    assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_rule,
-    merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id,
-    replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
-    validate_replication_config_structure, validate_replication_config_target_arns,
+    invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target,
+    unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
 };

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -85,6 +85,7 @@ const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
 const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;
+const BUCKET_METADATA_RELOAD_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// Error for a peer that reported `success = false` without an `error_info` payload.
 ///
@@ -1328,27 +1329,38 @@ impl PeerRestClient {
     }
 
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
-        self.finalize_result(
-            async {
-                let mut client = self.get_client().await?;
-                let mut request = Request::new(LoadBucketMetadataRequest {
-                    bucket: bucket.to_string(),
-                    scanner_maintenance_change,
-                });
-                set_tonic_mutation_body_digest(&mut request)?;
-
-                let response = client.load_bucket_metadata(request).await?.into_inner();
-                if !response.success {
-                    if let Some(msg) = response.error_info {
-                        return Err(Error::other(msg));
-                    }
-                    return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
-                }
-                Ok(())
+        let result = tokio::time::timeout(BUCKET_METADATA_RELOAD_TIMEOUT, async {
+            let result = self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
+            if let Err(err) = &result
+                && Self::is_network_like_error(err)
+            {
+                self.prepare_retry().await;
+                return self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await;
             }
-            .await,
-        )
+            result
+        })
         .await
+        .unwrap_or_else(|_| Err(Error::other(format!("load_bucket_metadata({bucket}) timed out"))));
+        self.finalize_result(result).await
+    }
+
+    async fn load_bucket_metadata_once(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
+        let mut client = self.get_client().await?;
+        let mut request = Request::new(LoadBucketMetadataRequest {
+            bucket: bucket.to_string(),
+            scanner_maintenance_change,
+        });
+        set_tonic_mutation_body_digest(&mut request)?;
+        request.set_timeout(BUCKET_METADATA_RELOAD_TIMEOUT);
+
+        let response = client.load_bucket_metadata(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
+        }
+        Ok(())
     }
 
     pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> {

crates/ecstore/src/io_support/bitrot.rs

@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/replication/src/config.rs

@@ -270,160 +270,6 @@ pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguratio
     arns
 }
 
-/// Deployment id extracted from a site-replication target ARN
-/// (`arn:{rustfs|minio}:replication::<deployment-id>:<bucket>`), or `None`
-/// for an operator-authored ARN.
-pub fn replication_target_arn_deployment_id(arn: &str) -> Option<String> {
-    let parts: Vec<_> = arn.split(':').collect();
-    if parts.len() == 6
-        && parts[0] == "arn"
-        && matches!(parts[1], "rustfs" | "minio")
-        && parts[2] == "replication"
-        && !parts[4].is_empty()
-    {
-        return Some(parts[4].to_string());
-    }
-
-    None
-}
-
-/// Rule id prefix the site-replication reconciler stamps on the rules it
-/// derives (`site-repl-<peer deployment id>`).
-pub const SITE_REPLICATION_RULE_ID_PREFIX: &str = "site-repl-";
-
-/// Whether `rule` carries a site-replication rule id (`site-repl-*`). The
-/// reconciler and the peer ingestion path treat the whole namespace as theirs
-/// on a site-replication bucket; the S3 edit path must not — rule ids are not
-/// reserved, so see [`site_replication_rule_deployment_id`].
-pub fn is_site_replication_rule(rule: &ReplicationRule) -> bool {
-    rule.id
-        .as_deref()
-        .is_some_and(|id| id.starts_with(SITE_REPLICATION_RULE_ID_PREFIX))
-}
-
-/// Deployment id of the peer a reconciler-derived rule replicates to, or
-/// `None` for any other rule. The reconciler builds each rule from one peer:
-/// the id is `site-repl-<deployment id>` and the destination ARN names that
-/// same deployment id — an operator-authored `site-repl-user` rule, or a
-/// `site-repl-<peer>` id pasted onto a foreign ARN, fails the agreement check.
-/// Callers that know the current peer set must also confirm the id is one of
-/// those peers before treating the rule as reconciler-owned.
-pub fn site_replication_rule_deployment_id(rule: &ReplicationRule) -> Option<&str> {
-    let deployment_id = rule.id.as_deref()?.strip_prefix(SITE_REPLICATION_RULE_ID_PREFIX)?;
-    (!deployment_id.is_empty()
-        && replication_target_arn_deployment_id(&rule.destination.bucket).as_deref() == Some(deployment_id))
-    .then_some(deployment_id)
-}
-
-/// Whether `rule` is one the local reconciler derived for a current remote
-/// site-replication peer in `peer_deployment_ids`. With an empty peer set
-/// (site replication disabled) nothing qualifies, so a bucket outside site
-/// replication keeps the verbatim S3 put/delete semantics.
-pub fn is_reconciler_owned_site_replication_rule(rule: &ReplicationRule, peer_deployment_ids: &HashSet<String>) -> bool {
-    site_replication_rule_deployment_id(rule).is_some_and(|deployment_id| peer_deployment_ids.contains(deployment_id))
-}
-
-/// Merge an incoming replication config into the local one.
-///
-/// `site-repl-*` rules encode the *holder's* outbound direction — their
-/// destination ARN names another site — so applying an external rule set
-/// verbatim replaces the local reverse rule with one this site can never
-/// satisfy (no bucket target backs it) and replication silently stops. Only
-/// operator-authored rules travel: the site-replication peer ingestion path
-/// and the S3 put/delete-bucket-replication path both keep the local site's
-/// `site-repl-*` rules through this merge. `incoming == None` models a
-/// delete of the operator-authored rules.
-pub fn merge_incoming_replication_config(
-    incoming: Option<ReplicationConfiguration>,
-    local: Option<ReplicationConfiguration>,
-) -> Option<ReplicationConfiguration> {
-    merge_replication_config_keeping_site_rules(incoming, local, is_site_replication_rule)
-}
-
-/// [`merge_incoming_replication_config`] for the S3 put/delete-bucket-replication
-/// path (issue #1948): only rules the local reconciler derived for a current
-/// peer in `peer_deployment_ids` survive as site rules; every other stored
-/// rule — including an operator-authored `site-repl-*` id — is operator state
-/// that the request replaces or deletes. An incoming rule whose id is a
-/// current peer's `site-repl-<id>` is dropped whatever its ARN: accepting it
-/// would duplicate the reconciler rule's id.
-pub fn merge_user_replication_config(
-    incoming: Option<ReplicationConfiguration>,
-    local: Option<ReplicationConfiguration>,
-    peer_deployment_ids: &HashSet<String>,
-) -> Option<ReplicationConfiguration> {
-    let incoming = incoming.map(|mut config| {
-        config.rules.retain(|rule| {
-            !rule
-                .id
-                .as_deref()
-                .and_then(|id| id.strip_prefix(SITE_REPLICATION_RULE_ID_PREFIX))
-                .is_some_and(|deployment_id| peer_deployment_ids.contains(deployment_id))
-        });
-        config
-    });
-    merge_replication_config_keeping_site_rules(incoming, local, |rule| {
-        is_reconciler_owned_site_replication_rule(rule, peer_deployment_ids)
-    })
-}
-
-fn merge_replication_config_keeping_site_rules(
-    incoming: Option<ReplicationConfiguration>,
-    local: Option<ReplicationConfiguration>,
-    is_site_rule: impl Fn(&ReplicationRule) -> bool,
-) -> Option<ReplicationConfiguration> {
-    let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default();
-    // Operator rules first, then the local site rules — the same order the
-    // site-replication reconciler produces, so its no-op check matches and
-    // the bucket metadata is written once per broadcast, not twice.
-    let mut rules: Vec<ReplicationRule> = incoming
-        .into_iter()
-        .flat_map(|config| config.rules)
-        .filter(|rule| !is_site_rule(rule))
-        .collect();
-    rules.extend(local.into_iter().flat_map(|config| config.rules).filter(&is_site_rule));
-
-    if rules.is_empty() {
-        return None;
-    }
-
-    assign_site_replication_rule_priorities(&mut rules, &is_site_rule);
-
-    // A site-replication ARN in `role` is the sender's, and the reconciler's
-    // per-peer target lookup reads it — carrying it over would pin the
-    // receiver's targets to the sender's identity.
-    let role = match replication_target_arn_deployment_id(&incoming_role) {
-        Some(_) => String::new(),
-        None => incoming_role,
-    };
-
-    Some(ReplicationConfiguration { role, rules })
-}
-
-/// Give the site rules in `rules` the lowest priorities no operator rule uses,
-/// in rule order, leaving every operator rule's priority untouched. Operator
-/// priorities decide which rule wins per target, so they are part of the
-/// submitted policy; site rules are derived state and only need to be unique
-/// (`validate_replication_config_structure` rejects duplicates). The result
-/// is a pure function of the rule list, so the site-replication reconciler,
-/// the peer ingestion merge and the S3 edit merge all converge on the same
-/// bytes and the reconciler's no-op check holds.
-pub fn assign_site_replication_rule_priorities(rules: &mut [ReplicationRule], is_site_rule: impl Fn(&ReplicationRule) -> bool) {
-    let taken: HashSet<i32> = rules
-        .iter()
-        .filter(|rule| !is_site_rule(rule))
-        .map(|rule| rule.priority.unwrap_or(0))
-        .collect();
-    let mut next = 1;
-    for rule in rules.iter_mut().filter(|rule| is_site_rule(rule)) {
-        while taken.contains(&next) {
-            next += 1;
-        }
-        rule.priority = Some(next);
-        next = next.saturating_add(1);
-    }
-}
-
 pub fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet<String> {
     let role = config.role.trim();
     if !role.is_empty() {
@@ -1698,103 +1544,4 @@ mod tests {
             "the child rule must win for target A while the overlapping child target B remains eligible"
         );
     }
-
-    #[test]
-    fn site_replication_rule_deployment_id_requires_id_and_arn_agreement() {
-        let reconciler_rule = replication_rule("site-repl-peer-dep", "arn:rustfs:replication::peer-dep:bucket");
-        assert_eq!(site_replication_rule_deployment_id(&reconciler_rule), Some("peer-dep"));
-
-        // A remote-target ARN carries the remote's deployment id (or a random
-        // uuid), never the operator's rule id.
-        let operator_named_rule = replication_rule("site-repl-user", "arn:minio:replication:us-east-1:2f1c-remote:bucket");
-        assert_eq!(site_replication_rule_deployment_id(&operator_named_rule), None);
-
-        let foreign_arn = replication_rule("site-repl-peer-dep", "arn:rustfs:replication::other-dep:bucket");
-        assert_eq!(site_replication_rule_deployment_id(&foreign_arn), None);
-
-        let empty_id = replication_rule("site-repl-", "arn:rustfs:replication::peer-dep:bucket");
-        assert_eq!(site_replication_rule_deployment_id(&empty_id), None);
-
-        let peers = HashSet::from(["peer-dep".to_string()]);
-        assert!(is_reconciler_owned_site_replication_rule(&reconciler_rule, &peers));
-        assert!(!is_reconciler_owned_site_replication_rule(&reconciler_rule, &HashSet::new()));
-        let removed_peer = replication_rule("site-repl-gone-dep", "arn:rustfs:replication::gone-dep:bucket");
-        assert!(!is_reconciler_owned_site_replication_rule(&removed_peer, &peers));
-    }
-
-    // The merge must not rewrite the operator's priorities: with the
-    // priority-5 rule listed first and renumbered 1 then 2, the priority-1
-    // delete-marker-disabled rule would win the replication decision.
-    #[test]
-    fn merge_keeps_operator_priorities_and_replication_decision() {
-        let user_arn = "arn:minio:replication:us-east-1:2f1c-remote:bucket";
-        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
-        let incoming = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![
-                delete_marker_rule("dm-enabled", user_arn, "logs/", 5, true),
-                delete_marker_rule("dm-disabled", user_arn, "logs/2026/", 1, false),
-            ],
-        };
-        let mut site_rule = delete_marker_rule("site-repl-peer-dep", peer_arn, "", 7, true);
-        site_rule.prefix = None;
-        let local = structure_config(vec![site_rule]);
-        let opts = ObjectOpts {
-            name: "logs/2026/app.log".to_string(),
-            op_type: ReplicationType::Delete,
-            delete_marker: true,
-            version_id: None,
-            ..Default::default()
-        };
-        let submitted: Vec<_> = incoming.filter_target_replication_decisions(&opts);
-
-        let peers = HashSet::from(["peer-dep".to_string()]);
-        let merged = merge_user_replication_config(Some(incoming.clone()), Some(local.clone()), &peers).expect("rules");
-
-        let priorities: Vec<_> = merged
-            .rules
-            .iter()
-            .map(|rule| (rule.id.as_deref().unwrap(), rule.priority))
-            .collect();
-        assert_eq!(
-            priorities,
-            vec![
-                ("dm-enabled", Some(5)),
-                ("dm-disabled", Some(1)),
-                ("site-repl-peer-dep", Some(2))
-            ],
-            "operator priorities are kept verbatim; the site rule takes the lowest free slot"
-        );
-        assert!(validate_replication_config_structure(&merged).is_ok());
-        let mut decisions = merged.filter_target_replication_decisions(&opts);
-        decisions.retain(|(arn, _)| arn == user_arn);
-        assert_eq!(decisions, submitted, "the merged config must replicate exactly as the operator submitted");
-        assert_eq!(decisions, vec![(user_arn.to_string(), true)]);
-
-        // The peer ingestion merge follows the same rule.
-        let merged = merge_incoming_replication_config(Some(incoming), Some(local)).expect("rules");
-        let priorities: Vec<_> = merged.rules.iter().map(|rule| rule.priority).collect();
-        assert_eq!(priorities, vec![Some(5), Some(1), Some(2)]);
-    }
-
-    #[test]
-    fn site_rule_priorities_skip_every_operator_priority() {
-        let mut rules = vec![
-            delete_marker_rule("a", "arn:a", "", 2, true),
-            delete_marker_rule("site-repl-x", "arn:rustfs:replication::x:b", "", 9, true),
-            delete_marker_rule("b", "arn:a", "", 1, true),
-            delete_marker_rule("site-repl-y", "arn:rustfs:replication::y:b", "", 9, true),
-            delete_marker_rule("c", "arn:a", "", 4, true),
-        ];
-        assign_site_replication_rule_priorities(&mut rules, is_site_replication_rule);
-        let priorities: Vec<_> = rules.iter().map(|rule| rule.priority).collect();
-        assert_eq!(priorities, vec![Some(2), Some(3), Some(1), Some(5), Some(4)]);
-        assert!(validate_replication_config_structure(&structure_config(rules.clone())).is_ok());
-
-        // Idempotent, so the reconciler's pass over an already-merged config
-        // is a byte-stable no-op rather than a rewrite every period.
-        let settled = rules.clone();
-        assign_site_replication_rule_priorities(&mut rules, is_site_replication_rule);
-        assert_eq!(rules, settled);
-    }
 }

crates/replication/src/lib.rs

@@ -32,11 +32,9 @@ pub use config::{
     ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
     REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
     ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError,
-    active_replication_rule_destination_arns, assign_site_replication_rule_priorities, invalid_replication_config_status_field,
-    is_reconciler_owned_site_replication_rule, is_site_replication_rule, merge_incoming_replication_config,
-    merge_user_replication_config, replication_target_arn_deployment_id, replication_target_arns,
-    should_remove_replication_target, site_replication_rule_deployment_id, unsupported_replication_config_field,
-    validate_replication_config_structure, validate_replication_config_target_arns,
+    active_replication_rule_destination_arns, invalid_replication_config_status_field, replication_target_arns,
+    should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure,
+    validate_replication_config_target_arns,
 };
 pub use delete::{
     DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,

rustfs/src/admin/handlers/site_replication.rs

@@ -31,10 +31,6 @@ use crate::admin::storage_api::bucket::metadata::{
 use crate::admin::storage_api::bucket::metadata_sys;
 use crate::admin::storage_api::bucket::quota::BucketQuota;
 use crate::admin::storage_api::bucket::replication;
-use crate::admin::storage_api::bucket::replication::{
-    assign_site_replication_rule_priorities, is_site_replication_rule, merge_incoming_replication_config,
-    replication_target_arn_deployment_id,
-};
 use crate::admin::storage_api::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials};
 use crate::admin::storage_api::bucket::target_sys::BucketTargetSys;
 use crate::admin::storage_api::bucket::utils::{deserialize, serialize};
@@ -1122,36 +1118,6 @@ async fn load_site_replication_state() -> S3Result<SiteReplicationState> {
     }
 }
 
-/// Whether this deployment participates in site replication (two or more
-/// peers in the persisted state). Read by the S3 interface layer to gate
-/// replication-config edits (MinIO `ErrReplicationDenyEditError` semantics,
-/// issue #1948); a state-read failure propagates so the gate fails closed.
-pub(crate) async fn site_replication_enabled() -> S3Result<bool> {
-    Ok(load_site_replication_state().await?.enabled())
-}
-
-/// Deployment ids of the remote peers the reconciler derives a
-/// `site-repl-<id>` rule for on every bucket (the same peer filter as
-/// `build_site_replication_config`); empty when site replication is not
-/// enabled. Read by the bucket usecase so an S3 replication-config edit keeps
-/// exactly the reconciler-owned rules (issue #1948); a state-read failure
-/// propagates so the edit fails closed.
-pub(crate) async fn site_replication_remote_peer_deployment_ids() -> S3Result<HashSet<String>> {
-    let state = load_site_replication_state().await?;
-    if !state.enabled() {
-        return Ok(HashSet::new());
-    }
-    let local_peer = current_local_runtime_peer(&state);
-    Ok(state
-        .peers
-        .values()
-        .filter(|peer| {
-            peer.deployment_id != local_peer.deployment_id && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
-        })
-        .map(|peer| peer.deployment_id.clone())
-        .collect())
-}
-
 async fn load_site_replication_state_no_lock(store: Arc<ECStore>) -> S3Result<SiteReplicationState> {
     match read_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await {
         Ok(data) => parse_site_replication_state(&data),
@@ -7796,6 +7762,20 @@ fn bucket_target_deployment_id(target: &BucketTarget) -> Option<String> {
     replication_target_arn_deployment_id(&target.arn)
 }
 
+fn replication_target_arn_deployment_id(arn: &str) -> Option<String> {
+    let parts: Vec<_> = arn.split(':').collect();
+    if parts.len() == 6
+        && parts[0] == "arn"
+        && matches!(parts[1], "rustfs" | "minio")
+        && parts[2] == "replication"
+        && !parts[4].is_empty()
+    {
+        return Some(parts[4].to_string());
+    }
+
+    None
+}
+
 fn prune_removed_site_replication_bucket_targets(
     existing: BucketTargets,
     removed_deployment_ids: &HashSet<String>,
@@ -7820,6 +7800,10 @@ fn prune_removed_site_replication_bucket_targets(
     (BucketTargets { targets }, removed)
 }
 
+fn is_site_replication_rule(rule: &ReplicationRule) -> bool {
+    rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-"))
+}
+
 /// Whether every `site-repl-*` rule on this bucket resolves to a live remote target.
 ///
 /// The rule set alone cannot answer this: a rule can be perfectly formed while the endpoint
@@ -7845,6 +7829,52 @@ async fn site_replication_targets_online(bucket: &str, replication_config_xml: &
     true
 }
 
+/// Merge a peer's replication config into the local one.
+///
+/// `site-repl-*` rules encode the *sender's* outbound direction — their destination ARN
+/// names the receiver — so applying a peer's rule set verbatim replaces the receiver's
+/// reverse rule with one pointing at itself. No bucket target can satisfy that ARN
+/// (`reconcile_site_replication_bucket_targets` skips the local peer), so the receiver
+/// silently stops replicating back: the one-directional symptom. Only operator-authored
+/// rules travel between sites; each site owns its own `site-repl-*` rules.
+fn merge_incoming_replication_config(
+    incoming: Option<ReplicationConfiguration>,
+    local: Option<ReplicationConfiguration>,
+) -> Option<ReplicationConfiguration> {
+    let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default();
+    // Operator rules first, then the local site rules — the same order
+    // `ensure_site_replication_bucket_replication_config_with_runtime` produces, so its
+    // no-op check matches and the bucket metadata is written once per broadcast, not twice.
+    let mut rules: Vec<ReplicationRule> = incoming
+        .into_iter()
+        .flat_map(|config| config.rules)
+        .filter(|rule| !is_site_replication_rule(rule))
+        .collect();
+    rules.extend(
+        local
+            .into_iter()
+            .flat_map(|config| config.rules)
+            .filter(is_site_replication_rule),
+    );
+
+    if rules.is_empty() {
+        return None;
+    }
+
+    for (index, rule) in rules.iter_mut().enumerate() {
+        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
+    }
+
+    // A site-replication ARN in `role` is the sender's, and `site_replication_target_arns_by_peer`
+    // reads it — carrying it over would pin the receiver's targets to the sender's identity.
+    let role = match replication_target_arn_deployment_id(&incoming_role) {
+        Some(_) => String::new(),
+        None => incoming_role,
+    };
+
+    Some(ReplicationConfiguration { role, rules })
+}
+
 /// Merge a peer's ILM expiry document into the local lifecycle config.
 ///
 /// Mirrors MinIO's `mergeWithCurrentLCConfig` with one hardening: incoming
@@ -8183,7 +8213,9 @@ fn prune_removed_site_replication_rules(
         return (None, removed);
     }
 
-    assign_site_replication_rule_priorities(&mut config.rules, is_site_replication_rule);
+    for (index, rule) in config.rules.iter_mut().enumerate() {
+        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
+    }
 
     (Some(config), removed)
 }
@@ -8357,10 +8389,9 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime(
         .cloned()
         .collect();
     rules.extend(desired.rules);
-    // Operator priorities are the operator's policy; only the derived rules
-    // take free slots, by the same function as the config merges so a merged
-    // write and this pass agree byte for byte.
-    assign_site_replication_rule_priorities(&mut rules, is_site_replication_rule);
+    for (index, rule) in rules.iter_mut().enumerate() {
+        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
+    }
 
     // Only a site-replication ARN in `role` is ours to drop — an operator-authored role is
     // part of the bucket's S3-visible configuration, and repairing a reverse rule must not
@@ -17058,7 +17089,7 @@ mod tests {
     }
 
     #[test]
-    fn test_prune_removed_site_replication_rules_removes_site_rule_and_keeps_operator_priority() {
+    fn test_prune_removed_site_replication_rules_removes_site_rule_and_reorders_priorities() {
         let removed_deployment_ids = HashSet::from(["removed-dep".to_string()]);
         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");
@@ -17075,9 +17106,9 @@ mod tests {
         assert!(updated.role.is_empty());
         assert_eq!(updated.rules.len(), 2);
         assert_eq!(updated.rules[0].id.as_deref(), Some("user-managed-rule"));
-        assert_eq!(updated.rules[0].priority, Some(9), "the operator's priority is policy and stays");
+        assert_eq!(updated.rules[0].priority, Some(1));
         assert_eq!(updated.rules[1].id.as_deref(), Some("site-repl-kept-dep"));
-        assert_eq!(updated.rules[1].priority, Some(1), "the derived rule moves to the lowest free slot");
+        assert_eq!(updated.rules[1].priority, Some(2));
     }
 
     #[test]

rustfs/src/admin/storage_api.rs

@@ -443,8 +443,6 @@ pub(crate) mod replication {
     pub(crate) use super::ecstore_bucket::replication::{
         REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS,
         REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
-        assign_site_replication_rule_priorities, is_site_replication_rule, merge_incoming_replication_config,
-        replication_target_arn_deployment_id,
     };
     pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
     pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;

rustfs/src/app/bucket_usecase.rs

@@ -38,9 +38,9 @@ use super::storage_api::bucket_usecase::bucket::{
     metadata_sys,
     policy_sys::PolicySys,
     replication::{
-        ReplicationTargetValidationError, invalid_replication_config_status_field, merge_user_replication_config,
-        replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
-        validate_replication_config_structure, validate_replication_config_target_arns,
+        ReplicationTargetValidationError, invalid_replication_config_status_field, replication_target_arns,
+        should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure,
+        validate_replication_config_target_arns,
     },
     target::{BucketTargetType, BucketTargets},
     utils::serialize,
@@ -513,13 +513,15 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
     }
 }
 
-fn notify_bucket_metadata_reload(
+async fn notify_bucket_metadata_reload(
     bucket: String,
     operation: &'static str,
     request_context: Option<request_context::RequestContext>,
     scanner_maintenance_change: bool,
 ) {
     record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change);
+    // Keep reload detached across request cancellation, but wait before a healthy peer can serve the previous config.
+    let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
     spawn_background_with_context(request_context, async move {
         if let Some(notification_sys) = current_notification_system() {
             let result = if scanner_maintenance_change {
@@ -531,7 +533,9 @@ fn notify_bucket_metadata_reload(
                 warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}");
             }
         }
+        let _ = completed_tx.send(());
     });
+    let _ = completed_rx.await;
 }
 
 fn record_local_scanner_maintenance_reload(bucket: &str, scanner_maintenance_change: bool) {
@@ -623,52 +627,11 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo
     validate_replication_config_targets(&targets, config)
 }
 
-/// Defense in depth for site-replication-managed buckets (issue #1948): an S3
-/// PutBucketReplication replaces the operator-authored rules but must not wipe
-/// the rules the reconciler derived for the current remote peers
-/// (`site_peer_deployment_ids`) — until its next pass (600s period) every
-/// peer link on this bucket would be silently dead. The same merge also drops
-/// incoming impostors of those rules. An empty peer set (site replication
-/// disabled) keeps the verbatim overwrite semantics: rule ids are not
-/// reserved, so an operator's own `site-repl-*` rule is ordinary state there.
-fn merge_user_replication_config_update(
-    incoming: ReplicationConfiguration,
-    existing: Option<ReplicationConfiguration>,
-    site_peer_deployment_ids: &HashSet<String>,
-) -> ReplicationConfiguration {
-    if site_peer_deployment_ids.is_empty() {
-        return incoming;
-    }
-    // `incoming` passed structure validation, so it holds at least one rule;
-    // `None` is only reachable when every incoming rule impersonates a
-    // reconciler rule, and then the stored reconciler rules are what remains.
-    merge_user_replication_config(Some(incoming.clone()), existing, site_peer_deployment_ids).unwrap_or(incoming)
-}
-
-/// Split of an S3 DeleteBucketReplication on the stored config (issue #1948):
-/// the operator-authored rules are removed, the rules the reconciler derived
-/// for the current remote peers survive (`None` means nothing survives and
-/// the config is deleted), and the returned ARNs are the ones whose bucket
-/// targets may be garbage-collected — never an ARN a surviving reconciler
-/// rule still points at.
-fn split_replication_config_for_user_delete(
-    config: ReplicationConfiguration,
-    site_peer_deployment_ids: &HashSet<String>,
-) -> (Option<ReplicationConfiguration>, HashSet<String>) {
-    let mut removable_arns = replication_target_arns(&config);
-    let remaining = merge_user_replication_config(None, Some(config), site_peer_deployment_ids);
-    if let Some(remaining) = remaining.as_ref() {
-        for rule in &remaining.rules {
-            removable_arns.remove(rule.destination.bucket.trim());
-        }
-    }
-    (remaining, removable_arns)
-}
-
-async fn replication_targets_without_arns(
+async fn replication_targets_without_config_targets(
     bucket: &str,
-    target_arns: &HashSet<String>,
+    config: &ReplicationConfiguration,
 ) -> S3Result<Option<(BucketTargets, usize)>> {
+    let target_arns = replication_target_arns(config);
     if target_arns.is_empty() {
         return Ok(None);
     }
@@ -679,7 +642,7 @@ async fn replication_targets_without_arns(
         Err(err) => return Err(ApiError::from(err).into()),
     };
 
-    let removed = remove_replication_targets_from_config_targets(&mut targets, target_arns);
+    let removed = remove_replication_targets_from_config_targets(&mut targets, &target_arns);
     if removed == 0 {
         return Ok(None);
     }
@@ -1517,7 +1480,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1549,7 +1512,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1581,7 +1544,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1613,7 +1576,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "policy");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1623,15 +1586,9 @@ impl DefaultBucketUsecase {
         Ok(S3Response::new(DeleteBucketPolicyOutput {}))
     }
 
-    /// `site_peers` is the set of remote site-replication peer deployment ids
-    /// (empty when site replication is disabled). The interface layer reads it
-    /// from the persisted state and fails closed on a read error, so this
-    /// usecase stays a pure function of its inputs (layer rule: app never
-    /// imports interface).
     pub async fn execute_delete_bucket_replication(
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
-        site_peers: HashSet<String>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
         let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
         let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
@@ -1651,29 +1608,15 @@ impl DefaultBucketUsecase {
             Err(StorageError::ConfigNotFound) => None,
             Err(err) => return Err(ApiError::from(err).into()),
         };
-        let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() {
-            let (remaining, removable_arns) = split_replication_config_for_user_delete(config.clone(), &site_peers);
-            let targets = replication_targets_without_arns(&bucket, &removable_arns).await?;
-            (remaining, targets)
+        let updated_targets = if let Some(config) = replication_config.as_ref() {
+            replication_targets_without_config_targets(&bucket, config).await?
         } else {
-            (None, None)
+            None
         };
 
-        match remaining_config {
-            // Site-replication rules and the targets backing them survive the
-            // S3 delete (issue #1948); only the operator-authored rules go.
-            Some(remaining) => {
-                let data = serialize_config(&remaining)?;
-                update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
-                    .await
-                    .map_err(ApiError::from)?;
-            }
-            None => {
-                delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id)
-                    .await
-                    .map_err(ApiError::from)?;
-            }
-        }
+        delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
         if let Some((targets, removed)) = updated_targets
             && let Err(err) =
                 write_replication_targets_after_config_delete(&bucket, &targets, removed, expected_incarnation_id).await
@@ -1691,7 +1634,7 @@ impl DefaultBucketUsecase {
         }
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1716,7 +1659,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false).await;
 
         let item = sr_bucket_meta_item(bucket.clone(), "tags");
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1749,7 +1692,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false).await;
 
         Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT))
     }
@@ -2204,7 +2147,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config");
         item.sse_config = Some(
@@ -2283,7 +2226,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         item.expiry_lc_config =
@@ -2368,7 +2311,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await;
 
         let region = resolve_notification_region(self.global_region(), request_region);
         let notify = current_notify_interface_for_context(self.context.as_deref());
@@ -2473,7 +2416,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "policy");
         item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?);
@@ -2508,7 +2451,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
         item.cors =
@@ -2520,11 +2463,9 @@ impl DefaultBucketUsecase {
         Ok(S3Response::new(PutBucketCorsOutput::default()))
     }
 
-    /// See [`Self::execute_delete_bucket_replication`] for `site_peers`.
     pub async fn execute_put_bucket_replication(
         &self,
         req: S3Request<PutBucketReplicationInput>,
-        site_peers: HashSet<String>,
     ) -> 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();
@@ -2548,20 +2489,13 @@ impl DefaultBucketUsecase {
 
         let targets_guard = lock_bucket_targets_metadata(&bucket).await;
         validate_bucket_replication_update(&bucket, &replication_configuration).await?;
-        let existing_config = match metadata_sys::get_replication_config(&bucket).await {
-            Ok((config, _)) => Some(config),
-            Err(StorageError::ConfigNotFound) => None,
-            Err(err) => return Err(ApiError::from(err).into()),
-        };
-        let replication_configuration =
-            merge_user_replication_config_update(replication_configuration, existing_config, &site_peers);
         let data = serialize_config(&replication_configuration)?;
         update_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, data, expected_incarnation_id)
             .await
             .map_err(ApiError::from)?;
         drop(targets_guard);
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config");
         item.replication_config = Some(
@@ -2601,7 +2535,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false).await;
 
         Ok(S3Response::new(PutPublicAccessBlockOutput::default()))
     }
@@ -2630,7 +2564,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "tags");
         item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
@@ -2663,7 +2597,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false);
+        notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false).await;
 
         let mut item = sr_bucket_meta_item(bucket.clone(), "version-config");
         item.versioning = Some(
@@ -3114,7 +3048,7 @@ mod tests {
                 "{method} should identify the bucket metadata operation in reload logs"
             );
             let expected_reload = format!(
-                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change});"
+                "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change}).await;"
             );
             assert!(
                 body.contains(&expected_reload),
@@ -3184,185 +3118,6 @@ mod tests {
         assert!(arns.contains(destination));
     }
 
-    fn replication_rule_with_id(arn: &str, id: &str, priority: i32) -> ReplicationRule {
-        let mut rule = replication_rule_for_target(arn);
-        rule.id = Some(id.to_string());
-        rule.priority = Some(priority);
-        rule
-    }
-
-    fn site_peers(deployment_ids: &[&str]) -> HashSet<String> {
-        deployment_ids.iter().map(|id| id.to_string()).collect()
-    }
-
-    #[test]
-    fn put_replication_merge_preserves_site_replication_rules() {
-        let existing = ReplicationConfiguration {
-            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 {
-            role: String::new(),
-            rules: vec![
-                replication_rule_with_id("arn:rustfs:replication:us-east-1:new:bucket", "new-user-rule", 1),
-                replication_rule_with_id("arn:rustfs:replication::forged-dep:bucket", "site-repl-peer-dep", 2),
-                replication_rule_with_id("arn:rustfs:replication::other-dep:bucket", "site-repl-other", 3),
-            ],
-        };
-
-        let merged = merge_user_replication_config_update(incoming, Some(existing), &site_peers(&["peer-dep"]));
-
-        let rules: Vec<_> = merged
-            .rules
-            .iter()
-            .map(|rule| (rule.id.as_deref().unwrap_or_default(), rule.destination.bucket.as_str()))
-            .collect();
-        assert_eq!(
-            rules,
-            vec![
-                ("new-user-rule", "arn:rustfs:replication:us-east-1:new:bucket"),
-                ("site-repl-other", "arn:rustfs:replication::other-dep:bucket"),
-                ("site-repl-peer-dep", "arn:rustfs:replication::peer-dep:bucket"),
-            ],
-            "user rules replaced, the reconciler rule for the current peer kept over the incoming impostor, \
-             a site-repl-* id that names no current peer is ordinary operator state"
-        );
-    }
-
-    // Rule ids do not reserve `site-repl-*`: outside site replication an
-    // owner's `site-repl-user` rule is ordinary state, so PUT stores it
-    // verbatim and DELETE removes it and garbage-collects its target.
-    #[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 {
-            role: String::new(),
-            rules: vec![replication_rule_with_id(user_arn, "site-repl-user", 1)],
-        };
-
-        let stored = merge_user_replication_config_update(incoming.clone(), None, &HashSet::new());
-        assert_eq!(stored, incoming, "PUT on a non-site-replication bucket is verbatim");
-
-        let (remaining, removable) = split_replication_config_for_user_delete(stored, &HashSet::new());
-        assert!(remaining.is_none(), "DELETE must remove the operator's site-repl-* rule");
-        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
-    }
-
-    // Under site replication only a rule the reconciler would derive — id
-    // `site-repl-<peer>` for a current peer, destination ARN naming the same
-    // peer — is reconciler-owned. Everything else is operator state.
-    #[test]
-    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 {
-            role: String::new(),
-            rules: vec![
-                replication_rule_with_id(user_arn, "site-repl-user", 1),
-                replication_rule_with_id(user_arn, "site-repl-peer-dep", 2),
-                replication_rule_with_id("arn:rustfs:replication::gone-dep:bucket", "site-repl-gone-dep", 3),
-                replication_rule_with_id(peer_arn, "site-repl-peer-dep", 4),
-            ],
-        };
-
-        let (remaining, removable) = split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]));
-
-        let remaining = remaining.expect("the reconciler-derived rule must survive");
-        assert_eq!(remaining.rules.len(), 1);
-        assert_eq!(remaining.rules[0].destination.bucket, peer_arn);
-        assert_eq!(
-            removable,
-            HashSet::from([user_arn.to_string(), "arn:rustfs:replication::gone-dep:bucket".to_string()]),
-            "targets of operator rules and of a removed peer are garbage-collected"
-        );
-    }
-
-    #[test]
-    fn put_replication_merge_returns_incoming_verbatim_without_site_rules() {
-        let existing = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![replication_rule_with_id(
-                "arn:rustfs:replication:us-east-1:old:bucket",
-                "old-user-rule",
-                7,
-            )],
-        };
-        let incoming = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![replication_rule_with_id(
-                "arn:rustfs:replication:us-east-1:new:bucket",
-                "new-user-rule",
-                5,
-            )],
-        };
-
-        let merged = merge_user_replication_config_update(incoming.clone(), Some(existing), &HashSet::new());
-
-        assert_eq!(merged.role, incoming.role);
-        assert_eq!(merged.rules, incoming.rules, "non-SR buckets keep the verbatim overwrite semantics");
-    }
-
-    #[test]
-    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 {
-            role: String::new(),
-            rules: vec![
-                replication_rule_with_id(user_arn, "user-rule", 1),
-                replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2),
-            ],
-        };
-
-        let (remaining, removable) = split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]));
-
-        let remaining = remaining.expect("site-replication rules must survive a user delete");
-        let ids: Vec<_> = remaining
-            .rules
-            .iter()
-            .map(|rule| rule.id.as_deref().unwrap_or_default())
-            .collect();
-        assert_eq!(ids, vec!["site-repl-peer-dep"]);
-        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
-    }
-
-    #[test]
-    fn delete_replication_split_protects_targets_shared_with_site_rules() {
-        let sr_arn = "arn:rustfs:replication::peer-dep:bucket";
-        let config = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![
-                replication_rule_with_id(sr_arn, "user-rule-on-sr-target", 1),
-                replication_rule_with_id(sr_arn, "site-repl-peer-dep", 2),
-            ],
-        };
-
-        let (remaining, removable) = split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]));
-
-        assert!(remaining.is_some());
-        assert!(
-            removable.is_empty(),
-            "a target still referenced by a surviving site-replication rule must not be removed"
-        );
-    }
-
-    #[test]
-    fn delete_replication_split_removes_everything_without_site_rules() {
-        let user_arn = "arn:rustfs:replication:us-east-1:user:bucket";
-        let config = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![replication_rule_with_id(user_arn, "user-rule", 1)],
-        };
-
-        let (remaining, removable) = split_replication_config_for_user_delete(config, &site_peers(&["peer-dep"]));
-
-        assert!(remaining.is_none(), "without site-replication rules the whole config is deleted");
-        assert_eq!(removable, HashSet::from([user_arn.to_string()]));
-    }
-
     fn replication_targets_with_arn(arns: &[&str]) -> BucketTargets {
         BucketTargets {
             targets: arns
@@ -3700,10 +3455,7 @@ mod tests {
         let req = build_request(input, Method::DELETE);
         let usecase = DefaultBucketUsecase::without_context();
 
-        let err = usecase
-            .execute_delete_bucket_replication(req, HashSet::new())
-            .await
-            .unwrap_err();
+        let err = usecase.execute_delete_bucket_replication(req).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
@@ -4789,7 +4541,7 @@ mod tests {
         let req = build_request(input, Method::PUT);
         let usecase = DefaultBucketUsecase::without_context();
 
-        let err = usecase.execute_put_bucket_replication(req, HashSet::new()).await.unwrap_err();
+        let err = usecase.execute_put_bucket_replication(req).await.unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InternalError);
     }
 
@@ -4807,7 +4559,7 @@ mod tests {
             .unwrap();
 
         let err = DefaultBucketUsecase::without_context()
-            .execute_put_bucket_replication(build_request(input, Method::PUT), HashSet::new())
+            .execute_put_bucket_replication(build_request(input, Method::PUT))
             .await
             .expect_err("unsupported fields must be rejected before store access");
 

rustfs/src/app/storage_api.rs

@@ -619,8 +619,6 @@ pub(crate) mod bucket {
 
         use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts;
 
-        pub(crate) use replication_contracts::merge_user_replication_config;
-
         type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
         pub(crate) type DeleteReplicationConfigSnapshot =
             crate::storage::storage_api::ecstore_bucket::replication::DeleteReplicationConfigSnapshot;

rustfs/src/storage/ecfs.rs

@@ -63,69 +63,6 @@ use crate::app::storage_api::object_usecase::bucket::replication::{
 };
 use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;
 
-#[cfg(test)]
-static SITE_REPLICATION_GATE_TEST_OVERRIDE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
-#[cfg(test)]
-const SITE_REPLICATION_GATE_FORCE_DISABLED: u8 = 1;
-#[cfg(test)]
-const SITE_REPLICATION_GATE_FORCE_ENABLED: u8 = 2;
-
-async fn site_replication_gate_enabled() -> S3Result<bool> {
-    #[cfg(test)]
-    match SITE_REPLICATION_GATE_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst) {
-        SITE_REPLICATION_GATE_FORCE_DISABLED => return Ok(false),
-        SITE_REPLICATION_GATE_FORCE_ENABLED => return Ok(true),
-        _ => {}
-    }
-    crate::admin::handlers::site_replication::site_replication_enabled().await
-}
-
-/// Remote site-replication peer deployment ids handed to the bucket usecase
-/// so an S3 replication-config edit keeps exactly the reconciler-owned rules
-/// (issue #1948). Read here, in the interface layer, because the usecase must
-/// not import the admin handlers (layer guard); a state-read failure
-/// propagates so the edit fails closed.
-async fn site_replication_peer_deployment_ids_for_edit() -> S3Result<std::collections::HashSet<String>> {
-    // While the gate override is in effect the test exercises the deny/allow
-    // branch, not the peer set; there is no persisted state to read.
-    #[cfg(test)]
-    if SITE_REPLICATION_GATE_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst) != 0 {
-        return Ok(std::collections::HashSet::new());
-    }
-    crate::admin::handlers::site_replication::site_replication_remote_peer_deployment_ids().await
-}
-
-/// MinIO `ErrReplicationDenyEditError`.
-fn replication_deny_edit_error() -> S3Error {
-    let mut err = S3Error::with_message(
-        S3ErrorCode::Custom("XMinioReplicationDenyEdit".into()),
-        "Sub-User is not allowed to edit Replication configuration",
-    );
-    err.set_status_code(StatusCode::BAD_REQUEST);
-    err
-}
-
-/// Site-replication gate for S3 replication-config edits (issue #1948).
-///
-/// On a site-replication deployment the bucket's replication config carries
-/// the operator-managed `site-repl-*` rules that keep every peer in sync, and
-/// a successful edit is broadcast to all peers — so a user holding only
-/// bucket-scoped `s3:PutReplicationConfiguration` could rewrite or erase
-/// replication net-wide. MinIO parity (`ErrReplicationDenyEditError`): only
-/// owner credentials (root or root-parented) may edit. Runs after the policy
-/// authorization in the access layer and only on the external S3 path — the
-/// reconciler and peer bucket-meta ingestion never route through these
-/// handlers.
-async fn deny_replication_config_edit_for_non_owner<T>(req: &S3Request<T>) -> S3Result<()> {
-    if crate::storage::access::req_info_ref(req)?.is_owner {
-        return Ok(());
-    }
-    if site_replication_gate_enabled().await? {
-        return Err(replication_deny_edit_error());
-    }
-    Ok(())
-}
-
 #[derive(Debug, Clone)]
 pub struct FS {
     /// This server's late-bound application-context slot (backlog#1052 S2).
@@ -563,10 +500,8 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
-        deny_replication_config_edit_for_non_owner(&req).await?;
-        let site_peers = site_replication_peer_deployment_ids_for_edit().await?;
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_replication(req, site_peers).await
+        usecase.execute_delete_bucket_replication(req).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -1418,10 +1353,8 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
-        deny_replication_config_edit_for_non_owner(&req).await?;
-        let site_peers = site_replication_peer_deployment_ids_for_edit().await?;
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_replication(req, site_peers).await
+        usecase.execute_put_bucket_replication(req).await
     }
 
     async fn put_bucket_request_payment(
@@ -1986,103 +1919,3 @@ impl S3 for FS {
         Box::pin(usecase.execute_upload_part_copy(req)).await
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::{
-        FS, SITE_REPLICATION_GATE_FORCE_DISABLED, SITE_REPLICATION_GATE_FORCE_ENABLED, SITE_REPLICATION_GATE_TEST_OVERRIDE,
-    };
-    use crate::storage::access::ReqInfo;
-    use http::Method;
-    use http::StatusCode;
-    use s3s::dto::{DeleteBucketReplicationInput, PutBucketReplicationInput, ReplicationConfiguration};
-    use s3s::{S3, S3Error, S3ErrorCode, S3Request};
-    use std::sync::atomic::Ordering;
-
-    fn replication_config_edit_request<T>(input: T, is_owner: bool) -> S3Request<T> {
-        let mut req = S3Request {
-            input,
-            method: Method::PUT,
-            uri: http::Uri::from_static("/"),
-            headers: http::HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-        req.extensions.insert(ReqInfo {
-            is_owner,
-            ..Default::default()
-        });
-        req
-    }
-
-    fn put_bucket_replication_input() -> PutBucketReplicationInput {
-        PutBucketReplicationInput {
-            bucket: "test-bucket".to_string(),
-            checksum_algorithm: None,
-            content_md5: None,
-            expected_bucket_owner: None,
-            replication_configuration: ReplicationConfiguration {
-                role: String::new(),
-                rules: Vec::new(),
-            },
-            token: None,
-        }
-    }
-
-    fn delete_bucket_replication_input() -> DeleteBucketReplicationInput {
-        DeleteBucketReplicationInput {
-            bucket: "test-bucket".to_string(),
-            expected_bucket_owner: None,
-        }
-    }
-
-    fn assert_replication_deny_edit(err: &S3Error) {
-        match err.code() {
-            S3ErrorCode::Custom(code) => assert_eq!(code, "XMinioReplicationDenyEdit"),
-            other => panic!("expected XMinioReplicationDenyEdit, got {other:?}"),
-        }
-        assert_eq!(err.status_code(), Some(StatusCode::BAD_REQUEST));
-    }
-
-    /// Single test on purpose: the branches share the process-wide gate
-    /// override, and parallel tests would race it.
-    #[tokio::test]
-    async fn replication_config_edit_gate_denies_only_non_owner_under_site_replication() {
-        let fs = FS::new();
-        SITE_REPLICATION_GATE_TEST_OVERRIDE.store(SITE_REPLICATION_GATE_FORCE_ENABLED, Ordering::SeqCst);
-
-        // Non-owner PUT/DELETE through the real S3 handlers: denied by the
-        // gate before the usecase (and thus the store) is ever touched.
-        let err = fs
-            .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), false))
-            .await
-            .expect_err("non-owner PutBucketReplication must be denied while site replication is enabled");
-        assert_replication_deny_edit(&err);
-        let err = fs
-            .delete_bucket_replication(replication_config_edit_request(delete_bucket_replication_input(), false))
-            .await
-            .expect_err("non-owner DeleteBucketReplication must be denied while site replication is enabled");
-        assert_replication_deny_edit(&err);
-
-        // Owner passes the gate (the usecase's empty-rules structure error
-        // proves the request reached the usecase instead of the deny path).
-        let err = fs
-            .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), true))
-            .await
-            .expect_err("owner request should pass the gate and fail later on config validation");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-
-        // Without site replication the policy check alone still governs the edit.
-        SITE_REPLICATION_GATE_TEST_OVERRIDE.store(SITE_REPLICATION_GATE_FORCE_DISABLED, Ordering::SeqCst);
-        let err = fs
-            .put_bucket_replication(replication_config_edit_request(put_bucket_replication_input(), false))
-            .await
-            .expect_err("non-owner request should pass the gate and fail later on config validation");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-
-        SITE_REPLICATION_GATE_TEST_OVERRIDE.store(0, Ordering::SeqCst);
-    }
-}