fix(replication): count force-delete outcomes in replication stats

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.

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
@@ -7444,6 +7518,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/app/object/put.rs

@@ -2092,8 +2092,7 @@ impl DefaultObjectUsecase {
             let put_admission = put_admission;
             let table_publication_guard = origin.take_table_publication_guard();
             let table_data_plane_publication_guards = origin.table_data_plane_publication_guards();
-            // Keep the large commit future off the request stack when propagating task-local context.
-            Box::pin(async move {
+            async move {
                 let _table_publication_guard = table_publication_guard;
                 let _table_data_plane_publication_guards = table_data_plane_publication_guards;
                 let _put_admission = put_admission;
@@ -2149,7 +2148,6 @@ impl DefaultObjectUsecase {
                                 result = "error",
                                 error_code = %err.code.as_str(),
                                 storage_error_code = ?diagnostic.storage_code,
-                                lock_failure_reason = diagnostic.lock_reason,
                                 io_error_kind = ?diagnostic.io_kind,
                                 rpc_error_code = ?diagnostic.rpc_code,
                                 pool_metadata_reason = diagnostic.pool_metadata.map(|context| context.reason),
@@ -2254,7 +2252,7 @@ impl DefaultObjectUsecase {
                 rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start);
 
                 Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned })
-            })
+            }
         };
         #[cfg(feature = "diagnostic-read-version-caller")]
         let put_commit_future = crate::app::storage_api::diagnostic::with_read_version_spawn_context(
@@ -2262,11 +2260,6 @@ impl DefaultObjectUsecase {
             crate::app::storage_api::diagnostic::ReadVersionContextBoundary::AppPutOwner,
             put_commit_future,
         );
-        #[cfg(not(feature = "diagnostic-read-version-caller"))]
-        let put_commit_future = rustfs_io_metrics::s3_http_metrics::with_s3_http_operation_context(
-            rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_context(),
-            put_commit_future,
-        );
         let put_commit = spawn_traced_join(put_commit_future);
         let commit_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let put_commit_result = if let Some(cancellation) = eager_put_commit_cancellation {
@@ -2347,68 +2340,6 @@ mod tests {
     use futures::StreamExt;
     use http::{HeaderMap, HeaderName, HeaderValue, Method};
 
-    #[test]
-    #[serial_test::serial]
-    fn put_commit_lock_failure_keeps_request_operation_in_spawned_owner() {
-        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
-        use metrics_util::debugging::{DebugValue, DebuggingRecorder};
-        use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard;
-        use rustfs_lock::LockError;
-
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("test runtime");
-        let (store, context) = runtime.block_on(Box::pin(real_store_test_context()));
-        let bucket = format!("put-lock-context-{}", Uuid::new_v4());
-        runtime
-            .block_on(Box::pin(store.make_bucket(&bucket, &MakeBucketOptions::default())))
-            .expect("test bucket");
-        let input = PutObjectInput::builder()
-            .bucket(bucket)
-            .key("test-object".to_string())
-            .body(Some(StreamingBlob::from_bytes(Bytes::from_static(b"payload"))))
-            .content_length(Some(7))
-            .build()
-            .expect("PUT input");
-        let usecase = DefaultObjectUsecase::with_context(Some(context));
-        let recorder = DebuggingRecorder::new();
-        let snapshotter = recorder.snapshotter();
-        let mut request = S3HttpRequestGuard::new("PUT");
-        let error = metrics::with_local_recorder(&recorder, || {
-            let fs = FS::new();
-            let mut future = Box::pin(PUT_STORAGE_TEST_FAILURE.scope(
-                (
-                    PutStorageTestStage::Commit,
-                    StorageError::Lock(LockError::timeout("private-bucket/private-key", Duration::from_secs(5))),
-                ),
-                Box::pin(usecase.execute_put_object(&fs, build_request(input, Method::PUT))),
-            ));
-            runtime.block_on(std::future::poll_fn(|cx| {
-                request.in_scope(|| std::future::Future::poll(future.as_mut(), cx))
-            }))
-        })
-        .expect_err("lock failure must reject the PUT");
-        assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
-
-        let rows = snapshotter.snapshot().into_vec();
-        let (key, _, _, value) = rows
-            .iter()
-            .find(|(key, _, _, _)| key.key().name() == "rustfs_s3_lock_failures_total")
-            .expect("spawned commit lock failure must be counted");
-        assert!(matches!(value, DebugValue::Counter(1)));
-        let labels = key
-            .key()
-            .labels()
-            .map(|label| (label.key(), label.value()))
-            .collect::<std::collections::BTreeMap<_, _>>();
-        assert_eq!(labels["op"], S3Operation::PutObject.as_str());
-        assert_eq!(labels["resource_class"], "object_namespace");
-        assert_eq!(labels["mode"], "unknown");
-        assert_eq!(labels["reason"], "timeout");
-        assert!(!labels.values().any(|value| value.contains("private")));
-    }
-
     #[tokio::test]
     #[serial_test::serial]
     async fn put_read_quorum_failures_are_normalized_at_each_stage() {

rustfs/src/error.rs

@@ -64,35 +64,18 @@ fn lock_resource_class(resource: Option<&str>) -> &'static str {
     }
 }
 
-fn lock_failure_reason(error: &LockError) -> &'static str {
-    match error {
-        LockError::Timeout { .. } => "timeout",
-        LockError::AlreadyLocked { .. } => "contention",
-        LockError::QuorumNotReached { .. } => "quorum",
-        LockError::InsufficientNodes { .. } => "insufficient_nodes",
-        LockError::Network { .. } => "network",
-        LockError::Internal { .. } => "internal",
-        _ => "other",
-    }
-}
-
 fn record_s3_lock_failure(err: &StorageError) {
     let record = |resource: Option<&str>, mode: &'static str, reason: &'static str| {
         rustfs_io_metrics::s3_http_metrics::record_s3_http_lock_failure(lock_resource_class(resource), mode, reason);
     };
     match err {
-        StorageError::Lock(lock_error) => {
-            let resource = match lock_error {
-                LockError::Timeout { resource, .. } | LockError::AlreadyLocked { resource, .. } => Some(resource.as_str()),
-                _ => None,
-            };
-            let reason = match lock_failure_reason(lock_error) {
-                "insufficient_nodes" => "quorum",
-                "network" | "internal" => "transport",
-                reason => reason,
-            };
-            record(resource, "unknown", reason);
-        }
+        StorageError::Lock(lock_error) => match lock_error {
+            LockError::Timeout { resource, .. } => record(Some(resource), "unknown", "timeout"),
+            LockError::AlreadyLocked { resource, .. } => record(Some(resource), "unknown", "contention"),
+            LockError::QuorumNotReached { .. } | LockError::InsufficientNodes { .. } => record(None, "unknown", "quorum"),
+            LockError::Network { .. } | LockError::Internal { .. } => record(None, "unknown", "transport"),
+            _ => record(None, "unknown", "other"),
+        },
         StorageError::NamespaceLockQuorumUnavailable {
             mode, bucket, object, ..
         } => {
@@ -326,7 +309,6 @@ impl std::error::Error for ApiError {
 #[derive(Debug, Default, PartialEq, Eq)]
 pub(crate) struct ApiErrorDiagnostic {
     pub storage_code: Option<StorageErrorCode>,
-    pub lock_reason: Option<&'static str>,
     pub io_kind: Option<std::io::ErrorKind>,
     pub io_os_error: Option<i32>,
     pub rpc_code: Option<tonic::Code>,
@@ -368,18 +350,6 @@ impl ApiErrorDiagnostic {
             };
             if let Some(storage) = error.downcast_ref::<StorageError>() {
                 diagnostic.storage_code = Some(storage.code());
-                match storage {
-                    StorageError::Lock(lock_error) => {
-                        diagnostic.lock_reason = Some(lock_failure_reason(lock_error));
-                    }
-                    StorageError::NamespaceLockQuorumUnavailable { .. } => {
-                        diagnostic.lock_reason = Some("quorum");
-                    }
-                    _ => {}
-                }
-            }
-            if let Some(lock_error) = error.downcast_ref::<LockError>() {
-                diagnostic.lock_reason = Some(lock_failure_reason(lock_error));
             }
             if let Some(status) = error.downcast_ref::<tonic::Status>() {
                 diagnostic.rpc_code = Some(status.code());
@@ -1293,91 +1263,6 @@ mod tests {
         assert!(!format!("{:?}", error.diagnostic()).contains("secret"));
     }
 
-    #[test]
-    fn api_error_diagnostic_classifies_lock_failures_without_exposing_resources() {
-        let cases = [
-            (
-                LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5)),
-                "timeout",
-            ),
-            (
-                LockError::AlreadyLocked {
-                    resource: "private-bucket/private-key".into(),
-                    owner: "private-owner".into(),
-                },
-                "contention",
-            ),
-            (
-                LockError::QuorumNotReached {
-                    required: 3,
-                    achieved: 2,
-                },
-                "quorum",
-            ),
-            (
-                LockError::InsufficientNodes {
-                    required: 3,
-                    available: 2,
-                },
-                "insufficient_nodes",
-            ),
-            (LockError::internal("private-transport-detail"), "internal"),
-            (LockError::network("private-network-detail", IoError::other("private-source")), "network"),
-            (
-                LockError::QueueFull {
-                    message: "private-queue-detail".into(),
-                },
-                "other",
-            ),
-        ];
-        for (failure, expected) in cases {
-            let api = ApiError::from(StorageError::Lock(failure));
-            assert_eq!(api.code, S3ErrorCode::ServiceUnavailable);
-            let diagnostic = api.diagnostic();
-            assert_eq!(diagnostic.storage_code, Some(StorageErrorCode::Lock));
-            assert_eq!(diagnostic.lock_reason, Some(expected));
-            assert!(!format!("{diagnostic:?}").contains("private"));
-            let s3: S3Error = api.into();
-            assert_eq!(s3.code(), &S3ErrorCode::ServiceUnavailable);
-            assert_eq!(super::ApiErrorDiagnostic::from_error(&s3).lock_reason, Some(expected));
-        }
-
-        let nested = ApiError::from(StorageError::Io(IoError::other(StorageError::Lock(LockError::timeout(
-            "private-bucket/private-key",
-            std::time::Duration::from_secs(5),
-        )))));
-        assert_eq!(nested.diagnostic().lock_reason, Some("timeout"));
-        assert!(!format!("{:?}", nested.diagnostic()).contains("private"));
-
-        let nested_lock = ApiError::from(StorageError::Lock(LockError::network(
-            "private-transport",
-            StorageError::Lock(LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5))),
-        )));
-        assert_eq!(nested_lock.diagnostic().storage_code, Some(StorageErrorCode::Lock));
-        assert_eq!(nested_lock.diagnostic().lock_reason, Some("timeout"));
-        assert!(!format!("{:?}", nested_lock.diagnostic()).contains("private"));
-
-        let nested_direct = ApiError::from(StorageError::Lock(LockError::network(
-            "private-transport",
-            LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5)),
-        )));
-        assert_eq!(nested_direct.diagnostic().lock_reason, Some("timeout"));
-        assert!(!format!("{:?}", nested_direct.diagnostic()).contains("private"));
-
-        let unavailable = ApiError::from(StorageError::NamespaceLockQuorumUnavailable {
-            mode: "write",
-            bucket: "private-bucket".into(),
-            object: "private-key".into(),
-            required: 3,
-            achieved: 2,
-        });
-        assert_eq!(unavailable.diagnostic().lock_reason, Some("quorum"));
-        assert!(!format!("{:?}", unavailable.diagnostic()).contains("private"));
-
-        let other = ApiError::from(StorageError::ErasureWriteQuorum);
-        assert_eq!(other.diagnostic().lock_reason, None);
-    }
-
     #[test]
     fn api_error_diagnostic_survives_s3_conversion_without_changing_response() {
         let api = ApiError::from(StorageError::Io(IoError::from_raw_os_error(30)));