perf(put): comprehensive PUT performance optimization (#3514)

* perf(put): add eager path metrics and isolation tooling

* fix(decommission): persist progress adaptively (#3497)

Persist decommission progress after either the existing time interval or a migrated-item threshold, and flush progress baselines after bucket and terminal-state saves.

Also stabilize the OIDC discovery mock used by the pre-commit gate.

* refactor: move bucket operations contract (#3507)

* fix(s3): handle multipart flexible checksums (#3508)

* fix(io-core): avoid blocking on pooled buffer return

* perf(put): add slow inflight diagnostics

* perf(put): fix 16KiB regression with threshold and pool bypass

- Lower SMALL_EAGER_PUT_MAX_SIZE from 256KB to 8KB so objects >8KiB
  use the streaming BufReader path (matches baseline behavior)
- Add POOL_BYPASE_MAX_SIZE (16KiB) to bypass BytesPool for very small
  objects, avoiding Small-tier Mutex contention under high concurrency
- Add read_small_put_body_exact_direct() for direct Vec<u8> allocation
- Fix stale test assertions to match new 8KB threshold

Root cause analysis: the 16KiB regression was primarily caused by
instrumentation overhead in set_disk.rs (4x Instant::now() + metrics
per PUT), not BytesPool contention. Lowering the threshold eliminates
the eager-path overhead for 16KiB+ objects.

* perf(put): gate stage metrics behind observability flag

Add put_stage_metrics_enabled() AtomicBool switch in io-metrics crate.
When disabled (default), record_put_object_path() and
record_put_object_stage_duration() are no-ops, avoiding unnecessary
histogram/counter macro overhead in the PUT hot path.

The flag is set to true during startup when OTEL metric export is
enabled (rustfs_obs::observability_metric_enabled() == true).

This eliminates the per-request metrics overhead that contributed
to the 16KiB PUT regression when metrics collection is not active.

* perf(put): comprehensive optimization - restore eager path, cache env, remove UUID

Change 1: Restore SMALL_EAGER_PUT_MAX_SIZE from 8KB to 1MB
- The try_lock() fix (d13a189e3) eliminates the blocking that caused
  service health timeouts under 512KiB c64 load
- Eager path with BytesPool is now safe for objects up to 1MB
- Recovers the eager path benefit for 32KiB-256KiB objects

Change 2: Adjust POOL_BYPASE_MAX_SIZE from 16KB to 4KB
- With eager path restored to 1MB, objects 4KB-1MB benefit from pool reuse
- Only ≤4KB objects bypass the pool (allocation cost negligible)

Change 3: Cache RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES via OnceLock
- Eliminates per-encode std::env::var() syscall
- Env var still works (read once at first use)

Change 4: Replace Uuid::new_v4() with Uuid::nil() in Erasure construction
- _id field is unused in hot paths (documented in code)
- Eliminates CSPRNG syscall per PUT request

Change 5: Add concurrency-aware buffer sizing to PUT path
- Reuses get_concurrency_aware_buffer_size() from GET path
- Reduces buffer size under high concurrency (0.4x at >8 concurrent)
- Lowers memory pressure for >1MB streaming PUTs

* chore: add pyroscope feature flag and clean up imports

- Add pyroscope feature flag forwarding to rustfs-obs
- Remove unused allow(non_upper_case_globals) in globals.rs
- Sort imports and fix Cargo.toml formatting consistency

* style: fix import ordering and code formatting

- Sort imports alphabetically in globals.rs, encode.rs
- Fix indentation in erasure_coding encode/erasure
- Clean up HashReader formatting in object_usecase.rs

* fix(test): use tokio::test for request_logging_layer tests

The tests call tokio::spawn via RequestContextLayer, which requires a
Tokio runtime. Changed from #[test] + futures::executor::block_on to
#[tokio::test] + .await, and replaced tracing::subscriber::with_default
with tracing::subscriber::set_default to support async.

* fix(bench): normalize no-space throughput/latency parsing in to_bps/to_ms

When a benchmark tool prints throughput without a separator (e.g. 123MiB/s),
awk '{print $2}' returns empty because the whole string is one field,
causing to_bps to return N/A and losing valid measurements in CSV output.

Insert a space between number and unit via sed before awk field splitting.
Same fix applied to to_ms for latency values like '50ms'.

Also add TODO comment on PUT path noting that get_concurrency_aware_buffer_size
reads ACTIVE_GET_REQUESTS instead of PUT concurrency (PR #3514 review).

Refs: PR #3514 review comments by chatgpt-codex-connector

* fix(metrics): correct POOL_BYPASS comments and separate PUT vs generic stage metrics

- Fix 3 comment-code mismatches: POOL_BYPASS_MAX_SIZE is 4KiB, not 16KiB
- Add generic record_stage_duration() with separate histogram
  (rustfs_internal_stage_duration_ms) for non-PUT paths
- Replace record_put_object_stage_duration with record_stage_duration in
  metacache_set, store_list_objects, and bucket_lifecycle_ops to avoid
  polluting PUT-specific dashboards with listing/lifecycle timings
- Fix flaky test: serialize tests mutating PUT_STAGE_METRICS_ENABLED with
  METRICS_FLAG_LOCK mutex and explicitly set desired state at test start

Refs: PR #3514 review comments by chatgpt-codex-connector

* style: apply cargo fmt to metacache_set.rs

---------

Co-authored-by: cxymds <cxymds@gmail.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/common/src/globals.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(non_upper_case_globals)] // FIXME
-
 use chrono::{DateTime, Utc};
 use std::collections::HashMap;
 use std::sync::LazyLock;

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -1416,6 +1416,10 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>) {
                     );
                 }
                 Err(err) => {
+                    rustfs_io_metrics::record_stage_duration(
+                        "lifecycle_free_version_recovery_failed",
+                        started_at.elapsed().as_secs_f64() * 1000.0,
+                    );
                     warn!(
                         event = EVENT_LIFECYCLE_WORKER_STATE,
                         component = LOG_COMPONENT_ECSTORE,

crates/ecstore/src/cache_value/metacache_set.rs

@@ -184,9 +184,19 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
             let mut need_fallback = false;
             let mut last_err = None;
             if let Some(disk) = opdisk {
+                let primary_walk_started = std::time::Instant::now();
                 match disk.walk_dir(wakl_opts, &mut wr).await {
-                    Ok(_res) => {}
+                    Ok(_res) => {
+                        rustfs_io_metrics::record_stage_duration(
+                            "metacache_walk_dir_primary",
+                            primary_walk_started.elapsed().as_secs_f64() * 1000.0,
+                        );
+                    }
                     Err(err) => {
+                        rustfs_io_metrics::record_stage_duration(
+                            "metacache_walk_dir_primary_failed",
+                            primary_walk_started.elapsed().as_secs_f64() * 1000.0,
+                        );
                         warn!(
                             event = EVENT_METACACHE_LISTING,
                             component = LOG_COMPONENT_ECSTORE,
@@ -237,6 +247,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                     return Err(err);
                 };
 
+                let fallback_walk_started = std::time::Instant::now();
                 match disk
                     .as_ref()
                     .walk_dir(
@@ -256,10 +267,18 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
                     .await
                 {
                     Ok(_r) => {
+                        rustfs_io_metrics::record_stage_duration(
+                            "metacache_walk_dir_fallback",
+                            fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
+                        );
                         need_fallback = false;
                         last_err = None;
                     }
                     Err(err) => {
+                        rustfs_io_metrics::record_stage_duration(
+                            "metacache_walk_dir_fallback_failed",
+                            fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
+                        );
                         error!(
                             event = EVENT_METACACHE_LISTING,
                             component = LOG_COMPONENT_ECSTORE,
@@ -538,7 +557,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
         Ok(())
     });
 
+    let merge_started = std::time::Instant::now();
     if let Err(err) = revjob.await.map_err(std::io::Error::other)? {
+        rustfs_io_metrics::record_stage_duration("metacache_merge_failed", merge_started.elapsed().as_secs_f64() * 1000.0);
         error!(
             event = EVENT_METACACHE_LISTING,
             component = LOG_COMPONENT_ECSTORE,
@@ -556,6 +577,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d
 
         return Err(err);
     }
+    rustfs_io_metrics::record_stage_duration("metacache_merge", merge_started.elapsed().as_secs_f64() * 1000.0);
 
     // The merge consumer can finish successfully before every producer finishes
     // (for example after reaching EOF quorum while a tolerated drive is stalled,

crates/ecstore/src/erasure_coding/encode.rs

@@ -31,6 +31,10 @@ const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCOD
 const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 32 * 1024 * 1024;
 const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS: usize = 32;
 
+/// Cached value of `RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES` env var.
+/// Read once at first use via `OnceLock` to avoid per-encode syscall.
+static CACHED_MAX_INFLIGHT_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+
 fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize {
     if expanded_block_bytes == 0 {
         return 1;
@@ -273,10 +277,12 @@ impl Erasure {
 
         // Bound queued encoded blocks by memory budget to avoid per-request spikes.
         let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count());
-        let max_inflight_bytes = rustfs_utils::get_env_usize(
-            ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
-            DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
-        );
+        let max_inflight_bytes = *CACHED_MAX_INFLIGHT_BYTES.get_or_init(|| {
+            rustfs_utils::get_env_usize(
+                ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
+                DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES,
+            )
+        });
         let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
         let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(inflight_blocks);
 
@@ -443,7 +449,7 @@ mod tests {
         ))];
 
         let erasure = Arc::new(Erasure::new(1, 0, 16));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"small payload".to_vec()));
+        let reader = tokio::io::BufReader::new(Cursor::new(b"small payload".to_vec()));
         let (_reader, written) = erasure.encode(reader, &mut writers, 1).await.unwrap();
 
         assert_eq!(written, b"small payload".len());
@@ -482,7 +488,7 @@ mod tests {
         ))];
 
         let erasure = Arc::new(Erasure::new(1, 0, 0));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"payload".to_vec()));
+        let reader = tokio::io::BufReader::new(Cursor::new(b"payload".to_vec()));
         let err = erasure
             .encode(reader, &mut writers, 1)
             .await
@@ -505,7 +511,7 @@ mod tests {
         ))];
 
         let erasure = Arc::new(Erasure::new(1, 0, 16));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(Vec::<u8>::new()));
+        let reader = tokio::io::BufReader::new(Cursor::new(Vec::<u8>::new()));
         let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, 1).await.unwrap();
 
         assert_eq!(total, 0);
@@ -537,7 +543,7 @@ mod tests {
 
         let payload = b"hello inline small";
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec()));
+        let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec()));
         let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, DATA_SHARDS).await.unwrap();
 
         assert_eq!(total, payload.len());
@@ -569,7 +575,7 @@ mod tests {
 
         let payload = b"hello single block";
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec()));
+        let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec()));
         let (_reader, total) = erasure
             .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS)
             .await
@@ -603,7 +609,7 @@ mod tests {
 
         let payload = vec![1u8; BLOCK_SIZE + 1];
         let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
-        let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload));
+        let reader = tokio::io::BufReader::new(Cursor::new(payload));
         let err = erasure
             .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS)
             .await

crates/ecstore/src/erasure_coding/erasure.rs

@@ -394,7 +394,7 @@ impl Erasure {
             encoder,
             legacy_encoder,
             uses_legacy,
-            _id: Uuid::new_v4(),
+            _id: Uuid::nil(), // Unused in hot paths; avoid CSPRNG syscall
         }
     }
 
@@ -631,15 +631,11 @@ impl Erasure {
     ) -> Result<usize, E>
     where
         R: AsyncRead + Send + Sync + Unpin,
-        F: FnMut(std::io::Result<Vec<Bytes>>) -> Fut + Send,
-        Fut: std::future::Future<Output = Result<(), E>> + Send,
+        F: FnMut(io::Result<Vec<Bytes>>) -> Fut + Send,
+        Fut: Future<Output = Result<(), E>> + Send,
     {
         if self.block_size == 0 {
-            on_block(Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "erasure block_size must be non-zero",
-            )))
-            .await?;
+            on_block(Err(io::Error::new(io::ErrorKind::InvalidInput, "erasure block_size must be non-zero"))).await?;
             return Ok(0);
         }
 
@@ -662,7 +658,7 @@ impl Erasure {
                     {
                         Ok(result) => result,
                         Err(err) => {
-                            on_block(Err(std::io::Error::other(format!("EC encode task failed: {err}")))).await?;
+                            on_block(Err(io::Error::other(format!("EC encode task failed: {err}")))).await?;
                             break;
                         }
                     };
@@ -673,7 +669,7 @@ impl Erasure {
                     warn!("encode_stream_callback_async read unexpected ok");
                     break;
                 }
-                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                     warn!("encode_stream_callback_async read unexpected eof");
                     break;
                 }
@@ -690,7 +686,6 @@ impl Erasure {
 
 #[cfg(test)]
 mod tests {
-
     use super::*;
 
     fn optional_shards(shards: &[Bytes]) -> Vec<Option<Vec<u8>>> {
@@ -1140,7 +1135,7 @@ mod tests {
         assert_eq!(total, 0);
         let observed = observed.lock().unwrap();
         let (kind, message) = observed.as_ref().expect("callback should be invoked once");
-        assert_eq!(*kind, std::io::ErrorKind::InvalidInput);
+        assert_eq!(*kind, io::ErrorKind::InvalidInput);
         assert!(message.contains("block_size"));
     }
 

crates/ecstore/src/set_disk.rs

@@ -137,6 +137,8 @@ const LOG_SUBSYSTEM_SET_DISK: &str = "set_disk";
 const EVENT_SET_DISK_MULTIPART: &str = "set_disk_multipart";
 const EVENT_SET_DISK_WRITE: &str = "set_disk_write";
 const EVENT_SET_DISK_HEAL: &str = "set_disk_heal";
+const EVENT_SET_DISK_COMMIT_TAIL_SLOW: &str = "set_disk_commit_tail_slow";
+const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000;
 
 use crate::rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _};
 
@@ -804,6 +806,16 @@ enum SmallWritePath {
     Pipeline,
 }
 
+impl SmallWritePath {
+    fn metric_label(&self) -> &'static str {
+        match self {
+            SmallWritePath::Inline => "write_inline",
+            SmallWritePath::SingleBlockNonInline => "write_single_block_non_inline",
+            SmallWritePath::Pipeline => "write_pipeline",
+        }
+    }
+}
+
 fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
     if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
         SmallWritePath::Inline
@@ -1061,6 +1073,7 @@ impl ObjectIO for SetDisks {
 
             let shard_file_size = erasure.shard_file_size(data.size());
             let shard_size = erasure.shard_size();
+            let writer_setup_stage_start = Instant::now();
             let writer_futs: Vec<_> = shuffle_disks
                 .iter()
                 .map(|disk_op| {
@@ -1107,6 +1120,10 @@ impl ObjectIO for SetDisks {
                 writers.push(w);
                 errors.push(e);
             }
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "set_disk_writer_setup",
+                writer_setup_stage_start.elapsed().as_secs_f64() * 1000.0,
+            );
 
             let nil_count = errors.iter().filter(|&e| e.is_none()).count();
             if nil_count < write_quorum {
@@ -1135,7 +1152,9 @@ impl ObjectIO for SetDisks {
             );
 
             let write_path = classify_small_write_path(is_inline_buffer, data.size(), fi.erasure.block_size);
+            rustfs_io_metrics::record_put_object_path(write_path.metric_label());
 
+            let encode_stage_start = Instant::now();
             let (reader, w_size) = match write_path {
                 SmallWritePath::Inline => match Arc::new(erasure)
                     .encode_inline_small(stream, &mut writers, write_quorum)
@@ -1165,6 +1184,10 @@ impl ObjectIO for SetDisks {
                     }
                 },
             };
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "set_disk_encode",
+                encode_stage_start.elapsed().as_secs_f64() * 1000.0,
+            );
 
             let _ = mem::replace(&mut data.stream, reader);
             // if let Err(err) = close_bitrot_writers(&mut writers).await {
@@ -1259,6 +1282,7 @@ impl ObjectIO for SetDisks {
                 object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?);
             }
 
+            let rename_stage_start = Instant::now();
             let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data(
                 &shuffle_disks,
                 RUSTFS_META_TMP_BUCKET,
@@ -1269,10 +1293,52 @@ impl ObjectIO for SetDisks {
                 write_quorum,
             )
             .await?;
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "set_disk_rename",
+                rename_stage_start.elapsed().as_secs_f64() * 1000.0,
+            );
+            let rename_stage_ms = rename_stage_start.elapsed().as_millis();
+            if rename_stage_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                warn!(
+                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    stage = "rename_data",
+                    bucket = %bucket,
+                    object = %object,
+                    tmp_dir = %tmp_dir,
+                    duration_ms = rename_stage_ms as u64,
+                    write_quorum,
+                    state = "slow",
+                    "SetDisk commit tail stage is slow"
+                );
+            }
 
             if let Some(old_dir) = op_old_dir {
+                let cleanup_stage_start = Instant::now();
                 self.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), write_quorum)
                     .await?;
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "set_disk_old_data_cleanup",
+                    cleanup_stage_start.elapsed().as_secs_f64() * 1000.0,
+                );
+                let cleanup_stage_ms = cleanup_stage_start.elapsed().as_millis();
+                if cleanup_stage_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                    warn!(
+                        event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        stage = "commit_rename_data_dir",
+                        bucket = %bucket,
+                        object = %object,
+                        tmp_dir = %tmp_dir,
+                        old_dir = %old_dir,
+                        duration_ms = cleanup_stage_ms as u64,
+                        write_quorum,
+                        state = "slow",
+                        "SetDisk commit tail stage is slow"
+                    );
+                }
             }
 
             drop(object_lock_guard); // drop object lock guard to release the lock
@@ -1307,6 +1373,23 @@ impl ObjectIO for SetDisks {
                 );
             }
 
+            let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
+            if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                warn!(
+                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    stage = "put_object_commit_tail",
+                    bucket = %bucket,
+                    object = %object,
+                    tmp_dir = %tmp_dir,
+                    duration_ms = total_commit_tail_ms as u64,
+                    write_quorum,
+                    state = "slow",
+                    "SetDisk commit tail is slow"
+                );
+            }
+
             Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
         }
         .await;

crates/ecstore/src/store_list_objects.rs

@@ -1158,6 +1158,7 @@ impl ECStore {
 
         tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await }.instrument(tracing::Span::current()));
 
+        let walk_started = std::time::Instant::now();
         let walk_results = join_all(futures).await;
         let mut errs = Vec::new();
         for walk_result in walk_results {
@@ -1166,6 +1167,10 @@ impl ECStore {
                 Err(err) => errs.push(Some(err.into())),
             }
         }
+        rustfs_io_metrics::record_stage_duration(
+            "store_list_objects_walk_internal",
+            walk_started.elapsed().as_secs_f64() * 1000.0,
+        );
 
         let result = walk_result_from_set_errors(&errs);
         if let Err(err) = &result {

crates/io-core/src/pool.rs

@@ -423,23 +423,31 @@ impl PoolTier {
         })
     }
 
-    /// Return a buffer to the pool for reuse.
+    /// Return a buffer to the pool for reuse without ever blocking the caller.
     fn return_buffer(&self, buffer: BytesMut) {
-        let mut available = self.available_buffers.lock().unwrap_or_else(|e| e.into_inner());
-        // Limit the size of the pool to prevent unbounded growth
-        if available.len() < self.max_buffers {
-            available.push(buffer);
-            if let Some(ref metrics) = *self.metrics.lock().unwrap_or_else(|e| e.into_inner()) {
+        let mut buffer = Some(buffer);
+
+        if let Ok(mut available) = self.available_buffers.try_lock()
+            && available.len() < self.max_buffers
+        {
+            available.push(buffer.take().expect("buffer should be present until returned"));
+            if let Ok(metrics) = self.metrics.try_lock()
+                && let Some(metrics) = metrics.as_ref()
+            {
                 metrics.available_buffers.fetch_add(1, Ordering::Relaxed);
             }
-        } else {
+        }
+
+        if let Some(buffer) = buffer {
             let released_bytes = buffer.capacity() as u64;
             self.tier_current_allocated_bytes
                 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
                     Some(current.saturating_sub(released_bytes))
                 })
                 .ok();
-            if let Some(ref metrics) = *self.metrics.lock().unwrap_or_else(|e| e.into_inner()) {
+            if let Ok(metrics) = self.metrics.try_lock()
+                && let Some(metrics) = metrics.as_ref()
+            {
                 metrics
                     .current_allocated_bytes
                     .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
@@ -448,7 +456,6 @@ impl PoolTier {
                     .ok();
             }
         }
-        // If pool is full, buffer is dropped and memory is freed
         rustfs_io_metrics::record_bytes_pool_allocated(self.name, self.tier_current_allocated_bytes.load(Ordering::Relaxed));
     }
 }
@@ -458,8 +465,6 @@ impl Drop for PooledBuffer {
     // buffer moves it exactly once into the pool when a tier still owns it.
     #[allow(unsafe_code)]
     fn drop(&mut self) {
-        // Return buffer to pool if tier reference exists.
-        // Otherwise, drop the standalone fallback buffer normally.
         let buffer = unsafe { ManuallyDrop::take(&mut self.buffer) };
         if let Some(ref tier) = self.tier {
             tier.return_buffer(buffer);

crates/io-metrics/src/lib.rs

@@ -49,7 +49,30 @@
 #[macro_use]
 extern crate metrics;
 
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+
+/// Global switch for detailed per-stage PUT metrics (path label, stage durations).
+/// When `false`, `record_put_object_path` and `record_put_object_stage_duration`
+/// become no-ops, and callers can skip the `Instant::now()` syscalls entirely.
+///
+/// Set to `true` during startup when OTEL metric export is enabled.
+static PUT_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
+
+/// Enable or disable detailed per-stage PUT metrics.
+///
+/// Called once during startup, typically gated by `rustfs_obs::observability_metric_enabled()`.
+pub fn set_put_stage_metrics_enabled(enabled: bool) {
+    PUT_STAGE_METRICS_ENABLED.store(enabled, Ordering::Relaxed);
+}
+
+/// Returns `true` if detailed per-stage PUT metrics are enabled.
+///
+/// Callers should check this before calling `Instant::now()` for stage timing
+/// to avoid unnecessary syscalls when metrics are disabled.
+#[inline(always)]
+pub fn put_stage_metrics_enabled() -> bool {
+    PUT_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
+}
 
 // Public modules
 pub mod adaptive_ttl;
@@ -409,9 +432,9 @@ pub fn record_get_object(duration_ms: f64, size_bytes: i64) {
 ///
 /// * `duration_ms` - Operation duration in milliseconds
 /// * `size_bytes` - Object size in bytes
-/// * `zero_copy_enabled` - Whether zero-copy was enabled for this operation
+/// * `zero_copy_eligible` - Whether the request was eligible for a zero-copy path
 #[inline(always)]
-pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: bool) {
+pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_eligible: bool) {
     counter!("rustfs_s3_put_object_total").increment(1);
     histogram!("rustfs_s3_put_object_duration_ms").record(duration_ms);
 
@@ -419,11 +442,37 @@ pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: b
         histogram!("rustfs_s3_put_object_size_bytes").record(size_bytes as f64);
     }
 
-    if zero_copy_enabled {
+    if zero_copy_eligible {
+        // Backward-compatible alias for historical dashboards.
         counter!("rustfs_s3_put_object_zero_copy_enabled_total").increment(1);
+        counter!("rustfs_s3_put_object_zero_copy_eligible_total").increment(1);
     }
 }
 
+#[inline(always)]
+pub fn record_put_object_path(path: &'static str) {
+    if !put_stage_metrics_enabled() {
+        return;
+    }
+    counter!("rustfs_s3_put_object_path_total", "path" => path).increment(1);
+}
+
+#[inline(always)]
+pub fn record_put_object_stage_duration(stage: &'static str, duration_ms: f64) {
+    if !put_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_s3_put_object_stage_duration_ms", "stage" => stage).record(duration_ms);
+}
+
+/// Record generic internal operation stage duration (non-PUT paths).
+/// Use this for metacache walks, listing, lifecycle, and other background
+/// operations that are NOT part of the PUT object hot path.
+#[inline(always)]
+pub fn record_stage_duration(stage: &'static str, duration_ms: f64) {
+    histogram!("rustfs_internal_stage_duration_ms", "stage" => stage).record(duration_ms);
+}
+
 /// Record ListObjects operation metrics.
 ///
 /// # Arguments
@@ -788,6 +837,10 @@ pub fn record_io_latency_p99(latency_ms: f64) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::Mutex;
+
+    // Serialize tests that mutate the process-global PUT_STAGE_METRICS_ENABLED flag.
+    static METRICS_FLAG_LOCK: Mutex<()> = Mutex::new(());
 
     #[test]
     fn test_record_zero_copy_read() {
@@ -824,6 +877,36 @@ mod tests {
         record_put_object(100.0, 512, false);
     }
 
+    #[test]
+    fn test_record_put_object_path_and_stage() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_put_stage_metrics_enabled(true);
+        record_put_object_path("small_eager");
+        record_put_object_path("write_inline");
+        record_put_object_stage_duration("ingress_prepare", 12.5);
+        record_put_object_stage_duration("set_disk_encode", 8.0);
+        set_put_stage_metrics_enabled(false);
+    }
+
+    #[test]
+    fn test_put_stage_metrics_disabled_by_default() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_put_stage_metrics_enabled(false);
+        // These should be no-ops (no panic, no recording)
+        record_put_object_path("small_eager");
+        record_put_object_stage_duration("set_disk_encode", 5.0);
+        // Still disabled
+        assert!(!put_stage_metrics_enabled());
+    }
+
+    #[test]
+    fn test_record_stage_duration_generic() {
+        // Generic stage duration should always record (no gating flag)
+        record_stage_duration("metacache_walk_dir_primary", 15.0);
+        record_stage_duration("store_list_objects_walk_internal", 8.5);
+        record_stage_duration("lifecycle_free_version_recovery_failed", 120.0);
+    }
+
     #[test]
     fn test_record_list_objects() {
         record_list_objects(50.0, 100, false);

crates/object-capacity/src/scan.rs

@@ -556,6 +556,7 @@ async fn get_dir_size_async(path: &Path) -> Result<CapacityScanResult, std::io::
                         subsystem = LOG_SUBSYSTEM_SCAN,
                         result = "partial",
                         root_path = ?path,
+                        file_count,
                         error = %err,
                         "capacity scan traversal failed"
                     );
@@ -590,6 +591,7 @@ async fn get_dir_size_async(path: &Path) -> Result<CapacityScanResult, std::io::
                         subsystem = LOG_SUBSYSTEM_SCAN,
                         result = "partial",
                         entry_path = ?entry.path(),
+                        file_count,
                         error = %err,
                         "capacity scan metadata failed"
                     );

rustfs/Cargo.toml

@@ -55,6 +55,7 @@ tracing-chunk-debug = []  # Enable per-chunk tracing in data plane (high noise,
 full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp"]
 manual-test-runners = []
 rio-v2 = ["rustfs-ecstore/rio-v2"]
+pyroscope = ["rustfs-obs/pyroscope"]
 
 [lints]
 workspace = true
@@ -79,7 +80,7 @@ rustfs-obs = { workspace = true }
 rustfs-policy = { workspace = true }
 rustfs-protocols = { workspace = true }
 rustfs-protos = { workspace = true }
-rustfs-rio.workspace = true
+rustfs-rio = { workspace = true }
 rustfs-s3-types = { workspace = true }
 rustfs-s3-ops = { workspace = true }
 rustfs-security-governance = { workspace = true }

rustfs/src/app/object_usecase.rs

@@ -86,6 +86,7 @@ use rustfs_filemeta::{
     ReplicationType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map,
     version_purge_statuses_map,
 };
+use rustfs_io_core::{BytesPool, PooledBuffer};
 use rustfs_io_metrics;
 use rustfs_lock::NamespaceLockGuard;
 use rustfs_notify::EventArgsBuilder;
@@ -118,6 +119,8 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use std::collections::HashMap;
 use std::ops::Add;
 use std::path::{Component, Path};
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -133,6 +136,11 @@ use uuid::Uuid;
 
 const ACCEPT_RANGES_BYTES: &str = "bytes";
 const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024;
+const LOG_COMPONENT_APP: &str = "app";
+const LOG_SUBSYSTEM_OBJECT: &str = "object";
+const EVENT_PUT_OBJECT_STORE_INFLIGHT_SLOW: &str = "put_object_store_inflight_slow";
+const EVENT_PUT_OBJECT_STORE_RETURNED: &str = "put_object_store_returned";
+const PUT_OBJECT_STORE_WARN_THRESHOLD: Duration = Duration::from_secs(5);
 static GET_OBJECT_BUFFER_THRESHOLD_WARNED: AtomicBool = AtomicBool::new(false);
 
 fn decoded_content_length_from_headers(headers: &HeaderMap) -> S3Result<Option<i64>> {
@@ -374,6 +382,33 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
     }
 }
 
+struct PooledBufferReader {
+    buffer: PooledBuffer,
+    len: usize,
+    pos: usize,
+}
+
+impl PooledBufferReader {
+    fn new(buffer: PooledBuffer, len: usize) -> Self {
+        Self { buffer, len, pos: 0 }
+    }
+}
+
+impl AsyncRead for PooledBufferReader {
+    fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        if self.pos >= self.len {
+            return Poll::Ready(Ok(()));
+        }
+
+        let remaining = self.len - self.pos;
+        let to_read = remaining.min(buf.remaining());
+        buf.put_slice(&self.buffer[self.pos..self.pos + to_read]);
+        self.pos += to_read;
+
+        Poll::Ready(Ok(()))
+    }
+}
+
 /// Determine if zero-copy write should be used for this PutObject operation.
 ///
 /// Zero-copy is beneficial for large objects without encryption or compression.
@@ -429,6 +464,105 @@ fn should_use_zero_copy(size: i64, headers: &HeaderMap) -> bool {
     true
 }
 
+fn has_put_sse_request_headers(headers: &HeaderMap) -> bool {
+    headers.get(AMZ_SERVER_SIDE_ENCRYPTION).is_some()
+        || headers.get(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM).is_some()
+        || headers.get(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID).is_some()
+}
+
+fn should_use_small_eager_put_path(
+    size: i64,
+    headers: &HeaderMap,
+    server_side_encryption_requested: bool,
+    should_compress: bool,
+    is_extract: bool,
+) -> bool {
+    const SMALL_EAGER_PUT_MAX_SIZE: i64 = 1024 * 1024;
+
+    if is_extract || should_compress || server_side_encryption_requested {
+        return false;
+    }
+
+    if size <= 0 || size > SMALL_EAGER_PUT_MAX_SIZE {
+        return false;
+    }
+
+    if has_put_sse_request_headers(headers) {
+        return false;
+    }
+
+    if request_uses_aws_chunked(headers) && decoded_content_length_from_headers(headers).ok().flatten().is_none() {
+        return false;
+    }
+
+    true
+}
+
+/// Objects at or below this size bypass BytesPool and use direct allocation.
+/// This avoids Small-tier Mutex contention under high concurrency for tiny objects
+/// where the allocation cost is negligible (≤4KiB memcpy).
+const POOL_BYPASS_MAX_SIZE: usize = 4 * 1024;
+
+async fn read_small_put_body_exact_pooled<R>(mut body: R, size: usize, pool: &BytesPool) -> S3Result<PooledBuffer>
+where
+    R: AsyncRead + Unpin,
+{
+    let mut buf = pool.acquire_buffer(size).await;
+    buf.resize(size, 0);
+    let mut filled = 0;
+
+    while filled < size {
+        let read = tokio::io::AsyncReadExt::read(&mut body, &mut buf[filled..size])
+            .await
+            .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?;
+        if read == 0 {
+            return Err(s3_error!(IncompleteBody));
+        }
+        filled += read;
+    }
+
+    let mut extra = [0u8; 1];
+    let extra_read = tokio::io::AsyncReadExt::read(&mut body, &mut extra)
+        .await
+        .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?;
+    if extra_read != 0 {
+        return Err(s3_error!(UnexpectedContent));
+    }
+
+    Ok(buf)
+}
+
+/// Read small PUT body into a directly-allocated buffer, bypassing BytesPool.
+/// Used for objects ≤4KiB where pool contention under high concurrency
+/// outweighs the allocation cost.
+async fn read_small_put_body_exact_direct<R>(mut body: R, size: usize) -> S3Result<std::io::Cursor<Vec<u8>>>
+where
+    R: AsyncRead + Unpin,
+{
+    let mut buf = vec![0u8; size];
+    let mut filled = 0;
+
+    while filled < size {
+        let read = tokio::io::AsyncReadExt::read(&mut body, &mut buf[filled..size])
+            .await
+            .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?;
+        if read == 0 {
+            return Err(s3_error!(IncompleteBody));
+        }
+        filled += read;
+    }
+
+    let mut extra = [0u8; 1];
+    let extra_read = tokio::io::AsyncReadExt::read(&mut body, &mut extra)
+        .await
+        .map_err(|err| ApiError::from(StorageError::other(err.to_string())))?;
+    if extra_read != 0 {
+        return Err(s3_error!(UnexpectedContent));
+    }
+
+    Ok(std::io::Cursor::new(buf))
+}
+
 fn object_seek_support_threshold() -> usize {
     static OBJECT_SEEK_SUPPORT_THRESHOLD: OnceLock<usize> = OnceLock::new();
     *OBJECT_SEEK_SUPPORT_THRESHOLD.get_or_init(|| {
@@ -1814,10 +1948,21 @@ impl DefaultObjectUsecase {
             return Err(s3_error!(UnexpectedContent));
         }
 
+        let ingress_stage_start = std::time::Instant::now();
+        let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64;
+        let server_side_encryption_requested =
+            server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some();
+
         // Apply adaptive buffer sizing based on file size for optimal streaming performance.
         // Uses workload profile configuration (enabled by default) to select appropriate buffer size.
         // Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile.
-        let buffer_size = get_buffer_size_opt_in(size);
+        // Concurrency-aware adjustment reduces buffer size under high concurrency to lower memory pressure.
+        // TODO: get_concurrency_aware_buffer_size reads ACTIVE_GET_REQUESTS (GET concurrency tracker),
+        // not PUT concurrency. Under pure PUT load the counter stays zero so buffers never shrink;
+        // unrelated GET load can shrink PUT buffers instead. Fix by adding ACTIVE_PUT_REQUESTS +
+        // PutObjectGuard and using PUT concurrency here. See PR #3514 review comment.
+        let base_buffer_size = get_buffer_size_opt_in(size);
+        let buffer_size = get_concurrency_aware_buffer_size(size, base_buffer_size);
 
         // Detect zero-copy opportunity before encryption/compression decisions
         // Zero-copy is beneficial for large unencrypted, uncompressed objects
@@ -1830,10 +1975,15 @@ impl DefaultObjectUsecase {
             debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key);
         }
 
-        let body = tokio::io::BufReader::with_capacity(
-            buffer_size,
-            StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
-        );
+        let use_small_eager_put_path =
+            should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false);
+        let put_path = if should_compress {
+            "stream_compressed"
+        } else if use_small_eager_put_path {
+            "small_eager"
+        } else {
+            "streaming"
+        };
 
         let store = get_validated_store(&bucket).await?;
 
@@ -1955,9 +2105,12 @@ impl DefaultObjectUsecase {
 
         let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query());
 
-        let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64;
         let mut write_plan = WritePlan::new();
         let mut reader = if should_compress {
+            let body = tokio::io::BufReader::with_capacity(
+                buffer_size,
+                StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
+            );
             let algorithm = CompressionAlgorithm::default();
             insert_str(
                 &mut metadata,
@@ -1985,7 +2138,35 @@ impl DefaultObjectUsecase {
             write_plan = write_plan.with_compression(algorithm);
             hrd
         } else {
-            HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
+            if use_small_eager_put_path {
+                if (actual_size as usize) <= POOL_BYPASS_MAX_SIZE {
+                    // Bypass BytesPool for very small objects to avoid Small-tier
+                    // Mutex contention under high concurrency. Direct allocation
+                    // for ≤4KiB is negligible cost.
+                    let eager_body = read_small_put_body_exact_direct(
+                        StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
+                        actual_size as usize,
+                    )
+                    .await?;
+                    HashReader::from_stream(eager_body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
+                } else {
+                    let pool = get_concurrency_manager().bytes_pool();
+                    let eager_body = read_small_put_body_exact_pooled(
+                        StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
+                        actual_size as usize,
+                        pool.as_ref(),
+                    )
+                    .await?;
+                    let eager_reader = PooledBufferReader::new(eager_body, actual_size as usize);
+                    HashReader::from_stream(eager_reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
+                }
+            } else {
+                let body = tokio::io::BufReader::with_capacity(
+                    buffer_size,
+                    StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
+                );
+                HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
+            }
         };
 
         if size >= 0 {
@@ -1995,6 +2176,11 @@ impl DefaultObjectUsecase {
 
             opts.want_checksum = reader.checksum();
         }
+        rustfs_io_metrics::record_put_object_path(put_path);
+        rustfs_io_metrics::record_put_object_stage_duration(
+            "ingress_prepare",
+            ingress_stage_start.elapsed().as_secs_f64() * 1000.0,
+        );
 
         let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject);
         let ssekms_context = extract_ssekms_context_from_headers(&req.headers)?;
@@ -2038,6 +2224,11 @@ impl DefaultObjectUsecase {
 
         let mt2 = metadata.clone();
         opts.user_defined.extend(metadata);
+        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
+        let request_id = request_context
+            .as_ref()
+            .map(|ctx| ctx.request_id.clone())
+            .unwrap_or_else(|| request_context::RequestContext::fallback().request_id);
 
         let repoptions =
             get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone());
@@ -2052,13 +2243,76 @@ impl DefaultObjectUsecase {
             );
         }
 
+        let store_put_watchdog = tokio_util::sync::CancellationToken::new();
+        spawn_traced({
+            let store_put_watchdog = store_put_watchdog.clone();
+            let request_id = request_id.clone();
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let put_path = put_path.to_string();
+            async move {
+                tokio::select! {
+                    _ = store_put_watchdog.cancelled() => {}
+                    _ = tokio::time::sleep(PUT_OBJECT_STORE_WARN_THRESHOLD) => {
+                        warn!(
+                            target: "rustfs::app::object_usecase",
+                            event = EVENT_PUT_OBJECT_STORE_INFLIGHT_SLOW,
+                            component = LOG_COMPONENT_APP,
+                            subsystem = LOG_SUBSYSTEM_OBJECT,
+                            request_id = %request_id,
+                            bucket = %bucket,
+                            key = %key,
+                            put_path = %put_path,
+                            object_size = actual_size,
+                            threshold_ms = PUT_OBJECT_STORE_WARN_THRESHOLD.as_millis() as u64,
+                            state = "store_put_pending",
+                            "PutObject store write remains in flight"
+                        );
+                    }
+                }
+            }
+        });
+
         let obj_info = match store
             .put_object(&bucket, &key, &mut reader, &opts)
             .await
             .map_err(ApiError::from)
         {
-            Ok(obj_info) => obj_info,
+            Ok(obj_info) => {
+                store_put_watchdog.cancel();
+                debug!(
+                    target: "rustfs::app::object_usecase",
+                    event = EVENT_PUT_OBJECT_STORE_RETURNED,
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_OBJECT,
+                    request_id = %request_id,
+                    bucket = %bucket,
+                    key = %key,
+                    put_path = put_path,
+                    object_size = actual_size,
+                    duration_ms = start_time.elapsed().as_millis() as u64,
+                    result = "success",
+                    "PutObject store write returned"
+                );
+                obj_info
+            }
             Err(err) => {
+                store_put_watchdog.cancel();
+                warn!(
+                    target: "rustfs::app::object_usecase",
+                    event = EVENT_PUT_OBJECT_STORE_RETURNED,
+                    component = LOG_COMPONENT_APP,
+                    subsystem = LOG_SUBSYSTEM_OBJECT,
+                    request_id = %request_id,
+                    bucket = %bucket,
+                    key = %key,
+                    put_path = put_path,
+                    object_size = actual_size,
+                    duration_ms = start_time.elapsed().as_millis() as u64,
+                    result = "error",
+                    error = %err,
+                    "PutObject store write returned"
+                );
                 let result: S3Result<S3Response<PutObjectOutput>> = Err(err.into());
                 let _ = helper.complete(&result);
                 return result;
@@ -5056,6 +5310,86 @@ mod tests {
         assert!(!should_use_zero_copy(2 * 1024 * 1024, &headers));
     }
 
+    #[test]
+    fn should_use_small_eager_put_path_allows_up_to_1mb() {
+        let headers = HeaderMap::new();
+
+        assert!(should_use_small_eager_put_path(1024, &headers, false, false, false));
+        assert!(should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false));
+        assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false));
+    }
+
+    #[test]
+    fn should_use_small_eager_put_path_rejects_sse_requests() {
+        let headers = HeaderMap::new();
+
+        assert!(!should_use_small_eager_put_path(1024, &headers, true, false, false));
+    }
+
+    #[test]
+    fn should_use_small_eager_put_path_rejects_compressible_objects() {
+        let headers = HeaderMap::new();
+
+        assert!(!should_use_small_eager_put_path(1024, &headers, false, true, false));
+    }
+
+    #[test]
+    fn should_use_small_eager_put_path_rejects_extract_requests() {
+        let headers = HeaderMap::new();
+
+        assert!(!should_use_small_eager_put_path(1024, &headers, false, false, true));
+    }
+
+    #[test]
+    fn should_use_small_eager_put_path_rejects_large_or_empty_objects() {
+        let headers = HeaderMap::new();
+
+        assert!(!should_use_small_eager_put_path(0, &headers, false, false, false));
+        assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false));
+    }
+
+    #[tokio::test]
+    async fn read_small_put_body_exact_pooled_reads_exact_bytes() {
+        let pool = get_concurrency_manager().bytes_pool();
+        let body = std::io::Cursor::new(b"hello".to_vec());
+
+        let buffer = read_small_put_body_exact_pooled(body, 5, pool.as_ref())
+            .await
+            .expect("pooled exact read should succeed");
+
+        assert_eq!(&buffer[..5], b"hello");
+    }
+
+    #[tokio::test]
+    async fn read_small_put_body_exact_pooled_rejects_short_body() {
+        let pool = get_concurrency_manager().bytes_pool();
+        let body = std::io::Cursor::new(b"hell".to_vec());
+
+        let err = match read_small_put_body_exact_pooled(body, 5, pool.as_ref()).await {
+            Ok(_) => panic!("short pooled body should fail"),
+            Err(err) => err,
+        };
+
+        assert_eq!(err.code(), &S3ErrorCode::IncompleteBody);
+    }
+
+    #[tokio::test]
+    async fn pooled_buffer_reader_keeps_buffer_alive_until_consumed() {
+        use tokio::io::AsyncReadExt;
+
+        let pool = get_concurrency_manager().bytes_pool();
+        let body = std::io::Cursor::new(b"hello".to_vec());
+        let buffer = read_small_put_body_exact_pooled(body, 5, pool.as_ref())
+            .await
+            .expect("pooled exact read should succeed");
+        let mut reader = PooledBufferReader::new(buffer, 5);
+        let mut out = Vec::new();
+
+        reader.read_to_end(&mut out).await.expect("pooled reader should be readable");
+
+        assert_eq!(out, b"hello");
+    }
+
     #[test]
     fn should_use_zero_copy_allows_large_unencrypted_binary_objects() {
         let mut headers = HeaderMap::new();

rustfs/src/server/http.rs

@@ -275,13 +275,31 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
 
         // Helper to configure socket with optimized parameters
         let configure_socket = |socket: &socket2::Socket| -> Result<()> {
-            socket.set_reuse_address(true)?;
+            if let Err(e) = socket.set_reuse_address(true) {
+                debug!(
+                    event = "socket_option_unavailable",
+                    component = LOG_COMPONENT_SERVER,
+                    subsystem = LOG_SUBSYSTEM_STARTUP,
+                    option = "SO_REUSEADDR",
+                    error = %e,
+                    "Socket option is unavailable"
+                );
+            }
 
             // Set the socket to non-blocking before passing it to Tokio.
             socket.set_nonblocking(true)?;
 
             // 1. Disable Nagle algorithm: Critical for 4KB Payload, achieving ultra-low latency
-            socket.set_tcp_nodelay(true)?;
+            if let Err(e) = socket.set_tcp_nodelay(true) {
+                debug!(
+                    event = "socket_option_unavailable",
+                    component = LOG_COMPONENT_SERVER,
+                    subsystem = LOG_SUBSYSTEM_STARTUP,
+                    option = "TCP_NODELAY",
+                    error = %e,
+                    "Socket option is unavailable"
+                );
+            }
 
             // 2. Enable SO_REUSEPORT for better multi-core scalability on supported platforms
             #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
@@ -297,11 +315,40 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
             }
 
             // 3. Set system-level TCP KeepAlive to protect long connections
-            socket.set_tcp_keepalive(&keepalive)?;
+            if let Err(e) = socket.set_tcp_keepalive(&keepalive) {
+                debug!(
+                    event = "socket_option_unavailable",
+                    component = LOG_COMPONENT_SERVER,
+                    subsystem = LOG_SUBSYSTEM_STARTUP,
+                    option = "TCP_KEEPALIVE",
+                    error = %e,
+                    "Socket option is unavailable"
+                );
+            }
 
-            // 4. Increase receive/send buffer to support BDP at GB-level throughput
-            socket.set_recv_buffer_size(4 * rustfs_config::MI_B)?;
-            socket.set_send_buffer_size(4 * rustfs_config::MI_B)?;
+            // 4. Increase receive/send buffer to support BDP at GB-level throughput.
+            // Some constrained local environments reject these socket options with
+            // EPERM/ENOPROTOOPT-style failures; log and continue in that case.
+            if let Err(e) = socket.set_recv_buffer_size(4 * rustfs_config::MI_B) {
+                debug!(
+                    event = "socket_option_unavailable",
+                    component = LOG_COMPONENT_SERVER,
+                    subsystem = LOG_SUBSYSTEM_STARTUP,
+                    option = "SO_RCVBUF",
+                    error = %e,
+                    "Socket option is unavailable"
+                );
+            }
+            if let Err(e) = socket.set_send_buffer_size(4 * rustfs_config::MI_B) {
+                debug!(
+                    event = "socket_option_unavailable",
+                    component = LOG_COMPONENT_SERVER,
+                    subsystem = LOG_SUBSYSTEM_STARTUP,
+                    option = "SO_SNDBUF",
+                    error = %e,
+                    "Socket option is unavailable"
+                );
+            }
 
             Ok(())
         };

rustfs/src/server/layer.rs

@@ -24,7 +24,9 @@ use crate::server::{
     has_path_prefix, is_admin_path, is_table_catalog_path,
 };
 use crate::storage::apply_cors_headers;
-use crate::storage::request_context::{RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers};
+use crate::storage::request_context::{
+    RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers, spawn_traced,
+};
 use bytes::Bytes;
 use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode, Uri};
 use http_body::Body;
@@ -38,17 +40,20 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use std::time::Instant;
+use std::time::{Duration, Instant};
+use tokio_util::sync::CancellationToken;
 use tower::{Layer, Service};
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
 use url::form_urlencoded;
 
 const HTTP_REQUEST_COMPLETED_EVENT: &str = "http_request_completed";
 const HTTP_REQUEST_FAILED_EVENT: &str = "http_request_failed";
+const HTTP_REQUEST_INFLIGHT_SLOW_EVENT: &str = "http_request_inflight_slow";
 const LOG_COMPONENT_SERVER: &str = "server";
 const LOG_SUBSYSTEM_HTTP: &str = "http";
 const REDACTED_QUERY_VALUE: &str = "redacted";
 const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/";
+const HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD: Duration = Duration::from_secs(5);
 
 pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String {
     let path = uri.path();
@@ -322,9 +327,39 @@ where
     fn call(&mut self, req: HttpRequest<B>) -> Self::Future {
         let context = RequestLogContext::from_request(&req);
         let mut inner = self.inner.clone();
+        let watchdog = CancellationToken::new();
+        let watchdog_context = context.clone();
+
+        spawn_traced({
+            let watchdog = watchdog.clone();
+            async move {
+                tokio::select! {
+                    _ = watchdog.cancelled() => {}
+                    _ = tokio::time::sleep(HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD) => {
+                        warn!(
+                            event = HTTP_REQUEST_INFLIGHT_SLOW_EVENT,
+                            component = LOG_COMPONENT_SERVER,
+                            subsystem = LOG_SUBSYSTEM_HTTP,
+                            request_id = %watchdog_context.request_id,
+                            trace_id = %watchdog_context.trace_id.as_deref().unwrap_or("unknown"),
+                            span_id = %watchdog_context.span_id.as_deref().unwrap_or("unknown"),
+                            peer_addr = %watchdog_context.peer_addr,
+                            method = %watchdog_context.method,
+                            uri = %watchdog_context.uri,
+                            duration_ms = watchdog_context.duration_ms(),
+                            active_requests = active_http_requests(),
+                            threshold_ms = HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD.as_millis() as u64,
+                            state = "response_pending",
+                            "HTTP request remains in flight"
+                        );
+                    }
+                }
+            }
+        });
 
         Box::pin(async move {
             let result = inner.call(req).await;
+            watchdog.cancel();
             match &result {
                 Ok(response) => context.log_response(response),
                 Err(error) => context.log_failure(error),
@@ -3054,8 +3089,8 @@ mod tests {
         assert_eq!(redact_sensitive_uri_query(&uri), "/rustfs/admin/v3/users?token=not-a-download-token");
     }
 
-    #[test]
-    fn request_logging_layer_emits_single_completion_event_with_standard_fields() {
+    #[tokio::test]
+    async fn request_logging_layer_emits_single_completion_event_with_standard_fields() {
         let writer = SharedWriter::default();
         let captured = writer.buffer.clone();
         let subscriber = Registry::default().with(
@@ -3067,25 +3102,25 @@ mod tests {
                 .with_writer(writer),
         );
 
-        tracing::subscriber::with_default(subscriber, || {
-            let mut service = tower::ServiceBuilder::new()
-                .layer(RequestContextLayer)
-                .layer(RequestLoggingLayer)
-                .service(StatusService::new(StatusCode::OK));
+        let _guard = tracing::subscriber::set_default(subscriber);
 
-            let mut request: Request<Full<Bytes>> = Request::builder()
-                .method(Method::GET)
-                .uri("/bucket/object.txt")
-                .header("x-request-id", "req-123")
-                .body(Full::from(Bytes::new()))
-                .expect("request");
-            request
-                .extensions_mut()
-                .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
+        let mut service = tower::ServiceBuilder::new()
+            .layer(RequestContextLayer)
+            .layer(RequestLoggingLayer)
+            .service(StatusService::new(StatusCode::OK));
 
-            let response = futures::executor::block_on(service.call(request)).expect("response");
-            assert_eq!(response.status(), StatusCode::OK);
-        });
+        let mut request: Request<Full<Bytes>> = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object.txt")
+            .header("x-request-id", "req-123")
+            .body(Full::from(Bytes::new()))
+            .expect("request");
+        request
+            .extensions_mut()
+            .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
+
+        let response = service.call(request).await.expect("response");
+        assert_eq!(response.status(), StatusCode::OK);
 
         let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs");
         assert_eq!(output.matches("HTTP request completed").count(), 1, "{output}");
@@ -3110,8 +3145,8 @@ mod tests {
         assert!(output.contains("duration_ms"), "{output}");
     }
 
-    #[test]
-    fn request_logging_layer_uses_request_context_trace_fields() {
+    #[tokio::test]
+    async fn request_logging_layer_uses_request_context_trace_fields() {
         let writer = SharedWriter::default();
         let captured = writer.buffer.clone();
         let subscriber = Registry::default().with(
@@ -3123,28 +3158,28 @@ mod tests {
                 .with_writer(writer),
         );
 
-        tracing::subscriber::with_default(subscriber, || {
-            let mut service = RequestLoggingLayer.layer(StatusService::new(StatusCode::INTERNAL_SERVER_ERROR));
+        let _guard = tracing::subscriber::set_default(subscriber);
 
-            let mut request = Request::builder()
-                .method(Method::GET)
-                .uri("/bucket/object.txt")
-                .body(())
-                .expect("request");
-            request.extensions_mut().insert(RequestContext {
-                request_id: "req-ctx".to_string(),
-                x_amz_request_id: "amz-ctx".to_string(),
-                trace_id: Some("trace-ctx".to_string()),
-                span_id: Some("span-ctx".to_string()),
-                start_time: Instant::now(),
-            });
-            request
-                .extensions_mut()
-                .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
+        let mut service = RequestLoggingLayer.layer(StatusService::new(StatusCode::INTERNAL_SERVER_ERROR));
 
-            let response = futures::executor::block_on(service.call(request)).expect("response");
-            assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        let mut request = Request::builder()
+            .method(Method::GET)
+            .uri("/bucket/object.txt")
+            .body(())
+            .expect("request");
+        request.extensions_mut().insert(RequestContext {
+            request_id: "req-ctx".to_string(),
+            x_amz_request_id: "amz-ctx".to_string(),
+            trace_id: Some("trace-ctx".to_string()),
+            span_id: Some("span-ctx".to_string()),
+            start_time: Instant::now(),
         });
+        request
+            .extensions_mut()
+            .insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
+
+        let response = service.call(request).await.expect("response");
+        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
 
         let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs");
         assert!(output.contains("http_request_completed"), "{output}");

rustfs/src/startup_services.rs

@@ -355,6 +355,7 @@ async fn init_observability_runtime(ctx: CancellationToken) {
     crate::allocator_reclaim::init_allocator_reclaim(ctx.clone());
 
     if rustfs_obs::observability_metric_enabled() {
+        rustfs_io_metrics::set_put_stage_metrics_enabled(true);
         init_metrics_runtime(ctx.clone());
         crate::memory_observability::init_memory_observability(ctx.clone());
         init_auto_tuner(ctx).await;

scripts/run_object_batch_bench_enhanced.sh

@@ -202,47 +202,56 @@ trim() {
 
 to_bps() {
   local human="$1"
+  local number unit factor
   if [[ "$human" == "N/A" || -z "$human" ]]; then
     echo "N/A"
     return
   fi
-  awk -v v="$human" '
-    function abs(x){return x<0?-x:x}
-    BEGIN{
-      if (match(v, /^([0-9]+(\.[0-9]+)?)\s*(GiB\/s|MiB\/s|KiB\/s|GB\/s|MB\/s|KB\/s|B\/s)$/, m)) {
-        n=m[1]; u=m[3];
-        if (u=="GiB/s") f=1024*1024*1024;
-        else if (u=="MiB/s") f=1024*1024;
-        else if (u=="KiB/s") f=1024;
-        else if (u=="GB/s") f=1000*1000*1000;
-        else if (u=="MB/s") f=1000*1000;
-        else if (u=="KB/s") f=1000;
-        else f=1;
-        printf "%.6f\n", n*f;
-      } else {
-        print "N/A";
-      }
-    }'
+
+  # Normalize "123MiB/s" → "123 MiB/s" when no space separator exists.
+  human="$(echo "$human" | sed -E 's/^([0-9]+(\.[0-9]+)?)([A-Za-zµ])/\1 \3/')"
+  number="$(echo "$human" | awk '{print $1}')"
+  unit="$(echo "$human" | awk '{print $2}')"
+  case "$unit" in
+    GiB/s) factor=1073741824 ;;
+    MiB/s) factor=1048576 ;;
+    KiB/s) factor=1024 ;;
+    GB/s) factor=1000000000 ;;
+    MB/s) factor=1000000 ;;
+    KB/s) factor=1000 ;;
+    B/s) factor=1 ;;
+    *)
+      echo "N/A"
+      return
+      ;;
+  esac
+
+  awk -v n="$number" -v f="$factor" 'BEGIN { printf "%.6f\n", n * f }'
 }
 
 to_ms() {
   local human="$1"
+  local number unit factor
   if [[ "$human" == "N/A" || -z "$human" ]]; then
     echo "N/A"
     return
   fi
-  awk -v v="$human" '
-    BEGIN{
-      if (match(v, /^([0-9]+(\.[0-9]+)?)\s*(ms|us|µs|s)$/, m)) {
-        n=m[1]; u=m[3];
-        if (u=="s") f=1000;
-        else if (u=="ms") f=1;
-        else f=0.001;
-        printf "%.6f\n", n*f;
-      } else {
-        print "N/A";
-      }
-    }'
+
+  # Normalize "50ms" → "50 ms" when no space separator exists.
+  human="$(echo "$human" | sed -E 's/^([0-9]+(\.[0-9]+)?)([A-Za-zµ])/\1 \3/')"
+  number="$(echo "$human" | awk '{print $1}')"
+  unit="$(echo "$human" | awk '{print $2}')"
+  case "$unit" in
+    s) factor=1000 ;;
+    ms) factor=1 ;;
+    us|µs) factor=0.001 ;;
+    *)
+      echo "N/A"
+      return
+      ;;
+  esac
+
+  awk -v n="$number" -v f="$factor" 'BEGIN { printf "%.6f\n", n * f }'
 }
 
 extract_first() {
@@ -255,20 +264,24 @@ extract_metrics() {
   local log_file="$1"
 
   local throughput reqps latency
-  throughput="$(extract_first '[0-9]+(\.[0-9]+)?\s*(GiB/s|MiB/s|KiB/s|GB/s|MB/s|KB/s|B/s)' "$log_file")"
-  reqps="$(extract_first '[0-9]+(\.[0-9]+)?\s*(req/s|ops/s|requests/s)' "$log_file")"
-  latency="$(extract_first '[0-9]+(\.[0-9]+)?\s*(ms|us|µs|s)\s*(avg|mean)' "$log_file")"
+  throughput="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(GiB/s|MiB/s|KiB/s|GB/s|MB/s|KB/s|B/s)' "$log_file")"
+  reqps="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(obj/s|req/s|ops/s|requests/s)' "$log_file")"
+  latency="$(rg -o 'Reqs:[[:space:]]+Avg:[[:space:]]+[0-9]+(\.[0-9]+)?(ms|us|µs|s)' "$log_file" | head -n1 | sed -E 's/^Reqs:[[:space:]]+Avg:[[:space:]]+//')"
 
   if [[ -z "$latency" ]]; then
-    latency="$(extract_first '[0-9]+(\.[0-9]+)?\s*(ms|us|µs|s)' "$log_file")"
+    latency="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(ms|us|µs|s)' "$log_file")"
   fi
 
   throughput="$(trim "${throughput:-N/A}")"
   reqps="$(trim "${reqps:-N/A}")"
   latency="$(trim "${latency:-N/A}")"
 
-  # Keep only "<num> <unit>" for latency if suffix avg/mean exists.
-  latency="$(echo "$latency" | awk '{print $1" "$2}')"
+  # Normalize to "<num> <unit>" shape for downstream conversion helpers.
+  if [[ "$latency" =~ ^([0-9]+(\.[0-9]+)?)(ms|us|µs|s)$ ]]; then
+    latency="${BASH_REMATCH[1]} ${BASH_REMATCH[3]}"
+  else
+    latency="$(echo "$latency" | awk '{print $1" "$2}')"
+  fi
   reqps_num="$(echo "$reqps" | awk '{print $1}')"
 
   echo "$throughput,${reqps_num:-N/A},$latency"
@@ -329,7 +342,7 @@ run_one_attempt() {
       printf '\n'
       echo "dry run" > "$log_file"
     else
-      if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then
+      if ! "${cmd[@]}" 2>&1 | tee "$log_file" >&2; then
         status="failed"
       fi
     fi
@@ -357,7 +370,7 @@ run_one_attempt() {
       printf '\n'
       echo "dry run" > "$log_file"
     else
-      if ! "${cmd[@]}" 2>&1 | tee "$log_file"; then
+      if ! "${cmd[@]}" 2>&1 | tee "$log_file" >&2; then
         status="failed"
       fi
     fi