diff --git a/crates/ecstore/src/erasure/coding/decode_reader.rs b/crates/ecstore/src/erasure/coding/decode_reader.rs index 21415021e..f8db2dccb 100644 --- a/crates/ecstore/src/erasure/coding/decode_reader.rs +++ b/crates/ecstore/src/erasure/coding/decode_reader.rs @@ -27,6 +27,8 @@ use std::io; use std::io::ErrorKind; use std::pin::Pin; use std::sync::Mutex; +#[cfg(test)] +use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context, Poll, ready}; use std::time::Instant; use tokio::io::{AsyncRead, ReadBuf}; @@ -38,6 +40,14 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT: usize = 2; const FILL_POLICY_SINGLE_INFLIGHT: &str = "single_inflight"; const FILL_POLICY_DUAL_INFLIGHT: &str = "dual_inflight"; +#[cfg(test)] +static SINGLE_INFLIGHT_CONSTRUCTIONS: AtomicU64 = AtomicU64::new(0); + +#[cfg(test)] +pub(crate) fn test_single_inflight_construction_count() -> u64 { + SINGLE_INFLIGHT_CONSTRUCTIONS.load(Ordering::Relaxed) +} + type FillTask = oneshot::Receiver; struct FillWorker { @@ -155,6 +165,23 @@ where Self::new_with_fill_policy_inner(source, engine, total_length, metrics_path, FillPolicy::from_env()) } + /// Construct the bounded reader without lookahead. + /// + /// Mid-size GETs are latency-sensitive and are already gated to a single + /// plain part. Keeping one stripe in flight avoids retaining a second + /// decoded output buffer while preserving the same source, reconstruction, + /// bitrot and cancellation semantics as the general streaming reader. + pub(crate) fn new_single_inflight_with_metrics_path( + source: S, + engine: E, + total_length: usize, + metrics_path: &'static str, + ) -> io::Result { + #[cfg(test)] + SINGLE_INFLIGHT_CONSTRUCTIONS.fetch_add(1, Ordering::Relaxed); + Self::new_with_fill_policy_inner(source, engine, total_length, metrics_path, FillPolicy::SingleInFlight) + } + fn new_with_fill_policy_inner( source: S, engine: E, @@ -602,7 +629,8 @@ where loop { if self.output_pos < self.output_buf.len() { - if self.prefetched_bufs.len() < self.fill_policy.max_inflight() + if self.fill_policy == FillPolicy::DualInFlight + && self.prefetched_bufs.len() < self.fill_policy.max_inflight() && self.prefetch_error.is_none() && self.remaining > 0 && let Poll::Ready(result) = self.poll_prefetch(cx) @@ -1620,6 +1648,149 @@ mod tests { assert_eq!(decoded, data); } + #[tokio::test] + async fn single_inflight_reader_reads_full_body_without_lookahead() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..96u8).collect::>(); + let read_count = Arc::new(AtomicUsize::new(0)); + let mut source = source_from_data(&erasure, &data, &[]); + source.read_count = Some(Arc::clone(&read_count)); + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path( + source, + engine, + data.len(), + GET_OBJECT_PATH_CODEC_STREAMING, + ) + .expect("single-inflight reader should be constructed"); + let mut decoded = Vec::new(); + + reader + .read_to_end(&mut decoded) + .await + .expect("single-inflight reader should decode the complete body"); + + assert_eq!(decoded, data); + assert_eq!( + read_count.load(Ordering::SeqCst), + 3, + "single-inflight must not read ahead after the final stripe" + ); + } + + #[tokio::test] + async fn single_inflight_reader_preserves_partial_reads() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..83u8).collect::>(); + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path( + source_from_data(&erasure, &data, &[]), + engine, + data.len(), + GET_OBJECT_PATH_CODEC_STREAMING, + ) + .expect("single-inflight reader should be constructed"); + let mut decoded = Vec::with_capacity(data.len()); + let mut chunk = [0u8; 3]; + + loop { + let read = reader.read(&mut chunk).await.expect("partial read should succeed"); + if read == 0 { + break; + } + decoded.extend_from_slice(&chunk[..read]); + } + + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn single_inflight_reader_does_not_prefetch_before_output_is_drained() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..96u8).collect::>(); + let read_count = Arc::new(AtomicUsize::new(0)); + let mut source = source_from_data(&erasure, &data, &[]); + source.read_count = Some(Arc::clone(&read_count)); + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path( + source, + engine, + data.len(), + GET_OBJECT_PATH_CODEC_STREAMING, + ) + .expect("single-inflight reader should be constructed"); + let mut first = [0u8; 3]; + + reader + .read_exact(&mut first) + .await + .expect("first partial read should succeed"); + + assert_eq!( + read_count.load(Ordering::SeqCst), + 1, + "single-inflight must not prefetch while output remains" + ); + assert_eq!(&first, &data[..3]); + } + + #[tokio::test] + async fn single_inflight_reader_reconstructs_degraded_body() { + let erasure = Erasure::new(4, 2, 32); + let data = (0..97u16).map(|value| value as u8).collect::>(); + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path( + source_from_data(&erasure, &data, &[1]), + engine, + data.len(), + GET_OBJECT_PATH_CODEC_STREAMING, + ) + .expect("single-inflight reader should be constructed"); + let mut decoded = Vec::new(); + + reader + .read_to_end(&mut decoded) + .await + .expect("a readable degraded stripe should be reconstructed"); + + assert_eq!(decoded, data); + } + + #[tokio::test] + async fn single_inflight_reader_surfaces_error_after_buffered_body() { + let erasure = Erasure::new(4, 2, 32); + let first_stripe = (0..32u8).collect::>(); + let first_state = source_from_data(&erasure, &first_stripe, &[]) + .stripes + .pop_front() + .expect("first stripe should exist"); + let source = VecStripeSource { + stripes: VecDeque::from([ + first_state, + StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards), + ]), + read_quorum: erasure.data_shards, + read_count: None, + }; + let engine = LegacyEcDecodeEngine::new(erasure); + let mut reader = ErasureDecodeReader::new_single_inflight_with_metrics_path( + source, + engine, + first_stripe.len() + 1, + GET_OBJECT_PATH_CODEC_STREAMING, + ) + .expect("single-inflight reader should be constructed"); + let mut decoded = Vec::new(); + + let error = reader + .read_to_end(&mut decoded) + .await + .expect_err("short source error should be returned after buffered bytes"); + + assert_eq!(error.kind(), ErrorKind::Other); + assert_eq!(decoded, first_stripe); + } + #[tokio::test] async fn erasure_decode_reader_stops_at_eof_for_empty_object() { let erasure = Erasure::new(4, 2, 32); @@ -1882,14 +2053,9 @@ mod tests { }; let engine = LegacyEcDecodeEngine::new(Erasure::new(1, 0, 32)); let task = tokio::spawn(async move { - let mut reader = ErasureDecodeReader::new_with_fill_policy( - source, - engine, - 1, - GET_OBJECT_PATH_CODEC_STREAMING, - FillPolicy::SingleInFlight, - ) - .expect("reader should be constructed"); + let mut reader = + ErasureDecodeReader::new_single_inflight_with_metrics_path(source, engine, 1, GET_OBJECT_PATH_CODEC_STREAMING) + .expect("reader should be constructed"); let mut first_read = [0u8; 1]; let _ = reader.read(&mut first_read).await; }); @@ -2226,7 +2392,7 @@ mod tests { engine, data.len(), GET_OBJECT_PATH_CODEC_STREAMING, - FillPolicy::SingleInFlight, + FillPolicy::DualInFlight, ) .expect("reader should be constructed"); let mut first_read = [0u8; 1]; @@ -2235,13 +2401,11 @@ mod tests { assert_eq!(read, first_read.len()); assert_eq!(first_read[0], data[0]); - timeout(Duration::from_secs(1), async { - while read_count.load(Ordering::SeqCst) < 2 { - yield_now().await; - } - }) - .await - .expect("reader should start reading the next stripe before the current output buffer is fully consumed"); + assert_eq!( + read_count.load(Ordering::SeqCst), + 2, + "dual-inflight reader should prefetch the next stripe before returning the first byte" + ); } #[tokio::test] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 046a5190f..c49958849 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -2357,6 +2357,7 @@ enum GetCodecStreamingFallbackReason { ReadQuorumNotSafe, MultipartPartLimit, CopySourceDemandBound, + InvalidMetadataShape, } impl GetCodecStreamingFallbackReason { @@ -2379,6 +2380,7 @@ impl GetCodecStreamingFallbackReason { Self::ReadQuorumNotSafe => "read_quorum_not_safe", Self::MultipartPartLimit => "multipart_part_limit", Self::CopySourceDemandBound => "copy_source_demand_bound", + Self::InvalidMetadataShape => "invalid_metadata_shape", } } } @@ -2438,6 +2440,7 @@ enum GetDirectMemoryFallbackReason { Remote, ObjectInfoMultipart, FileInfoMultipart, + MetadataShape, InvalidSize, SizeMismatch, AboveThreshold, @@ -2463,6 +2466,7 @@ impl GetDirectMemoryFallbackReason { Self::Remote => "remote", Self::ObjectInfoMultipart => "object_info_multipart", Self::FileInfoMultipart => "file_info_multipart", + Self::MetadataShape => "metadata_shape", Self::InvalidSize => "invalid_size", Self::SizeMismatch => "size_mismatch", Self::AboveThreshold => "above_threshold", @@ -2511,10 +2515,42 @@ fn record_get_object_reader_path_observation( object_class: GetCodecStreamingObjectClass, size_bucket: &'static str, ) { + #[cfg(test)] + LAST_GET_OBJECT_READER_PATH.store( + match path { + crate::set_disk::read::GET_OBJECT_PATH_MID_SIZE_STREAMING => 1, + GET_OBJECT_PATH_DIRECT_MEMORY => 2, + GET_OBJECT_PATH_INLINE_DIRECT => 3, + GET_OBJECT_PATH_BODY_CACHE => 4, + GET_OBJECT_PATH_CODEC_STREAMING => 5, + GET_OBJECT_PATH_REMOTE_TRANSITION => 6, + GET_OBJECT_PATH_EMPTY => 7, + _ => 255, + }, + Ordering::Relaxed, + ); rustfs_io_metrics::record_get_object_reader_path(path); rustfs_io_metrics::record_get_object_reader_path_by_size(path, object_class.as_str(), size_bucket); } +#[cfg(test)] +static LAST_GET_OBJECT_READER_PATH: AtomicU64 = AtomicU64::new(0); + +#[cfg(test)] +pub(crate) fn reset_test_get_object_reader_path() { + LAST_GET_OBJECT_READER_PATH.store(0, Ordering::Relaxed); +} + +#[cfg(test)] +pub(crate) fn test_get_object_reader_selected_mid_size() -> bool { + LAST_GET_OBJECT_READER_PATH.load(Ordering::Relaxed) == 1 +} + +#[cfg(test)] +pub(crate) fn test_get_object_reader_path_id() -> u64 { + LAST_GET_OBJECT_READER_PATH.load(Ordering::Relaxed) +} + fn classify_get_codec_streaming_object_class( range: &Option, object_info: &ObjectInfo, @@ -2559,6 +2595,26 @@ fn get_small_object_direct_memory_decision_with_threshold( opts: &ObjectOptions, enabled: bool, threshold: usize, +) -> GetDirectMemoryDecision { + get_small_object_direct_memory_decision_with_threshold_and_plan( + range, + object_info, + fi, + opts, + enabled, + threshold, + ReadPathPlan::new(object_info, fi), + ) +} + +fn get_small_object_direct_memory_decision_with_threshold_and_plan( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, + enabled: bool, + threshold: usize, + plan: ReadPathPlan, ) -> GetDirectMemoryDecision { if !enabled { return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Disabled); @@ -2617,14 +2673,24 @@ fn get_small_object_direct_memory_decision_with_threshold( return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::FileInfoMultipart); } - let Ok(object_size) = usize::try_from(fi.size) else { - return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize); + if object_info.size != fi.size { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch); + } + let Some(shape) = plan.shape() else { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::MetadataShape); }; + let object_size = shape.object_size; if object_size == 0 { return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize); } - if object_info.size != fi.size { - return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch); + if !plan.is_plain() { + if object_info.is_encrypted() || fi.metadata.keys().any(|key| is_object_encryption_marker(key)) { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Encrypted); + } + if object_info.is_compressed() || fi.is_compressed() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Compressed); + } + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Remote); } if object_size > threshold { return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::AboveThreshold); @@ -2633,6 +2699,7 @@ fn get_small_object_direct_memory_decision_with_threshold( GetDirectMemoryDecision::Use { object_size } } +#[allow(dead_code, reason = "asserted by this file's gate tests")] fn get_small_object_direct_memory_decision( range: &Option, object_info: &ObjectInfo, @@ -2667,6 +2734,7 @@ fn should_prefer_codec_streaming_data_blocks_first_reader_setup( max_size > 0 && object_size <= max_size } +#[allow(dead_code, reason = "asserted by this file's gate tests")] fn get_codec_streaming_reader_gate( bucket: &str, object: &str, @@ -2675,6 +2743,29 @@ fn get_codec_streaming_reader_gate( object_info: &ObjectInfo, fi: &FileInfo, lock_optimization_enabled: bool, +) -> GetCodecStreamingGate { + get_codec_streaming_reader_gate_with_plan( + bucket, + object, + part_number, + object_class, + object_info, + fi, + lock_optimization_enabled, + ReadPathPlan::new(object_info, fi), + ) +} + +#[allow(clippy::too_many_arguments, reason = "keeps the hot-path gate inputs explicit")] +fn get_codec_streaming_reader_gate_with_plan( + bucket: &str, + object: &str, + part_number: Option, + object_class: GetCodecStreamingObjectClass, + object_info: &ObjectInfo, + fi: &FileInfo, + lock_optimization_enabled: bool, + plan: ReadPathPlan, ) -> GetCodecStreamingGate { let config = get_codec_streaming_config(); @@ -2790,6 +2881,22 @@ fn get_codec_streaming_reader_gate( }; } } + if object_class == GetCodecStreamingObjectClass::PlainSinglePart { + let Some(_shape) = plan.shape() else { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMetadataShape), + prefer_data_blocks_first_reader_setup: false, + }; + }; + if !plan.is_plain() { + return GetCodecStreamingGate { + object_class, + decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMetadataShape), + prefer_data_blocks_first_reader_setup: false, + }; + } + } let Ok(min_size) = i64::try_from(config.min_size) else { return GetCodecStreamingGate { object_class, @@ -4074,6 +4181,80 @@ fn object_fits_single_block(object_size: i64, block_size: usize) -> bool { } } +/// The common metadata contract consumed by all bounded GET fast paths. +/// +/// `ObjectInfo` is assembled from `FileInfo`, but callers may also provide a +/// prepared snapshot or metadata from an older peer. Never let either copy +/// independently decide that a request is safe: a disagreement must fall +/// back to the regular reader. The returned size is the only size used for +/// fast-path allocation and reader setup. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) struct ReadPathShape { + pub(super) object_size: usize, +} + +impl ReadPathShape { + pub(super) fn is_plain(self, object_info: &ObjectInfo, fi: &FileInfo) -> bool { + !object_info.is_encrypted() + && !fi.metadata.keys().any(|key| is_object_encryption_marker(key)) + && !object_info.is_compressed() + && !fi.is_compressed() + && !object_info.is_remote() + && !fi.is_remote() + } +} + +/// Request-local metadata decision reused by all small-object GET gates. +/// +/// The metadata pair is immutable for the lifetime of `get_object_reader`, so +/// validating it once avoids repeating part-array and transform scans on every +/// fast-path predicate while keeping the trust boundary fail-closed. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct ReadPathPlan { + shape: Option, + plain: bool, +} + +impl ReadPathPlan { + fn new(object_info: &ObjectInfo, fi: &FileInfo) -> Self { + let shape = read_path_shape(object_info, fi); + let plain = shape.is_some_and(|shape| shape.is_plain(object_info, fi)); + Self { shape, plain } + } + + const fn shape(self) -> Option { + self.shape + } + + const fn is_plain(self) -> bool { + self.plain + } +} + +/// Validate the single-part geometry shared by inline, direct-memory, and +/// bounded mid-size readers. This is deliberately fail-closed: a stale or +/// mixed-version `ObjectInfo`/`FileInfo` pair must use the legacy path rather +/// than risk allocating or decoding with a mismatched size. +pub(super) fn read_path_shape(object_info: &ObjectInfo, fi: &FileInfo) -> Option { + if object_info.parts.len() != 1 || fi.parts.len() != 1 || object_info.size < 0 || fi.size < 0 || object_info.size != fi.size { + return None; + } + + let object_part = object_info.parts.first()?; + let file_part = fi.parts.first()?; + let object_size = usize::try_from(fi.size).ok()?; + if object_part.number != file_part.number + || object_part.size != file_part.size + || object_part.actual_size != file_part.actual_size + || file_part.size != object_size + || file_part.actual_size != fi.size + { + return None; + } + + Some(ReadPathShape { object_size }) +} + fn should_use_inline_small_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool { is_inline_buffer && object_fits_single_block(object_size, block_size) } @@ -4082,35 +4263,37 @@ fn should_use_single_block_non_inline_fast_path(is_inline_buffer: bool, object_s !is_inline_buffer && object_fits_single_block(object_size, block_size) } +#[allow(dead_code, reason = "asserted by this file's gate tests")] fn should_use_inline_fast_path( range: &Option, object_info: &ObjectInfo, fi: &FileInfo, opts: &ObjectOptions, +) -> bool { + should_use_inline_fast_path_with_plan(range, object_info, fi, opts, ReadPathPlan::new(object_info, fi)) +} + +fn should_use_inline_fast_path_with_plan( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, + plan: ReadPathPlan, ) -> bool { if !object_info.is_inline_fast_path_eligible() || fi.data.is_none() || range.is_some() || opts.part_number.is_some() { return false; } + let Some(shape) = plan.shape() else { + return false; + }; // The persisted marker is authoritative for the storage decision, but it // is still untrusted metadata at this boundary. Revalidate its geometry so // a stale/corrupt marker cannot route an oversized payload into the - // in-memory decoder. The persisted marker is the writer's policy decision; - // reloading the current storage-class config here would add a hot-path - // snapshot load and could make an already durable inline object unreadable - // after an operator changes the admission policy. The independent - // direct-memory reader remains capped at 128 KiB below. - let Some(part) = fi.parts.first() else { - return false; - }; - // Plain inline metadata must describe one coherent object. In particular, - // do not trust `fi.size` for the allocation below when a corrupt part has - // a different `actual_size`; compressed/unknown `actual_size` objects are - // excluded earlier by the plain-object checks. - if fi.size < 0 - || fi.size > INLINE_FAST_PATH_MAX_OBJECT_SIZE - || object_info.size != fi.size - || part.actual_size != fi.size + // in-memory decoder. The independent direct-memory reader remains capped + // at 128 KiB below. + if !plan.is_plain() + || shape.object_size > usize::try_from(INLINE_FAST_PATH_MAX_OBJECT_SIZE).expect("inline fast path limit fits usize") || fi.erasure.data_blocks == 0 || fi.erasure.block_size == 0 { @@ -11022,6 +11205,41 @@ mod tests { )); } + #[test] + fn read_path_shape_rejects_mismatched_metadata_copies_and_transforms() { + let (object_info, fi, _) = direct_memory_test_metadata(1024); + let shape = read_path_shape(&object_info, &fi).expect("matching metadata should have a valid shape"); + assert_eq!(shape.object_size, 1024); + assert!(shape.is_plain(&object_info, &fi)); + + let mut bad_part_number = object_info.clone(); + Arc::make_mut(&mut bad_part_number.parts)[0].number = 2; + assert!(read_path_shape(&bad_part_number, &fi).is_none()); + + let mut bad_part_size = fi.clone(); + bad_part_size.parts[0].size += 1; + assert!(read_path_shape(&object_info, &bad_part_size).is_none()); + + let mut bad_actual_size = object_info.clone(); + Arc::make_mut(&mut bad_actual_size.parts)[0].actual_size += 1; + assert!(read_path_shape(&bad_actual_size, &fi).is_none()); + + let mut compressed = fi.clone(); + insert_str(&mut compressed.metadata, SUFFIX_COMPRESSION, "zstd".to_string()); + let compressed_shape = read_path_shape(&object_info, &compressed).expect("geometry remains valid"); + assert!(!compressed_shape.is_plain(&object_info, &compressed)); + + let mut encrypted = fi; + encrypted + .metadata + .insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); + assert!( + !read_path_shape(&object_info, &encrypted) + .expect("geometry remains valid") + .is_plain(&object_info, &encrypted) + ); + } + #[test] fn inline_fast_path_rejects_part_number_requests() { let (mut object_info, mut fi, opts) = direct_memory_test_metadata(1024); @@ -11170,6 +11388,13 @@ mod tests { get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 128 * 1024), GetDirectMemoryDecision::Use { object_size: 1024 } ); + + let mut corrupt_part = fi; + corrupt_part.parts[0].actual_size += 1; + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &corrupt_part, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::MetadataShape) + ); } #[test] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 34286c744..699243632 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -30,32 +30,34 @@ use super::super::{ GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE, GET_STAGE_LOCK_ACQUIRE, GET_STAGE_METADATA, GET_STAGE_OBJECT_INFO, GET_STAGE_PATH_DECISION, - GET_STAGE_READER_SETUP, GenericError, GetCodecStreamingDecision, GetDirectMemoryDecision, GetObjectReader, HTTPRangeSpec, - HashAlgorithm, HashMap, HashReader, HashSet, HeaderMap, HealChannelPriority, InstanceContext, Instant, LOG_COMPONENT_ECSTORE, - LOG_SUBSYSTEM_SET_DISK, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, ObjectKey, ObjectLockConfigSnapshot, - ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin, PutObjReader, - RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReaderImpl, ReplicateDecision, ReplicationObjectBridge, Result, - SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, - SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError, TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, - WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE, - adaptive_duplex_buffer_size, build_get_object_info, build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, - can_try_inline_data_shards_direct, check_object_lock_delete, check_object_lock_for_deletion_with_state, - check_object_lock_retention_update, classify_get_codec_streaming_object_class, classify_put_write_path, - classify_storage_error, collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug, - delete_file_info_version_id, disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker, - finish_set_disk_read_lock, get_codec_streaming_reader_gate, get_object_body_cache_hook, get_raw_etag, - get_small_object_direct_memory_decision, get_stage_timer_if_enabled, get_str, - get_transitioned_object_reader_with_tier_manager, inline_erasure_shard_file_offset, inline_erasure_shard_size, insert_str, - is_deadlock_detection_enabled, is_err_object_not_found, is_err_version_not_found, is_explicit_null_version, - is_get_codec_streaming_base_enabled, is_lock_optimization_enabled, issue3031_diag_enabled, join_all, - known_put_object_storage_size, path_join_buf, put_restore_opts, record_compression_total_memory, - record_get_codec_streaming_gate_decision, record_get_direct_memory_decision, record_get_object_pipeline_failure, - record_get_object_pipeline_failure_for_path, record_get_object_reader_path_observation, record_get_stage_duration_if_enabled, - record_lock_acquire, reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate, - require_restore_operation_id, resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result, - resolve_write_layout, restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event, + GET_STAGE_READER_SETUP, GenericError, GetCodecStreamingDecision, GetCodecStreamingFallbackReason, GetDirectMemoryDecision, + GetObjectReader, HTTPRangeSpec, HashAlgorithm, HashMap, HashReader, HashSet, HeaderMap, HealChannelPriority, InstanceContext, + Instant, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, ObjectKey, + ObjectLockConfigSnapshot, ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin, + PutObjReader, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReadPathPlan, ReaderImpl, ReplicateDecision, + ReplicationObjectBridge, Result, SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE, + SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError, + TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, WriteLayout, X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, + X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_RESTORE, adaptive_duplex_buffer_size, build_get_object_info, + build_inline_bitrot_readers, build_inline_bitrot_readers_from_refs, can_try_inline_data_shards_direct, + check_object_lock_delete, check_object_lock_for_deletion_with_state, check_object_lock_retention_update, + classify_get_codec_streaming_object_class, classify_put_write_path, classify_storage_error, + collect_inline_data_shard_fileinfos_by_index, contains_key_str, create_bitrot_writer, debug, delete_file_info_version_id, + disk, ensure_delete_commit_locks_held, error, explicit_delete_removed_marker, finish_set_disk_read_lock, + get_codec_streaming_reader_gate_with_plan, get_object_body_cache_hook, get_raw_etag, + get_small_object_direct_memory_decision_with_threshold_and_plan, get_small_object_direct_memory_threshold, + get_stage_timer_if_enabled, get_str, get_transitioned_object_reader_with_tier_manager, inline_erasure_shard_file_offset, + inline_erasure_shard_size, insert_str, is_deadlock_detection_enabled, is_err_object_not_found, is_err_version_not_found, + is_explicit_null_version, is_get_codec_streaming_base_enabled, is_get_small_object_direct_memory_enabled, + is_lock_optimization_enabled, issue3031_diag_enabled, join_all, known_put_object_storage_size, path_join_buf, + put_restore_opts, record_compression_total_memory, record_get_codec_streaming_gate_decision, + record_get_direct_memory_decision, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, + record_get_object_reader_path_observation, record_get_stage_duration_if_enabled, record_lock_acquire, + reduce_write_quorum_errs, release_materialized_read_lock, replication_write_may_pass_worm_gate, require_restore_operation_id, + resolve_delete_version_state, resolve_tiered_decommission_write_quorum_result, resolve_write_layout, + restore_commit_operation_id_from_metadata, restore_operation_id_from_metadata, send_event, set_disk_delete_creates_delete_marker, should_force_delete_marker_for_missing_version, - should_persist_encryption_original_size, should_preserve_delete_replication_state, should_use_inline_fast_path, + should_persist_encryption_original_size, should_preserve_delete_replication_state, should_use_inline_fast_path_with_plan, take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn, }; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; @@ -109,6 +111,7 @@ fn is_get_mid_size_streaming_enabled() -> bool { /// codec-streaming gate: only a whole, plain, single-part object is eligible. /// Ranges, transforms, remote objects, multipart reads, copy-source reads and /// special movement/version requests retain their existing legacy semantics. +#[allow(dead_code, reason = "asserted by this file's gate tests")] fn get_mid_size_streaming_object_size( range: &Option, object_info: &ObjectInfo, @@ -127,6 +130,7 @@ fn get_mid_size_streaming_object_size( ) } +#[allow(dead_code, reason = "asserted by this file's gate tests")] fn get_mid_size_streaming_object_size_with_flags( range: &Option, object_info: &ObjectInfo, @@ -135,6 +139,26 @@ fn get_mid_size_streaming_object_size_with_flags( lock_optimization_enabled: bool, mid_size_enabled: bool, codec_base_enabled: bool, +) -> Option { + get_mid_size_streaming_object_size_with_flags_and_plan( + range, + object_info, + opts, + lock_optimization_enabled, + mid_size_enabled, + codec_base_enabled, + super::super::ReadPathPlan::new(object_info, fi), + ) +} + +fn get_mid_size_streaming_object_size_with_flags_and_plan( + range: &Option, + object_info: &ObjectInfo, + opts: &ObjectOptions, + lock_optimization_enabled: bool, + mid_size_enabled: bool, + codec_base_enabled: bool, + plan: super::super::ReadPathPlan, ) -> Option { if !mid_size_enabled || !codec_base_enabled @@ -149,26 +173,18 @@ fn get_mid_size_streaming_object_size_with_flags( || object_info.delete_marker || object_info.metadata_only || object_info.version_only - || object_info.is_encrypted() - || object_info.is_compressed() - || object_info.is_remote() || crate::set_disk::get_object_read_policy() != super::super::GetObjectReadPolicy::Default - || object_info.parts.len() != 1 - || fi.parts.len() != 1 - || object_info.size != fi.size { return None; } - let object_size = usize::try_from(fi.size).ok()?; - let object_part = object_info.parts.first()?; - let file_part = fi.parts.first()?; - if object_part.number != file_part.number || file_part.size != object_size || file_part.actual_size != fi.size { + let shape = plan.shape()?; + if !plan.is_plain() { return None; } (GET_MID_SIZE_STREAMING_MIN_SIZE..=GET_MID_SIZE_STREAMING_MAX_SIZE) - .contains(&object_size) - .then_some(object_size) + .contains(&shape.object_size) + .then_some(shape.object_size) } #[cfg(all(test, feature = "test-util"))] @@ -1721,6 +1737,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { return Ok(reader); } + // All remaining local fast paths share this immutable, fail-closed + // metadata decision. Build it once after empty/remote exits so those + // requests do not pay for part and transform scans they cannot use. + let read_path_plan = ReadPathPlan::new(&object_info, fi); + // Inline data fast path: skip duplex pipe for small inline objects. // Uses the shared predicate from ObjectInfo; additionally checks that // inline data is actually present and neither range nor partNumber is @@ -1729,7 +1750,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { && fi.data.is_some() && range.is_none() && opts.part_number.is_none() - && should_use_inline_fast_path(&range, &object_info, fi, opts); + && should_use_inline_fast_path_with_plan(&range, &object_info, fi, opts, read_path_plan); if use_inline_fast_path { let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let data_shards = fi.erasure.data_blocks; @@ -1892,27 +1913,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let codec_streaming_gate = get_codec_streaming_reader_gate( - bucket, - object, - opts.part_number, - object_class, - &object_info, - fi, - lock_optimization_enabled, - ); - record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start); - if object_info.is_remote() { - if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_gate.decision { - record_get_codec_streaming_gate_decision( - codec_streaming_gate.object_class, - codec_streaming_gate.decision, - size_bucket, - ); - rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); - } + let decision = GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote); + record_get_codec_streaming_gate_decision(object_class, decision, size_bucket); + rustfs_io_metrics::record_get_object_codec_streaming_fallback(GetCodecStreamingFallbackReason::Remote.as_str()); record_get_object_reader_path_observation(GET_OBJECT_PATH_REMOTE_TRANSITION, object_class, size_bucket); let mut opts = opts.clone(); if object_info.parts.len() == 1 { @@ -1932,6 +1936,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), bucket, object)); } + // Metadata resolution and the remote-tier branch are complete here. + // Keep the rollout/configuration gate deferred until the request + // really needs codec streaming so an opted-out codec path cannot add + // fixed cost to the inline/direct-memory/mid-size hot paths. // App-layer object data cache probe: metadata (etag/size) is resolved // but no data shards have been read yet, so a hit skips the erasure // read, bitrot verify and decode entirely. The hook validates object @@ -1978,7 +1986,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, fi, opts); + let direct_memory_decision = get_small_object_direct_memory_decision_with_threshold_and_plan( + &range, + &object_info, + fi, + opts, + is_get_small_object_direct_memory_enabled(), + get_small_object_direct_memory_threshold(), + read_path_plan, + ); record_get_direct_memory_decision(object_class, direct_memory_decision, size_bucket); if let GetDirectMemoryDecision::Use { object_size } = direct_memory_decision { if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo( @@ -2064,7 +2080,17 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // rollout remains off by default because its worker overhead is not a // win for tiny objects. A failed setup degrades to the existing codec // gate/legacy path before any response bytes are returned. - if get_mid_size_streaming_object_size(&range, &object_info, fi, opts, lock_optimization_enabled).is_some() { + if get_mid_size_streaming_object_size_with_flags_and_plan( + &range, + &object_info, + opts, + lock_optimization_enabled, + is_get_mid_size_streaming_enabled(), + is_get_codec_streaming_base_enabled(), + read_path_plan, + ) + .is_some() + { match Self::get_object_mid_size_reader_with_fileinfo( bucket, object, @@ -2096,6 +2122,19 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } + let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); + let codec_streaming_gate = get_codec_streaming_reader_gate_with_plan( + bucket, + object, + opts.part_number, + object_class, + &object_info, + fi, + lock_optimization_enabled, + read_path_plan, + ); + record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start); + match codec_streaming_gate.decision { GetCodecStreamingDecision::Use => { match Self::get_object_decode_reader_with_fileinfo( @@ -9639,7 +9678,7 @@ mod replication_lww_tests { mod inline_put_commit_path_tests { use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; - use crate::config::storageclass::lookup_config_for_pools_without_env; + use crate::config::storageclass::{INLINE_BLOCK_ENV, lookup_config_for_pools, lookup_config_for_pools_without_env}; use crate::disk::ReadOptions; use rustfs_config::server_config::KVS; use serial_test::serial; @@ -9706,6 +9745,62 @@ mod inline_put_commit_path_tests { assert_eq!(restored, payload); } + #[tokio::test] + #[serial] + async fn get_object_reader_wires_mid_size_to_single_inflight() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "mid-size-reader-wiring"; + let object = "object.bin"; + let payload: Vec = (0..256 * 1024).map(|index| (index % 251) as u8).collect(); + make_bucket(&disk_stores, bucket).await; + let storage_class = temp_env::with_var(INLINE_BLOCK_ENV, Some("1KiB"), || lookup_config_for_pools(&KVS::new(), &[4])) + .expect("test storage class should resolve"); + set_disks.set_test_storage_class_config(storage_class); + + let mut writer = PutObjReader::from_vec(payload.clone()); + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, Some("true")), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("on")), + (rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true")), + ], + async { + set_disks + .put_object(bucket, object, &mut writer, &ObjectOptions::default()) + .await + .expect("mid-size wiring fixture should commit"); + + crate::set_disk::reset_test_get_object_reader_path(); + let single_inflight_before = crate::set_disk::coding::decode_reader::test_single_inflight_construction_count(); + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("mid-size wiring GET should succeed"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("mid-size wiring reader should stream"); + + assert_eq!(restored, payload); + assert!( + crate::set_disk::test_get_object_reader_selected_mid_size(), + "full get_object_reader path must select mid-size streaming (path id {})", + crate::set_disk::test_get_object_reader_path_id() + ); + assert!( + crate::set_disk::coding::decode_reader::test_single_inflight_construction_count() > single_inflight_before, + "mid-size get_object_reader wiring must construct SingleInFlight" + ); + }, + ) + .await; + } + #[tokio::test] async fn repeated_gets_reuse_the_set_erasure_shell() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 61bceb78d..6962d2189 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -1370,6 +1370,7 @@ impl SetDisks { prefer_data_blocks_first_reader_setup, get_codec_streaming_metrics_path(), false, + false, ) .await } @@ -1408,6 +1409,7 @@ impl SetDisks { prefer_data_blocks_first_reader_setup, GET_OBJECT_PATH_MID_SIZE_STREAMING, true, + true, ) .await } @@ -1428,6 +1430,7 @@ impl SetDisks { prefer_data_blocks_first_reader_setup: bool, metrics_path: &'static str, allow_inplace_legacy_fallback: bool, + single_inflight: bool, ) -> Result { let erasure = erasure_cache.get_for_file_info(fi)?; let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); @@ -1451,6 +1454,7 @@ impl SetDisks { metrics_object_class, metrics_size_bucket, prefer_data_blocks_first_reader_setup, + single_inflight, // Single-part objects keep the whole-request fallback: a degraded // sole part is detected before any byte streams, so the caller can // still hand the request to the legacy duplex path unchanged. @@ -1504,6 +1508,7 @@ impl SetDisks { metrics_object_class, metrics_size_bucket, false, + false, // The first part stays eager and keeps the whole-request fallback: // if part 1 is already degraded, the entire GET drops to the legacy // duplex path before a single byte is streamed (semantics unchanged). @@ -1551,6 +1556,7 @@ impl SetDisks { ctx.metrics_object_class, ctx.metrics_size_bucket, false, + false, // backlog#879: later parts have already streamed earlier bytes, // so a whole-request fallback is impossible here. Degrade this // part in place to a legacy per-part decode reader instead of @@ -1584,6 +1590,7 @@ impl SetDisks { metrics_object_class: &'static str, metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, + single_inflight: bool, allow_inplace_legacy_fallback: bool, metrics_path: &'static str, ) -> Result { @@ -1704,11 +1711,23 @@ impl SetDisks { .with_deferred_parity_handles(deferred_stripe_handles) .with_deferred_parity_reopeners(deferred_reopeners); let engine = build_get_codec_streaming_decode_engine(erasure.clone())?; - let reader = - coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?; - Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new( - coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path), - ))) + let reader = if single_inflight { + coding::decode_reader::ErasureDecodeReader::new_single_inflight_with_metrics_path( + source, + engine, + part_length, + metrics_path, + )? + } else { + coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)? + }; + if single_inflight { + Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(reader))) + } else { + Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new( + coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path), + ))) + } } } @@ -4876,6 +4895,7 @@ mod tests { "test-size-bucket", false, false, + false, get_codec_streaming_metrics_path(), ) .await; @@ -4897,6 +4917,7 @@ mod tests { "test-size-bucket", false, false, + false, get_codec_streaming_metrics_path(), ) .await;