chore(gateway): merge main into integration

A replicated force-delete that a target rejected (for example a peer whose
credentials lack s3:ForceDeleteObject) was only logged; the per-target
replication counters never saw it, so `replication status` reported zero
failures while the sites had already diverged.

Build a ReplicatedTargetInfo for every target outcome of
replicate_force_delete_to_targets (missing client, offline target,
rejected remove, success) with a Pending previous status, and feed each
through ReplicationStats::update so rejections count as failures and
accepted force-deletes count as replicated.

Document that the managed site-replicator-0 policy must name
s3:ForceDeleteObject explicitly and that a bucket-replication target with
its own credentials needs the same grant.

.config/test-count-baseline.txt

@@ -10,7 +10,7 @@
 # intentionally removed. Regenerate with:
 #   scripts/check_test_count_ratchet.sh --update
 e2e_test	993	20
-rustfs	6085	64
+rustfs	6087	64
 rustfs-audit	70	0
 rustfs-checksums	26	0
 rustfs-common	36	0
@@ -20,7 +20,7 @@ rustfs-connect-symbol-catalog	6	0
 rustfs-credentials	24	0
 rustfs-crypto	67	0
 rustfs-data-usage	56	0
-rustfs-ecstore	7415	2
+rustfs-ecstore	7416	2
 rustfs-extension-schema	23	0
 rustfs-filemeta	353	0
 rustfs-heal	1033	3

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

@@ -3282,10 +3282,18 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
 
     let mut join_set = JoinSet::new();
     let mut all_succeeded = true;
+    let mut target_infos = Vec::with_capacity(tgt_arns.len());
 
     for arn in tgt_arns {
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else {
             all_succeeded = false;
+            target_infos.push(force_delete_target_info(
+                dobj.op_type,
+                &arn,
+                "",
+                false,
+                Err(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()),
+            ));
             debug!(
                 event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
@@ -3311,6 +3319,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
 
         let bucket = bucket.clone();
         let object_name = object_name.clone();
+        let op_type = dobj.op_type;
 
         join_set.spawn(async move {
             if ReplicationTargetStore::target_is_offline(&tgt_client).await {
@@ -3335,9 +3344,16 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
                     user_agent: "Internal: [Replication]".to_string(),
                     ..Default::default()
                 });
-                return false;
+                return force_delete_target_info(
+                    op_type,
+                    &tgt_client.arn,
+                    &tgt_client.endpoint,
+                    tgt_client.secure,
+                    Err(FORCE_DELETE_TARGET_OFFLINE_ERROR.to_string()),
+                );
             }
 
+            let started = Instant::now();
             if let Err(e) = tgt_client
                 .remove_object(&tgt_client.bucket, &object_name, None, replication_force_delete_remove_options())
                 .await
@@ -3364,16 +3380,25 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
                     user_agent: "Internal: [Replication]".to_string(),
                     ..Default::default()
                 });
-                return false;
+                return force_delete_target_info(
+                    op_type,
+                    &tgt_client.arn,
+                    &tgt_client.endpoint,
+                    tgt_client.secure,
+                    Err(e.to_string()),
+                );
             }
 
-            true
+            force_delete_target_info(op_type, &tgt_client.arn, &tgt_client.endpoint, tgt_client.secure, Ok(started.elapsed()))
         });
     }
 
     while let Some(result) = join_set.join_next().await {
         match result {
-            Ok(success) => all_succeeded &= success,
+            Ok(info) => {
+                all_succeeded &= info.replication_status == ReplicationStatusType::Completed;
+                target_infos.push(info);
+            }
             Err(error) => {
                 all_succeeded = false;
                 error!(
@@ -3390,6 +3415,8 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
         }
     }
 
+    record_force_delete_target_stats(bucket, &target_infos).await;
+
     if all_succeeded
         && let Some(operation_id) = dobj.delete_object.force_delete_id
         && let Err(error) = super::replication_pool::complete_force_delete_intent(storage, operation_id).await
@@ -3409,6 +3436,53 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
     all_succeeded
 }
 
+const FORCE_DELETE_TARGET_OFFLINE_ERROR: &str = "replication target is offline";
+
+/// One target's outcome of a replicated force-delete in the shape the
+/// replication statistics consume. A force-delete carries no per-version
+/// replication state, so the previous status is always `Pending`: that is
+/// the transition `ReplicationStats::update` counts as a failure, whereas an
+/// `Empty` previous status would drop the outcome and leave `replication
+/// status` reporting zero failures while the peer still holds the objects.
+fn force_delete_target_info(
+    op_type: ReplicationType,
+    arn: &str,
+    endpoint: &str,
+    secure: bool,
+    outcome: std::result::Result<std::time::Duration, String>,
+) -> ReplicatedTargetInfo {
+    let mut rinfo = ReplicatedTargetInfo {
+        arn: arn.to_string(),
+        op_type,
+        endpoint: endpoint.to_string(),
+        secure,
+        prev_replication_status: ReplicationStatusType::Pending,
+        ..Default::default()
+    };
+    match outcome {
+        Ok(duration) => {
+            rinfo.replication_status = ReplicationStatusType::Completed;
+            rinfo.duration = duration;
+        }
+        Err(error) => {
+            rinfo.replication_status = ReplicationStatusType::Failed;
+            rinfo.error = Some(error);
+        }
+    }
+    rinfo
+}
+
+async fn record_force_delete_target_stats(bucket: &str, targets: &[ReplicatedTargetInfo]) {
+    let Some(stats) = runtime_sources::replication_stats() else {
+        return;
+    };
+    for target in targets {
+        stats
+            .update(bucket, target, target.replication_status.clone(), target.prev_replication_status.clone())
+            .await;
+    }
+}
+
 fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo {
     let mut rinfo = dobj
         .delete_object
@@ -7445,6 +7519,51 @@ mod tests {
         server.join().expect("test HTTP server should finish");
     }
 
+    /// A force-delete the peer rejects must reach the replication failure
+    /// counters. A replicated force-delete used to only log the rejection, so
+    /// `replication status` reported zero failures while the sites diverged.
+    #[tokio::test]
+    async fn force_delete_target_outcomes_reach_replication_stats() {
+        let stats = super::super::replication_state::ReplicationStats::new();
+        let denied = force_delete_target_info(
+            ReplicationType::Delete,
+            "arn:rustfs:replication::denied:bucket",
+            "http://peer-a:9000",
+            false,
+            Err("remove_object failed: AccessDenied".to_string()),
+        );
+        assert_eq!(denied.replication_status, ReplicationStatusType::Failed);
+        assert_eq!(denied.prev_replication_status, ReplicationStatusType::Pending);
+        assert_eq!(denied.error.as_deref(), Some("remove_object failed: AccessDenied"));
+
+        let accepted = force_delete_target_info(
+            ReplicationType::Heal,
+            "arn:rustfs:replication::accepted:bucket",
+            "https://peer-b:9000",
+            true,
+            Ok(std::time::Duration::from_millis(3)),
+        );
+        assert_eq!(accepted.replication_status, ReplicationStatusType::Completed);
+        assert!(accepted.error.is_none());
+
+        for target in [&denied, &accepted] {
+            stats
+                .update(
+                    "bucket",
+                    target,
+                    target.replication_status.clone(),
+                    target.prev_replication_status.clone(),
+                )
+                .await;
+        }
+
+        let bucket_stats = stats.get("bucket").await;
+        let denied_stat = bucket_stats.stats.get(&denied.arn).expect("denied target: stats entry");
+        assert_eq!((denied_stat.failed.count, denied_stat.replicated_count), (1, 0));
+        let accepted_stat = bucket_stats.stats.get(&accepted.arn).expect("accepted target: stats entry");
+        assert_eq!((accepted_stat.failed.count, accepted_stat.replicated_count), (0, 1));
+    }
+
     /// AWS answers `HEAD ?versionId=<source marker id>` with a bodiless 400.
     /// The marker is verified through the latest version instead: a
     /// versionless 404 with `x-amz-delete-marker: true` proves it propagated.

docs/operations/site-replication-operations.md

@@ -289,6 +289,16 @@ receiving site. If a group ended up disabled on one site only, re-enable it
 there explicitly with `mc admin group enable`; a snapshot or repair will not
 do it.
 
+The managed `site-replicator-0` service account carries an embedded policy
+that the server owns. It names `s3:ForceDeleteObject` explicitly because a
+replicated force-delete (Console folder delete, `x-rustfs-force-delete: true`)
+is authorized on the peer like any other force-delete and `s3:*` does not
+grant that action. The reconcile tick refreshes an account whose stored policy
+predates the grant; do not edit the policy by hand, the next repair rewrites
+it. A bucket-replication target configured with its own credentials needs the
+same `s3:ForceDeleteObject` grant on the target side, or its force-deletes are
+rejected and counted as replication failures.
+
 Treat IAM divergence as a security incident: a user deleted on one site can
 remain usable on an unreachable peer until replay or repair completes. A peer
 whose IAM entry is escalated does not receive scheduled snapshots either -

rustfs/src/admin/handlers/site_replication.rs

@@ -172,6 +172,7 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
         "s3:PutObject",
         "s3:DeleteObject",
         "s3:DeleteObjectVersion",
+        "s3:ForceDeleteObject",
         "s3:ReplicateObject",
         "s3:ReplicateDelete",
         "s3:ReplicateTags",
@@ -194,6 +195,29 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
     .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("parse site replicator policy failed: {e}")))
 }
 
+/// A replicated force-delete needs an explicit `s3:ForceDeleteObject` on the peer; `s3:*` does not grant it.
+/// Accounts created before the action was added to the policy lack it and must be refreshed in place.
+async fn site_replicator_policy_allows_force_delete(policy: Option<&Policy>) -> bool {
+    let Some(policy) = policy else {
+        return false;
+    };
+    let groups: Option<Vec<String>> = None;
+    let claims = HashMap::new();
+    let conditions = HashMap::new();
+    let args = rustfs_policy::policy::Args {
+        account: SITE_REPLICATOR_SERVICE_ACCOUNT,
+        groups: &groups,
+        action: Action::S3Action(rustfs_policy::policy::action::S3Action::ForceDeleteObjectAction),
+        conditions: &conditions,
+        is_owner: false,
+        claims: &claims,
+        deny_only: false,
+        bucket: "bucket",
+        object: "prefix/",
+    };
+    policy.is_allowed(&args).await
+}
+
 // Lock order: lifecycle -> bucket-mutation admission -> per-bucket mutation
 // -> bucket operation -> repair admission -> state -> per-bucket metadata.
 // "state" is the distributed state-object lock in
@@ -1657,10 +1681,39 @@ async fn reconcile_site_replicator_service_account() -> S3Result<()> {
         Ok(_) => {
             // Never rebuild on an unread parent: `unwrap_or_default` here would compare an
             // empty string against a real parent and repair a healthy account on every boot.
-            let Ok((credentials, _)) = iam_sys.get_service_account(access_key).await else {
+            let Ok((credentials, stored_policy)) = iam_sys.get_service_account(access_key).await else {
                 return Ok(());
             };
             if credentials.parent_user == parent_user {
+                if site_replicator_policy_allows_force_delete(stored_policy.as_ref()).await {
+                    return Ok(());
+                }
+
+                iam_sys
+                    .update_service_account(
+                        access_key,
+                        UpdateServiceAccountOpts {
+                            session_policy: Some(session_policy),
+                            secret_key: None,
+                            name: None,
+                            description: None,
+                            expiration: None,
+                            status: None,
+                            parent_user: None,
+                            allow_site_replicator_account: true,
+                        },
+                    )
+                    .await
+                    .map_err(ApiError::from)?;
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "service_account_repaired",
+                    reason = "outdated_policy",
+                    parent = %parent_user,
+                    "admin site replication state"
+                );
                 return Ok(());
             }
 
@@ -10661,6 +10714,31 @@ mod tests {
         assert!(!policy.is_allowed(&put_policy_args).await);
     }
 
+    // A replicated force-delete (Console folder delete, `x-rustfs-force-delete: true`) is rejected by the
+    // peer with AccessDenied unless the replication account names `s3:ForceDeleteObject` explicitly.
+    #[tokio::test]
+    async fn test_site_replicator_policy_allows_force_delete_replication() {
+        let policy = site_replicator_service_account_policy().expect("site replicator policy should parse");
+        assert!(site_replicator_policy_allows_force_delete(Some(&policy)).await);
+    }
+
+    // Accounts created before the action was added keep their stored policy until the reconciler refreshes it.
+    #[tokio::test]
+    async fn test_outdated_site_replicator_policy_is_not_treated_as_current() {
+        assert!(!site_replicator_policy_allows_force_delete(None).await);
+
+        let outdated: Policy = serde_json::from_value(serde_json::json!({
+            "Version": "2012-10-17",
+            "Statement": [{
+                "Effect": "Allow",
+                "Action": ["s3:DeleteObject", "s3:DeleteObjectVersion", "s3:ReplicateDelete", "s3:*"],
+                "Resource": ["arn:aws:s3:::*/*"]
+            }]
+        }))
+        .expect("outdated policy should parse");
+        assert!(!site_replicator_policy_allows_force_delete(Some(&outdated)).await);
+    }
+
     // The replication service account must be able to carry object-lock metadata to the peer.
     // Without these actions the peer answers AccessDenied for any replicated object that has
     // retention or a legal hold, so a WORM-protected object never reaches the replica at all,