fix(replication): keep operator rule priorities across site rule merges

Merging stored site-replication rules into a PutBucketReplication body
renumbered every rule 1..n in list order, rewriting the submitted policy:
overlapping same-target rules submitted as priority 5 then 1 became 1
then 2, so the delete-marker-disabled rule won the replication decision.
The reconciler and the peer-removal prune renumbered the same way.

Operator priorities now stay verbatim everywhere; only the reconciler's
derived rules move, to the lowest priorities no operator rule uses, via
one pure helper shared by the S3 edit merge, the peer ingestion merge,
the reconciler pass and the prune. Being a pure function of the rule
list it is idempotent, so the reconciler's no-op check still holds after
a merged write, and an on-disk config in the historical layout (operator
rules 1..k, site rules k+1..n) yields the same bytes, so nothing is
rewritten on upgrade.

crates/ecstore/src/api/mod.rs

@@ -196,15 +196,16 @@ pub mod bucket {
             ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
             ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
             ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            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,
+            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,
         };
     }
 

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

@@ -47,8 +47,10 @@ 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,
-    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,
+    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,
 };
 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,6 +16,8 @@ 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,
-    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,
+    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,
 };

crates/replication/src/config.rs

@@ -265,6 +265,160 @@ 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() {
@@ -1539,4 +1693,103 @@ 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,9 +32,11 @@ 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, 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,
+    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,
 };
 pub use delete::{
     DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,

rustfs/src/admin/handlers/site_replication.rs

@@ -31,6 +31,10 @@ 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};
@@ -137,11 +141,6 @@ const SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE: usize = 100;
 const SITE_REPL_RESYNC_MAX_PAGE_SIZE: usize = 1000;
 const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
 const SITE_REPLICATION_PEER_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
-/// Bound on waiting for the lifecycle lock (below). 3x the peer request
-/// timeout: outlives one full peer round of a healthy concurrent lifecycle
-/// operation, while converting a holder wedged on unreachable peers into a
-/// retryable 503 for the waiter instead of an unbounded hang.
-const SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
 const SITE_REPLICATION_PEER_ERROR_DETAIL_LIMIT: usize = 256;
 const SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT: usize = 32;
 const MAX_PEER_CA_CERT_PEM_SIZE: usize = 256 * 1024;
@@ -392,17 +391,9 @@ struct SiteReplicationLifecycleGuard {
 }
 
 impl SiteReplicationLifecycleGuard {
-    /// Bounded acquire: a holder wedged on unreachable peers (each probe
-    /// costs up to [`SITE_REPLICATION_PEER_REQUEST_TIMEOUT`]) must not hang
-    /// every other lifecycle operation indefinitely, so waiters get a
-    /// retryable 503 after [`SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT`].
-    async fn acquire() -> S3Result<Self> {
-        match tokio::time::timeout(SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT, SITE_REPLICATION_LIFECYCLE_LOCK.lock()).await {
-            Ok(guard) => Ok(Self { _guard: guard }),
-            Err(_) => Err(S3Error::with_message(
-                S3ErrorCode::ServiceUnavailable,
-                "another site replication lifecycle operation is in progress; retry later".to_string(),
-            )),
+    async fn acquire() -> Self {
+        Self {
+            _guard: SITE_REPLICATION_LIFECYCLE_LOCK.lock().await,
         }
     }
 
@@ -1130,6 +1121,36 @@ 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),
@@ -2128,27 +2149,6 @@ async fn remote_add_preflight_info(site: &PeerSite) -> S3Result<SiteReplicationA
     add_preflight_info_from_sr_info(site, info, idp_settings)
 }
 
-/// Preflight every site in an add request while the lifecycle lock is held.
-/// Probes run concurrently (matching the other peer fan-outs in this file):
-/// k unreachable sites cost roughly one peer request timeout, not k of them.
-/// Results (and the first error, if any) are reported in request order.
-async fn add_preflight_infos(
-    sites: &[PeerSite],
-    current_state: &SiteReplicationState,
-    local_peer: &PeerInfo,
-) -> S3Result<Vec<SiteReplicationAddPreflightInfo>> {
-    futures::future::join_all(sites.iter().map(|site| async move {
-        if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) {
-            local_add_preflight_info(current_state, local_peer, site).await
-        } else {
-            remote_add_preflight_info(site).await
-        }
-    }))
-    .await
-    .into_iter()
-    .collect()
-}
-
 fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], local_peer: &PeerInfo) -> S3Result<()> {
     let mut deployment_ids = HashSet::new();
     let mut local_seen = false;
@@ -7782,20 +7782,6 @@ 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,10 +7806,6 @@ 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
@@ -7849,52 +7831,6 @@ 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
@@ -8233,9 +8169,7 @@ fn prune_removed_site_replication_rules(
         return (None, removed);
     }
 
-    for (index, rule) in config.rules.iter_mut().enumerate() {
-        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
-    }
+    assign_site_replication_rule_priorities(&mut config.rules, is_site_replication_rule);
 
     (Some(config), removed)
 }
@@ -8409,9 +8343,10 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime(
         .cloned()
         .collect();
     rules.extend(desired.rules);
-    for (index, rule) in rules.iter_mut().enumerate() {
-        rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX));
-    }
+    // 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);
 
     // 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
@@ -9892,7 +9827,7 @@ impl Operation for SiteReplicationAddHandler {
         let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationAddAction).await?;
         reject_site_replicator_on_public_admin(&cred)?;
         let replicate_ilm_expiry = sr_add_replicate_ilm_expiry(&req.uri);
-        let lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+        let lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
         // Everything up to the commit below is preflight: peer probes, IAM
         // work and the join fan-out all talk to the network, so none of it may
         // run inside the state transaction. The snapshot read here is what the
@@ -9907,7 +9842,14 @@ impl Operation for SiteReplicationAddHandler {
         // inject it so the add preflight (which requires the local deployment) succeeds. No-op for `mc`.
         ensure_local_site_present(&mut sites, &local_peer);
         validate_add_sites(&sites, &local_peer)?;
-        let preflight_infos = add_preflight_infos(&sites, &current_state, &local_peer).await?;
+        let mut preflight_infos = Vec::with_capacity(sites.len());
+        for site in &sites {
+            if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) {
+                preflight_infos.push(local_add_preflight_info(&current_state, &local_peer, site).await?);
+            } else {
+                preflight_infos.push(remote_add_preflight_info(site).await?);
+            }
+        }
         validate_add_preflight_topology(&preflight_infos, &local_peer)?;
         let expected_updated_at = current_state.updated_at;
         require_add_peer_tls_capability(&sites, &local_peer).await?;
@@ -10115,7 +10057,7 @@ impl Operation for SiteReplicationRemoveHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationRemoveAction).await?;
         reject_site_replicator_on_public_admin(&cred)?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
         // The request body is read before the bucket-op guard and the state
         // transaction: a client that stalls mid-body must hold neither the
         // state-object lock nor the write half of the bucket-op RwLock (which
@@ -10313,7 +10255,7 @@ where
     F: FnOnce(SRPeerJoinReq) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = S3Result<()>> + Send + 'static,
 {
-    let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+    let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
     admit_peer_join_across_nodes(local_endpoint, join_req, defer_sync_state_enable, apply_iam).await
 }
 
@@ -11128,7 +11070,7 @@ impl Operation for SRPeerRemoveHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         validate_site_replication_admin_request(&req, AdminAction::SiteReplicationRemoveAction).await?;
         let remove_req: SRRemoveReq = read_site_replication_json(req, "", false).await?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
         let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
         let removed_deployment_ids = update_site_replication_state(move |state| {
             if pending_endpoint_refresh(state).is_some() {
@@ -11172,7 +11114,7 @@ impl Operation for SiteReplicationResyncOpHandler {
         let operation = query.get("operation").cloned().unwrap_or_default();
         let resolved_store = object_store_from_req(&req);
         let requested_peer: PeerInfo = read_site_replication_json(req, "", false).await?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
         let (peer, existing_status) = {
             let state = load_site_replication_state().await?;
             let local_peer = current_local_runtime_peer(&state);
@@ -11464,7 +11406,7 @@ impl Operation for SRRotateServiceAccountHandler {
         // mid-repair and race its own IAM write against the reconciler's
         // stale one. (The removed process mutex used to provide this
         // exclusion as a side effect.)
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
         let local_endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
         let rotation_parent = cred.access_key.clone();
         let (pending_rotation, local_peer, previous_access_key) = update_site_replication_state_when_changed(move |state| {
@@ -13957,9 +13899,7 @@ mod tests {
     async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() {
         let token;
         {
-            let lifecycle = SiteReplicationLifecycleGuard::acquire()
-                .await
-                .expect("acquire lifecycle guard");
+            let lifecycle = SiteReplicationLifecycleGuard::acquire().await;
             let guard = SiteReplicationAddInProgressGuard::start(lifecycle, HashSet::from(["legacy-bucket".to_string()]))
                 .expect("start site replication add guard");
             token = guard.token.to_string();
@@ -14015,18 +13955,14 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn test_add_lifecycle_allows_callback_before_remove_writer() {
-        let lifecycle = SiteReplicationLifecycleGuard::acquire()
-            .await
-            .expect("acquire lifecycle guard");
+        let lifecycle = SiteReplicationLifecycleGuard::acquire().await;
         let add_guard =
             SiteReplicationAddInProgressGuard::start(lifecycle, HashSet::new()).expect("start site replication add guard");
         let (started_tx, started_rx) = tokio::sync::oneshot::channel();
         let (entered_tx, mut entered_rx) = tokio::sync::oneshot::channel();
         let remove = tokio::spawn(async move {
             let _ = started_tx.send(());
-            let _lifecycle = SiteReplicationLifecycleGuard::acquire()
-                .await
-                .expect("acquire lifecycle guard");
+            let _lifecycle = SiteReplicationLifecycleGuard::acquire().await;
             let _bucket_op = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
             let _ = entered_tx.send(());
         });
@@ -14046,111 +13982,6 @@ mod tests {
         entered_rx.await.expect("remove entered lifecycle");
     }
 
-    /// Deleting either constant (or "simplifying" the client builders to
-    /// inline values) removes the only bound on how long a lifecycle
-    /// operation can be wedged per unreachable peer (#1889 C1 / #1952 C2).
-    #[test]
-    fn test_peer_timeout_constants_bound_unreachable_peer_probes() {
-        assert_eq!(SITE_REPLICATION_PEER_REQUEST_TIMEOUT, Duration::from_secs(10));
-        assert_eq!(SITE_REPLICATION_PEER_CONNECT_TIMEOUT, Duration::from_secs(3));
-        assert!(
-            SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT >= SITE_REPLICATION_PEER_REQUEST_TIMEOUT,
-            "a waiter must not give up before the holder's single wedged peer probe can finish"
-        );
-    }
-
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_lifecycle_guard_acquire_times_out_with_retryable_503() {
-        let holder = SiteReplicationLifecycleGuard::acquire().await.expect("first acquire");
-        let err =
-            match tokio::time::timeout(SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT * 2, SiteReplicationLifecycleGuard::acquire())
-                .await
-                .expect("bounded acquire must not hang while the lock is held")
-            {
-                Ok(_) => panic!("acquire while the lock is held should time out"),
-                Err(err) => err,
-            };
-        assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
-
-        drop(holder);
-        tokio::time::timeout(Duration::from_secs(1), SiteReplicationLifecycleGuard::acquire())
-            .await
-            .expect("acquire after release must not wait")
-            .expect("acquire after release");
-    }
-
-    #[derive(Clone)]
-    struct PreflightFanoutTestState {
-        metainfo_barrier: Arc<tokio::sync::Barrier>,
-    }
-
-    async fn preflight_fanout_test_handler(State(state): State<PreflightFanoutTestState>, uri: Uri) -> (StatusCode, String) {
-        if uri.path().ends_with("/site-replication/metainfo") {
-            state.metainfo_barrier.wait().await;
-        }
-        (StatusCode::OK, "{}".to_string())
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn test_add_preflight_probes_sites_concurrently() {
-        temp_env::async_with_vars(
-            [(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))],
-            add_preflight_probes_sites_concurrently_inner(),
-        )
-        .await;
-    }
-
-    async fn add_preflight_probes_sites_concurrently_inner() {
-        const REMOTE_SITES: usize = 3;
-        let listener = match TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
-            Err(err) => panic!("bind preflight test server: {err}"),
-        };
-        let endpoint = format!("http://{}", listener.local_addr().expect("preflight test address"));
-        let state = PreflightFanoutTestState {
-            metainfo_barrier: Arc::new(tokio::sync::Barrier::new(REMOTE_SITES)),
-        };
-        let server = tokio::spawn(async move {
-            axum::serve(listener, Router::new().fallback(any(preflight_fanout_test_handler)).with_state(state))
-                .await
-                .expect("serve preflight test requests");
-        });
-
-        let sites: Vec<PeerSite> = (0..REMOTE_SITES)
-            .map(|index| PeerSite {
-                name: format!("site-{index}"),
-                endpoint: endpoint.clone(),
-                access_key: "test-access".to_string(),
-                secret_key: "test-secret".to_string(),
-                ..Default::default()
-            })
-            .collect();
-        let local_peer = PeerInfo {
-            deployment_id: "local".to_string(),
-            endpoint: "http://192.0.2.1:9000".to_string(),
-            ..Default::default()
-        };
-        let current_state = SiteReplicationState::default();
-
-        // Each site's metainfo request parks on a barrier that only releases
-        // once every site's request has arrived: serial probing never sends
-        // the second request and dies on the peer request timeout, so
-        // finishing well inside that timeout proves the probes overlap —
-        // which is what caps k unreachable sites at one timeout, not k.
-        let infos = tokio::time::timeout(
-            SITE_REPLICATION_PEER_REQUEST_TIMEOUT / 2,
-            add_preflight_infos(&sites, &current_state, &local_peer),
-        )
-        .await
-        .expect("preflight probes must fan out concurrently, not serially")
-        .expect("preflight infos");
-        assert_eq!(infos.len(), REMOTE_SITES);
-        server.abort();
-    }
-
     #[test]
     fn test_merge_add_sites_propagates_replicate_ilm_expiry() {
         let state = merge_add_sites(
@@ -17138,7 +16969,7 @@ mod tests {
     }
 
     #[test]
-    fn test_prune_removed_site_replication_rules_removes_site_rule_and_reorders_priorities() {
+    fn test_prune_removed_site_replication_rules_removes_site_rule_and_keeps_operator_priority() {
         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");
@@ -17155,9 +16986,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(1));
+        assert_eq!(updated.rules[0].priority, Some(9), "the operator's priority is policy and stays");
         assert_eq!(updated.rules[1].id.as_deref(), Some("site-repl-kept-dep"));
-        assert_eq!(updated.rules[1].priority, Some(2));
+        assert_eq!(updated.rules[1].priority, Some(1), "the derived rule moves to the lowest free slot");
     }
 
     #[test]

rustfs/src/admin/storage_api.rs

@@ -443,6 +443,8 @@ 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, 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, merge_user_replication_config,
+        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,
@@ -66,6 +66,7 @@ use super::storage_api::bucket_usecase::{
 };
 use crate::admin::handlers::site_replication::{
     site_replication_bucket_meta_hook, site_replication_delete_bucket_hook, site_replication_make_bucket_hook,
+    site_replication_remote_peer_deployment_ids,
 };
 use crate::app::object_data_cache::invalidate_object_data_cache_bucket_after_delete;
 use crate::app::runtime_sources::{
@@ -623,11 +624,52 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo
     validate_replication_config_targets(&targets, config)
 }
 
-async fn replication_targets_without_config_targets(
+/// 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(
     bucket: &str,
-    config: &ReplicationConfiguration,
+    target_arns: &HashSet<String>,
 ) -> S3Result<Option<(BucketTargets, usize)>> {
-    let target_arns = replication_target_arns(config);
     if target_arns.is_empty() {
         return Ok(None);
     }
@@ -638,7 +680,7 @@ async fn replication_targets_without_config_targets(
         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);
     }
@@ -1604,15 +1646,30 @@ impl DefaultBucketUsecase {
             Err(StorageError::ConfigNotFound) => None,
             Err(err) => return Err(ApiError::from(err).into()),
         };
-        let updated_targets = if let Some(config) = replication_config.as_ref() {
-            replication_targets_without_config_targets(&bucket, config).await?
+        let (remaining_config, updated_targets) = if let Some(config) = replication_config.as_ref() {
+            let site_peers = site_replication_remote_peer_deployment_ids().await?;
+            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)
         } else {
-            None
+            (None, None)
         };
 
-        delete_bucket_config_for_incarnation(&bucket, BUCKET_REPLICATION_CONFIG, expected_incarnation_id)
-            .await
-            .map_err(ApiError::from)?;
+        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)?;
+            }
+        }
         if let Some((targets, removed)) = updated_targets
             && let Err(err) =
                 write_replication_targets_after_config_delete(&bucket, &targets, removed, expected_incarnation_id).await
@@ -2485,6 +2542,14 @@ 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 site_peers = site_replication_remote_peer_deployment_ids().await?;
+        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
@@ -3114,6 +3179,185 @@ 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

rustfs/src/app/storage_api.rs

@@ -614,6 +614,8 @@ 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,6 +63,54 @@ 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
+}
+
+/// 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).
@@ -500,6 +548,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
         let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_delete_bucket_replication(req).await
     }
@@ -1353,6 +1402,7 @@ impl S3 for FS {
         &self,
         req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
         let usecase = s3_api::bucket_usecase_for(self);
         usecase.execute_put_bucket_replication(req).await
     }
@@ -1919,3 +1969,103 @@ 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);
+    }
+}