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	6083	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

.github/workflows/rustfs-fault-tolerance-test.yml

@@ -197,21 +197,21 @@ jobs:
           jq -e --arg testing_sha "${AUTO_TESTING_SHA}" '
             .schema_version == 1 and
             .auto_testing_sha == $testing_sha and
-            .selected_scenarios == ["A", "B", "C", "C2", "D", "E"] and
+            .selected_scenarios == ["A", "B", "C", "C2", "D", "E", "F"] and
             .complete == true and
-            .counts.expected == 44 and
-            .counts.seen == 44 and
+            .counts.expected == 54 and
+            .counts.seen == 54 and
             .counts.harness_errors == 0 and
-            (.expected_cases | length) == 44 and
-            (.expected_cases | unique | length) == 44 and
-            (.seen_cases | length) == 44 and
-            (.seen_cases | unique | length) == 44 and
+            (.expected_cases | length) == 54 and
+            (.expected_cases | unique | length) == 54 and
+            (.seen_cases | length) == 54 and
+            (.seen_cases | unique | length) == 54 and
             (.seen_cases | sort) == (.expected_cases | sort) and
             (.missing_cases | length) == 0 and
             (.harness_errors | length) == 0 and
-            (.cases | length) == 44 and
+            (.cases | length) == 54 and
             ([.cases[].case_id] | sort) == (.expected_cases | sort) and
-            ([.cases[].case_id] | unique | length) == 44 and
+            ([.cases[].case_id] | unique | length) == 54 and
             all(.cases[]; .verdict == "pass" or .verdict == "known-divergence" or .verdict == "UNEXPECTED") and
             .counts.unexpected == ([.cases[] | select(.verdict == "UNEXPECTED")] | length) and
             .counts.known_divergence == ([.cases[] | select(.verdict == "known-divergence")] | length)

.github/workflows/rustfs-functional-chain.yml

@@ -281,12 +281,20 @@ jobs:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
           PERFORMANCE_OPTIONAL: ${{ contains(fromJSON(steps.selection.outputs.optional_suites), 'performance') }}
         run: |
-          if python3 scripts/check_functional_runners.py pf-testing; then
-            echo "performance_ready=online" >> "$GITHUB_OUTPUT"
-          else
-            echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
-            test "$PERFORMANCE_OPTIONAL" = true
-          fi
+          # TEMPORARY: the pf-testing environment is broken, so the
+          # performance lane is disabled until it is repaired. Forcing
+          # offline keeps the chain on its designed skip path: the lane job
+          # is skipped, aggregate receives --allow-skipped performance
+          # (required=false in scheduled-validations.json), and the report
+          # records the optional-lane skip with its reason.
+          # Restore the probe below once the environment is healthy again.
+          # if python3 scripts/check_functional_runners.py pf-testing; then
+          #   echo "performance_ready=online" >> "$GITHUB_OUTPUT"
+          # else
+          #   echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
+          #   test "$PERFORMANCE_OPTIONAL" = true
+          # fi
+          echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
 
   validation:
     if: ${{ github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && (inputs.validation_case != '' || inputs.validation_tools_only || inputs.validation_native_tools || inputs.validation_shared_tools || inputs.validation_manager_tools || inputs.validation_cleanup_run_id != '' || inputs.validation_cleanup_retained_evidence) }}

.github/workflows/rustfs-heal-test.yml

@@ -55,7 +55,8 @@ env:
 jobs:
   heal-test:
     runs-on: smoke-testing
-    timeout-minutes: 60
+    # 90-minute heal run (see the step below) plus setup, report and cleanup.
+    timeout-minutes: 120
     # Standalone manual run, or one link of the nightly functional chain
     # (storage -> heal -> pool). Pool expansion no longer re-runs heal.
     if: ${{ inputs.chain_manifest != '' || github.event_name == 'workflow_dispatch' }}
@@ -154,7 +155,7 @@ jobs:
           ./auto-testing/rustfs_heal_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
 
       - name: Run heal test (write -> outage -> heal -> verify)
-        timeout-minutes: 45
+        timeout-minutes: 90
         id: test
         # Collect reports and backlog evidence before the final suite verdict.
         continue-on-error: true

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,

rustfs/src/app/object/put.rs

@@ -2092,7 +2092,8 @@ 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();
-            async move {
+            // Keep the large commit future off the request stack when propagating task-local context.
+            Box::pin(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;
@@ -2148,6 +2149,7 @@ 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),
@@ -2252,7 +2254,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(
@@ -2260,6 +2262,11 @@ 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 {
@@ -2340,6 +2347,68 @@ 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,18 +64,35 @@ 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) => 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::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::NamespaceLockQuorumUnavailable {
             mode, bucket, object, ..
         } => {
@@ -309,6 +326,7 @@ 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>,
@@ -350,6 +368,18 @@ 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());
@@ -1273,6 +1303,91 @@ 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)));

scripts/test_functional_chain.py

@@ -919,7 +919,10 @@ class WorkflowTimeoutTests(unittest.TestCase):
             with self.subTest(suite=suite):
                 source = (candidate.ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
                 job = yaml_block(source.splitlines(), job_id, 2)
-                job_timeout = 360 if suite == "pool-expand" else 60
+                # heal: a progressing heal outran the shared 45-minute step
+                # cap (chain 40796474356), so it gets a 90-minute run budget
+                # and a 120-minute job budget for setup, report and cleanup.
+                job_timeout = 360 if suite == "pool-expand" else 120 if suite == "heal" else 60
                 self.assertIn(f"    timeout-minutes: {job_timeout}", job)
                 steps = named_steps(job)
                 primary = [step for step in steps.values() if any(
@@ -933,6 +936,8 @@ class WorkflowTimeoutTests(unittest.TestCase):
                         self.assertIsNotNone(timeout_input, event)
                         self.assertIn("        default: '240'", timeout_input)
                         self.assertIn("        required: false", timeout_input)
+                elif suite == "heal":
+                    self.assertIn("        timeout-minutes: 90", primary[0])
                 else:
                     self.assertIn("        timeout-minutes: 45", primary[0])
                 cleanup = "Cleanup environment"

scripts/test_security_workflow.py

@@ -540,10 +540,10 @@ class FaultToleranceWorkflowContractTests(unittest.TestCase):
             "GITHUB_RUN_ID": "314159",
         }
 
-    def write_result(self, *, seen: int = 44, complete: bool = True,
+    def write_result(self, *, seen: int = 54, complete: bool = True,
                      duplicate: bool = False, testing_sha: str | None = None,
                      unrelated_seen: bool = False, missing_verdict: bool = False) -> None:
-        expected = [f"CASE-{index:02d}" for index in range(44)]
+        expected = [f"CASE-{index:02d}" for index in range(54)]
         seen_cases = expected[:seen]
         if unrelated_seen:
             seen_cases = [f"OTHER-{index:02d}" for index in range(seen)]
@@ -555,14 +555,14 @@ class FaultToleranceWorkflowContractTests(unittest.TestCase):
         payload = {
             "schema_version": 1,
             "auto_testing_sha": testing_sha or self.TESTING_SHA,
-            "selected_scenarios": ["A", "B", "C", "C2", "D", "E"],
+            "selected_scenarios": ["A", "B", "C", "C2", "D", "E", "F"],
             "complete": complete,
             "expected_cases": expected,
             "seen_cases": seen_cases,
             "missing_cases": expected[seen:],
             "harness_errors": [],
             "counts": {
-                "expected": 44,
+                "expected": 54,
                 "seen": seen,
                 "unexpected": 1,
                 "known_divergence": 0,
@@ -1186,7 +1186,6 @@ class TableSuiteTests(unittest.TestCase):
         self.assertIn("runs-on: smoke-testing", source)
         # New gate semantics: case failures stay green, harness breakdowns red.
         self.assertRegex(source, r"(?m)^        continue-on-error: true")
-        self.assertIn("# Red only for harness/environment breakdowns; case failures stay green.", source)
         self.assertIn("HARNESS_OK", source)
         self.assertIn("Product result: ${CASES_PASS} passed, ${CASES_FAIL} failed", source)
         self.assertIn('python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}"', source)