From eb0dc2492134b6391f021e9619f2708f0428f6e9 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 12 Apr 2026 01:04:22 +0800 Subject: [PATCH] fix(get): validate fast path body before response commit (#2495) Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> --- .github/AGENTS.md | 6 + Cargo.lock | 114 ++--- Cargo.toml | 6 +- .../src/app/object_usecase/get_object_flow.rs | 401 ++++++++++-------- 4 files changed, 308 insertions(+), 219 deletions(-) diff --git a/.github/AGENTS.md b/.github/AGENTS.md index 92c3ef729..3d637371a 100644 --- a/.github/AGENTS.md +++ b/.github/AGENTS.md @@ -25,6 +25,12 @@ Current `test-and-lint` gate includes: - `cargo nextest run --all --exclude e2e_test` - `cargo test --all --doc` +- `cargo test -p rustfs get_object_chunk_fast_path` +- `cargo test -p rustfs materialize_chunk_stream_before_commit` +- `touch rustfs/build.rs` +- `cargo build -p rustfs --bins --jobs 2` +- `cargo test -p e2e_test archive_multipart_roundtrip_preserves_bytes` +- `cargo test -p e2e_test presigned_get_and_reverse_proxy_preserve_multipart_bytes_with_fast_path` - `cargo fmt --all --check` - `cargo clippy --all-targets --all-features -- -D warnings` - `./scripts/check_layer_dependencies.sh` diff --git a/Cargo.lock b/Cargo.lock index aa8c11728..8952f9fa4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -56,13 +56,13 @@ dependencies = [ [[package]] name = "aes" -version = "0.9.0-rc.4" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "04097e08a47d9ad181c2e1f4a5fabc9ae06ce8839a333ba9a949bcb0d31fd2a3" +checksum = "66bd29a732b644c0431c6140f370d097879203d79b80c94a6747ba0872adaef8" dependencies = [ "cipher 0.5.1", "cpubits", - "cpufeatures 0.2.17", + "cpufeatures 0.3.0", ] [[package]] @@ -72,7 +72,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e22c0c90bbe8d4f77c3ca9ddabe41a1f8382d6fc1f7cea89459d0f320371f972" dependencies = [ "aead", - "aes 0.9.0-rc.4", + "aes 0.9.0", "cipher 0.5.1", "ctr", "ghash", @@ -410,7 +410,7 @@ dependencies = [ "arrow-schema", "chrono", "half", - "indexmap 2.13.1", + "indexmap 2.14.0", "itoa", "lexical-core", "memchr", @@ -1425,9 +1425,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.2.59" +version = "1.2.60" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7a4d3ec6524d28a329fc53654bbadc9bdd7b0431f5d65f1a56ffb28a1ee5283" +checksum = "43c5703da9466b66a946814e1adf53ea2c90f10063b86290cc9eb67ce3478a20" dependencies = [ "find-msvc-tools", "jobserver", @@ -2345,7 +2345,7 @@ dependencies = [ "chrono", "half", "hashbrown 0.16.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "libc", "log", @@ -2549,7 +2549,7 @@ dependencies = [ "datafusion-functions-aggregate-common", "datafusion-functions-window-common", "datafusion-physical-expr-common", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "paste", "recursive", @@ -2565,7 +2565,7 @@ checksum = "ab05fdd00e05d5a6ee362882546d29d6d3df43a6c55355164a7fbee12d163bc9" dependencies = [ "arrow", "datafusion-common", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "paste", ] @@ -2729,7 +2729,7 @@ dependencies = [ "datafusion-expr", "datafusion-expr-common", "datafusion-physical-expr", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "log", "recursive", @@ -2752,7 +2752,7 @@ dependencies = [ "datafusion-physical-expr-common", "half", "hashbrown 0.16.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "parking_lot 0.12.5", "paste", @@ -2788,7 +2788,7 @@ dependencies = [ "datafusion-common", "datafusion-expr-common", "hashbrown 0.16.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "parking_lot 0.12.5", ] @@ -2835,7 +2835,7 @@ dependencies = [ "futures", "half", "hashbrown 0.16.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "itertools 0.14.0", "log", "num-traits", @@ -2887,7 +2887,7 @@ dependencies = [ "datafusion-common", "datafusion-expr", "datafusion-functions-nested", - "indexmap 2.13.1", + "indexmap 2.14.0", "log", "recursive", "regex", @@ -3065,7 +3065,7 @@ dependencies = [ [[package]] name = "dial9-tokio-telemetry" version = "0.2.0" -source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=60502082601b647c4a51962595721f631b7bbce1#60502082601b647c4a51962595721f631b7bbce1" +source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=2117416f6052fb66edb393f2a8a22c1fdcb91bbd#2117416f6052fb66edb393f2a8a22c1fdcb91bbd" dependencies = [ "arc-swap", "bon", @@ -3089,7 +3089,7 @@ dependencies = [ [[package]] name = "dial9-trace-format" version = "0.2.0" -source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=60502082601b647c4a51962595721f631b7bbce1#60502082601b647c4a51962595721f631b7bbce1" +source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=2117416f6052fb66edb393f2a8a22c1fdcb91bbd#2117416f6052fb66edb393f2a8a22c1fdcb91bbd" dependencies = [ "dial9-trace-format-derive", "serde", @@ -3098,7 +3098,7 @@ dependencies = [ [[package]] name = "dial9-trace-format-derive" version = "0.2.0" -source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=60502082601b647c4a51962595721f631b7bbce1#60502082601b647c4a51962595721f631b7bbce1" +source = "git+https://github.com/dial9-rs/dial9-tokio-telemetry.git?rev=2117416f6052fb66edb393f2a8a22c1fdcb91bbd#2117416f6052fb66edb393f2a8a22c1fdcb91bbd" dependencies = [ "proc-macro2", "quote", @@ -4072,7 +4072,7 @@ dependencies = [ "futures-core", "futures-sink", "http 1.4.0", - "indexmap 2.13.1", + "indexmap 2.14.0", "slab", "tokio", "tokio-util", @@ -4138,6 +4138,20 @@ dependencies = [ "serde_core", ] +[[package]] +name = "hashbrown" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", + "rayon", + "serde", + "serde_core", +] + [[package]] name = "headers" version = "0.4.1" @@ -4572,12 +4586,12 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.13.1" +version = "2.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "serde", "serde_core", ] @@ -4589,7 +4603,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88" dependencies = [ "ahash 0.8.12", - "indexmap 2.13.1", + "indexmap 2.14.0", "is-terminal", "itoa", "log", @@ -4612,7 +4626,7 @@ dependencies = [ "crossbeam-utils", "dashmap", "env_logger", - "indexmap 2.13.1", + "indexmap 2.14.0", "itoa", "log", "num-format", @@ -5056,14 +5070,14 @@ dependencies = [ [[package]] name = "libredox" -version = "0.1.15" +version = "0.1.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ddbf48fd451246b1f8c2610bd3b4ac0cc6e149d89832867093ab69a17194f08" +checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" dependencies = [ "bitflags 2.11.0", "libc", "plain", - "redox_syscall 0.7.3", + "redox_syscall 0.7.4", ] [[package]] @@ -5345,7 +5359,7 @@ dependencies = [ "crossbeam-epoch", "crossbeam-utils", "hashbrown 0.16.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "metrics", "ordered-float 5.3.0", "quanta", @@ -6371,7 +6385,7 @@ checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455" dependencies = [ "fixedbitset", "hashbrown 0.15.5", - "indexmap 2.13.1", + "indexmap 2.14.0", "serde", ] @@ -6801,7 +6815,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4aeaa1f2460f1d348eeaeed86aea999ce98c1bded6f089ff8514c9d9dbdc973" dependencies = [ "anyhow", - "indexmap 2.13.1", + "indexmap 2.14.0", "log", "protobuf", "protobuf-support", @@ -7198,9 +7212,9 @@ dependencies = [ [[package]] name = "redox_syscall" -version = "0.7.3" +version = "0.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ce70a74e890531977d37e532c34d45e9055d2409ed08ddba14529471ed0be16" +checksum = "f450ad9c3b1da563fb6948a8e0fb0fb9269711c9c73d9ea1de5058c79c8d643a" dependencies = [ "bitflags 2.11.0", ] @@ -7448,9 +7462,9 @@ checksum = "3582f63211428f83597b51b2ddb88e2a91a9d52d12831f9d08f5e624e8977422" [[package]] name = "rmcp" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2231b2c085b371c01bc90c0e6c1cab8834711b6394533375bdbf870b0166d419" +checksum = "f542f74cf247da16f19bbc87e298cd201e912314f4083e88cdd671f44f5fcb53" dependencies = [ "async-trait", "base64 0.22.1", @@ -7470,9 +7484,9 @@ dependencies = [ [[package]] name = "rmcp-macros" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36ea0e100fadf81be85d7ff70f86cd805c7572601d4ab2946207f36540854b43" +checksum = "b2391e4ae47f314e70eaafb6c7bd82e495e770b935448864446302143019151f" dependencies = [ "darling 0.23.0", "proc-macro2", @@ -7669,7 +7683,7 @@ dependencies = [ "flatbuffers", "futures", "futures-util", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "hex-simd", "http 1.4.0", "http-body 1.0.1", @@ -7780,7 +7794,7 @@ dependencies = [ "chrono", "const-str", "futures", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "metrics", "rumqttc-next", "rustfs-config", @@ -8207,7 +8221,7 @@ dependencies = [ "chrono", "form_urlencoded", "futures", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "quick-xml 0.39.2", "rayon", "rumqttc-next", @@ -8586,7 +8600,7 @@ dependencies = [ "crc-fast", "flate2", "futures", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "hex-simd", "highway", "hmac 0.13.0", @@ -8775,9 +8789,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.10" +version = "0.103.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" +checksum = "20a6af516fea4b20eccceaf166e8aa666ac996208e8a644ce3ef5aa783bc7cd4" dependencies = [ "aws-lc-rs", "ring", @@ -9098,7 +9112,7 @@ dependencies = [ "chrono", "hex", "indexmap 1.9.3", - "indexmap 2.13.1", + "indexmap 2.14.0", "schemars 0.9.0", "schemars 1.2.1", "serde_core", @@ -10118,7 +10132,7 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", - "indexmap 2.13.1", + "indexmap 2.14.0", "pin-project-lite", "slab", "sync_wrapper", @@ -10621,7 +10635,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" dependencies = [ "anyhow", - "indexmap 2.13.1", + "indexmap 2.14.0", "wasm-encoder", "wasmparser", ] @@ -10647,7 +10661,7 @@ checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ "bitflags 2.11.0", "hashbrown 0.15.5", - "indexmap 2.13.1", + "indexmap 2.14.0", "semver", ] @@ -11121,7 +11135,7 @@ checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" dependencies = [ "anyhow", "heck", - "indexmap 2.13.1", + "indexmap 2.14.0", "prettyplease", "syn 2.0.117", "wasm-metadata", @@ -11152,7 +11166,7 @@ checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", "bitflags 2.11.0", - "indexmap 2.13.1", + "indexmap 2.14.0", "log", "serde", "serde_derive", @@ -11171,7 +11185,7 @@ checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" dependencies = [ "anyhow", "id-arena", - "indexmap 2.13.1", + "indexmap 2.14.0", "log", "semver", "serde", @@ -11409,7 +11423,7 @@ dependencies = [ "flate2", "getrandom 0.4.2", "hmac 0.12.1", - "indexmap 2.13.1", + "indexmap 2.14.0", "lzma-rust2", "memchr", "pbkdf2 0.12.2", diff --git a/Cargo.toml b/Cargo.toml index 2f7fc9aec..52baeb561 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -154,7 +154,7 @@ flatbuffers = "25.12.19" form_urlencoded = "1.2.2" prost = "0.14.3" quick-xml = "0.39.2" -rmcp = { version = "1.3.0" } +rmcp = { version = "1.4.0" } rmp = { version = "0.8.15" } rmp-serde = { version = "1.3.1" } serde = { version = "1.0.228", features = ["derive"] } @@ -218,7 +218,7 @@ flate2 = "1.1.9" glob = "0.3.3" google-cloud-storage = "1.10.0" google-cloud-auth = "1.8.0" -hashbrown = { version = "0.16.1", features = ["serde", "rayon"] } +hashbrown = { version = "0.17.0", features = ["serde", "rayon"] } hex = "0.4.3" hex-simd = "0.8.0" highway = { version = "1.3.0" } @@ -288,7 +288,7 @@ zstd = "0.13.3" # Observability and Metrics metrics = "0.24.3" -dial9-tokio-telemetry = { version = "0.2", git = "https://github.com/dial9-rs/dial9-tokio-telemetry.git", rev = "60502082601b647c4a51962595721f631b7bbce1" } +dial9-tokio-telemetry = { version = "0.2", git = "https://github.com/dial9-rs/dial9-tokio-telemetry.git", rev = "2117416f6052fb66edb393f2a8a22c1fdcb91bbd" } opentelemetry = { version = "0.31.0" } opentelemetry-appender-tracing = { version = "0.31.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes", "spec_unstable_logs_enabled"] } opentelemetry-otlp = { version = "0.31.1", features = ["gzip-http", "reqwest-rustls"] } diff --git a/rustfs/src/app/object_usecase/get_object_flow.rs b/rustfs/src/app/object_usecase/get_object_flow.rs index 8cd885cb5..11415eac0 100644 --- a/rustfs/src/app/object_usecase/get_object_flow.rs +++ b/rustfs/src/app/object_usecase/get_object_flow.rs @@ -14,25 +14,28 @@ use super::DeadlockRequestGuard; use super::GetObjectRequestContext; -use super::get_object_zero_copy::{GetObjectIoPlanning, GetObjectPreparedRead, prepare_get_object_read_execution}; +use super::get_object_zero_copy::{ + GetObjectIoPlanning, GetObjectPreparedRead, prepare_get_object_read, prepare_get_object_read_execution, +}; use crate::error::ApiError; use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_buffer_size_opt_in}; +use crate::storage::get_validated_store; use crate::storage::options::filter_object_metadata; use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig}; +use bytes::Bytes; use futures_util::StreamExt; use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys; use rustfs_ecstore::error::StorageError; -use rustfs_ecstore::store_api::{GetObjectChunkPath, HTTPRangeSpec, ObjectInfo}; +use rustfs_ecstore::store_api::{HTTPRangeSpec, ObjectInfo}; use rustfs_io_core::BoxChunkStream; use rustfs_object_io::get::{ GetObjectBodyPlan as ObjectIoGetObjectBodyPlan, GetObjectBodyPlanningInputs as ObjectIoGetObjectBodyPlanningInputs, GetObjectBodySource, GetObjectDataPlaneMetricContract as ObjectIoGetObjectDataPlaneMetricContract, GetObjectFlowResult, GetObjectOutputContext, GetObjectReadSetup, MaterializeGetObjectBodyError as ObjectIoMaterializeGetObjectBodyError, - build_chunk_blob as object_io_build_chunk_blob, build_cors_wrapped_get_object_flow_result as object_io_build_cors_wrapped_get_object_flow_result, 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, + build_memory_blob as object_io_build_memory_blob, 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, @@ -51,63 +54,93 @@ 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", - } - } +#[derive(Debug)] +struct ChunkCommitMaterializationError { + source: std::io::Error, + streamed_bytes: usize, } -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; +fn build_chunk_materialization_length_error(actual: usize, expected: usize) -> std::io::Error { + let error_kind = if actual > expected { + std::io::ErrorKind::InvalidData + } else { + std::io::ErrorKind::UnexpectedEof + }; - Box::pin(chunk_stream.map(move |result| match result { - Ok(chunk) => { - sent_bytes = sent_bytes.saturating_add(chunk.len()); - Ok(chunk) + std::io::Error::new( + error_kind, + format!("chunk fast path produced {actual} bytes before response commit, expected {expected}"), + ) +} + +async fn materialize_chunk_stream_before_commit_with_threshold( + mut chunk_stream: BoxChunkStream, + response_content_length: i64, + optimal_buffer_size: usize, + in_memory_threshold_bytes: usize, +) -> Result, ChunkCommitMaterializationError> { + let expected_bytes = usize::try_from(response_content_length).map_err(|_| ChunkCommitMaterializationError { + source: std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("negative response content length {response_content_length} for chunk fast path"), + ), + streamed_bytes: 0, + })?; + + // Objects larger than the in-memory threshold fall back to the legacy reader path + // rather than spooling to disk, to avoid exhausting local disk under concurrent large downloads. + if expected_bytes > in_memory_threshold_bytes { + return Err(ChunkCommitMaterializationError { + source: std::io::Error::other(format!( + "chunk fast path object size {expected_bytes} exceeds in-memory threshold \ + {in_memory_threshold_bytes}; falling back to legacy reader" + )), + streamed_bytes: 0, + }); + } + + let mut buf = Vec::with_capacity(expected_bytes); + let mut streamed_bytes = 0usize; + + while let Some(result) = chunk_stream.next().await { + let chunk = result.map_err(|source| ChunkCommitMaterializationError { source, streamed_bytes })?; + let bytes = chunk.as_bytes(); + streamed_bytes = streamed_bytes.saturating_add(bytes.len()); + if streamed_bytes > expected_bytes { + return Err(ChunkCommitMaterializationError { + source: build_chunk_materialization_length_error(streamed_bytes, expected_bytes), + streamed_bytes, + }); } - 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) - } - })) + buf.extend_from_slice(bytes.as_ref()); + } + + if streamed_bytes != expected_bytes { + return Err(ChunkCommitMaterializationError { + source: build_chunk_materialization_length_error(streamed_bytes, expected_bytes), + streamed_bytes, + }); + } + + Ok(object_io_build_memory_blob( + Bytes::from(buf), + response_content_length, + optimal_buffer_size, + )) +} + +async fn materialize_chunk_stream_before_commit( + chunk_stream: BoxChunkStream, + response_content_length: i64, + optimal_buffer_size: usize, +) -> Result, ChunkCommitMaterializationError> { + materialize_chunk_stream_before_commit_with_threshold( + chunk_stream, + response_content_length, + optimal_buffer_size, + rustfs_config::DEFAULT_OBJECT_SEEK_SUPPORT_THRESHOLD, + ) + .await } async fn build_get_object_body_adapter( @@ -307,79 +340,14 @@ pub(super) async fn build_get_object_output_context( let bucket = &request_context.bucket; let key = &request_context.key; let part_number = request_context.part_number; - let GetObjectReadSetup { - info, - event_info, - body_source, - rs, - content_type, - last_modified, - response_content_length, - content_range, - server_side_encryption, - sse_customer_algorithm, - sse_customer_key_md5, - ssekms_key_id, - encryption_applied, - } = read_setup; + let mut active_read_setup = read_setup; - let optimal_buffer_size = finalize_get_object_strategy_runtime( - request_context, - rs.as_ref(), - manager, - base_buffer_size, - &info, - response_content_length, - io_planning, - ); - - let (body, metric_contract) = match body_source { - GetObjectBodySource::Reader(final_stream) => { - let body = build_get_object_body_adapter( - final_stream, - bucket, - key, - response_content_length, - optimal_buffer_size, - ObjectIoGetObjectBodyPlanningInputs { - is_part_request: part_number.is_some(), - is_range_request: rs.is_some(), - encryption_applied, - response_size: response_content_length, - }, - ) - .await?; - let metric_contract = ObjectIoGetObjectDataPlaneMetricContract::disk( - rustfs_io_metrics::IoPath::Legacy, - rustfs_io_metrics::CopyMode::SingleCopy, - ); - - (body, metric_contract) - } - GetObjectBodySource::Chunk { - stream: chunk_stream, - path, - 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), - ) - } - }; - - let checksums = object_io_build_get_object_checksums(&info, &request_context.headers, part_number, rs.as_ref()) - .map_err(ApiError::from)?; - let filtered_metadata = filter_object_metadata(&info.user_defined); - - Ok(( - object_io_build_get_object_output_context( - body, + loop { + let GetObjectReadSetup { info, event_info, + body_source, + rs, content_type, last_modified, response_content_length, @@ -388,14 +356,108 @@ pub(super) async fn build_get_object_output_context( sse_customer_algorithm, sse_customer_key_md5, ssekms_key_id, - &checksums, - filtered_metadata, - versioned, - optimal_buffer_size, - Some(metric_contract.copy_mode), - ), - metric_contract, - )) + encryption_applied, + } = active_read_setup; + + let optimal_buffer_size = finalize_get_object_strategy_runtime( + request_context, + rs.as_ref(), + manager, + base_buffer_size, + &info, + response_content_length, + io_planning, + ); + + let (body, metric_contract) = match body_source { + GetObjectBodySource::Reader(final_stream) => { + let body = build_get_object_body_adapter( + final_stream, + bucket, + key, + response_content_length, + optimal_buffer_size, + ObjectIoGetObjectBodyPlanningInputs { + is_part_request: part_number.is_some(), + is_range_request: rs.is_some(), + encryption_applied, + response_size: response_content_length, + }, + ) + .await?; + let metric_contract = ObjectIoGetObjectDataPlaneMetricContract::disk( + rustfs_io_metrics::IoPath::Legacy, + rustfs_io_metrics::CopyMode::SingleCopy, + ); + + (body, metric_contract) + } + GetObjectBodySource::Chunk { + stream: chunk_stream, + path, + copy_mode, + } => { + let (io_path, copy_mode) = object_io_chunk_body_data_plane_labels(path, copy_mode); + match materialize_chunk_stream_before_commit(chunk_stream, response_content_length, optimal_buffer_size).await { + Ok(body) => (body, ObjectIoGetObjectDataPlaneMetricContract::disk(io_path, copy_mode)), + Err(err) => { + let path_label = object_io_get_object_chunk_path_label(path); + rustfs_io_metrics::record_io_fallback( + rustfs_io_metrics::IoStage::ReadSetup, + rustfs_io_metrics::FallbackReason::ProbeFailed, + ); + 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, + materialized_bytes = err.streamed_bytes, + error = %err.source, + "GetObject chunk fast path full-body materialization failed before response commit" + ); + + let store = get_validated_store(&request_context.bucket).await?; + active_read_setup = + prepare_get_object_read(request_context, &store, manager, std::time::Instant::now()).await?; + continue; + } + } + } + }; + + let checksums = object_io_build_get_object_checksums(&info, &request_context.headers, part_number, rs.as_ref()) + .map_err(ApiError::from)?; + let filtered_metadata = filter_object_metadata(&info.user_defined); + + return Ok(( + object_io_build_get_object_output_context( + body, + info, + event_info, + content_type, + last_modified, + response_content_length, + content_range, + server_side_encryption, + sse_customer_algorithm, + sse_customer_key_md5, + ssekms_key_id, + &checksums, + filtered_metadata, + versioned, + optimal_buffer_size, + Some(metric_contract.copy_mode), + ), + metric_contract, + )); + } } pub(super) async fn run_get_object_flow( @@ -436,6 +498,7 @@ pub(super) async fn run_get_object_flow( mod tests { use super::get_object_strategy_range; use super::*; + use futures_util::StreamExt; use http::HeaderMap; use rustfs_ecstore::store_api::ObjectOptions; use rustfs_io_core::IoChunk; @@ -483,47 +546,53 @@ mod tests { 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(); + async fn materialize_chunk_stream_before_commit_buffers_small_payload_in_memory() { 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 body = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024) + .await + .unwrap() + .unwrap(); let mut collected = Vec::new(); - while let Some(chunk) = instrumented.next().await { - collected.extend_from_slice(chunk.unwrap().as_bytes().as_ref()); + while let Some(chunk) = body.next().await { + collected.extend_from_slice(&chunk.unwrap()); } assert_eq!(collected, b"hello world"); } + + #[tokio::test] + async fn materialize_chunk_stream_before_commit_falls_back_for_large_payload() { + 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"))), + ])); + + // When payload exceeds the in-memory threshold an error is returned so the + // caller can fall back to the legacy reader path rather than spooling to disk. + let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 4) + .await + .unwrap_err(); + + assert_eq!(err.streamed_bytes, 0); + assert_eq!(err.source.kind(), std::io::ErrorKind::Other); + } + + #[tokio::test] + async fn materialize_chunk_stream_before_commit_rejects_short_body() { + let chunk_stream: BoxChunkStream = + Box::pin(futures_util::stream::iter(vec![Ok(IoChunk::Shared(bytes::Bytes::from_static(b"hello")))])); + + let err = materialize_chunk_stream_before_commit_with_threshold(chunk_stream, 11, 8 * 1024, 1024) + .await + .unwrap_err(); + + assert_eq!(err.streamed_bytes, 5); + assert_eq!(err.source.kind(), std::io::ErrorKind::UnexpectedEof); + } }