fix(replication): skip site-replication rules in replica delete fan-out (#9138)

* fix(replication): skip site-replication rules in replica delete fan-out

A site that receives a replicated delete forwards it to its own targets
so chained replication keeps propagating deletes. The fan-out used the
whole bucket config, including derived site-replication rules, so in a
site-replication mesh a replicated purge was echoed back to its origin
and duplicated to every other peer. The echo could remove the origin's
pending-purge version before its own delivery state was written back.

Ignore derived site-replication rules when computing the replica delete
fan-out. The origin already delivers and retries every delete to each
peer itself. Operator rules still fan out, so chained replication is
unchanged.

* fix(replication): keep operator rules matching the site-repl shape

Rule ids and target ARN shapes are not reserved for site replication:
an ordinary bucket target ARN carries its target id, so an operator rule
named site-repl-<target id> matched the derived shape and was dropped
from replica delete fan-out, breaking an A -> B -> C chain.

Skip a rule only when its deployment id is a current remote
site-replication peer, reading the peer set only when a rule has the
derived shape. If the state cannot be read, skip nothing.

rustfs/src/app/object/delete.rs

@@ -295,16 +295,32 @@ fn enrich_delete_replication_state_if_needed(
     }
 }
 
-fn should_schedule_replica_delete_replication(
+/// The current peer set is read only when a rule has the derived shape. If it
+/// cannot be read, no rule is skipped: a redundant echo is safer than a chained
+/// target missing the delete.
+async fn replica_delete_fanout_state(
     snapshot: &DeleteReplicationConfigSnapshot,
+    bucket: &str,
     replication_source: &ObjectInfo,
     version_id: Option<Uuid>,
-) -> bool {
-    let Some(config) = snapshot.replication_config() else {
-        return false;
+) -> Option<ReplicationState> {
+    let config = snapshot.replication_config()?;
+    let site_peer_deployment_ids = if has_site_replication_shaped_rule(config) {
+        match crate::site_replication::site_replication_remote_peer_deployment_ids().await {
+            Ok(peers) => peers,
+            Err(error) => {
+                warn!(
+                    bucket = %bucket,
+                    error = %error,
+                    "failed to load site replication peers; forwarding replica delete along every rule"
+                );
+                HashSet::new()
+            }
+        }
+    } else {
+        HashSet::new()
     };
-
-    delete_replication_state_from_config(config, replication_source, version_id, true).is_some()
+    replica_delete_fanout_state_from_config(config, replication_source, version_id, &site_peer_deployment_ids)
 }
 
 fn validate_undo_delete_version(expected: Option<&str>, requested: Option<&str>) -> S3Result<()> {
@@ -1347,12 +1363,21 @@ impl DefaultObjectUsecase {
         let deleted_delete_marker_version = deleted_replication_info.is_some_and(|info| info.delete_marker);
 
         let delete_replication_version_id = delete_replication_version_id(deleted_object_source, deleted_delete_marker_version);
-        let schedule_delete_replication = if opts.replication_request && replica {
-            should_schedule_replica_delete_replication(
-                &delete_config_snapshot,
-                replication_state_source,
-                delete_replication_version_id,
+        let replica_fanout_state = if opts.replication_request && replica {
+            Some(
+                replica_delete_fanout_state(
+                    &delete_config_snapshot,
+                    &bucket,
+                    replication_state_source,
+                    delete_replication_version_id,
+                )
+                .await,
             )
+        } else {
+            None
+        };
+        let schedule_delete_replication = if let Some(fanout_state) = &replica_fanout_state {
+            fanout_state.is_some()
         } else {
             should_schedule_delete_replication(
                 &opts,
@@ -1381,8 +1406,12 @@ impl DefaultObjectUsecase {
                 replication_state: None,
                 ..Default::default()
             };
-            set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
-            enrich_delete_replication_state_if_needed(&delete_config_snapshot, &mut deleted_object, replication_state_source);
+            if let Some(Some(fanout_state)) = &replica_fanout_state {
+                set_deleted_object_replication_state(&mut deleted_object, fanout_state);
+            } else {
+                set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
+                enrich_delete_replication_state_if_needed(&delete_config_snapshot, &mut deleted_object, replication_state_source);
+            }
             schedule_replication_delete(deleted_object, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
         }
 
@@ -3583,6 +3612,83 @@ mod tests {
         assert!(state.purge_targets.contains_key(&arn));
     }
 
+    #[test]
+    fn replica_delete_fanout_skips_site_replication_peers_but_keeps_chain_targets() {
+        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
+        let chain_arn = "arn:aws:s3:::downstream-bucket".to_string();
+        let site_rule = crate::site_replication::build_site_replication_rule(peer_arn, 10, "site-repl-peer-dep");
+        let mut chain_rule = site_rule.clone();
+        chain_rule.id = Some("chain".to_string());
+        chain_rule.priority = Some(1);
+        chain_rule.destination.bucket = chain_arn.clone();
+        let site_only = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![site_rule.clone()],
+        };
+        let site_and_chain = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![site_rule, chain_rule],
+        };
+        let marker = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            delete_marker: true,
+            replication_status: ReplicationStatusType::Replica,
+            ..Default::default()
+        };
+        let version = ObjectInfo {
+            delete_marker: false,
+            ..marker.clone()
+        };
+        // An ordinary bucket target ARN carries its target id, so an operator
+        // rule named `site-repl-<target id>` has the derived shape.
+        let shaped_operator_arn = "arn:minio:replication:us-east-1:target-1:downstream".to_string();
+        let mut shaped_operator_rule =
+            crate::site_replication::build_site_replication_rule(&shaped_operator_arn, 1, "site-repl-target-1");
+        shaped_operator_rule.priority = Some(1);
+        let shaped_operator_only = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![shaped_operator_rule],
+        };
+        assert!(has_site_replication_shaped_rule(&shaped_operator_only));
+        let site_peers = HashSet::from(["peer-dep".to_string()]);
+        let no_site_peers = HashSet::new();
+        let removed_peer_only = HashSet::from(["other-dep".to_string()]);
+        let purged_version = Some(Uuid::new_v4());
+        let fanout_targets = |state: ReplicationState, version_id: Option<Uuid>| {
+            if version_id.is_some() {
+                state.purge_targets.keys().cloned().collect::<Vec<_>>()
+            } else {
+                state.targets.keys().cloned().collect::<Vec<_>>()
+            }
+        };
+
+        for (source, version_id) in [(&marker, None), (&marker, purged_version), (&version, purged_version)] {
+            assert!(
+                delete_replication_state_from_config(&site_only, source, version_id, true).is_some(),
+                "the site-replication rule alone would forward the replica delete"
+            );
+            assert!(
+                replica_delete_fanout_state_from_config(&site_only, source, version_id, &site_peers).is_none(),
+                "a replica delete must not echo back across the site-replication mesh"
+            );
+
+            let state = replica_delete_fanout_state_from_config(&site_and_chain, source, version_id, &site_peers)
+                .expect("a replica delete must still follow a chained operator rule");
+            assert!(!state.replicate_decision_str.contains(peer_arn));
+            assert_eq!(fanout_targets(state, version_id), vec![chain_arn.clone()]);
+
+            for peers in [&no_site_peers, &removed_peer_only, &site_peers] {
+                let state = replica_delete_fanout_state_from_config(&shaped_operator_only, source, version_id, peers)
+                    .expect("an operator rule naming no current site peer must keep forwarding replica deletes");
+                assert_eq!(fanout_targets(state, version_id), vec![shaped_operator_arn.clone()]);
+            }
+            let state = replica_delete_fanout_state_from_config(&site_only, source, version_id, &no_site_peers)
+                .expect("with site replication disabled a site-repl rule is an operator rule");
+            assert_eq!(fanout_targets(state, version_id), vec![peer_arn.to_string()]);
+        }
+    }
+
     #[test]
     fn replica_delete_enrichment_must_not_reuse_upstream_targets() {
         let upstream_state = ReplicationState {

rustfs/src/app/object/mod.rs

@@ -29,8 +29,9 @@ use super::storage_api::object_usecase::access::{
 #[cfg(test)]
 use super::storage_api::object_usecase::bucket::quota::BucketQuota;
 use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker;
+use super::storage_api::object_usecase::bucket::replication::ReplicationState;
 #[cfg(test)]
-use super::storage_api::object_usecase::bucket::replication::{ReplicationState, replication_statuses_map};
+use super::storage_api::object_usecase::bucket::replication::replication_statuses_map;
 use super::storage_api::object_usecase::bucket::{
     VersioningConfigExt as _,
     lifecycle::{
@@ -49,10 +50,11 @@ use super::storage_api::object_usecase::bucket::{
     replication::{
         DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent,
         delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
-        force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, load_delete_config_snapshot,
-        must_replicate_object, persist_force_delete_intent, record_replication_proxy, schedule_object_replication,
-        schedule_replication_delete, schedule_replication_deletes, set_deleted_object_replication_state,
-        should_schedule_delete_replication, should_use_existing_delete_replication_info,
+        force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, has_site_replication_shaped_rule,
+        load_delete_config_snapshot, must_replicate_object, persist_force_delete_intent, record_replication_proxy,
+        replica_delete_fanout_state_from_config, schedule_object_replication, schedule_replication_delete,
+        schedule_replication_deletes, set_deleted_object_replication_state, should_schedule_delete_replication,
+        should_use_existing_delete_replication_info,
     },
     tagging::decode_tags,
     validate_restore_request,

rustfs/src/app/storage_api.rs

@@ -708,7 +708,6 @@ pub(crate) mod bucket {
         pub(crate) type DeleteReplicationConfigSnapshot =
             crate::storage::storage_api::ecstore_bucket::replication::DeleteReplicationConfigSnapshot;
         pub(crate) type ReplicateDecision = replication_contracts::ReplicateDecision;
-        #[cfg(test)]
         pub(crate) type ReplicationState = replication_contracts::ReplicationState;
         pub(crate) type ReplicationStatusType = replication_contracts::ReplicationStatusType;
         pub(crate) type ReplicationTargetValidationError = replication_contracts::ReplicationTargetValidationError;
@@ -959,6 +958,52 @@ pub(crate) mod bucket {
             replication_contracts::delete_replication_state_from_config(config, &source)
         }
 
+        /// Replication state for forwarding a replica delete (an inbound
+        /// replication request) to this site's own targets.
+        ///
+        /// Operator rules with ReplicaModifications keep forwarding, so chained
+        /// replication A -> B -> C propagates deletes (rustfs#2609). Derived
+        /// site-replication rules never do: site replication is a full mesh
+        /// where the origin delivers, and MRF-retries, every delete to each
+        /// peer itself. Forwarding along a `site-repl-<deploymentId>` rule only
+        /// echoes the delete back to the origin, where it removes the origin's
+        /// pending-purge version before its own delivery state is written back,
+        /// and duplicates it to every other peer (rustfs#9121). MinIO never
+        /// re-replicates a replication request at all.
+        ///
+        /// Rule ids and target ARN shapes are not reserved: an ordinary bucket
+        /// target ARN also carries a target id, so an operator rule can match
+        /// the derived shape. Only a rule naming a current remote peer in
+        /// `site_peer_deployment_ids` is skipped; with site replication
+        /// disabled the set is empty and every rule forwards.
+        pub(crate) fn replica_delete_fanout_state_from_config(
+            config: &s3s::dto::ReplicationConfiguration,
+            obj_info: &crate::storage::storage_api::StorageObjectInfo,
+            version_id: Option<Uuid>,
+            site_peer_deployment_ids: &HashSet<String>,
+        ) -> Option<replication_contracts::ReplicationState> {
+            let is_site_peer_rule = |rule: &s3s::dto::ReplicationRule| {
+                replication_contracts::site_replication_rule_deployment_id(rule)
+                    .is_some_and(|deployment_id| site_peer_deployment_ids.contains(deployment_id))
+            };
+            if !config.rules.iter().any(is_site_peer_rule) {
+                return delete_replication_state_from_config(config, obj_info, version_id, true);
+            }
+
+            let mut operator_config = config.clone();
+            operator_config.rules.retain(|rule| !is_site_peer_rule(rule));
+            delete_replication_state_from_config(&operator_config, obj_info, version_id, true)
+        }
+
+        /// Whether `config` has a rule with the derived site-replication shape,
+        /// so the replica-delete fan-out needs the current peer set.
+        pub(crate) fn has_site_replication_shaped_rule(config: &s3s::dto::ReplicationConfiguration) -> bool {
+            config
+                .rules
+                .iter()
+                .any(|rule| replication_contracts::site_replication_rule_deployment_id(rule).is_some())
+        }
+
         pub(crate) fn replication_target_arns(config: &s3s::dto::ReplicationConfiguration) -> HashSet<String> {
             replication_contracts::replication_target_arns(config)
         }

rustfs/src/site_replication/state.rs

@@ -368,6 +368,18 @@ pub(crate) async fn site_replication_deployment_ids() -> S3Result<HashSet<String
     Ok(state.peers.values().map(|peer| peer.deployment_id.clone()).collect())
 }
 
+/// Deployment ids of the remote peers this site's reconciler derives a
+/// `site-repl-<id>` rule for; empty when site replication is not enabled.
+/// Read on the replica-delete path so only rules naming a current peer are
+/// treated as reconciler-owned.
+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());
+    }
+    Ok(remote_peer_deployment_ids(&state, &current_local_runtime_peer(&state)))
+}
+
 pub(crate) 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),