diff --git a/crates/config/src/constants/zero_copy.rs b/crates/config/src/constants/zero_copy.rs index ff23d6059..087f49acf 100644 --- a/crates/config/src/constants/zero_copy.rs +++ b/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 // ============================================================================= diff --git a/crates/e2e_test/src/archive_download_integrity_test.rs b/crates/e2e_test/src/archive_download_integrity_test.rs index 7bbc1d6dd..d5015e844 100644 --- a/crates/e2e_test/src/archive_download_integrity_test.rs +++ b/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 { + buf.windows(4).position(|window| window == b"\r\n\r\n") + } + + async fn read_proxy_request(stream: &mut tokio::net::TcpStream) -> Result<(), Box> { + 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>>), Box> { + 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 = 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> { + 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::().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::()?) + .header( + HOST, + object_url + .parse::()? + .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::().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(()) + } } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index c75a8041f..f605a92ff 100644 --- a/crates/io-metrics/src/lib.rs +++ b/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); diff --git a/crates/io-metrics/src/metric_names.rs b/crates/io-metrics/src/metric_names.rs index 6c611b6aa..8387e6332 100644 --- a/crates/io-metrics/src/metric_names.rs +++ b/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"; } diff --git a/crates/object-io/src/get.rs b/crates/object-io/src/get.rs index 4e0d695af..c776ee669 100644 --- a/crates/object-io/src/get.rs +++ b/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, diff --git a/rustfs/src/app/object_usecase/get_object_flow.rs b/rustfs/src/app/object_usecase/get_object_flow.rs index c112ff24f..8cd885cb5 100644 --- a/rustfs/src/app/object_usecase/get_object_flow.rs +++ b/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( 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"); + } } diff --git a/rustfs/src/app/object_usecase/get_object_zero_copy.rs b/rustfs/src/app/object_usecase/get_object_zero_copy.rs index bd2bb02ce..a799cf822 100644 --- a/rustfs/src/app/object_usecase/get_object_zero_copy.rs +++ b/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 { + 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::)); + 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); + } +} diff --git a/rustfs/src/app/object_usecase/zero_copy_tests.rs b/rustfs/src/app/object_usecase/zero_copy_tests.rs index fe0701a6e..65b262e3b 100644 --- a/rustfs/src/app/object_usecase/zero_copy_tests.rs +++ b/rustfs/src/app/object_usecase/zero_copy_tests.rs @@ -325,6 +325,16 @@ fn build_request(input: T, method: Method) -> S3Request { } } +async fn execute_get_object_with_fast_path_enabled( + usecase: &DefaultObjectUsecase, + req: S3Request, +) -> S3Result> { + 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();