feat(ecstore): observe PUT commit lock admission (#6319)

* feat(ecstore): observe PUT commit lock admission

Co-Authored-By: heihutu <heihutu@gmail.com>

* update h2 v0.4.18

* test(e2e): box SSE-KMS negative errors

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

Cargo.lock

@@ -4757,9 +4757,9 @@ dependencies = [
 
 [[package]]
 name = "h2"
-version = "0.4.17"
+version = "0.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9f877e75f39e9827ec50a572dd592684ac28c029578726c85f1b2aa6ab807449"
+checksum = "839c0e8a181239723652be9062bb56ca5bf5f64011f73b623f6f4fc59086a228"
 dependencies = [
  "atomic-waker",
  "bytes",

crates/config/README.md

@@ -115,6 +115,15 @@ Current guidance:
   - enables KMS readiness enforcement for `/health/ready`.
   - default is `false`.
 
+## Object lock admission environment variables
+
+- `RUSTFS_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS`
+  - experimental same-object PUT commit namespace-lock admission budget.
+  - default is `0`, which disables this override and keeps `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT` behavior.
+  - when set, only `put_object_commit` write-lock acquisition is bounded by this millisecond budget; other namespace lock users keep the global object-lock timeout.
+  - timeout returns S3 `SlowDown`, so clients should use normal SDK retry handling.
+  - this is not a fdatasync or group-commit switch. Track fdatasync batching separately with `rustfs_s3_put_object_rename_fdatasync_batch_files`.
+
 ## Drive timeout environment variables
 
 - `RUSTFS_DRIVE_METADATA_TIMEOUT_SECS`

crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs

@@ -39,6 +39,7 @@ use std::time::Duration;
 use tracing::info;
 
 type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+type S3OperationResult<T> = Result<T, Box<aws_sdk_s3::Error>>;
 
 const ALLOWED_KEY: &str = "kms-matrix-allowed-key";
 const OTHER_KEY: &str = "kms-matrix-other-key";
@@ -130,7 +131,7 @@ fn policy_document(statements: Vec<serde_json::Value>) -> String {
     serde_json::json!({ "Version": "2012-10-17", "Statement": statements }).to_string()
 }
 
-async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> Result<(), Box<aws_sdk_s3::Error>> {
+async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> S3OperationResult<()> {
     client
         .put_object()
         .bucket(BUCKET)
@@ -303,7 +304,7 @@ async fn sse_kms_per_key_authorization_negative_matrix() -> TestResult {
             .send()
             .await
             .map(|_| ())
-            .map_err(aws_sdk_s3::Error::from),
+            .map_err(|err| Box::new(aws_sdk_s3::Error::from(err))),
         "SSE-KMS read by an identity holding no kms grant",
     );
 
@@ -317,7 +318,7 @@ async fn sse_kms_per_key_authorization_negative_matrix() -> TestResult {
             .send()
             .await
             .map(|_| ())
-            .map_err(aws_sdk_s3::Error::from),
+            .map_err(|err| Box::new(aws_sdk_s3::Error::from(err))),
         "SSE-KMS read by an identity holding kms:GenerateDataKey but not kms:Decrypt",
     );
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -90,6 +90,12 @@ use uuid::Uuid;
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
 
+pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
+pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
+pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
+pub type PutObjectTaggingSdkError = Box<SdkError<PutObjectTaggingError>>;
+pub type DeleteObjectTaggingSdkError = Box<SdkError<DeleteObjectTaggingError>>;
+
 pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock<BucketTargetSys> = OnceLock::new();
 
 fn replication_target_versioning_enabled(versioning: Option<&BucketVersioningStatus>) -> bool {
@@ -1968,7 +1974,7 @@ impl TargetClient {
         bucket: &str,
         object: &str,
         version_id: Option<String>,
-    ) -> Result<HeadObjectOutput, Box<SdkError<HeadObjectError>>> {
+    ) -> Result<HeadObjectOutput, HeadObjectSdkError> {
         // Announce the replication check so a RustFS target returns SSE-C
         // object metadata (etag/size) without the customer key the replication
         // worker cannot hold; otherwise SSE-C replicas never converge on HEAD.
@@ -2019,7 +2025,7 @@ impl TargetClient {
         range: Option<String>,
         part_number: Option<i32>,
         extra_headers: HeaderMap,
-    ) -> Result<HeadObjectOutput, Box<SdkError<HeadObjectError>>> {
+    ) -> Result<HeadObjectOutput, HeadObjectSdkError> {
         let headers = proxy_outbound_headers(extra_headers);
         self.client
             .head_object()
@@ -2048,7 +2054,7 @@ impl TargetClient {
         range: Option<String>,
         part_number: Option<i32>,
         extra_headers: HeaderMap,
-    ) -> Result<GetObjectOutput, Box<SdkError<GetObjectError>>> {
+    ) -> Result<GetObjectOutput, GetObjectSdkError> {
         let headers = proxy_outbound_headers(extra_headers);
         self.client
             .get_object()
@@ -2071,7 +2077,7 @@ impl TargetClient {
         bucket: &str,
         object: &str,
         version_id: Option<String>,
-    ) -> Result<GetObjectTaggingOutput, Box<SdkError<GetObjectTaggingError>>> {
+    ) -> Result<GetObjectTaggingOutput, GetObjectTaggingSdkError> {
         let headers = proxy_outbound_headers(HeaderMap::new());
         self.client
             .get_object_tagging()
@@ -2093,7 +2099,7 @@ impl TargetClient {
         object: &str,
         version_id: Option<String>,
         tagging: SdkTagging,
-    ) -> Result<PutObjectTaggingOutput, Box<SdkError<PutObjectTaggingError>>> {
+    ) -> Result<PutObjectTaggingOutput, PutObjectTaggingSdkError> {
         let headers = proxy_outbound_headers(HeaderMap::new());
         self.client
             .put_object_tagging()
@@ -2115,7 +2121,7 @@ impl TargetClient {
         bucket: &str,
         object: &str,
         version_id: Option<String>,
-    ) -> Result<DeleteObjectTaggingOutput, Box<SdkError<DeleteObjectTaggingError>>> {
+    ) -> Result<DeleteObjectTaggingOutput, DeleteObjectTaggingSdkError> {
         let headers = proxy_outbound_headers(HeaderMap::new());
         self.client
             .delete_object_tagging()

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

@@ -52,8 +52,8 @@ use super::replication_storage_boundary::{
     ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
 };
 use super::replication_target_boundary::{
-    ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore,
+    ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
-    SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error,
+    ReplicationTargetStore, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, is_replication_target_offline_error,
     replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options,
     replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted,
     replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version,
@@ -214,7 +214,7 @@ async fn head_object_for_worker(
     target_bucket: &str,
     object: &str,
     version_id: Option<String>,
-) -> std::result::Result<HeadObjectOutput, Box<SdkError<HeadObjectError>>> {
+) -> std::result::Result<HeadObjectOutput, HeadObjectSdkError> {
     target_client.head_object(target_bucket, object, version_id).await
 }
 
@@ -233,7 +233,7 @@ async fn mark_replication_target_offline_if_needed(target_client: &Arc<TargetCli
 async fn head_object_fallback(
     tgt_client: &TargetClient,
     object: &str,
-) -> std::result::Result<Option<HeadObjectOutput>, Box<SdkError<HeadObjectError>>> {
+) -> std::result::Result<Option<HeadObjectOutput>, HeadObjectSdkError> {
     match head_object_for_worker(tgt_client, &tgt_client.bucket, object, None).await {
         Ok(oi) => Ok(Some(oi)),
         Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => Ok(None),
@@ -1152,11 +1152,11 @@ fn spawn_resync_walk_task<S: ReplicationStorage>(
 /// updating the per-object status counters and returning the accounted size
 /// together with any verification error.
 async fn verify_resync_head_result(
-    head_result: std::result::Result<HeadObjectOutput, Box<SdkError<HeadObjectError>>>,
+    head_result: std::result::Result<HeadObjectOutput, HeadObjectSdkError>,
     roi: &ReplicateObjectInfo,
     st: &mut TargetReplicationResyncStatus,
     target_client: &Arc<TargetClient>,
-) -> (i64, Option<Box<SdkError<HeadObjectError>>>) {
+) -> (i64, Option<HeadObjectSdkError>) {
     match head_result {
         Ok(_) => {
             st.replicated_count += 1;
@@ -1275,7 +1275,7 @@ async fn resync_worker_process_object<S: ReplicationStorage>(
             "Processed resync object"
         );
     }
-    st.error = err.as_ref().and_then(|err| resync_target_error_detail(err));
+    st.error = err.as_ref().and_then(|err| resync_target_error_detail(err.as_ref()));
 
     st
 }
@@ -2467,7 +2467,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
             Ok(_) => {}
             Err(e) => {
                 let non_retryable = matches!(
-                    &*e,
+                    e.as_ref(),
                     SdkError::ServiceError(service_err)
                         if is_retryable_delete_replication_head_error(
                             service_err.err().is_not_found(),

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

@@ -36,7 +36,8 @@ use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 
 pub(crate) use crate::bucket::bucket_target_sys::{
-    AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, resolve_read_api_version_id,
+    AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
+    resolve_read_api_version_id,
 };
 #[cfg(test)]
 pub(crate) use crate::bucket::target::BucketTarget;

crates/ecstore/src/set_disk/mod.rs

@@ -1539,6 +1539,50 @@ fn put_object_commit_lock_timeout_override_enabled(op: &'static str) -> bool {
     op == "put_object_commit" && get_put_object_commit_lock_acquire_timeout_override_ms() != 0
 }
 
+fn put_object_commit_lock_admission_budget_label() -> &'static str {
+    match get_put_object_commit_lock_acquire_timeout_override_ms() {
+        0 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED,
+        1..=250 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+        251..=500 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS,
+        501..=1000 => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS,
+        _ => rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS,
+    }
+}
+
+fn record_put_object_commit_lock_admission(op: &'static str, outcome: &'static str) {
+    if op != "put_object_commit" || !rustfs_io_metrics::put_stage_metrics_enabled() {
+        return;
+    }
+    rustfs_io_metrics::record_put_object_commit_lock_admission(put_object_commit_lock_admission_budget_label(), outcome);
+}
+
+fn put_object_commit_lock_acquire_error_outcome(op: &'static str, err: &rustfs_lock::error::LockError) -> &'static str {
+    if put_object_commit_lock_timeout_override_enabled(op) && matches!(err, rustfs_lock::error::LockError::Timeout { .. }) {
+        rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN
+    } else {
+        rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR
+    }
+}
+
+fn resolve_put_object_commit_lock_acquire_result(
+    set: &SetDisks,
+    op: &'static str,
+    bucket: &str,
+    object: &str,
+    result: std::result::Result<rustfs_lock::namespace::NamespaceLockGuard, rustfs_lock::error::LockError>,
+) -> Result<rustfs_lock::namespace::NamespaceLockGuard> {
+    match result {
+        Ok(guard) => {
+            record_put_object_commit_lock_admission(op, rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED);
+            Ok(guard)
+        }
+        Err(err) => {
+            record_put_object_commit_lock_admission(op, put_object_commit_lock_acquire_error_outcome(op, &err));
+            Err(map_put_object_commit_lock_acquire_error(set, op, bucket, object, err))
+        }
+    }
+}
+
 fn map_put_object_commit_lock_acquire_error(
     set: &SetDisks,
     op: &'static str,
@@ -3355,10 +3399,13 @@ impl SetDisks {
         let ns_lock = self.new_ns_lock(bucket, object).await?;
         let acquire_start = Instant::now();
         let acquire_timeout = get_put_object_commit_lock_acquire_timeout(op);
-        let guard = ns_lock
+        let guard = resolve_put_object_commit_lock_acquire_result(
-            .get_write_lock(acquire_timeout)
+            self,
-            .await
+            op,
-            .map_err(|e| map_put_object_commit_lock_acquire_error(self, op, bucket, object, e))?;
+            bucket,
+            object,
+            ns_lock.get_write_lock(acquire_timeout).await,
+        )?;
         Self::record_put_object_commit_namespace_lock_wait(op, acquire_start);
         let owner = diag_enabled.then(|| ns_lock.owner().to_string());
         self.log_object_lock_acquire_if_slow(
@@ -3397,7 +3444,12 @@ impl SetDisks {
         let acquire = ns_lock.get_write_lock(acquire_timeout);
         tokio::pin!(acquire);
         let mut on_pending = Some(on_pending);
-        let guard = futures::future::poll_fn(|cx| match std::future::Future::poll(acquire.as_mut(), cx) {
+        let guard = resolve_put_object_commit_lock_acquire_result(
+            self,
+            op,
+            bucket,
+            object,
+            futures::future::poll_fn(|cx| match std::future::Future::poll(acquire.as_mut(), cx) {
                 std::task::Poll::Pending => {
                     if let Some(on_pending) = on_pending.take() {
                         on_pending();
@@ -3406,8 +3458,8 @@ impl SetDisks {
                 }
                 std::task::Poll::Ready(result) => std::task::Poll::Ready(result),
             })
-        .await
+            .await,
-        .map_err(|e| map_put_object_commit_lock_acquire_error(self, op, bucket, object, e))?;
+        )?;
         Self::record_put_object_commit_namespace_lock_wait(op, acquire_start);
         let owner = diag_enabled.then(|| ns_lock.owner().to_string());
         self.log_object_lock_acquire_if_slow(
@@ -5782,6 +5834,81 @@ mod tests {
             .sum()
     }
 
+    fn put_object_commit_lock_admission_count(
+        rows: &[(
+            metrics_util::CompositeKey,
+            Option<metrics::Unit>,
+            Option<metrics::SharedString>,
+            DebugValue,
+        )],
+        budget: &'static str,
+        outcome: &'static str,
+    ) -> u64 {
+        rows.iter()
+            .filter(|(composite, _, _, _)| {
+                composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total"
+                    && composite
+                        .key()
+                        .labels()
+                        .any(|label| label.key() == "budget" && label.value() == budget)
+                    && composite
+                        .key()
+                        .labels()
+                        .any(|label| label.key() == "outcome" && label.value() == outcome)
+            })
+            .map(|(_, _, _, value)| match value {
+                DebugValue::Counter(count) => *count,
+                _ => 0,
+            })
+            .sum()
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_budget_labels_are_bounded() {
+        let cases = [
+            ("0", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED),
+            ("250", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS),
+            ("251", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS),
+            ("500", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS),
+            ("501", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS),
+            ("1000", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS),
+            ("1001", rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS),
+        ];
+        for (timeout_ms, expected) in cases {
+            temp_env::with_vars(
+                [(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some(timeout_ms))],
+                || {
+                    assert_eq!(put_object_commit_lock_admission_budget_label(), expected);
+                },
+            );
+        }
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_error_outcomes_are_bounded() {
+        let timeout = LockError::timeout("bucket/object", Duration::from_millis(1));
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || {
+            assert_eq!(
+                put_object_commit_lock_acquire_error_outcome("put_object_commit", &timeout),
+                rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN
+            );
+            assert_eq!(
+                put_object_commit_lock_acquire_error_outcome("complete_multipart_upload_commit", &timeout),
+                rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR
+            );
+        });
+
+        let internal = LockError::internal("simulated lock manager error");
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || {
+            assert_eq!(
+                put_object_commit_lock_acquire_error_outcome("put_object_commit", &internal),
+                rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR
+            );
+        });
+    }
+
     #[test]
     #[serial]
     fn put_object_commit_namespace_lock_wait_metric_is_wired_to_both_write_lock_paths() {
@@ -5905,6 +6032,231 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_records_acquired_and_timeout() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should start");
+
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("1"))], || {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || {
+                rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+                runtime.block_on(async {
+                    let ctx = Arc::new(InstanceContext::new());
+                    ctx.update_erasure_type(SetupType::Erasure).await;
+                    let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await;
+                    let held_guard = set
+                        .acquire_write_lock_diag("put_object_commit", "bucket", "object")
+                        .await
+                        .expect("holder acquire should succeed");
+                    let err = match set.acquire_write_lock_diag("put_object_commit", "bucket", "object").await {
+                        Ok(_) => panic!("contended PUT commit acquire should return SlowDown"),
+                        Err(err) => err,
+                    };
+                    assert!(matches!(err, StorageError::SlowDown));
+                    drop(held_guard);
+                    rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+                });
+            });
+
+            let rows = snapshotter.snapshot().into_vec();
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED,
+                ),
+                1
+            );
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN,
+                ),
+                1
+            );
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_records_disabled_budget_acquired() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should start");
+
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("0"))], || {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || {
+                rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+                runtime.block_on(async {
+                    let ctx = Arc::new(InstanceContext::new());
+                    ctx.update_erasure_type(SetupType::Erasure).await;
+                    let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await;
+                    let guard = set
+                        .acquire_write_lock_diag("put_object_commit", "bucket", "object")
+                        .await
+                        .expect("PUT commit acquire should succeed with default timeout");
+                    drop(guard);
+                    rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+                });
+            });
+
+            let rows = snapshotter.snapshot().into_vec();
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED,
+                ),
+                1
+            );
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_skips_non_put_commit_ops() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should start");
+
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("250"))], || {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || {
+                rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+                runtime.block_on(async {
+                    let ctx = Arc::new(InstanceContext::new());
+                    ctx.update_erasure_type(SetupType::Erasure).await;
+                    let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await;
+                    let guard = set
+                        .acquire_write_lock_diag("complete_multipart_upload_commit", "bucket", "object")
+                        .await
+                        .expect("non-PUT commit acquire should succeed");
+                    drop(guard);
+                    rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+                });
+            });
+
+            let rows = snapshotter.snapshot().into_vec();
+            assert_eq!(
+                rows.iter()
+                    .filter(|(composite, _, _, _)| {
+                        composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total"
+                    })
+                    .count(),
+                0
+            );
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_records_lock_error() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should start");
+
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("250"))], || {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || {
+                rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+                runtime.block_on(async {
+                    let healthy: Arc<dyn LockClient> =
+                        Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+                    let failing: Arc<dyn LockClient> = Arc::new(FailingClient);
+                    let ctx = Arc::new(InstanceContext::new());
+                    ctx.update_erasure_type(SetupType::DistErasure).await;
+                    let set = make_test_set_disks_with_ctx(vec![healthy, failing], ctx).await;
+                    assert!(
+                        set.acquire_write_lock_diag("put_object_commit", "bucket", "object")
+                            .await
+                            .is_err(),
+                        "one healthy locker must not satisfy the PUT commit write quorum"
+                    );
+                    rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+                });
+            });
+
+            let rows = snapshotter.snapshot().into_vec();
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR,
+                ),
+                1
+            );
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN,
+                ),
+                0
+            );
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn put_object_commit_lock_admission_records_pending_hook_acquired() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should start");
+
+        temp_env::with_vars([(rustfs_config::ENV_PUT_COMMIT_NAMESPACE_LOCK_ACQUIRE_TIMEOUT_MS, Some("500"))], || {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || {
+                rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+                runtime.block_on(async {
+                    let ctx = Arc::new(InstanceContext::new());
+                    ctx.update_erasure_type(SetupType::Erasure).await;
+                    let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await;
+                    let held_guard = set
+                        .acquire_write_lock_diag("put_object_commit", "bucket", "object")
+                        .await
+                        .expect("holder acquire should succeed");
+                    let (pending_tx, pending_rx) = tokio::sync::oneshot::channel();
+                    let pending_acquire =
+                        set.acquire_write_lock_diag_with_pending_hook("put_object_commit", "bucket", "object", move || {
+                            let _ = pending_tx.send(());
+                        });
+                    let release_holder = async {
+                        pending_rx.await.expect("pending hook should fire");
+                        drop(held_guard);
+                    };
+                    let (pending_guard, ()) = tokio::join!(pending_acquire, release_holder);
+                    drop(pending_guard.expect("pending-hook PUT commit acquire should succeed"));
+                    rustfs_io_metrics::set_put_stage_metrics_enabled(false);
+                });
+            });
+
+            let rows = snapshotter.snapshot().into_vec();
+            assert_eq!(
+                put_object_commit_lock_admission_count(
+                    &rows,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS,
+                    rustfs_io_metrics::PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED,
+                ),
+                2
+            );
+        });
+    }
+
     #[tokio::test]
     async fn new_ns_lock_shares_clients_without_changing_quorum() {
         let healthy: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));

crates/io-metrics/src/lib.rs

@@ -121,6 +121,16 @@ pub const PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC: &str = "set_disk_rename_ba
 pub const PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC: &str = "set_disk_rename_ancestor_dir_fsync";
 pub const PUT_STAGE_SET_DISK_RENAME_RENAME_SYSCALL: &str = "set_disk_rename_rename_syscall";
 
+pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED: &str = "disabled";
+pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS: &str = "le_250ms";
+pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS: &str = "le_500ms";
+pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS: &str = "le_1000ms";
+pub const PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS: &str = "gt_1000ms";
+
+pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED: &str = "acquired";
+pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN: &str = "timeout_slowdown";
+pub const PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR: &str = "lock_error";
+
 pub const PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL: &str = "serial";
 pub const PUT_RENAME_FDATASYNC_BATCH_MODE_PARALLEL: &str = "parallel";
 pub const PUT_RENAME_FDATASYNC_GROUP_WAIT_ROLE_LEADER: &str = "leader";
@@ -2060,6 +2070,14 @@ pub fn record_put_object_stage_duration_from(stage: &'static str, started_at: Op
     }
 }
 
+#[inline(always)]
+pub fn record_put_object_commit_lock_admission(budget: &'static str, outcome: &'static str) {
+    if !put_stage_metrics_enabled() {
+        return;
+    }
+    counter!("rustfs_s3_put_object_commit_namespace_lock_admission_total", "budget" => budget, "outcome" => outcome).increment(1);
+}
+
 #[inline(always)]
 fn put_stage_count_value(value: usize) -> f64 {
     match u32::try_from(value) {
@@ -3204,6 +3222,83 @@ mod tests {
         assert!(stages.iter().all(|stage| recorded.contains(*stage)));
     }
 
+    #[test]
+    fn put_commit_lock_admission_labels_are_static_and_gated() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let budgets = [
+            PUT_COMMIT_LOCK_ADMISSION_BUDGET_DISABLED,
+            PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+            PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS,
+            PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_1000MS,
+            PUT_COMMIT_LOCK_ADMISSION_BUDGET_GT_1000MS,
+        ];
+        let outcomes = [
+            PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED,
+            PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN,
+            PUT_COMMIT_LOCK_ADMISSION_OUTCOME_LOCK_ERROR,
+        ];
+        assert_eq!(budgets.iter().copied().collect::<HashSet<_>>().len(), budgets.len());
+        assert_eq!(outcomes.iter().copied().collect::<HashSet<_>>().len(), outcomes.len());
+        assert!(budgets.iter().chain(outcomes.iter()).all(|label| {
+            !label.contains('/')
+                && !label.contains('{')
+                && !label.contains('}')
+                && !label.contains(' ')
+                && label
+                    .chars()
+                    .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_')
+        }));
+
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        metrics::with_local_recorder(&recorder, || {
+            set_put_stage_metrics_enabled(false);
+            record_put_object_commit_lock_admission(
+                PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN,
+            );
+
+            set_put_stage_metrics_enabled(true);
+            record_put_object_commit_lock_admission(
+                PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS,
+                PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN,
+            );
+            record_put_object_commit_lock_admission(
+                PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS,
+                PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED,
+            );
+            set_put_stage_metrics_enabled(false);
+        });
+
+        let rows = snapshotter.snapshot().into_vec();
+        assert_eq!(
+            counter_total(&rows, "rustfs_s3_put_object_commit_namespace_lock_admission_total"),
+            Some(2)
+        );
+        let label_sets = rows
+            .iter()
+            .filter(|(composite, _, _, _)| {
+                composite.kind() == MetricKind::Counter
+                    && composite.key().name() == "rustfs_s3_put_object_commit_namespace_lock_admission_total"
+            })
+            .map(|(composite, _, _, _)| {
+                composite
+                    .key()
+                    .labels()
+                    .map(|label| (label.key().to_string(), label.value().to_string()))
+                    .collect::<HashSet<_>>()
+            })
+            .collect::<Vec<_>>();
+        assert!(label_sets.contains(&HashSet::from([
+            ("budget".to_string(), PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_250MS.to_string()),
+            ("outcome".to_string(), PUT_COMMIT_LOCK_ADMISSION_OUTCOME_TIMEOUT_SLOWDOWN.to_string(),),
+        ])));
+        assert!(label_sets.contains(&HashSet::from([
+            ("budget".to_string(), PUT_COMMIT_LOCK_ADMISSION_BUDGET_LE_500MS.to_string()),
+            ("outcome".to_string(), PUT_COMMIT_LOCK_ADMISSION_OUTCOME_ACQUIRED.to_string()),
+        ])));
+    }
+
     #[test]
     fn put_rename_code_level_metrics_are_static_and_gated() {
         let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());

crates/s3select-query/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 pub mod data_source;
 pub mod dispatcher;
 pub mod execution;