fix(get-object): harden GET fast path against mid-stream regressions (#2472)

crates/config/src/constants/zero_copy.rs

@@ -17,6 +17,24 @@
 //! This module defines environment variables and default values for zero-copy
 //! read operations, which use memory mapping (mmap) to avoid data copying.
 
+// =============================================================================
+// GET Fast Path Configuration
+// =============================================================================
+
+/// Environment variable for the GetObject chunk fast path master switch.
+///
+/// When disabled, `GetObject` bypasses the chunk-streaming fast path entirely and
+/// always uses the legacy reader path. This provides an operational stopgap for
+/// regressions in the streaming data plane while keeping zero-copy internals
+/// configurable independently for future opt-in validation.
+pub const ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE: &str = "RUSTFS_OBJECT_GET_CHUNK_FAST_PATH_ENABLE";
+
+/// Default: GetObject chunk fast path is disabled.
+///
+/// The legacy reader path remains the safe default until the chunk-streaming
+/// path has sufficient regression coverage for full-body delivery semantics.
+pub const DEFAULT_OBJECT_GET_CHUNK_FAST_PATH_ENABLE: bool = false;
+
 // =============================================================================
 // Zero-Copy Configuration
 // =============================================================================

crates/e2e_test/src/archive_download_integrity_test.rs

@@ -14,7 +14,7 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path};
+    use crate::common::{RustFSTestEnvironment, init_logging, local_http_client, rustfs_binary_path};
     use aws_sdk_s3::primitives::ByteStream;
     use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
     use http::header::{CONTENT_TYPE, HOST};
@@ -28,6 +28,8 @@ mod tests {
     use std::io::{Cursor, Write};
     use std::process::Command;
     use time::OffsetDateTime;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
     use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions};
 
     const ARCHIVE_TEST_BUCKET: &str = "archive-download-integrity";
@@ -116,6 +118,75 @@ mod tests {
             .await?)
     }
 
+    fn find_header_terminator(buf: &[u8]) -> Option<usize> {
+        buf.windows(4).position(|window| window == b"\r\n\r\n")
+    }
+
+    async fn read_proxy_request(stream: &mut tokio::net::TcpStream) -> Result<(), Box<dyn Error + Send + Sync>> {
+        let mut buffer = Vec::new();
+        let mut chunk = [0_u8; 4096];
+
+        loop {
+            let read = stream.read(&mut chunk).await?;
+            if read == 0 {
+                return Err("proxy request ended before headers were fully received".into());
+            }
+            buffer.extend_from_slice(&chunk[..read]);
+            if find_header_terminator(&buffer).is_some() {
+                return Ok(());
+            }
+        }
+    }
+
+    async fn spawn_reverse_proxy_to_presigned_url(
+        target_url: String,
+    ) -> Result<(String, tokio::task::JoinHandle<Result<(), Box<dyn Error + Send + Sync>>>), Box<dyn Error + Send + Sync>> {
+        let listener = TcpListener::bind("127.0.0.1:0").await?;
+        let address = listener.local_addr()?;
+        let proxy_url = format!("http://{address}/");
+
+        let handle = tokio::spawn(async move {
+            let (mut downstream, _) = listener.accept().await?;
+            read_proxy_request(&mut downstream).await?;
+
+            let upstream_response: Result<reqwest::Response, reqwest::Error> = local_http_client().get(&target_url).send().await;
+            let (status, body, content_type) = match upstream_response {
+                Ok(response) => {
+                    let status = response.status();
+                    let content_type = response
+                        .headers()
+                        .get("content-type")
+                        .and_then(|value| value.to_str().ok())
+                        .map(str::to_string);
+                    match response.bytes().await {
+                        Ok(body) => (status, body.to_vec(), content_type),
+                        Err(err) => {
+                            let body = format!("upstream body read failed: {err}").into_bytes();
+                            (StatusCode::BAD_GATEWAY, body, Some("text/plain".to_string()))
+                        }
+                    }
+                }
+                Err(err) => {
+                    let body = format!("upstream request failed: {err}").into_bytes();
+                    (StatusCode::BAD_GATEWAY, body, Some("text/plain".to_string()))
+                }
+            };
+
+            let mut response_head = format!("HTTP/1.1 {}\r\ncontent-length: {}\r\nconnection: close\r\n", status, body.len());
+            if let Some(content_type) = content_type {
+                response_head.push_str(&format!("content-type: {content_type}\r\n"));
+            }
+            response_head.push_str("\r\n");
+
+            downstream.write_all(response_head.as_bytes()).await?;
+            downstream.write_all(&body).await?;
+            downstream.shutdown().await?;
+            Ok(())
+        });
+
+        Ok((proxy_url, handle))
+    }
+
     async fn signed_put_request_with_headers(
         url: &str,
         access_key: &str,
@@ -326,4 +397,130 @@ mod tests {
         env.stop_server();
         Ok(())
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_presigned_get_and_reverse_proxy_preserve_multipart_bytes_with_fast_path()
+    -> Result<(), Box<dyn Error + Send + Sync>> {
+        init_logging();
+        let mut env = RustFSTestEnvironment::new().await?;
+        start_rustfs_server_with_env(&mut env, &[("RUSTFS_OBJECT_GET_CHUNK_FAST_PATH_ENABLE", "true")]).await?;
+        env.create_test_bucket(MULTIPART_ARCHIVE_TEST_BUCKET).await?;
+
+        let client = env.create_s3_client();
+        let payload = random_bytes(MULTIPART_PART_SIZE + 768 * 1024);
+        let zip_bytes = build_zip_bytes(&[("payload.bin", payload.as_slice())])?;
+        assert!(zip_bytes.len() > MULTIPART_PART_SIZE, "zip payload must exceed multipart threshold");
+
+        let create_output = client
+            .create_multipart_upload()
+            .bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
+            .key("presigned-multipart-bundle.zip")
+            .content_type("application/zip")
+            .send()
+            .await?;
+        let upload_id = create_output.upload_id().expect("multipart upload id");
+
+        let first_part = zip_bytes[..MULTIPART_PART_SIZE].to_vec();
+        let second_part = zip_bytes[MULTIPART_PART_SIZE..].to_vec();
+
+        let upload_part_1 = client
+            .upload_part()
+            .bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
+            .key("presigned-multipart-bundle.zip")
+            .upload_id(upload_id)
+            .part_number(1)
+            .body(ByteStream::from(first_part))
+            .send()
+            .await?;
+
+        let upload_part_2 = client
+            .upload_part()
+            .bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
+            .key("presigned-multipart-bundle.zip")
+            .upload_id(upload_id)
+            .part_number(2)
+            .body(ByteStream::from(second_part))
+            .send()
+            .await?;
+
+        let completed_upload = CompletedMultipartUpload::builder()
+            .parts(
+                CompletedPart::builder()
+                    .part_number(1)
+                    .e_tag(upload_part_1.e_tag().unwrap_or_default())
+                    .build(),
+            )
+            .parts(
+                CompletedPart::builder()
+                    .part_number(2)
+                    .e_tag(upload_part_2.e_tag().unwrap_or_default())
+                    .build(),
+            )
+            .build();
+
+        client
+            .complete_multipart_upload()
+            .bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
+            .key("presigned-multipart-bundle.zip")
+            .upload_id(upload_id)
+            .multipart_upload(completed_upload)
+            .send()
+            .await?;
+
+        let object_url = format!("{}/{}/{}", env.url, MULTIPART_ARCHIVE_TEST_BUCKET, "presigned-multipart-bundle.zip");
+        let direct_response =
+            presigned_get_request_with_accept_encoding(&object_url, &env.access_key, &env.secret_key, "identity").await?;
+        assert_eq!(direct_response.status(), StatusCode::OK);
+        assert_eq!(
+            direct_response
+                .headers()
+                .get("content-length")
+                .and_then(|value| value.to_str().ok())
+                .and_then(|value| value.parse::<usize>().ok()),
+            Some(zip_bytes.len())
+        );
+        let direct_body = direct_response.bytes().await?;
+        assert_eq!(direct_body.len(), zip_bytes.len());
+        assert_eq!(direct_body.as_ref(), zip_bytes.as_slice());
+
+        let signed = pre_sign_v4(
+            http::Request::builder()
+                .method(http::Method::GET)
+                .uri(object_url.parse::<http::Uri>()?)
+                .header(
+                    HOST,
+                    object_url
+                        .parse::<http::Uri>()?
+                        .authority()
+                        .ok_or("request URL missing authority")?
+                        .to_string(),
+                )
+                .body(Body::empty())?,
+            &env.access_key,
+            &env.secret_key,
+            "",
+            "us-east-1",
+            600,
+            OffsetDateTime::now_utc(),
+        );
+        let (proxy_url, proxy_handle) = spawn_reverse_proxy_to_presigned_url(signed.uri().to_string()).await?;
+        let proxied_response: reqwest::Response = local_http_client().get(&proxy_url).send().await?;
+        assert_eq!(proxied_response.status(), StatusCode::OK);
+        assert_eq!(
+            proxied_response
+                .headers()
+                .get("content-length")
+                .and_then(|value| value.to_str().ok())
+                .and_then(|value| value.parse::<usize>().ok()),
+            Some(zip_bytes.len())
+        );
+        let proxied_body: bytes::Bytes = proxied_response.bytes().await?;
+        assert_eq!(proxied_body.len(), zip_bytes.len());
+        assert_eq!(proxied_body.as_ref(), zip_bytes.as_slice());
+
+        proxy_handle.await??;
+        env.stop_server();
+        Ok(())
+    }
 }

crates/io-metrics/src/lib.rs

@@ -193,6 +193,8 @@ impl IoStage {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum FallbackReason {
     Unknown,
+    FeatureDisabled,
+    ProbeFailed,
     MmapDisabled,
     MmapUnavailable,
     SmallObject,
@@ -213,6 +215,8 @@ impl FallbackReason {
     pub const fn as_str(self) -> &'static str {
         match self {
             Self::Unknown => "unknown",
+            Self::FeatureDisabled => "feature_disabled",
+            Self::ProbeFailed => "probe_failed",
             Self::MmapDisabled => "mmap_disabled",
             Self::MmapUnavailable => "mmap_unavailable",
             Self::SmallObject => "small_object",
@@ -347,6 +351,59 @@ pub fn record_io_fallback(stage: IoStage, reason: FallbackReason) {
     .increment(1);
 }
 
+/// Record a selected GET chunk fast path.
+#[inline(always)]
+pub fn record_get_object_fast_path_selected(path: &'static str, copy_mode: CopyMode, promised_bytes: i64) {
+    counter!(
+        metric_names::data_plane::GET_FAST_PATH_SELECTED_TOTAL,
+        "path" => path.to_string(),
+        "copy_mode" => copy_mode.as_str().to_string()
+    )
+    .increment(1);
+
+    if promised_bytes >= 0 {
+        histogram!(metric_names::data_plane::GET_FAST_PATH_PROMISED_BYTES).record(promised_bytes as f64);
+    }
+}
+
+/// Record a failed GET chunk fast path probe before the response is committed.
+#[inline(always)]
+pub fn record_get_object_fast_path_probe_failed(path: &'static str, copy_mode: CopyMode, promised_bytes: i64) {
+    counter!(
+        metric_names::data_plane::GET_FAST_PATH_PROBE_FAILED_TOTAL,
+        "path" => path.to_string(),
+        "copy_mode" => copy_mode.as_str().to_string()
+    )
+    .increment(1);
+
+    if promised_bytes >= 0 {
+        histogram!(metric_names::data_plane::GET_FAST_PATH_PROMISED_BYTES).record(promised_bytes as f64);
+    }
+}
+
+/// Record a GET chunk fast path mid-stream error after headers have already been committed.
+#[inline(always)]
+pub fn record_get_object_fast_path_midstream_error(
+    path: &'static str,
+    copy_mode: CopyMode,
+    error_kind: &'static str,
+    sent_bytes: usize,
+    promised_bytes: i64,
+) {
+    counter!(
+        metric_names::data_plane::GET_FAST_PATH_MIDSTREAM_ERROR_TOTAL,
+        "path" => path.to_string(),
+        "copy_mode" => copy_mode.as_str().to_string(),
+        "error_kind" => error_kind.to_string()
+    )
+    .increment(1);
+
+    histogram!(metric_names::data_plane::GET_FAST_PATH_MIDSTREAM_SENT_BYTES).record(sent_bytes as f64);
+    if promised_bytes >= 0 {
+        histogram!(metric_names::data_plane::GET_FAST_PATH_PROMISED_BYTES).record(promised_bytes as f64);
+    }
+}
+
 /// Record the currently active mmap bytes held by LocalDisk chunk streams.
 #[inline(always)]
 pub fn record_local_disk_active_mmap_bytes(active_bytes: usize) {
@@ -875,6 +932,8 @@ mod tests {
     #[test]
     fn test_fallback_reason_as_str_values_stable() {
         assert_eq!(FallbackReason::Unknown.as_str(), "unknown");
+        assert_eq!(FallbackReason::FeatureDisabled.as_str(), "feature_disabled");
+        assert_eq!(FallbackReason::ProbeFailed.as_str(), "probe_failed");
         assert_eq!(FallbackReason::MmapDisabled.as_str(), "mmap_disabled");
         assert_eq!(FallbackReason::MmapUnavailable.as_str(), "mmap_unavailable");
         assert_eq!(FallbackReason::SmallObject.as_str(), "small_object");
@@ -928,6 +987,13 @@ mod tests {
         record_local_disk_compat_collect(3, 16384);
     }
 
+    #[test]
+    fn test_record_get_object_fast_path_metrics() {
+        record_get_object_fast_path_selected("direct", CopyMode::TrueZeroCopy, 8192);
+        record_get_object_fast_path_probe_failed("bridge", CopyMode::SingleCopy, 4096);
+        record_get_object_fast_path_midstream_error("direct", CopyMode::Reconstructed, "unexpected_eof", 2048, 8192);
+    }
+
     #[test]
     fn test_record_put_object_attempted_fast_path() {
         record_put_object_attempted_fast_path(1024 * 1024);

crates/io-metrics/src/metric_names.rs

@@ -54,4 +54,19 @@ pub mod data_plane {
 
     /// Size distribution for transformed PUT selections.
     pub const PUT_TRANSFORM_SIZE_BYTES: &str = "rustfs.io.put.transform.size.bytes";
+
+    /// Total number of selected GET chunk fast paths.
+    pub const GET_FAST_PATH_SELECTED_TOTAL: &str = "rustfs.io.get.fast_path.selected_total";
+
+    /// Total number of GET chunk fast path probe failures before response commit.
+    pub const GET_FAST_PATH_PROBE_FAILED_TOTAL: &str = "rustfs.io.get.fast_path.probe_failed_total";
+
+    /// Total number of GET chunk fast path mid-stream errors after response commit.
+    pub const GET_FAST_PATH_MIDSTREAM_ERROR_TOTAL: &str = "rustfs.io.get.fast_path.midstream_error_total";
+
+    /// Byte distribution promised by GET chunk fast path selections or failures.
+    pub const GET_FAST_PATH_PROMISED_BYTES: &str = "rustfs.io.get.fast_path.promised.bytes";
+
+    /// Byte distribution already sent when a GET chunk fast path fails mid-stream.
+    pub const GET_FAST_PATH_MIDSTREAM_SENT_BYTES: &str = "rustfs.io.get.fast_path.midstream_sent.bytes";
 }

crates/object-io/src/get.rs

@@ -143,6 +143,14 @@ pub fn chunk_body_data_plane_labels(
     )
 }
 
+#[must_use]
+pub const fn get_object_chunk_path_label(path: GetObjectChunkPath) -> &'static str {
+    match path {
+        GetObjectChunkPath::Direct => "direct",
+        GetObjectChunkPath::Bridge => "bridge",
+    }
+}
+
 pub fn get_object_chunk_fast_path_guard(
     has_sse_customer_key: bool,
     has_sse_customer_key_md5: bool,

rustfs/src/app/object_usecase/get_object_flow.rs

@@ -19,9 +19,11 @@ use crate::error::ApiError;
 use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_buffer_size_opt_in};
 use crate::storage::options::filter_object_metadata;
 use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig};
+use futures_util::StreamExt;
 use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
 use rustfs_ecstore::error::StorageError;
-use rustfs_ecstore::store_api::{HTTPRangeSpec, ObjectInfo};
+use rustfs_ecstore::store_api::{GetObjectChunkPath, HTTPRangeSpec, ObjectInfo};
+use rustfs_io_core::BoxChunkStream;
 use rustfs_object_io::get::{
     GetObjectBodyPlan as ObjectIoGetObjectBodyPlan, GetObjectBodyPlanningInputs as ObjectIoGetObjectBodyPlanningInputs,
     GetObjectBodySource, GetObjectDataPlaneMetricContract as ObjectIoGetObjectDataPlaneMetricContract, GetObjectFlowResult,
@@ -31,6 +33,7 @@ use rustfs_object_io::get::{
     build_get_object_checksums as object_io_build_get_object_checksums,
     build_get_object_output_context as object_io_build_get_object_output_context,
     chunk_body_data_plane_labels as object_io_chunk_body_data_plane_labels,
+    get_object_chunk_path_label as object_io_get_object_chunk_path_label,
     materialize_get_object_body as object_io_materialize_get_object_body, plan_get_object_body as object_io_plan_get_object_body,
     plan_get_object_strategy_layout as object_io_plan_get_object_strategy_layout,
 };
@@ -48,6 +51,65 @@ pub(super) struct GetObjectBootstrap {
     pub(super) _deadlock_request_guard: DeadlockRequestGuard,
 }
 
+fn classify_get_object_midstream_error(err: &std::io::Error) -> &'static str {
+    let lower = err.to_string().to_ascii_lowercase();
+    if lower.contains("bitrot") {
+        "bitrot"
+    } else if lower.contains("decode") {
+        "decode"
+    } else {
+        match err.kind() {
+            std::io::ErrorKind::UnexpectedEof => "unexpected_eof",
+            std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionAborted => "channel_closed",
+            _ => "io_other",
+        }
+    }
+}
+
+fn instrument_get_object_chunk_stream(
+    request_context: &GetObjectRequestContext,
+    chunk_stream: BoxChunkStream,
+    path: GetObjectChunkPath,
+    copy_mode: rustfs_io_metrics::CopyMode,
+    response_content_length: i64,
+) -> BoxChunkStream {
+    let bucket = request_context.bucket.clone();
+    let key = request_context.key.clone();
+    let version_id = request_context.opts.version_id.clone();
+    let path_label = object_io_get_object_chunk_path_label(path);
+    let mut sent_bytes = 0usize;
+
+    Box::pin(chunk_stream.map(move |result| match result {
+        Ok(chunk) => {
+            sent_bytes = sent_bytes.saturating_add(chunk.len());
+            Ok(chunk)
+        }
+        Err(err) => {
+            let error_kind = classify_get_object_midstream_error(&err);
+            rustfs_io_metrics::record_get_object_fast_path_midstream_error(
+                path_label,
+                copy_mode,
+                error_kind,
+                sent_bytes,
+                response_content_length,
+            );
+            warn!(
+                bucket = %bucket,
+                key = %key,
+                version_id = ?version_id,
+                path = path_label,
+                copy_mode = copy_mode.as_str(),
+                promised_bytes = response_content_length,
+                sent_bytes,
+                error_kind,
+                error = %err,
+                "GetObject chunk fast path failed mid-stream after response commit"
+            );
+            Err(err)
+        }
+    }))
+}
+
 async fn build_get_object_body_adapter<R>(
     final_stream: R,
     bucket: &str,
@@ -300,6 +362,8 @@ pub(super) async fn build_get_object_output_context(
             copy_mode,
         } => {
             let (io_path, copy_mode) = object_io_chunk_body_data_plane_labels(path, copy_mode);
+            let chunk_stream =
+                instrument_get_object_chunk_stream(request_context, chunk_stream, path, copy_mode, response_content_length);
             (
                 object_io_build_chunk_blob(chunk_stream),
                 ObjectIoGetObjectDataPlaneMetricContract::disk(io_path, copy_mode),
@@ -374,6 +438,7 @@ mod tests {
     use super::*;
     use http::HeaderMap;
     use rustfs_ecstore::store_api::ObjectOptions;
+    use rustfs_io_core::IoChunk;
 
     fn sample_range(start: i64, end: i64) -> HTTPRangeSpec {
         HTTPRangeSpec {
@@ -417,4 +482,48 @@ mod tests {
         assert_eq!(strategy_range.start, 0);
         assert_eq!(strategy_range.end, 511);
     }
+
+    #[test]
+    fn classify_get_object_midstream_error_maps_expected_variants() {
+        assert_eq!(
+            classify_get_object_midstream_error(&std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "eof")),
+            "unexpected_eof"
+        );
+        assert_eq!(
+            classify_get_object_midstream_error(&std::io::Error::new(std::io::ErrorKind::BrokenPipe, "closed")),
+            "channel_closed"
+        );
+        assert_eq!(
+            classify_get_object_midstream_error(&std::io::Error::other("bitrot verification failed")),
+            "bitrot"
+        );
+        assert_eq!(
+            classify_get_object_midstream_error(&std::io::Error::other("decode chunk failed")),
+            "decode"
+        );
+    }
+
+    #[tokio::test]
+    async fn instrument_get_object_chunk_stream_preserves_payload() {
+        let request_context = sample_request_context();
+        let chunk_stream: BoxChunkStream = Box::pin(futures_util::stream::iter(vec![
+            Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello"))),
+            Ok(IoChunk::Shared(bytes::Bytes::from_static(b" world"))),
+        ]));
+
+        let mut instrumented = instrument_get_object_chunk_stream(
+            &request_context,
+            chunk_stream,
+            GetObjectChunkPath::Direct,
+            rustfs_io_metrics::CopyMode::SharedBytes,
+            11,
+        );
+
+        let mut collected = Vec::new();
+        while let Some(chunk) = instrumented.next().await {
+            collected.extend_from_slice(chunk.unwrap().as_bytes().as_ref());
+        }
+
+        assert_eq!(collected, b"hello world");
+    }
 }

rustfs/src/app/object_usecase/get_object_zero_copy.rs

@@ -20,21 +20,56 @@ use crate::storage::{
     DecryptionRequest, check_preconditions, get_validated_store, sse_decryption, validate_sse_headers_for_read,
     validate_ssec_for_read,
 };
+use futures_util::{StreamExt, stream};
 use http::HeaderMap;
 use rustfs_concurrency::GetObjectQueueSnapshot;
 use rustfs_ecstore::store_api::{ObjectIO, ObjectOperations};
+use rustfs_io_core::{BoxChunkStream, IoChunk};
 use rustfs_object_io::get::{
     ChunkReadDecision, ChunkReadPlanError, GetObjectEncryptionState as ObjectIoGetObjectEncryptionState, GetObjectReadSetup,
     build_reader_read_setup as object_io_build_reader_read_setup,
     finalize_chunk_read_setup as object_io_finalize_chunk_read_setup,
-    get_object_chunk_fast_path_guard as object_io_get_object_chunk_fast_path_guard, plan_chunk_read as object_io_plan_chunk_read,
-    plan_legacy_read as object_io_plan_legacy_read,
+    get_object_chunk_fast_path_guard as object_io_get_object_chunk_fast_path_guard,
+    get_object_chunk_path_label as object_io_get_object_chunk_path_label, map_chunk_copy_mode as object_io_map_chunk_copy_mode,
+    plan_chunk_read as object_io_plan_chunk_read, plan_legacy_read as object_io_plan_legacy_read,
 };
 use rustfs_rio::{Reader, WarpReader};
 use s3s::{S3Error, S3ErrorCode, S3Result, s3_error};
 use std::time::Duration;
 use tracing::{debug, warn};
 
+fn get_object_chunk_fast_path_enabled() -> bool {
+    rustfs_utils::get_env_bool(
+        rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE,
+        rustfs_config::DEFAULT_OBJECT_GET_CHUNK_FAST_PATH_ENABLE,
+    )
+}
+
+async fn probe_chunk_stream_before_commit(
+    mut chunk_stream: BoxChunkStream,
+    response_content_length: i64,
+) -> Result<BoxChunkStream, rustfs_io_metrics::FallbackReason> {
+    if response_content_length <= 0 {
+        return Ok(chunk_stream);
+    }
+
+    let mut prefetched = Vec::new();
+
+    loop {
+        match chunk_stream.next().await {
+            Some(Ok(chunk)) => {
+                let chunk_len = chunk.len();
+                prefetched.push(chunk);
+                if chunk_len > 0 {
+                    let prefix = stream::iter(prefetched.into_iter().map(Ok::<IoChunk, std::io::Error>));
+                    return Ok(Box::pin(prefix.chain(chunk_stream)));
+                }
+            }
+            Some(Err(_)) | None => return Err(rustfs_io_metrics::FallbackReason::ProbeFailed),
+        }
+    }
+}
+
 pub(super) struct GetObjectIoPlanning<'a> {
     pub(super) _disk_permit: tokio::sync::SemaphorePermit<'a>,
     pub(super) permit_wait_duration: Duration,
@@ -229,17 +264,25 @@ pub(super) async fn prepare_get_object_read_execution<'a>(
     let store = get_validated_store(&request_context.bucket).await?;
 
     let read_start = std::time::Instant::now();
-    let read_setup = match object_io_get_object_chunk_fast_path_guard(
-        request_context.sse_customer_key.is_some(),
-        request_context.sse_customer_key_md5.is_some(),
-    ) {
-        Ok(()) => match prepare_get_object_chunk_read(request_context, &store, manager, read_start).await? {
-            Some(read_setup) => read_setup,
-            None => prepare_get_object_read(request_context, &store, manager, read_start).await?,
-        },
-        Err(fallback) => {
-            rustfs_io_metrics::record_io_fallback(fallback.stage, fallback.reason);
-            prepare_get_object_read(request_context, &store, manager, read_start).await?
+    let read_setup = if !get_object_chunk_fast_path_enabled() {
+        rustfs_io_metrics::record_io_fallback(
+            rustfs_io_metrics::IoStage::ReadSetup,
+            rustfs_io_metrics::FallbackReason::FeatureDisabled,
+        );
+        prepare_get_object_read(request_context, &store, manager, read_start).await?
+    } else {
+        match object_io_get_object_chunk_fast_path_guard(
+            request_context.sse_customer_key.is_some(),
+            request_context.sse_customer_key_md5.is_some(),
+        ) {
+            Ok(()) => match prepare_get_object_chunk_read(request_context, &store, manager, read_start).await? {
+                Some(read_setup) => read_setup,
+                None => prepare_get_object_read(request_context, &store, manager, read_start).await?,
+            },
+            Err(fallback) => {
+                rustfs_io_metrics::record_io_fallback(fallback.stage, fallback.reason);
+                prepare_get_object_read(request_context, &store, manager, read_start).await?
+            }
         }
     };
 
@@ -293,6 +336,7 @@ pub(super) async fn prepare_get_object_chunk_read(
         Err(ChunkReadPlanError::Io(err)) => return Err(ApiError::from(err).into()),
     };
     let rs = plan.rs.clone();
+    let response_content_length = plan.response_content_length;
 
     let read_duration = read_start.elapsed();
     manager.record_disk_operation(info.size as u64, read_duration, true).await;
@@ -318,8 +362,94 @@ pub(super) async fn prepare_get_object_chunk_read(
             return Ok(None);
         }
     };
+    let path_label = object_io_get_object_chunk_path_label(chunk_result.path);
+    let copy_mode = object_io_map_chunk_copy_mode(chunk_result.copy_mode);
+    let chunk_result = match probe_chunk_stream_before_commit(chunk_result.stream, response_content_length).await {
+        Ok(stream) => rustfs_ecstore::store_api::GetObjectChunkResult {
+            stream,
+            path: chunk_result.path,
+            copy_mode: chunk_result.copy_mode,
+        },
+        Err(reason) => {
+            rustfs_io_metrics::record_io_fallback(rustfs_io_metrics::IoStage::ReadSetup, reason);
+            rustfs_io_metrics::record_get_object_fast_path_probe_failed(path_label, copy_mode, response_content_length);
+            warn!(
+                bucket = %request_context.bucket,
+                key = %request_context.key,
+                version_id = ?request_context.opts.version_id,
+                path = path_label,
+                copy_mode = copy_mode.as_str(),
+                promised_bytes = response_content_length,
+                fallback_reason = reason.as_str(),
+                "GetObject chunk fast path probe failed before response commit"
+            );
+            return Ok(None);
+        }
+    };
+
     let setup_result = object_io_finalize_chunk_read_setup(info, event_info, chunk_result, plan);
+    rustfs_io_metrics::record_get_object_fast_path_selected(path_label, copy_mode, response_content_length);
     rustfs_io_metrics::record_io_path_selected("get", setup_result.io_path);
 
     Ok(Some(setup_result.read_setup))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{get_object_chunk_fast_path_enabled, probe_chunk_stream_before_commit};
+    use bytes::Bytes;
+    use futures_util::{StreamExt, stream};
+    use rustfs_io_core::{BoxChunkStream, IoChunk};
+
+    #[test]
+    fn get_object_chunk_fast_path_defaults_to_disabled() {
+        temp_env::with_var_unset(rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE, || {
+            assert!(!get_object_chunk_fast_path_enabled());
+        });
+    }
+
+    #[test]
+    fn get_object_chunk_fast_path_can_be_explicitly_enabled() {
+        temp_env::with_var(rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE, Some("true"), || {
+            assert!(get_object_chunk_fast_path_enabled());
+        });
+    }
+
+    #[tokio::test]
+    async fn probe_chunk_stream_before_commit_preserves_prefetched_payload() {
+        let stream: BoxChunkStream = Box::pin(stream::iter(vec![
+            Ok(IoChunk::Shared(Bytes::from_static(b"hello "))),
+            Ok(IoChunk::Shared(Bytes::from_static(b"world"))),
+        ]));
+
+        let mut probed = probe_chunk_stream_before_commit(stream, 11).await.unwrap();
+        let mut collected = Vec::new();
+        while let Some(chunk) = probed.next().await {
+            collected.extend_from_slice(chunk.unwrap().as_bytes().as_ref());
+        }
+
+        assert_eq!(collected, b"hello world");
+    }
+
+    #[tokio::test]
+    async fn probe_chunk_stream_before_commit_rejects_midstream_failure_before_first_chunk() {
+        let stream: BoxChunkStream = Box::pin(stream::iter(vec![Err(std::io::Error::other("probe failed"))]));
+
+        let err = match probe_chunk_stream_before_commit(stream, 1).await {
+            Ok(_) => panic!("expected probe failure"),
+            Err(err) => err,
+        };
+        assert_eq!(err, rustfs_io_metrics::FallbackReason::ProbeFailed);
+    }
+
+    #[tokio::test]
+    async fn probe_chunk_stream_before_commit_rejects_unexpected_empty_stream() {
+        let stream: BoxChunkStream = Box::pin(stream::empty());
+
+        let err = match probe_chunk_stream_before_commit(stream, 1).await {
+            Ok(_) => panic!("expected probe failure"),
+            Err(err) => err,
+        };
+        assert_eq!(err, rustfs_io_metrics::FallbackReason::ProbeFailed);
+    }
+}

rustfs/src/app/object_usecase/zero_copy_tests.rs

@@ -325,6 +325,16 @@ fn build_request<T>(input: T, method: Method) -> S3Request<T> {
     }
 }
 
+async fn execute_get_object_with_fast_path_enabled(
+    usecase: &DefaultObjectUsecase,
+    req: S3Request<GetObjectInput>,
+) -> S3Result<S3Response<GetObjectOutput>> {
+    temp_env::with_var(rustfs_config::ENV_OBJECT_GET_CHUNK_FAST_PATH_ENABLE, Some("true"), || async move {
+        usecase.execute_get_object(req).await
+    })
+    .await
+}
+
 #[tokio::test]
 async fn execute_get_object_rejects_zero_part_number() {
     let input = GetObjectInput::builder()
@@ -383,7 +393,7 @@ async fn prepare_get_object_chunk_read_marks_direct_path_for_single_disk_store()
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(payload.len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -476,7 +486,7 @@ async fn execute_get_object_range_marks_direct_path_for_single_disk_store() {
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(63 * 1024 + 1));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -552,7 +562,7 @@ async fn execute_get_object_range_marks_direct_path_for_multi_disk_store_without
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some((range_end - range_start + 1) as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -620,7 +630,7 @@ async fn execute_get_object_range_marks_reconstructed_path_for_multi_disk_store_
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some((range_end - range_start + 1) as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -669,7 +679,7 @@ async fn execute_get_object_part_number_marks_direct_path_for_single_disk_store(
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(parts[0].len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -717,7 +727,7 @@ async fn execute_get_object_whole_multipart_marks_direct_path_for_single_disk_st
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(parts.iter().map(|part| part.len() as i64).sum()));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -774,7 +784,7 @@ async fn execute_get_object_part_number_marks_reconstructed_path_for_multi_disk_
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(parts[0].len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -825,7 +835,7 @@ async fn execute_get_object_part_number_marks_reconstructed_path_for_second_mult
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(parts[1].len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -876,7 +886,7 @@ async fn execute_get_object_part_number_marks_reconstructed_path_for_final_multi
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(parts[2].len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -953,7 +963,7 @@ async fn execute_get_object_whole_multipart_marks_reconstructed_path_for_missing
     );
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some(expected.len() as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();
@@ -1018,7 +1028,7 @@ async fn execute_get_object_multipart_range_marks_reconstructed_path_for_missing
     }
 
     let usecase = DefaultObjectUsecase::without_context();
-    let response = usecase.execute_get_object(req).await.unwrap();
+    let response = execute_get_object_with_fast_path_enabled(&usecase, req).await.unwrap();
     assert_eq!(response.output.content_length, Some((range_end - range_start + 1) as i64));
     let mut body = response.output.body.expect("expected body");
     let mut collected = Vec::new();