diff --git a/.gitignore b/.gitignore
index 0e8bd841b..0dc9afc37 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,6 +52,7 @@ docs/*
 !docs/architecture/
 !docs/architecture/**
 docs/heal-scanner-logging-governance.md
+docs/benchmark/rustfs-target-bench/
 .codegraph/*
 .docker/compat/data/*
 .docker/compat/kms/*
@@ -69,4 +70,4 @@ tmp/
 crates/*/docs
 fuzz/target
 outputs
-worktrees/*
\ No newline at end of file
+worktrees/*
diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs
index 2bd8689b8..5fc4da587 100644
--- a/crates/ecstore/src/set_disk/read.rs
+++ b/crates/ecstore/src/set_disk/read.rs
@@ -1154,7 +1154,7 @@ impl SetDisks {
             // Default: enabled (true) for performance
             let use_zero_copy = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE);
 
-            let reader_setup_stage_start = Instant::now();
+            let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
             let mut readers = Vec::with_capacity(disks.len());
             let mut errors = Vec::with_capacity(disks.len());
             for (idx, disk_op) in disks.iter().enumerate() {
@@ -1186,7 +1186,11 @@ impl SetDisks {
                     }
                 }
             }
-            rustfs_io_metrics::record_get_object_shard_reader_setup_duration(reader_setup_stage_start.elapsed().as_secs_f64());
+            if let Some(reader_setup_stage_start) = reader_setup_stage_start {
+                rustfs_io_metrics::record_get_object_shard_reader_setup_duration(
+                    reader_setup_stage_start.elapsed().as_secs_f64(),
+                );
+            }
 
             let nil_count = errors.iter().filter(|&e| e.is_none()).count();
             if nil_count < erasure.data_shards {
@@ -1281,9 +1285,11 @@ impl SetDisks {
             //     "read part {} part_offset {},part_length {},part_size {}  ",
             //     part_number, part_offset, part_length, part_size
             // );
-            let decode_stage_start = Instant::now();
+            let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
             let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await;
-            rustfs_io_metrics::record_get_object_decode_duration(decode_stage_start.elapsed().as_secs_f64());
+            if let Some(decode_stage_start) = decode_stage_start {
+                rustfs_io_metrics::record_get_object_decode_duration(decode_stage_start.elapsed().as_secs_f64());
+            }
             debug!(
                 bucket,
                 object,
@@ -1396,7 +1402,7 @@ impl SetDisks {
         let use_zero_copy = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE);
         let till_offset = erasure.shard_file_offset(0, part_length, part_size);
 
-        let reader_setup_stage_start = Instant::now();
+        let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
         let mut readers = Vec::with_capacity(disks.len());
         let mut errors = Vec::with_capacity(disks.len());
         for (idx, disk_op) in disks.iter().enumerate() {
@@ -1428,11 +1434,13 @@ impl SetDisks {
                 }
             }
         }
-        rustfs_io_metrics::record_get_object_stage_duration(
-            GET_OBJECT_PATH_CODEC_STREAMING,
-            GET_STAGE_READER_SETUP,
-            reader_setup_stage_start.elapsed().as_secs_f64(),
-        );
+        if let Some(reader_setup_stage_start) = reader_setup_stage_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                GET_OBJECT_PATH_CODEC_STREAMING,
+                GET_STAGE_READER_SETUP,
+                reader_setup_stage_start.elapsed().as_secs_f64(),
+            );
+        }
 
         let available_shards = errors.iter().filter(|err| err.is_none()).count();
         if available_shards < erasure.data_shards {
diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs
index 2b9303fe7..af50079b8 100644
--- a/crates/io-metrics/src/lib.rs
+++ b/crates/io-metrics/src/lib.rs
@@ -57,6 +57,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 ///
 /// Set to `true` during startup when OTEL metric export is enabled.
 static PUT_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
+static GET_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
 
 /// Enable or disable detailed per-stage PUT metrics.
 ///
@@ -65,6 +66,10 @@ pub fn set_put_stage_metrics_enabled(enabled: bool) {
     PUT_STAGE_METRICS_ENABLED.store(enabled, Ordering::Relaxed);
 }
 
+pub fn set_get_stage_metrics_enabled(enabled: bool) {
+    GET_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
@@ -74,6 +79,11 @@ pub fn put_stage_metrics_enabled() -> bool {
     PUT_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
 }
 
+#[inline(always)]
+pub fn get_stage_metrics_enabled() -> bool {
+    GET_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
+}
+
 // Public modules
 pub mod adaptive_ttl;
 pub mod autotuner;
@@ -274,6 +284,9 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte
 /// Record the streaming strategy chosen for a GetObject response body.
 #[inline(always)]
 pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1);
     histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string())
         .record(usize_to_f64(buffer_size_bytes));
@@ -290,6 +303,9 @@ pub fn record_get_object_response_handoff(
     response_size_bytes: i64,
     duration_secs: f64,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!(
         "rustfs_io_get_object_response_handoff_total",
         "strategy" => strategy.to_string(),
@@ -338,6 +354,9 @@ pub fn record_get_object_io_state(
     load_level: &str,
     buffer_multiplier: f64,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     histogram!("rustfs_io_disk_permit_wait_duration_seconds").record(permit_wait_secs);
     gauge!("rustfs_io_queue_utilization_percent").set(queue_utilization_percent);
     gauge!("rustfs_io_queue_permits_in_use").set(permits_in_use as f64);
@@ -349,30 +368,45 @@ pub fn record_get_object_io_state(
 /// Record GetObject phase duration for the current read path.
 #[inline(always)]
 pub fn record_get_object_stage_duration(path: &'static str, stage: &'static str, duration_secs: f64) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     histogram!("rustfs_io_get_object_stage_duration_seconds", "path" => path, "stage" => stage).record(duration_secs);
 }
 
 /// Record the selected GetObject reader path.
 #[inline(always)]
 pub fn record_get_object_reader_path(path: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_reader_path_total", "path" => path).increment(1);
 }
 
 /// Record why the codec streaming reader was not selected.
 #[inline(always)]
 pub fn record_get_object_codec_streaming_fallback(reason: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_codec_streaming_fallback_total", "reason" => reason).increment(1);
 }
 
 /// Record one decoded reader stripe processed by a GetObject read path.
 #[inline(always)]
 pub fn record_get_object_reader_stripe(path: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_reader_stripes_total", "path" => path).increment(1);
 }
 
 /// Record bytes emitted by a GetObject reader path.
 #[inline(always)]
 pub fn record_get_object_reader_bytes(path: &'static str, bytes: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!("rustfs_io_get_object_reader_bytes_total", "path" => path).increment(bytes);
 }
@@ -380,6 +414,9 @@ pub fn record_get_object_reader_bytes(path: &'static str, bytes: usize) {
 /// Record one reader buffer produced by a GetObject read path.
 #[inline(always)]
 pub fn record_get_object_reader_buffer(path: &'static str, role: &'static str, bytes: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     histogram!("rustfs_io_get_object_reader_buffer_bytes", "path" => path, "role" => role).record(usize_to_f64(bytes));
 }
 
@@ -392,6 +429,9 @@ pub fn record_get_object_reader_copy(
     output_remaining_before: usize,
     duration_secs: f64,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     let bytes_counter = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!("rustfs_io_get_object_reader_copy_chunks_total", "path" => path).increment(1);
     counter!("rustfs_io_get_object_reader_copy_bytes_total", "path" => path).increment(bytes_counter);
@@ -412,6 +452,9 @@ pub fn record_get_object_reader_poll(
     filled_bytes: usize,
     duration_secs: f64,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     let filled_bytes_counter = u64::try_from(filled_bytes).unwrap_or(u64::MAX);
     counter!("rustfs_io_get_object_reader_poll_total", "path" => path, "outcome" => outcome).increment(1);
     counter!("rustfs_io_get_object_reader_poll_filled_bytes_total", "path" => path, "outcome" => outcome)
@@ -426,12 +469,18 @@ pub fn record_get_object_reader_poll(
 /// Record a bounded prefetch outcome for a GetObject reader path.
 #[inline(always)]
 pub fn record_get_object_reader_prefetch(path: &'static str, outcome: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_reader_prefetch_total", "path" => path, "outcome" => outcome).increment(1);
 }
 
 /// Record how long a GetObject reader spent waiting for a prefetch/fill result.
 #[inline(always)]
 pub fn record_get_object_reader_prefetch_wait(path: &'static str, duration_secs: f64) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     histogram!("rustfs_io_get_object_reader_prefetch_wait_seconds", "path" => path).record(duration_secs);
 }
 
@@ -444,6 +493,9 @@ pub fn record_get_object_shard_read(
     bytes: usize,
     duration_secs: f64,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!("rustfs_io_get_object_shard_read_total", "path" => path, "role" => role, "outcome" => outcome).increment(1);
     counter!("rustfs_io_get_object_shard_read_bytes_total", "path" => path, "role" => role, "outcome" => outcome)
@@ -466,6 +518,9 @@ pub fn record_get_object_shard_read_fanout(
     successful: usize,
     failed: usize,
 ) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     histogram!("rustfs_io_get_object_shard_read_scheduled", "path" => path).record(shard_read_fanout_to_f64(scheduled));
     histogram!("rustfs_io_get_object_shard_read_completed", "path" => path).record(shard_read_fanout_to_f64(completed));
     histogram!("rustfs_io_get_object_shard_read_successful", "path" => path).record(shard_read_fanout_to_f64(successful));
@@ -499,6 +554,9 @@ pub fn record_get_object_duplex_backpressure_duration(duration_secs: f64) {
 /// Record GetObject read pipeline failures using bounded labels.
 #[inline(always)]
 pub fn record_get_object_pipeline_failure(stage: &'static str, reason: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_pipeline_failures_total", "path" => "legacy_duplex", "stage" => stage, "reason" => reason)
         .increment(1);
 }
@@ -506,6 +564,9 @@ pub fn record_get_object_pipeline_failure(stage: &'static str, reason: &'static
 /// Record GetObject read pipeline failures for an explicit bounded path label.
 #[inline(always)]
 pub fn record_get_object_pipeline_failure_for_path(path: &'static str, stage: &'static str, reason: &'static str) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
     counter!("rustfs_io_get_object_pipeline_failures_total", "path" => path, "stage" => stage, "reason" => reason).increment(1);
 }
 
@@ -1161,6 +1222,52 @@ mod tests {
         assert!(!put_stage_metrics_enabled());
     }
 
+    #[test]
+    fn test_record_get_object_path_and_stage() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_get_stage_metrics_enabled(true);
+        record_get_object_stage_duration("s3_handler", "request_context", 0.001);
+        record_get_object_reader_path("codec_streaming");
+        record_get_object_codec_streaming_fallback("range");
+        record_get_object_reader_stripe("codec_streaming");
+        record_get_object_reader_bytes("codec_streaming", 1024);
+        record_get_object_reader_buffer("codec_streaming", "output", 1024);
+        record_get_object_reader_copy("codec_streaming", 512, 8192, 1024, 0.0001);
+        record_get_object_reader_poll("codec_streaming", "ready_data", 8192, 512, 0.0002);
+        record_get_object_reader_prefetch("codec_streaming", "stored");
+        record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
+        record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
+        record_get_object_shard_reader_setup_duration(0.003);
+        record_get_object_decode_duration(0.004);
+        record_get_object_duplex_backpressure_duration(0.005);
+        record_get_object_pipeline_failure("decode", "read_quorum");
+        record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum");
+        set_get_stage_metrics_enabled(false);
+    }
+
+    #[test]
+    fn test_get_stage_metrics_disabled_by_default() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_get_stage_metrics_enabled(false);
+        record_get_object_stage_duration("s3_handler", "request_context", 0.001);
+        record_get_object_reader_path("codec_streaming");
+        record_get_object_codec_streaming_fallback("range");
+        record_get_object_reader_stripe("codec_streaming");
+        record_get_object_reader_bytes("codec_streaming", 1024);
+        record_get_object_reader_buffer("codec_streaming", "output", 1024);
+        record_get_object_reader_copy("codec_streaming", 512, 8192, 1024, 0.0001);
+        record_get_object_reader_poll("codec_streaming", "ready_data", 8192, 512, 0.0002);
+        record_get_object_reader_prefetch("codec_streaming", "stored");
+        record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
+        record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
+        record_get_object_shard_reader_setup_duration(0.003);
+        record_get_object_decode_duration(0.004);
+        record_get_object_duplex_backpressure_duration(0.005);
+        record_get_object_pipeline_failure("decode", "read_quorum");
+        record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum");
+        assert!(!get_stage_metrics_enabled());
+    }
+
     #[test]
     fn test_record_stage_duration_generic() {
         // Generic stage duration should always record (no gating flag)
diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs
index 484c97429..670e6bfa8 100644
--- a/rustfs/src/app/object_usecase.rs
+++ b/rustfs/src/app/object_usecase.rs
@@ -143,6 +143,7 @@ use s3s::dto::{
     ServerSideEncryption, StorageClass, StreamingBlob, TaggingHeader, Timestamp, TimestampFormat, WebsiteRedirectLocation,
 };
 use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH};
+use s3s::stream::{ByteStream, RemainingLength};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 
 const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;
@@ -169,8 +170,13 @@ use super::storage_api::object_usecase::{
     StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader,
 };
 
+type S3StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
+
 const ACCEPT_RANGES_BYTES: &str = "bytes";
 const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024;
+const MEDIUM_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 8 * 1024 * 1024;
+const HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 4 * 1024 * 1024;
+const VERY_HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 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";
@@ -395,6 +401,14 @@ pin_project! {
     }
 }
 
+pin_project! {
+    struct GetObjectReaderStream<R> {
+        #[pin]
+        inner: ReaderStream<R>,
+        remaining: usize,
+    }
+}
+
 impl MemoryTrackedBytesStream {
     fn new(bytes: Bytes, guard: Option<rustfs_io_metrics::MemoryGaugeGuard>) -> Self {
         Self {
@@ -405,6 +419,18 @@ impl MemoryTrackedBytesStream {
     }
 }
 
+impl<R> GetObjectReaderStream<R>
+where
+    R: AsyncRead,
+{
+    fn new(reader: R, capacity: usize, remaining: usize) -> Self {
+        Self {
+            inner: ReaderStream::with_capacity(reader, capacity),
+            remaining,
+        }
+    }
+}
+
 impl futures::Stream for MemoryTrackedBytesStream {
     type Item = std::io::Result<Bytes>;
 
@@ -419,6 +445,50 @@ impl futures::Stream for MemoryTrackedBytesStream {
     }
 }
 
+impl<R> futures::Stream for GetObjectReaderStream<R>
+where
+    R: AsyncRead,
+{
+    type Item = Result<Bytes, S3StdError>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+        if *this.remaining == 0 {
+            return Poll::Ready(None);
+        }
+
+        match this.inner.as_mut().poll_next(cx) {
+            Poll::Ready(Some(Ok(mut bytes))) => {
+                if bytes.len() > *this.remaining {
+                    bytes.truncate(*this.remaining);
+                }
+                *this.remaining -= bytes.len();
+                if bytes.is_empty() {
+                    Poll::Ready(None)
+                } else {
+                    Poll::Ready(Some(Ok(bytes)))
+                }
+            }
+            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(Box::new(err)))),
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
+}
+
+impl<R> ByteStream for GetObjectReaderStream<R>
+where
+    R: AsyncRead,
+{
+    fn remaining_length(&self) -> RemainingLength {
+        RemainingLength::new_exact(self.remaining)
+    }
+}
+
 impl<R> ExtractArchiveEtagReader<R> {
     fn new(inner: R, etag: Arc<Mutex<Option<String>>>) -> Self {
         Self {
@@ -744,14 +814,56 @@ fn object_seek_support_threshold() -> usize {
     })
 }
 
+fn object_seek_support_concurrency_thresholds() -> (usize, usize) {
+    static OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS: OnceLock<(usize, usize)> = OnceLock::new();
+    *OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS.get_or_init(|| {
+        let medium = rustfs_utils::get_env_usize(
+            rustfs_config::ENV_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD,
+            rustfs_config::DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD,
+        )
+        .max(1);
+        let high = rustfs_utils::get_env_usize(
+            rustfs_config::ENV_OBJECT_HIGH_CONCURRENCY_THRESHOLD,
+            rustfs_config::DEFAULT_OBJECT_HIGH_CONCURRENCY_THRESHOLD,
+        )
+        .max(medium + 1);
+        (medium, high)
+    })
+}
+
+fn concurrency_aware_seek_support_threshold(configured_threshold: i64, concurrent_requests: usize) -> i64 {
+    let (medium_threshold, high_threshold) = object_seek_support_concurrency_thresholds();
+    let effective_threshold = configured_threshold.min(MAX_GET_OBJECT_MEMORY_BUFFER_BYTES);
+
+    if concurrent_requests >= high_threshold.saturating_mul(2) {
+        return effective_threshold.min(VERY_HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES);
+    }
+    if concurrent_requests >= high_threshold {
+        return effective_threshold.min(HIGH_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES);
+    }
+    if concurrent_requests >= medium_threshold {
+        return effective_threshold.min(MEDIUM_CONCURRENCY_GET_OBJECT_MEMORY_BUFFER_BYTES);
+    }
+
+    effective_threshold
+}
+
 fn should_buffer_get_object_in_memory(
     info: &ObjectInfo,
     response_content_length: i64,
     part_number: Option<usize>,
     has_range: bool,
+    concurrent_requests: usize,
 ) -> bool {
     let configured_threshold = object_seek_support_threshold() as i64;
-    should_buffer_get_object_in_memory_with_threshold(info, response_content_length, part_number, has_range, configured_threshold)
+    should_buffer_get_object_in_memory_with_threshold(
+        info,
+        response_content_length,
+        part_number,
+        has_range,
+        configured_threshold,
+        concurrent_requests,
+    )
 }
 
 fn should_buffer_get_object_in_memory_with_threshold(
@@ -760,12 +872,13 @@ fn should_buffer_get_object_in_memory_with_threshold(
     part_number: Option<usize>,
     has_range: bool,
     configured_threshold: i64,
+    concurrent_requests: usize,
 ) -> bool {
     if part_number.is_some() || has_range || response_content_length <= 0 || configured_threshold <= 0 {
         return false;
     }
 
-    let effective_threshold = configured_threshold.min(MAX_GET_OBJECT_MEMORY_BUFFER_BYTES);
+    let effective_threshold = concurrency_aware_seek_support_threshold(configured_threshold, concurrent_requests);
     if configured_threshold > MAX_GET_OBJECT_MEMORY_BUFFER_BYTES
         && GET_OBJECT_BUFFER_THRESHOLD_WARNED
             .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
@@ -1609,16 +1722,19 @@ impl DefaultObjectUsecase {
         let expected = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX);
         let (stream_buffer_size, buffer_source) =
             resolve_reader_stream_buffer_size(stream_buffer_size, get_reader_stream_buffer_size_override());
-        rustfs_io_metrics::record_get_object_stream_strategy(
-            stream_strategy.as_str(),
-            stream_buffer_size,
-            response_content_length,
-        );
-        let handoff_start = std::time::Instant::now();
+        let get_stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        if get_stage_metrics_enabled {
+            rustfs_io_metrics::record_get_object_stream_strategy(
+                stream_strategy.as_str(),
+                stream_buffer_size,
+                response_content_length,
+            );
+        }
+        let handoff_start = get_stage_metrics_enabled.then(std::time::Instant::now);
         #[cfg(feature = "tracing-chunk-debug")]
         let stream = {
             let mut emitted = 0usize;
-            ReaderStream::with_capacity(reader, stream_buffer_size).inspect(move |item| match item {
+            GetObjectReaderStream::new(reader, stream_buffer_size, expected).inspect(move |item| match item {
                 Ok(bytes) => {
                     emitted += bytes.len();
                     tracing::debug!(emitted, expected, chunk_len = bytes.len(), "GetObject ReaderStream emitted bytes");
@@ -1634,15 +1750,17 @@ impl DefaultObjectUsecase {
             })
         };
         #[cfg(not(feature = "tracing-chunk-debug"))]
-        let stream = ReaderStream::with_capacity(reader, stream_buffer_size);
-        let blob = StreamingBlob::wrap(bytes_stream(stream, expected));
-        rustfs_io_metrics::record_get_object_response_handoff(
-            stream_strategy.as_str(),
-            buffer_source,
-            stream_buffer_size,
-            response_content_length,
-            handoff_start.elapsed().as_secs_f64(),
-        );
+        let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected);
+        let blob = StreamingBlob::new(stream);
+        if let Some(handoff_start) = handoff_start {
+            rustfs_io_metrics::record_get_object_response_handoff(
+                stream_strategy.as_str(),
+                buffer_source,
+                stream_buffer_size,
+                response_content_length,
+                handoff_start.elapsed().as_secs_f64(),
+            );
+        }
         Some(blob)
     }
 
@@ -1793,17 +1911,32 @@ impl DefaultObjectUsecase {
     ) -> S3Result<GetObjectPreparedRead<'a>> {
         let h = req.headers.clone();
         let io_planning = Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?;
-        let store_lookup_start = std::time::Instant::now();
+        let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
         let store = get_validated_store(bucket).await?;
-        rustfs_io_metrics::record_get_object_stage_duration(
-            "s3_handler",
-            "store_lookup",
-            store_lookup_start.elapsed().as_secs_f64(),
-        );
+        if let Some(store_lookup_start) = store_lookup_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "store_lookup",
+                store_lookup_start.elapsed().as_secs_f64(),
+            );
+        }
 
         let read_start = std::time::Instant::now();
-        let read_setup =
-            Self::prepare_get_object_read(req, &store, manager, bucket, key, rs, h, opts, part_number, read_start).await?;
+        let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start);
+        let read_setup = Self::prepare_get_object_read(
+            req,
+            &store,
+            manager,
+            bucket,
+            key,
+            rs,
+            h,
+            opts,
+            part_number,
+            read_start,
+            read_stage_start,
+        )
+        .await?;
 
         Ok(GetObjectPreparedRead { io_planning, read_setup })
     }
@@ -1820,16 +1953,19 @@ impl DefaultObjectUsecase {
         opts: &ObjectOptions,
         part_number: Option<usize>,
         read_start: std::time::Instant,
+        read_stage_start: Option<std::time::Instant>,
     ) -> S3Result<GetObjectReadSetup> {
         let reader = store
             .get_object_reader(bucket, key, rs.clone(), h, opts)
             .await
             .map_err(map_get_object_reader_error)?;
-        rustfs_io_metrics::record_get_object_stage_duration(
-            "s3_handler",
-            "store_reader_setup",
-            read_start.elapsed().as_secs_f64(),
-        );
+        if let Some(read_stage_start) = read_stage_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "store_reader_setup",
+                read_stage_start.elapsed().as_secs_f64(),
+            );
+        }
 
         let info = reader.object_info;
 
@@ -2098,6 +2234,7 @@ impl DefaultObjectUsecase {
         response_content_length: i64,
         optimal_buffer_size: usize,
         enable_readahead: bool,
+        concurrent_requests: usize,
         part_number: Option<usize>,
         has_range: bool,
         encryption_applied: bool,
@@ -2107,7 +2244,7 @@ impl DefaultObjectUsecase {
     {
         if encryption_applied {
             let should_buffer_encrypted_object =
-                should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range);
+                should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range, concurrent_requests);
 
             if should_buffer_encrypted_object {
                 let mut buf = Vec::with_capacity(response_content_length as usize);
@@ -2139,7 +2276,7 @@ impl DefaultObjectUsecase {
         }
 
         let should_provide_seek_support =
-            should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range);
+            should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range, concurrent_requests);
 
         if should_provide_seek_support {
             let mut buf = Vec::with_capacity(response_content_length as usize);
@@ -2882,6 +3019,7 @@ impl DefaultObjectUsecase {
             response_content_length,
             optimal_buffer_size,
             enable_readahead,
+            concurrent_requests,
             part_number,
             rs.is_some(),
             encryption_applied,
@@ -2974,13 +3112,15 @@ impl DefaultObjectUsecase {
         let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
         // mc get 3
 
-        let request_context_start = std::time::Instant::now();
+        let request_context_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
         let request_context = Self::prepare_get_object_request_context(&req).await?;
-        rustfs_io_metrics::record_get_object_stage_duration(
-            "s3_handler",
-            "request_context",
-            request_context_start.elapsed().as_secs_f64(),
-        );
+        if let Some(request_context_start) = request_context_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "request_context",
+                request_context_start.elapsed().as_secs_f64(),
+            );
+        }
         let GetObjectRequestContext {
             bucket,
             key,
@@ -3025,14 +3165,16 @@ impl DefaultObjectUsecase {
             encryption_applied,
         } = read_setup;
 
-        let versioning_start = std::time::Instant::now();
+        let versioning_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
         let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-        rustfs_io_metrics::record_get_object_stage_duration(
-            "s3_handler",
-            "versioning_lookup",
-            versioning_start.elapsed().as_secs_f64(),
-        );
-        let output_build_start = std::time::Instant::now();
+        if let Some(versioning_start) = versioning_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "versioning_lookup",
+                versioning_start.elapsed().as_secs_f64(),
+            );
+        }
+        let output_build_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
         let output_context = self
             .build_get_object_output_context(
                 &req,
@@ -3060,11 +3202,13 @@ impl DefaultObjectUsecase {
                 versioned,
             )
             .await?;
-        rustfs_io_metrics::record_get_object_stage_duration(
-            "s3_handler",
-            "output_build",
-            output_build_start.elapsed().as_secs_f64(),
-        );
+        if let Some(output_build_start) = output_build_start {
+            rustfs_io_metrics::record_get_object_stage_duration(
+                "s3_handler",
+                "output_build",
+                output_build_start.elapsed().as_secs_f64(),
+            );
+        }
         let GetObjectOutputContext {
             output,
             event_info,
@@ -5520,7 +5664,7 @@ mod tests {
         let configured_threshold = 20_i64 * 1024 * 1024 * 1024;
         let response_len = 80_i64 * 1024 * 1024;
         let should_buffer =
-            should_buffer_get_object_in_memory_with_threshold(&info, response_len, None, false, configured_threshold);
+            should_buffer_get_object_in_memory_with_threshold(&info, response_len, None, false, configured_threshold, 1);
 
         assert!(
             !should_buffer,
@@ -5538,21 +5682,24 @@ mod tests {
             1024 * 1024,
             None,
             false,
-            configured_threshold
+            configured_threshold,
+            1
         ));
         assert!(!should_buffer_get_object_in_memory_with_threshold(
             &info,
             1024 * 1024,
             Some(1),
             false,
-            configured_threshold
+            configured_threshold,
+            1
         ));
         assert!(!should_buffer_get_object_in_memory_with_threshold(
             &info,
             1024 * 1024,
             None,
             true,
-            configured_threshold
+            configured_threshold,
+            1
         ));
     }
 
@@ -5566,14 +5713,16 @@ mod tests {
             configured_threshold,
             None,
             false,
-            configured_threshold
+            configured_threshold,
+            1
         ));
         assert!(!should_buffer_get_object_in_memory_with_threshold(
             &info,
             configured_threshold + 1,
             None,
             false,
-            configured_threshold
+            configured_threshold,
+            1
         ));
     }
 
@@ -5587,16 +5736,49 @@ mod tests {
             0,
             None,
             false,
-            configured_threshold
+            configured_threshold,
+            1
         ));
         assert!(!should_buffer_get_object_in_memory_with_threshold(
             &info,
             -1,
             None,
             false,
-            configured_threshold
+            configured_threshold,
+            1
+        ));
+        assert!(!should_buffer_get_object_in_memory_with_threshold(&info, 1024, None, false, 0, 1));
+    }
+
+    #[test]
+    fn should_buffer_get_object_in_memory_reduces_threshold_under_concurrency() {
+        let info = ObjectInfo::default();
+        let configured_threshold = 10_i64 * 1024 * 1024;
+
+        assert!(should_buffer_get_object_in_memory_with_threshold(
+            &info,
+            configured_threshold,
+            None,
+            false,
+            configured_threshold,
+            1
+        ));
+        assert!(!should_buffer_get_object_in_memory_with_threshold(
+            &info,
+            configured_threshold,
+            None,
+            false,
+            configured_threshold,
+            32
+        ));
+        assert!(should_buffer_get_object_in_memory_with_threshold(
+            &info,
+            4_i64 * 1024 * 1024,
+            None,
+            false,
+            configured_threshold,
+            rustfs_config::DEFAULT_OBJECT_HIGH_CONCURRENCY_THRESHOLD
         ));
-        assert!(!should_buffer_get_object_in_memory_with_threshold(&info, 1024, None, false, 0));
     }
 
     struct ReadProbeReader {
@@ -5627,6 +5809,7 @@ mod tests {
             18_i64 * 1024 * 1024 * 1024,
             128 * 1024,
             true,
+            1,
             None,
             false,
             false,
@@ -5659,6 +5842,7 @@ mod tests {
             18_i64 * 1024 * 1024 * 1024,
             128 * 1024,
             true,
+            1,
             None,
             false,
             true,
@@ -5861,6 +6045,40 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::UnexpectedContent);
     }
 
+    #[tokio::test]
+    async fn get_object_reader_stream_tracks_remaining_length() {
+        let mut stream = GetObjectReaderStream::new(std::io::Cursor::new(b"hello".to_vec()), 2, 5);
+
+        assert_eq!(stream.remaining_length().exact(), Some(5));
+
+        let first = stream
+            .next()
+            .await
+            .expect("reader stream should emit first chunk")
+            .expect("first chunk should read");
+
+        assert_eq!(first.as_ref(), b"he");
+        assert_eq!(stream.remaining_length().exact(), Some(3));
+    }
+
+    #[tokio::test]
+    async fn get_object_reader_stream_truncates_to_expected_length() {
+        let stream = GetObjectReaderStream::new(std::io::Cursor::new(b"hello!".to_vec()), 64, 5);
+
+        let chunks = stream
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect("reader stream should read");
+        let body = chunks.into_iter().fold(Vec::new(), |mut acc, chunk| {
+            acc.extend_from_slice(&chunk);
+            acc
+        });
+
+        assert_eq!(body, b"hello");
+    }
+
     #[tokio::test]
     async fn pooled_buffer_reader_keeps_buffer_alive_until_consumed() {
         use tokio::io::AsyncReadExt;
diff --git a/rustfs/src/startup_observability.rs b/rustfs/src/startup_observability.rs
index 99705151c..ce7009557 100644
--- a/rustfs/src/startup_observability.rs
+++ b/rustfs/src/startup_observability.rs
@@ -23,6 +23,7 @@ pub(crate) async fn init_observability_runtime(ctx: CancellationToken) {
 
     if startup_runtime_sources::observability_metric_enabled() {
         startup_runtime_sources::set_put_stage_metrics_enabled(true);
+        startup_runtime_sources::set_get_stage_metrics_enabled(true);
         startup_runtime_sources::init_metrics_runtime(ctx.clone());
         crate::memory_observability::init_memory_observability(ctx.clone());
         init_auto_tuner(ctx).await;
diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs
index 76dcfc46f..5fb7d32a6 100644
--- a/rustfs/src/startup_runtime_sources.rs
+++ b/rustfs/src/startup_runtime_sources.rs
@@ -80,6 +80,10 @@ pub(crate) fn set_put_stage_metrics_enabled(enabled: bool) {
     rustfs_io_metrics::set_put_stage_metrics_enabled(enabled);
 }
 
+pub(crate) fn set_get_stage_metrics_enabled(enabled: bool) {
+    rustfs_io_metrics::set_get_stage_metrics_enabled(enabled);
+}
+
 pub(crate) fn init_tls_metrics() {
     rustfs_tls_runtime::init_tls_metrics();
 }
diff --git a/scripts/fuzz/run.sh b/scripts/fuzz/run.sh
index 3eeb8a3c1..656c86f99 100755
--- a/scripts/fuzz/run.sh
+++ b/scripts/fuzz/run.sh
@@ -1,5 +1,19 @@
 #!/bin/sh
 
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 # Unified fuzz runner script.
 #
 # Modes:
diff --git a/scripts/run_get_metrics_gate_smoke.sh b/scripts/run_get_metrics_gate_smoke.sh
new file mode 100755
index 000000000..0dc08df97
--- /dev/null
+++ b/scripts/run_get_metrics_gate_smoke.sh
@@ -0,0 +1,514 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+ENHANCED_BENCH="${PROJECT_ROOT}/scripts/run_object_batch_bench_enhanced.sh"
+
+ADDRESS="127.0.0.1:19031"
+ACCESS_KEY="rustfsadmin"
+SECRET_KEY="rustfsadmin"
+REGION="us-east-1"
+SIZE="10MiB"
+CONCURRENCY=32
+DURATION="10s"
+ROUNDS=3
+ROUND_COOLDOWN_SECS=10
+RETRY_PER_ROUND=1
+SEED_DURATION="5s"
+SEED_CONCURRENCY=8
+HEALTH_TIMEOUT_SECS=60
+OUT_DIR=""
+DATA_ROOT=""
+BUCKET=""
+RUSTFS_BIN="${PROJECT_ROOT}/target/release/rustfs"
+WARP_BIN="warp"
+BASELINE_CSV=""
+DRY_RUN=false
+SKIP_BUILD=false
+
+SERVER_PID=""
+
+usage() {
+  cat <<'USAGE'
+Usage:
+  scripts/run_get_metrics_gate_smoke.sh [options]
+
+Purpose:
+  Start one local single-node multi-disk RustFS server with observability
+  export disabled, seed 10MiB objects, and run a focused warp GET benchmark.
+
+Options:
+  --address <host:port>          RustFS listen address (default: 127.0.0.1:19031)
+  --access-key <value>           Access key (default: rustfsadmin)
+  --secret-key <value>           Secret key (default: rustfsadmin)
+  --region <value>               Region (default: us-east-1)
+  --bucket <name>                Benchmark bucket (default: auto-generated)
+  --size <label>                 Object size label (default: 10MiB)
+  --concurrency <n>              warp GET concurrency (default: 32)
+  --duration <duration>          warp GET duration per round (default: 10s)
+  --rounds <n>                   Benchmark rounds (default: 3)
+  --round-cooldown-secs <n>      Cooldown after each round (default: 10)
+  --retry-per-round <n>          Failed-round retries (default: 1)
+  --seed-duration <duration>     warp PUT duration for object seeding (default: 5s)
+  --seed-concurrency <n>         warp PUT concurrency for object seeding (default: 8)
+  --health-timeout-secs <n>      Health wait timeout (default: 60)
+  --out-dir <path>               Output directory (default: target/bench/get-metrics-gate-<timestamp>)
+  --data-root <path>             Data root for d1..d4 (default: /private/tmp/get-metrics-gate-<timestamp>)
+  --rustfs-bin <path>            RustFS binary (default: target/release/rustfs)
+  --warp-bin <path>              warp binary (default: warp)
+  --baseline-csv <path>          Optional baseline median_summary.csv for delta output
+  --skip-build                   Skip cargo build --release -p rustfs --bin rustfs
+  --dry-run                      Print commands only
+  -h, --help                     Show help
+USAGE
+}
+
+die() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+require_cmd() {
+  if ! command -v "$1" >/dev/null 2>&1; then
+    die "command not found: $1"
+  fi
+}
+
+validate_positive_int() {
+  local value="$1"
+  local name="$2"
+  if ! [[ "$value" =~ ^[0-9]+$ ]] || [[ "$value" -le 0 ]]; then
+    die "$name must be a positive integer, got: $value"
+  fi
+}
+
+validate_non_negative_int() {
+  local value="$1"
+  local name="$2"
+  if ! [[ "$value" =~ ^[0-9]+$ ]]; then
+    die "$name must be a non-negative integer, got: $value"
+  fi
+}
+
+endpoint_url() {
+  echo "http://${ADDRESS}"
+}
+
+normalize_warp_host() {
+  local raw="$1"
+  raw="${raw#http://}"
+  raw="${raw#https://}"
+  raw="${raw%%/*}"
+  raw="${raw%%\?*}"
+  raw="${raw%%\#*}"
+  echo "$raw"
+}
+
+parse_args() {
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --address) ADDRESS="$2"; shift 2 ;;
+      --access-key) ACCESS_KEY="$2"; shift 2 ;;
+      --secret-key) SECRET_KEY="$2"; shift 2 ;;
+      --region) REGION="$2"; shift 2 ;;
+      --bucket) BUCKET="$2"; shift 2 ;;
+      --size) SIZE="$2"; shift 2 ;;
+      --concurrency) CONCURRENCY="$2"; shift 2 ;;
+      --duration) DURATION="$2"; shift 2 ;;
+      --rounds) ROUNDS="$2"; shift 2 ;;
+      --round-cooldown-secs) ROUND_COOLDOWN_SECS="$2"; shift 2 ;;
+      --retry-per-round) RETRY_PER_ROUND="$2"; shift 2 ;;
+      --seed-duration) SEED_DURATION="$2"; shift 2 ;;
+      --seed-concurrency) SEED_CONCURRENCY="$2"; shift 2 ;;
+      --health-timeout-secs) HEALTH_TIMEOUT_SECS="$2"; shift 2 ;;
+      --out-dir) OUT_DIR="$2"; shift 2 ;;
+      --data-root) DATA_ROOT="$2"; shift 2 ;;
+      --rustfs-bin) RUSTFS_BIN="$2"; shift 2 ;;
+      --warp-bin) WARP_BIN="$2"; shift 2 ;;
+      --baseline-csv) BASELINE_CSV="$2"; shift 2 ;;
+      --skip-build) SKIP_BUILD=true; shift ;;
+      --dry-run) DRY_RUN=true; shift ;;
+      -h|--help) usage; exit 0 ;;
+      *)
+        usage >&2
+        die "unknown arg: $1"
+        ;;
+    esac
+  done
+}
+
+validate_args() {
+  [[ -n "$ADDRESS" ]] || die "--address must not be empty"
+  [[ -n "$ACCESS_KEY" ]] || die "--access-key must not be empty"
+  [[ -n "$SECRET_KEY" ]] || die "--secret-key must not be empty"
+  [[ -n "$REGION" ]] || die "--region must not be empty"
+  [[ -n "$SIZE" ]] || die "--size must not be empty"
+  validate_positive_int "$CONCURRENCY" "--concurrency"
+  validate_positive_int "$ROUNDS" "--rounds"
+  validate_positive_int "$RETRY_PER_ROUND" "--retry-per-round"
+  validate_positive_int "$SEED_CONCURRENCY" "--seed-concurrency"
+  validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout-secs"
+  validate_non_negative_int "$ROUND_COOLDOWN_SECS" "--round-cooldown-secs"
+
+  [[ -x "$ENHANCED_BENCH" ]] || die "benchmark script is not executable: $ENHANCED_BENCH"
+  require_cmd curl
+  require_cmd git
+  require_cmd "$WARP_BIN"
+  if [[ "$DRY_RUN" != "true" && "$SKIP_BUILD" != "true" ]]; then
+    require_cmd cargo
+  fi
+  if [[ -n "$BASELINE_CSV" && ! -f "$BASELINE_CSV" ]]; then
+    die "--baseline-csv does not exist: $BASELINE_CSV"
+  fi
+}
+
+setup_paths() {
+  local timestamp
+  timestamp="$(date +%Y%m%d-%H%M%S)"
+  if [[ -z "$OUT_DIR" ]]; then
+    OUT_DIR="${PROJECT_ROOT}/target/bench/get-metrics-gate-${timestamp}"
+  fi
+  if [[ -z "$DATA_ROOT" ]]; then
+    DATA_ROOT="/private/tmp/get-metrics-gate-${timestamp}"
+  fi
+  if [[ -z "$BUCKET" ]]; then
+    BUCKET="rustfs-get-metrics-${timestamp}"
+  fi
+
+  mkdir -p "$OUT_DIR" "$DATA_ROOT"/d1 "$DATA_ROOT"/d2 "$DATA_ROOT"/d3 "$DATA_ROOT"/d4
+}
+
+build_rustfs_if_needed() {
+  if [[ "$DRY_RUN" == "true" || "$SKIP_BUILD" == "true" ]]; then
+    return
+  fi
+  cargo build --release -p rustfs --bin rustfs
+}
+
+write_manifest() {
+  local git_head
+  git_head="$(git -C "$PROJECT_ROOT" rev-parse HEAD)"
+
+  cat >"${OUT_DIR}/manifest.env" <<EOF
+git_head=${git_head}
+endpoint=$(endpoint_url)
+address=${ADDRESS}
+bucket=${BUCKET}
+region=${REGION}
+size=${SIZE}
+concurrency=${CONCURRENCY}
+duration=${DURATION}
+rounds=${ROUNDS}
+round_cooldown_secs=${ROUND_COOLDOWN_SECS}
+retry_per_round=${RETRY_PER_ROUND}
+seed_duration=${SEED_DURATION}
+seed_concurrency=${SEED_CONCURRENCY}
+rustfs_bin=${RUSTFS_BIN}
+warp_bin=${WARP_BIN}
+data_root=${DATA_ROOT}
+RUST_LOG=off
+RUSTFS_OBS_LOGGER_LEVEL=off
+RUSTFS_OBS_TRACES_EXPORT_ENABLED=false
+RUSTFS_OBS_METRICS_EXPORT_ENABLED=false
+RUSTFS_OBS_LOGS_EXPORT_ENABLED=false
+RUSTFS_OBS_PROFILING_EXPORT_ENABLED=false
+RUSTFS_OBS_USE_STDOUT=false
+RUSTFS_OBS_LOG_STDOUT_ENABLED=false
+RUSTFS_SCANNER_ENABLED=false
+RUSTFS_CONSOLE_ENABLE=false
+RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true
+EOF
+}
+
+stop_server() {
+  if [[ -n "$SERVER_PID" ]]; then
+    if kill -0 "$SERVER_PID" >/dev/null 2>&1; then
+      kill "$SERVER_PID" >/dev/null 2>&1 || true
+      wait "$SERVER_PID" >/dev/null 2>&1 || true
+    fi
+    SERVER_PID=""
+  fi
+}
+
+wait_for_health() {
+  local health_url
+  health_url="$(endpoint_url)/health"
+
+  for ((attempt = 1; attempt <= HEALTH_TIMEOUT_SECS; attempt++)); do
+    if curl -fsS --noproxy '*' --connect-timeout 2 --max-time 3 "$health_url" >/dev/null 2>&1; then
+      return
+    fi
+    if [[ -n "$SERVER_PID" ]] && ! kill -0 "$SERVER_PID" >/dev/null 2>&1; then
+      tail -n 80 "${OUT_DIR}/rustfs.log" >&2 || true
+      die "RustFS exited before health check passed"
+    fi
+    sleep 1
+  done
+
+  tail -n 80 "${OUT_DIR}/rustfs.log" >&2 || true
+  die "RustFS health check timed out"
+}
+
+start_server() {
+  if [[ "$DRY_RUN" == "true" ]]; then
+    echo "[DRY-RUN] start RustFS at $(endpoint_url)"
+    return
+  fi
+
+  [[ -x "$RUSTFS_BIN" ]] || die "RustFS binary is not executable: $RUSTFS_BIN"
+
+  (
+    export RUST_LOG=off
+    export RUSTFS_OBS_LOGGER_LEVEL=off
+    export RUSTFS_OBS_TRACES_EXPORT_ENABLED=false
+    export RUSTFS_OBS_METRICS_EXPORT_ENABLED=false
+    export RUSTFS_OBS_LOGS_EXPORT_ENABLED=false
+    export RUSTFS_OBS_PROFILING_EXPORT_ENABLED=false
+    export RUSTFS_OBS_USE_STDOUT=false
+    export RUSTFS_OBS_LOG_STDOUT_ENABLED=false
+    export RUSTFS_SCANNER_ENABLED=false
+    export RUSTFS_CONSOLE_ENABLE=false
+    export RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true
+    export RUSTFS_ADDRESS="$ADDRESS"
+    export RUSTFS_ACCESS_KEY="$ACCESS_KEY"
+    export RUSTFS_SECRET_KEY="$SECRET_KEY"
+    export RUSTFS_RPC_SECRET="rustfs-get-metrics-gate-rpc-secret"
+    export RUSTFS_REGION="$REGION"
+    exec "$RUSTFS_BIN" server \
+      "${DATA_ROOT}/d1" \
+      "${DATA_ROOT}/d2" \
+      "${DATA_ROOT}/d3" \
+      "${DATA_ROOT}/d4"
+  ) >"${OUT_DIR}/rustfs.log" 2>&1 &
+
+  SERVER_PID="$!"
+  wait_for_health
+}
+
+run_bench() {
+  local cmd=(
+    "$ENHANCED_BENCH"
+    --tool warp
+    --endpoint "$(endpoint_url)"
+    --access-key "$ACCESS_KEY"
+    --secret-key "$SECRET_KEY"
+    --bucket "$BUCKET"
+    --region "$REGION"
+    --warp-bin "$WARP_BIN"
+    --warp-mode get
+    --sizes "$SIZE"
+    --concurrency "$CONCURRENCY"
+    --duration "$DURATION"
+    --rounds "$ROUNDS"
+    --retry-per-round "$RETRY_PER_ROUND"
+    --round-cooldown-secs "$ROUND_COOLDOWN_SECS"
+    --out-dir "${OUT_DIR}/warp"
+  )
+
+  if [[ -n "$BASELINE_CSV" ]]; then
+    cmd+=(--baseline-csv "$BASELINE_CSV")
+  fi
+  if [[ "$DRY_RUN" == "true" ]]; then
+    cmd+=(--dry-run)
+  fi
+
+  "${cmd[@]}"
+}
+
+to_bps() {
+  local human="$1"
+  local number unit factor
+  if [[ "$human" == "N/A" || -z "$human" ]]; then
+    echo "N/A"
+    return
+  fi
+
+  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
+
+  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_final_get_metrics() {
+  local log_file="$1"
+  local avg_line reqs_line throughput_human reqps latency_human
+
+  avg_line="$(awk '/^[[:space:]]*\* Average:/ { line=$0 } END { print line }' "$log_file")"
+  reqs_line="$(awk '/^[[:space:]]*\* Reqs: Avg:/ { line=$0 } END { print line }' "$log_file")"
+
+  throughput_human="$(
+    echo "$avg_line" \
+      | sed -nE 's/^[[:space:]]*\* Average: ([0-9]+(\.[0-9]+)? [A-Za-z\/]+), ([0-9]+(\.[0-9]+)?) obj\/s$/\1/p'
+  )"
+  reqps="$(
+    echo "$avg_line" \
+      | sed -nE 's/^[[:space:]]*\* Average: ([0-9]+(\.[0-9]+)? [A-Za-z\/]+), ([0-9]+(\.[0-9]+)?) obj\/s$/\3/p'
+  )"
+  latency_human="$(
+    echo "$reqs_line" \
+      | sed -nE 's/^[[:space:]]*\* Reqs: Avg: ([0-9]+(\.[0-9]+)?)(ms|us|µs|s),.*$/\1 \3/p'
+  )"
+
+  echo "${throughput_human:-N/A},${reqps:-N/A},${latency_human:-N/A}"
+}
+
+median_from_numbers() {
+  local values="$1"
+  local count
+
+  count="$(printf '%s\n' "$values" | awk 'NF{c++} END{print c+0}')"
+  if [[ "$count" -eq 0 ]]; then
+    echo "N/A"
+    return
+  fi
+
+  printf '%s\n' "$values" | awk 'NF' | sort -n | awk '
+    {a[NR]=$1}
+    END{
+      n=NR
+      if (n==0) { print "N/A"; exit }
+      if (n%2==1) {
+        printf "%.6f\n", a[(n+1)/2]
+      } else {
+        printf "%.6f\n", (a[n/2]+a[n/2+1])/2
+      }
+    }'
+}
+
+rebuild_round_results() {
+  local round_csv="${OUT_DIR}/warp/round_results.csv"
+  local tmp_csv
+  tmp_csv="$(mktemp)"
+
+  while IFS=',' read -r size tool round attempt concurrency status throughput_human throughput_bps reqps latency_human latency_ms log_file; do
+    if [[ "$size" == "size" ]]; then
+      echo "$size,$tool,$round,$attempt,$concurrency,$status,$throughput_human,$throughput_bps,$reqps,$latency_human,$latency_ms,$log_file" >> "$tmp_csv"
+      continue
+    fi
+
+    if [[ "$status" == "ok" ]]; then
+      local metrics
+      metrics="$(extract_final_get_metrics "$log_file")"
+      throughput_human="$(echo "$metrics" | cut -d',' -f1)"
+      reqps="$(echo "$metrics" | cut -d',' -f2)"
+      latency_human="$(echo "$metrics" | cut -d',' -f3)"
+      throughput_bps="$(to_bps "$throughput_human")"
+      latency_ms="$(to_ms "$latency_human")"
+    fi
+
+    echo "$size,$tool,$round,$attempt,$concurrency,$status,$throughput_human,$throughput_bps,$reqps,$latency_human,$latency_ms,$log_file" >> "$tmp_csv"
+  done < "$round_csv"
+
+  mv "$tmp_csv" "$round_csv"
+}
+
+rebuild_median_summary() {
+  local round_csv="${OUT_DIR}/warp/round_results.csv"
+  local median_csv="${OUT_DIR}/warp/median_summary.csv"
+  local ok_rounds fail_rounds t_vals r_vals l_vals m_t m_r m_l
+
+  ok_rounds="$(awk -F',' 'NR>1 && $6=="ok" {c++} END{print c+0}' "$round_csv")"
+  fail_rounds="$(awk -F',' 'NR>1 && $6!="ok" {c++} END{print c+0}' "$round_csv")"
+  t_vals="$(awk -F',' 'NR>1 && $6=="ok" && $8!="N/A" {print $8}' "$round_csv")"
+  r_vals="$(awk -F',' 'NR>1 && $6=="ok" && $9!="N/A" {print $9}' "$round_csv")"
+  l_vals="$(awk -F',' 'NR>1 && $6=="ok" && $11!="N/A" {print $11}' "$round_csv")"
+
+  m_t="$(median_from_numbers "$t_vals")"
+  m_r="$(median_from_numbers "$r_vals")"
+  m_l="$(median_from_numbers "$l_vals")"
+
+  {
+    echo "size,tool,concurrency,successful_rounds,failed_rounds,median_throughput_bps,median_reqps,median_latency_ms"
+    echo "$SIZE,warp,$CONCURRENCY,$ok_rounds,$fail_rounds,$m_t,$m_r,$m_l"
+  } > "$median_csv"
+}
+
+rebuild_baseline_compare() {
+  local compare_csv="${OUT_DIR}/warp/baseline_compare.csv"
+  if [[ -z "$BASELINE_CSV" ]]; then
+    return
+  fi
+
+  echo "size,tool,concurrency,new_median_reqps,baseline_median_reqps,delta_reqps_pct,new_median_latency_ms,baseline_median_latency_ms,delta_latency_pct,new_median_throughput_bps,baseline_median_throughput_bps,delta_throughput_pct" > "$compare_csv"
+
+  awk -F',' '
+    NR==FNR {
+      if (FNR==1) next
+      b_req=$7
+      b_lat=$8
+      b_thr=$6
+      next
+    }
+    FNR==2 {
+      n_thr=$6
+      n_req=$7
+      n_lat=$8
+      dr="N/A"; dl="N/A"; dt="N/A"
+      if (b_req!="N/A" && n_req!="N/A" && b_req+0!=0) dr=sprintf("%.2f", ((n_req-b_req)/b_req)*100)
+      if (b_lat!="N/A" && n_lat!="N/A" && b_lat+0!=0) dl=sprintf("%.2f", ((n_lat-b_lat)/b_lat)*100)
+      if (b_thr!="N/A" && n_thr!="N/A" && b_thr+0!=0) dt=sprintf("%.2f", ((n_thr-b_thr)/b_thr)*100)
+      print $1 "," $2 "," $3 "," n_req "," b_req "," dr "," n_lat "," b_lat "," dl "," n_thr "," b_thr "," dt
+    }
+  ' "$BASELINE_CSV" "${OUT_DIR}/warp/median_summary.csv" >> "$compare_csv"
+}
+
+postprocess_results() {
+  rebuild_round_results
+  rebuild_median_summary
+  rebuild_baseline_compare
+}
+
+main() {
+  parse_args "$@"
+  validate_args
+  setup_paths
+  write_manifest
+  build_rustfs_if_needed
+
+  trap stop_server EXIT INT TERM
+
+  echo "Output dir: $OUT_DIR"
+  start_server
+  run_bench
+  postprocess_results
+  stop_server
+}
+
+main "$@"