diff --git a/Cargo.lock b/Cargo.lock index af2a80f9e..ead4d05e9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -62,7 +62,7 @@ checksum = "f8eb277bec05f56a0e0591f155a484cbd0f4f07ff2905051a48c72f004f7ed58" dependencies = [ "cipher 0.5.2", "cpubits", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "zeroize", ] @@ -323,13 +323,13 @@ checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" [[package]] name = "argon2" -version = "0.6.0-rc.8" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7af50940b73bf4e16c15c448a2b121c63f2d68e3e54b6a8731673cb4aa0cdff5" +checksum = "134c52ddac6d63c576bef8168db10c83c49c26444ecbc68060fef078925a901c" dependencies = [ "base64ct", "blake2", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "password-hash", ] @@ -1024,7 +1024,7 @@ dependencies = [ "http 0.2.12", "http 1.5.0", "http-body 1.1.0", - "lru 0.18.2", + "lru 0.18.3", "percent-encoding", "regex-lite", "sha2 0.11.0", @@ -1945,7 +1945,7 @@ checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cipher 0.5.2", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "rand_core 0.10.1", "zeroize", ] @@ -2283,9 +2283,9 @@ dependencies = [ [[package]] name = "convert_case" -version = "0.11.0" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "affbf0190ed2caf063e3def54ff444b449371d55c58e513a95ab98eca50adb49" +checksum = "1af709f1f33454bf52eadfc8c78b3b9ef9cb26fb54d16dc9cd9a7299f899fd1b" dependencies = [ "unicode-segmentation", ] @@ -2348,9 +2348,9 @@ dependencies = [ [[package]] name = "cpufeatures" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" dependencies = [ "libc", ] @@ -2611,7 +2611,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23" dependencies = [ "cfg-if", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "curve25519-dalek-derive", "digest 0.11.3", "fiat-crypto 0.3.0", @@ -5885,7 +5885,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8f198d1db720e4940b5a493201d199d9f24f568f8f746bd13706243a2f71598" dependencies = [ "cfg-if", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", ] [[package]] @@ -6103,9 +6103,9 @@ checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" [[package]] name = "libredox" -version = "0.1.20" +version = "0.1.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" +checksum = "d7955dfc218a8afb29dfeffd540e3a6e96baeb94fe7138228dd7cc6937fbbf96" dependencies = [ "libc", ] @@ -6209,9 +6209,9 @@ dependencies = [ [[package]] name = "lru" -version = "0.18.2" +version = "0.18.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" +checksum = "0d317b4b9eb398e6acce275758ec6125535505e7a146fb1a9b8bda2451b0ff4c" dependencies = [ "hashbrown 0.17.1", ] @@ -6659,7 +6659,7 @@ dependencies = [ "futures-sink", "futures-util", "keyed_priority_queue", - "lru 0.18.2", + "lru 0.18.3", "mysql_common", "percent-encoding", "rand 0.10.2", @@ -7371,9 +7371,9 @@ checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" [[package]] name = "owo-colors" -version = "4.3.0" +version = "4.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d211803b9b6b570f68772237e415a029d5a50c65d382910b879fb19d3271f94d" +checksum = "13c45bb4a6ae1280ec0803b1ef9d3455eb50f01efbbe1447ab020f1d54fba9d8" [[package]] name = "p12-keystore" @@ -7924,7 +7924,7 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e2d0073b297041425c7c3df6eb4792d598a15323fe63346852b092eca02904c" dependencies = [ - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "universal-hash", "zeroize", ] @@ -7936,7 +7936,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0fa31d631f2b2cb2a544d0aa321ce847a94764d701ca2becc411138b93d49cd" dependencies = [ "cpubits", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "universal-hash", "zeroize", ] @@ -10816,7 +10816,7 @@ dependencies = [ "blake2", "brotli", "bytes", - "convert_case 0.11.0", + "convert_case 0.12.0", "crc-fast", "criterion", "flate2", @@ -11474,7 +11474,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" dependencies = [ "cfg-if", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "digest 0.11.3", ] @@ -11502,7 +11502,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", - "cpufeatures 0.3.0", + "cpufeatures 0.3.1", "digest 0.11.3", ] @@ -12865,9 +12865,9 @@ dependencies = [ [[package]] name = "twox-hash" -version = "2.1.3" +version = "2.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8464ec13c3691491391d9fce00f6416c9a48e46972f72d7865688be2080192c9" +checksum = "5283634e518fe9e82c7b20520bb4bc209009fd16c82077c802f8111ecbb0117a" [[package]] name = "typed-path" diff --git a/Cargo.toml b/Cargo.toml index 35fb5e512..550899e18 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -198,7 +198,7 @@ serde_urlencoded = "0.7.1" # have incompatible APIs. Keep them exact-pinned and monitor upstream for stable # releases. aes-gcm = { version = "=0.11.1" } -argon2 = { version = "=0.6.0-rc.8" } +argon2 = { version = "=0.6.0" } blake2 = "=0.11.0" chacha20poly1305 = { version = "=0.11.0" } crc-fast = "1.10.0" @@ -247,7 +247,7 @@ base64-simd = "0.8.0" brotli = "8.0.4" clap = { version = "4.6.6" } const-str = { version = "1.1.0" } -convert_case = "0.11.0" +convert_case = "0.12.0" criterion = { version = "0.8" } crossbeam-queue = "0.3.13" crossbeam-channel = "0.5.16" diff --git a/crates/ecstore/src/config/storageclass.rs b/crates/ecstore/src/config/storageclass.rs index 361b15702..751b88340 100644 --- a/crates/ecstore/src/config/storageclass.rs +++ b/crates/ecstore/src/config/storageclass.rs @@ -246,16 +246,7 @@ impl Config { } let shard_size = shard_size as usize; - // Keep the historical two-data-shard object budget while preventing - // wider EC layouts from multiplying the maximum inline object size. - // Use div_ceil to match the shard_file_size calculation (which also uses - // div_ceil), avoiding a 1-byte rounding discrepancy that prevents inline - // for objects right at the threshold. - let inline_block = if self.initialized && self.inline_block_explicit { - self.inline_block - } else { - DEFAULT_INLINE_OBJECT_BUDGET.div_ceil(data_shards).min(DEFAULT_INLINE_BLOCK) - }; + let inline_block = self.effective_inline_block(data_shards); if versioned { shard_size <= inline_block / 8 @@ -264,6 +255,27 @@ impl Config { } } + /// Returns the per-shard inline budget used by both write admission and + /// legacy read fallback. + /// + /// The default budget is scaled by the number of data shards so a wider EC + /// layout does not silently increase the maximum inline object size. An + /// explicitly configured `inline_block` remains a fixed per-shard limit for + /// compatibility with deployments that opted into the historical policy. + pub(crate) fn effective_inline_block(&self, data_shards: usize) -> usize { + if data_shards == 0 { + return 0; + } + + if self.initialized && self.inline_block_explicit { + self.inline_block + } else { + // Keep the historical two-data-shard object budget while preventing + // wider EC layouts from multiplying the maximum inline object size. + DEFAULT_INLINE_OBJECT_BUDGET.div_ceil(data_shards).min(DEFAULT_INLINE_BLOCK) + } + } + pub fn inline_block(&self) -> usize { if !self.initialized { DEFAULT_INLINE_BLOCK @@ -602,6 +614,51 @@ mod tests { } } + #[test] + fn should_inline_keeps_ec8_and_ec12_object_boundaries_consistent() { + let config = Config::default(); + let object_sizes = [128 * 1024, 256 * 1024, 512 * 1024, 1024 * 1024, 4 * 1024 * 1024]; + + for (data_shards, parity_shards) in [(8, 4), (12, 4)] { + let erasure = crate::erasure::coding::Erasure::new(data_shards, parity_shards, 1024 * 1024); + let mut previous = true; + for object_size in object_sizes { + let shard_size = erasure.shard_file_size(object_size); + let inline = config.should_inline(shard_size, data_shards, false); + + // The effective policy is monotonic across object sizes. This + // table covers the boundaries that previously exposed the + // fixed-shard read-ahead mismatch, including the 1 MiB case. + assert!(!inline || previous, "inline decision must not re-enable at {object_size} bytes"); + previous = inline; + } + + assert!( + !config.should_inline(erasure.shard_file_size(1024 * 1024), data_shards, false), + "1 MiB must use the non-inline path for EC{data_shards}+{parity_shards}" + ); + } + } + + #[test] + fn effective_inline_block_scales_default_budget_and_preserves_explicit_limit() { + let config = Config::default(); + assert_eq!(config.effective_inline_block(8), 32 * 1024); + assert_eq!(config.effective_inline_block(12), 21_846); + assert_eq!(config.effective_inline_block(0), 0); + + let explicit = lookup_config_for_pools_with_env( + &KVS::new(), + &[12], + StorageClassEnvOverrides { + inline_block: Some("128KiB".to_string()), + ..Default::default() + }, + ) + .expect("explicit inline block should resolve"); + assert_eq!(explicit.effective_inline_block(12), 128 * 1024); + } + #[test] fn explicit_inline_block_preserves_fixed_per_shard_rollback() { let overrides = StorageClassEnvOverrides { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 3d591a06d..b0aa53299 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::config::storageclass::DEFAULT_INLINE_BLOCK; use crate::crash_inject::{self, CrashPoint}; use crate::data_usage::local_snapshot::ensure_data_usage_layout; use crate::diagnostics::get::{ @@ -10410,8 +10409,14 @@ impl DiskAPI for LocalDisk { fi.data = None; } - let inline = fi.transition_status.is_empty() && fi.data_dir.is_some() && fi.parts.len() == 1; - if inline && fi.shard_file_size(fi.parts[0].actual_size) < DEFAULT_INLINE_BLOCK as i64 { + // Keep this compatibility read-ahead decision on the same policy + // as PUT's inline admission. In particular, do not use the old + // fixed 128 KiB shard limit: a non-inline object in a wider EC + // layout can have a smaller shard and would otherwise be copied + // out of part.1 during every metadata read. Such objects remain + // fully readable through the normal EC reader below. + let storage_class_config = runtime_sources::storage_class_config_snapshot(); + if should_read_legacy_inline_part(&fi, storage_class_config.as_ref()) { let part_path = path_join_buf(&[ path, fi.data_dir.map_or_else(|| "".to_string(), |dir| dir.to_string()).as_str(), @@ -10913,6 +10918,21 @@ impl DiskAPI for LocalDisk { } } +/// Whether a legacy object without the inline marker should have its external +/// part materialized into `FileInfo.data` for compatibility with the old GET +/// fast path. The marker-bearing path is handled by `read_raw`/`get_file_info`; +/// this is only a conservative fallback for old metadata. +fn should_read_legacy_inline_part(fi: &FileInfo, storage_class_config: &crate::config::storageclass::Config) -> bool { + if !fi.transition_status.is_empty() || fi.data_dir.is_none() || fi.parts.len() != 1 || fi.inline_data() { + return false; + } + + let part = &fi.parts[0]; + let shard_size = fi.shard_file_size(part.actual_size); + let versioned = fi.versioned || fi.version_id.is_some_and(|version_id| !version_id.is_nil()); + storage_class_config.should_inline(shard_size, fi.erasure.data_blocks, versioned) +} + impl LocalDisk { pub(crate) async fn rename_data_borrowed( &self, @@ -11049,6 +11069,46 @@ mod test { file_info } + #[test] + fn legacy_inline_read_ahead_matches_writer_policy_for_ec_layouts() { + let config = crate::config::storageclass::Config::default(); + let object_sizes = [128 * 1024_i64, 256 * 1024, 512 * 1024, 1024 * 1024, 4 * 1024 * 1024]; + + for (data_shards, parity_shards) in [(8, 4), (12, 4)] { + for object_size in object_sizes { + let mut fi = FileInfo::new("object", data_shards, parity_shards); + fi.data_dir = Some(Uuid::from_u128(1)); + fi.parts = vec![ObjectPartInfo { + number: 1, + size: usize::try_from(object_size).expect("test object size should fit usize"), + actual_size: object_size, + ..Default::default() + }]; + + let writer_decision = config.should_inline(fi.shard_file_size(object_size), data_shards, false); + assert_eq!( + should_read_legacy_inline_part(&fi, &config), + writer_decision, + "legacy read-ahead must match PUT for EC{data_shards}+{parity_shards}, size={object_size}" + ); + } + } + + let mut ec12 = FileInfo::new("object", 12, 4); + ec12.data_dir = Some(Uuid::from_u128(1)); + ec12.parts = vec![ObjectPartInfo { + number: 1, + size: 1024 * 1024, + actual_size: 1024 * 1024, + ..Default::default() + }]; + assert!( + ec12.shard_file_size(1024 * 1024) < crate::config::storageclass::DEFAULT_INLINE_BLOCK as i64, + "the regression guard must exercise the old fixed 128 KiB read-ahead boundary" + ); + assert!(!should_read_legacy_inline_part(&ec12, &config)); + } + fn test_meta(fi: FileInfo) -> Vec { let mut meta = FileMeta::default(); meta.add_version(fi).expect("test metadata should accept file info"); diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 280dd10cc..9ea4b8324 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -752,12 +752,18 @@ impl ObjectInfo { .any(|key| rustfs_utils::http::is_object_encryption_marker(key)) } - /// Maximum inline size for non-versioned objects (128 KiB). - /// Matches `DEFAULT_INLINE_BLOCK` in `storageclass.rs`. + /// Historical non-versioned inline size reference. + /// + /// Inline admission is layout-specific now; callers must not use this + /// constant to decide whether an object is eligible for the fast path. + #[deprecated(note = "inline eligibility is layout-specific; use persisted metadata and the read-path policy")] pub const INLINE_MAX_SIZE: i64 = 128 * 1024; - /// Maximum inline size for versioned objects (16 KiB). - /// Matches `DEFAULT_INLINE_BLOCK / 8` in `storageclass.rs`. + /// Historical versioned inline size reference. + /// + /// Inline admission is layout-specific now; callers must not use this + /// constant to decide whether an object is eligible for the fast path. + #[deprecated(note = "inline eligibility is layout-specific; use persisted metadata and the read-path policy")] pub const INLINE_MAX_SIZE_VERSIONED: i64 = 16 * 1024; /// Returns `true` when this object qualifies for the inline data fast path. @@ -765,10 +771,12 @@ impl ObjectInfo { /// The inline fast path decodes erasure-coded data entirely in memory, /// bypassing disk I/O, duplex pipes, and the disk-read semaphore. /// - /// The `inlined` flag is the primary signal — PUT sets it through the - /// captured storage-class snapshot's `Config::should_inline`, which applies - /// the correct version-aware threshold (128 KiB non-versioned, 16 KiB versioned). - /// The size check below is a safety net using the same thresholds. + /// The persisted `inlined` flag is the canonical size-policy decision. PUT + /// sets it through the captured storage-class snapshot's effective policy, + /// which is layout- and version-aware. Reapplying a fixed object-size limit + /// here would disagree with that policy for wider EC layouts and explicit + /// inline configurations. The direct-memory reader retains its own bounded + /// 128 KiB allocation gate at the call site. /// /// Additional conditions: /// - Single part @@ -779,14 +787,8 @@ impl ObjectInfo { if !self.inlined { return false; } - // Apply the same version-aware threshold as PUT (storageclass.rs). - let max_size = if self.version_id.is_some() { - Self::INLINE_MAX_SIZE_VERSIONED - } else { - Self::INLINE_MAX_SIZE - }; self.parts.len() == 1 - && self.size <= max_size + && self.size >= 0 && !self.is_encrypted() && !self.is_compressed() && self.transitioned_object.tier.is_empty() @@ -1382,14 +1384,14 @@ mod tests { } #[test] - fn inline_fast_path_eligibility_preserves_exact_versioned_boundaries() { + fn inline_fast_path_eligibility_follows_persisted_marker() { for (case, size, versioned, expected) in [ ("unversioned below", 128 * 1024 - 1, false, true), ("unversioned exact", 128 * 1024, false, true), - ("unversioned above", 128 * 1024 + 1, false, false), + ("unversioned above", 128 * 1024 + 1, false, true), ("versioned below", 16 * 1024 - 1, true, true), ("versioned exact", 16 * 1024, true, true), - ("versioned above", 16 * 1024 + 1, true, false), + ("versioned above", 16 * 1024 + 1, true, true), ] { assert_eq!( inline_fast_path_object(size, versioned).is_inline_fast_path_eligible(), @@ -1399,9 +1401,29 @@ mod tests { } } + #[test] + fn inline_fast_path_marker_allows_ec8_and_ec12_layout_specific_256kib_objects() { + for data_blocks in [8, 12] { + let object = ObjectInfo { + size: 256 * 1024, + data_blocks, + parity_blocks: 4, + inlined: true, + version_id: Some(Uuid::from_u128(1)), + parts: Arc::new(vec![ObjectPartInfo::default()]), + ..Default::default() + }; + + assert!( + object.is_inline_fast_path_eligible(), + "the persisted inline marker must be authoritative for EC{data_blocks}+4" + ); + } + } + #[test] fn inline_fast_path_eligibility_rejects_incompatible_object_shapes() { - let mut object = inline_fast_path_object(ObjectInfo::INLINE_MAX_SIZE, false); + let mut object = inline_fast_path_object(128 * 1024, false); object.inlined = false; assert!(!object.is_inline_fast_path_eligible(), "non-inline objects must fall back"); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 2fa44581f..046a5190f 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -740,6 +740,10 @@ const ENV_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY: &str = "RUSTFS_GET_SMALL_OBJECT const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY: bool = true; const ENV_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: &str = "RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD"; const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 1024; +// Bound the in-memory inline decoder independently from the configurable +// storage-class admission policy. This protects its Vec allocation while +// allowing explicitly configured EC layouts to inline objects above 128 KiB. +const INLINE_FAST_PATH_MAX_OBJECT_SIZE: i64 = 1024 * 1024; // --- Metadata Early-Stop Configuration --- @@ -2285,6 +2289,14 @@ fn get_codec_streaming_config() -> GetCodecStreamingConfig { } } +/// Shared kill-switch and compatibility guard for all non-duplex GET readers. +/// Size-specific readers may have independent rollout gates, but they must not +/// bypass these emergency controls. +pub(super) fn is_get_codec_streaming_base_enabled() -> bool { + let config = get_codec_streaming_config(); + config.enabled && config.body_compat_confirmed && config.header_compat_confirmed +} + fn get_codec_streaming_engine() -> GetCodecStreamingEngine { get_codec_streaming_config().engine } @@ -4076,7 +4088,35 @@ fn should_use_inline_fast_path( fi: &FileInfo, opts: &ObjectOptions, ) -> bool { - object_info.is_inline_fast_path_eligible() && fi.data.is_some() && range.is_none() && opts.part_number.is_none() + if !object_info.is_inline_fast_path_eligible() || fi.data.is_none() || range.is_some() || opts.part_number.is_some() { + 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 + || fi.erasure.data_blocks == 0 + || fi.erasure.block_size == 0 + { + return false; + } + true } enum SmallWritePath { @@ -10995,6 +11035,68 @@ mod tests { assert!(!should_use_inline_fast_path(&None, &object_info, &fi, &part_opts)); } + #[test] + fn inline_fast_path_allows_layout_specific_ec8_and_ec12_256kib_objects() { + for (data_blocks, parity_blocks) in [(8, 4), (12, 4)] { + let object_size = 256 * 1024_i64; + let mut fi = FileInfo::new("bucket/object", data_blocks, parity_blocks); + fi.size = object_size; + fi.data = Some(Bytes::from_static(b"payload")); + fi.add_object_part(1, String::new(), object_size as usize, None, object_size, None, None); + let object_info = ObjectInfo { + size: object_size, + data_blocks, + parity_blocks, + inlined: true, + parts: Arc::new(fi.parts.clone()), + ..Default::default() + }; + + assert!(should_use_inline_fast_path(&None, &object_info, &fi, &ObjectOptions::default())); + } + } + + #[test] + fn inline_fast_path_rejects_oversized_or_corrupt_inline_markers() { + for (data_blocks, parity_blocks) in [(1, 0), (8, 4), (12, 4)] { + let object_size = 4 * 1024 * 1024_i64; + let mut fi = FileInfo::new("bucket/object", data_blocks, parity_blocks); + fi.size = object_size; + fi.data = Some(Bytes::from_static(b"payload")); + fi.add_object_part(1, String::new(), object_size as usize, None, object_size, None, None); + let object_info = ObjectInfo { + size: object_size, + data_blocks, + parity_blocks, + inlined: true, + parts: Arc::new(fi.parts.clone()), + ..Default::default() + }; + + assert!( + object_info.is_inline_fast_path_eligible(), + "the marker and object shape alone must not be the final trust boundary" + ); + assert!(!should_use_inline_fast_path(&None, &object_info, &fi, &ObjectOptions::default())); + } + + // A malformed erasure geometry must fail closed before shard-size + // arithmetic, even when an inline marker and payload are present. + let (mut object_info, mut fi, opts) = direct_memory_test_metadata(4 * 1024 * 1024); + object_info.inlined = true; + fi.data = Some(Bytes::from_static(b"payload")); + fi.erasure.data_blocks = 0; + assert!(!should_use_inline_fast_path(&None, &object_info, &fi, &opts)); + + // A mismatched plain part size must not allow the large object size to + // reach the inline decoder's `Vec::with_capacity` allocation. + let (mut object_info, mut fi, opts) = direct_memory_test_metadata(4 * 1024 * 1024); + object_info.inlined = true; + fi.data = Some(Bytes::from_static(b"payload")); + fi.parts[0].actual_size = 0; + assert!(!should_use_inline_fast_path(&None, &object_info, &fi, &opts)); + } + #[test] fn small_object_direct_memory_decision_reports_bounded_reasons() { let (object_info, fi, opts) = direct_memory_test_metadata(1024); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 1e5c84afe..ee1aa7029 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -47,13 +47,13 @@ use super::super::{ 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_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, + 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, 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, take_prepared_get_object_metadata, to_object_err, try_read_inline_data_shards_direct, warn, @@ -68,7 +68,7 @@ use crate::set_disk::coding; use crate::set_disk::core::io_primitives::GetCodecStreamingReaderBuildOutcome; use crate::set_disk::mem; use crate::set_disk::metadata_sys; -use crate::set_disk::read::GetObjectDownstreamWriter; +use crate::set_disk::read::{GET_OBJECT_PATH_MID_SIZE_STREAMING, GetObjectDownstreamWriter}; use crate::set_disk::runtime_sources; use crate::storage_api_contracts::multipart::MultipartOperations; use crate::storage_api_contracts::object::ObjectIO; @@ -80,6 +80,97 @@ use rustfs_rio::TryGetIndex; use rustfs_utils::http::HeaderExt; use tokio::io::AsyncWriteExt; +// Keep the mid-size reader separate from the codec-streaming rollout. The +// latter is intentionally opt-in because its per-stripe worker can regress +// tiny objects; this bounded range is the gap between direct-memory GETs and +// the legacy duplex reader. +const ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE: &str = "RUSTFS_GET_MID_SIZE_STREAMING_ENABLE"; +const DEFAULT_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE: bool = true; +const GET_MID_SIZE_STREAMING_MIN_SIZE: usize = 128 * 1024 + 1; +const GET_MID_SIZE_STREAMING_MAX_SIZE: usize = 1024 * 1024; + +fn is_get_mid_size_streaming_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE) + } + #[cfg(not(test))] + { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE) + }) + } +} + +/// Return the object size when the bounded non-duplex reader is safe. +/// +/// This predicate deliberately has a narrower contract than the general +/// 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. +fn get_mid_size_streaming_object_size( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, + lock_optimization_enabled: bool, +) -> Option { + get_mid_size_streaming_object_size_with_flags( + range, + object_info, + fi, + opts, + lock_optimization_enabled, + is_get_mid_size_streaming_enabled(), + is_get_codec_streaming_base_enabled(), + ) +} + +fn get_mid_size_streaming_object_size_with_flags( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, + lock_optimization_enabled: bool, + mid_size_enabled: bool, + codec_base_enabled: bool, +) -> Option { + if !mid_size_enabled + || !codec_base_enabled + || !lock_optimization_enabled + || range.is_some() + || opts.part_number.is_some() + || opts.version_id.is_some() + || opts.incl_free_versions + || opts.skip_free_version + || opts.data_movement + || opts.raw_data_movement_read + || 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 { + return None; + } + (GET_MID_SIZE_STREAMING_MIN_SIZE..=GET_MID_SIZE_STREAMING_MAX_SIZE) + .contains(&object_size) + .then_some(object_size) +} + #[cfg(all(test, feature = "test-util"))] use super::super::GetObjectMetadataCacheEntry; #[cfg(test)] @@ -1634,7 +1725,12 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // Uses the shared predicate from ObjectInfo; additionally checks that // inline data is actually present and neither range nor partNumber is // in flight. - if should_use_inline_fast_path(&range, &object_info, fi, opts) { + let use_inline_fast_path = object_info.is_inline_fast_path_eligible() + && fi.data.is_some() + && range.is_none() + && opts.part_number.is_none() + && should_use_inline_fast_path(&range, &object_info, fi, opts); + 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; @@ -1963,6 +2059,43 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { return Ok(reader); } + // Mid-size plain objects use a bounded, non-duplex reader. Keep this + // path independent from the general codec-streaming rollout: that + // 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() { + match Self::get_object_mid_size_reader_with_fileinfo( + bucket, + object, + Arc::clone(&self.erasure_cache), + fi, + files, + disks, + self.set_index, + self.pool_index, + opts.skip_verify_bitrot, + object_class.as_str(), + size_bucket, + false, + ) + .await? + { + GetCodecStreamingReaderBuildOutcome::Reader(stream) => { + record_get_object_reader_path_observation(GET_OBJECT_PATH_MID_SIZE_STREAMING, object_class, size_bucket); + let (mut reader, _offset, _length) = + get_object_reader_with_context(&self.ctx, stream, range, &object_info, opts, &h).await?; + reader.body_source = body_source; + return Ok(finish_set_disk_read_lock(reader, read_lock_guard.take(), bucket, object)); + } + GetCodecStreamingReaderBuildOutcome::Fallback(_) => { + // The setup found a degraded-but-readable layout. Let the + // established codec gate and then legacy path handle it; + // this preserves whole-request fallback semantics. + } + } + } + match codec_streaming_gate.decision { GetCodecStreamingDecision::Use => { match Self::get_object_decode_reader_with_fileinfo( @@ -8198,6 +8331,190 @@ mod erasure_construction_tests { } } +#[cfg(test)] +mod mid_size_streaming_gate_tests { + use super::*; + use rustfs_filemeta::ObjectPartInfo; + use serial_test::serial; + + fn plain_metadata(size: usize) -> (ObjectInfo, FileInfo) { + let size_i64 = i64::try_from(size).expect("test size fits i64"); + let mut fi = FileInfo::new("object", 2, 2); + fi.size = size_i64; + fi.add_object_part(1, "etag".to_string(), size, fi.mod_time, size_i64, None, None); + let object_info = ObjectInfo { + size: size_i64, + parts: Arc::new(vec![ObjectPartInfo { + number: 1, + size, + actual_size: size_i64, + ..Default::default() + }]), + ..Default::default() + }; + (object_info, fi) + } + + #[test] + #[serial] + fn direct_memory_boundary_stays_out_of_mid_size_streaming() { + let (object_info, fi) = plain_metadata(128 * 1024); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true), + None + ); + } + + #[test] + #[serial] + fn mid_size_streaming_accepts_object_just_above_direct_memory_ceiling() { + let (object_info, fi) = plain_metadata(128 * 1024 + 1); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true), + Some(128 * 1024 + 1) + ); + } + + #[test] + #[serial] + fn mid_size_streaming_includes_one_mib_and_rejects_larger_objects() { + let (object_info, fi) = plain_metadata(1024 * 1024); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true), + Some(1024 * 1024) + ); + + let (large_info, large_fi) = plain_metadata(1024 * 1024 + 1); + assert_eq!( + get_mid_size_streaming_object_size_with_flags( + &None, + &large_info, + &large_fi, + &ObjectOptions::default(), + true, + true, + true + ), + None + ); + } + + #[test] + #[serial] + fn mid_size_streaming_rejects_ranges_and_transformed_objects() { + let (object_info, fi) = plain_metadata(256 * 1024); + let range = Some(HTTPRangeSpec { + is_suffix_length: false, + start: 0, + end: 1, + }); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&range, &object_info, &fi, &ObjectOptions::default(), true, true, true), + None + ); + + let mut encrypted = object_info; + encrypted.user_defined = Arc::new(HashMap::from([("x-minio-encryption-key".to_string(), "opaque".to_string())])); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &encrypted, &fi, &ObjectOptions::default(), true, true, true), + None + ); + } + + #[test] + #[serial] + fn mid_size_streaming_rejects_inconsistent_part_geometry() { + let (object_info, mut fi) = plain_metadata(256 * 1024); + fi.parts[0].size += 1; + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true), + None + ); + + let (mut object_info, fi) = plain_metadata(256 * 1024); + object_info.parts = Arc::new(vec![ObjectPartInfo { + number: 2, + size: fi.parts[0].size, + actual_size: fi.parts[0].actual_size, + ..Default::default() + }]); + assert_eq!( + get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true), + None + ); + } + + #[test] + #[serial] + fn mid_size_streaming_can_be_disabled_without_affecting_direct_memory_gate() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, Some("false")), + (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")), + ], + || { + let (object_info, fi) = plain_metadata(256 * 1024); + assert_eq!( + get_mid_size_streaming_object_size(&None, &object_info, &fi, &ObjectOptions::default(), true), + None + ); + }, + ); + } + + #[test] + #[serial] + fn mid_size_streaming_respects_codec_compatibility_kill_switches() { + let (object_info, fi) = plain_metadata(256 * 1024); + for (name, value) in [ + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, "false"), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, "false"), + (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, "false"), + ] { + temp_env::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")), + (name, Some(value)), + ], + || { + assert_eq!( + get_mid_size_streaming_object_size(&None, &object_info, &fi, &ObjectOptions::default(), true), + None + ); + }, + ); + } + } + + #[test] + #[serial] + fn mid_size_streaming_rejects_copy_source_policy() { + let (object_info, fi) = plain_metadata(256 * 1024); + let result = tokio::runtime::Runtime::new() + .expect("test runtime should initialize") + .block_on(crate::set_disk::with_get_object_read_policy( + crate::set_disk::GetObjectReadPolicy::CopySource, + async { + get_mid_size_streaming_object_size_with_flags( + &None, + &object_info, + &fi, + &ObjectOptions::default(), + true, + true, + true, + ) + }, + )); + assert_eq!(result, None); + } +} + #[cfg(test)] mod object_encryption_resolver_wiring_tests { use super::*; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 794be00d9..61bceb78d 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -52,6 +52,8 @@ use std::{ }; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +pub(super) const GET_OBJECT_PATH_MID_SIZE_STREAMING: &str = "mid_size_streaming"; + #[cfg(test)] use super::DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES; #[cfg(test)] @@ -1340,6 +1342,78 @@ impl SetDisks { #[allow(clippy::too_many_arguments)] #[hotpath::measure(impl_type = "SetDisks")] pub(super) async fn get_object_decode_reader_with_fileinfo( + bucket: &str, + object: &str, + erasure_cache: Arc, + fi: &FileInfo, + files: &[FileInfo], + disks: &[Option], + set_index: usize, + pool_index: usize, + skip_verify_bitrot: bool, + metrics_object_class: &'static str, + metrics_size_bucket: &'static str, + prefer_data_blocks_first_reader_setup: bool, + ) -> Result { + Self::get_object_decode_reader_with_fileinfo_inner( + bucket, + object, + erasure_cache, + fi, + files, + disks, + set_index, + pool_index, + skip_verify_bitrot, + metrics_object_class, + metrics_size_bucket, + prefer_data_blocks_first_reader_setup, + get_codec_streaming_metrics_path(), + false, + ) + .await + } + + /// Build the bounded mid-size reader while allowing a degraded part to + /// reuse the shard readers it just opened for an in-place legacy fallback. + /// This avoids opening every shard twice when a healthy quorum requires + /// reconstruction. + #[allow(clippy::too_many_arguments)] + pub(super) async fn get_object_mid_size_reader_with_fileinfo( + bucket: &str, + object: &str, + erasure_cache: Arc, + fi: &FileInfo, + files: &[FileInfo], + disks: &[Option], + set_index: usize, + pool_index: usize, + skip_verify_bitrot: bool, + metrics_object_class: &'static str, + metrics_size_bucket: &'static str, + prefer_data_blocks_first_reader_setup: bool, + ) -> Result { + Self::get_object_decode_reader_with_fileinfo_inner( + bucket, + object, + erasure_cache, + fi, + files, + disks, + set_index, + pool_index, + skip_verify_bitrot, + metrics_object_class, + metrics_size_bucket, + prefer_data_blocks_first_reader_setup, + GET_OBJECT_PATH_MID_SIZE_STREAMING, + true, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + async fn get_object_decode_reader_with_fileinfo_inner( bucket: &str, object: &str, erasure_cache: Arc, @@ -1352,6 +1426,8 @@ impl SetDisks { metrics_object_class: &'static str, metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, + metrics_path: &'static str, + allow_inplace_legacy_fallback: 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); @@ -1378,7 +1454,8 @@ impl SetDisks { // 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. - false, + allow_inplace_legacy_fallback, + metrics_path, ) .await; } @@ -1431,6 +1508,7 @@ impl SetDisks { // if part 1 is already degraded, the entire GET drops to the legacy // duplex path before a single byte is streamed (semantics unchanged). false, + metrics_path, ) .await? { @@ -1452,6 +1530,7 @@ impl SetDisks { skip_verify_bitrot, metrics_object_class, metrics_size_bucket, + metrics_path, }); let builder: LazyPartBuilder = Box::new(move |remaining_index| { let ctx = Arc::clone(&ctx); @@ -1477,13 +1556,14 @@ impl SetDisks { // part in place to a legacy per-part decode reader instead of // failing the stream mid-flight. true, + ctx.metrics_path, ) .await }) }); Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new( - LazyMultipartCodecStreamingReader::new(first_reader, total_parts, builder, get_codec_streaming_metrics_path()), + LazyMultipartCodecStreamingReader::new(first_reader, total_parts, builder, metrics_path), ))) } @@ -1504,14 +1584,8 @@ impl SetDisks { metrics_object_class: &'static str, metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, - // backlog#879: when the codec streaming fast path cannot serve this part - // (a shard is missing and reconstruction is required), `false` preserves - // the historical whole-request fallback by returning `Fallback`, while - // `true` degrades in place — building a legacy per-part decode reader that - // reuses the shard readers already opened here. Only lazily-built later - // parts pass `true`, so the eager first-part fallback semantics are - // untouched and the common read path is never affected. allow_inplace_legacy_fallback: bool, + metrics_path: &'static str, ) -> Result { if part_length > part_size { return Err(Error::other("codec streaming reader part length exceeds part size")); @@ -1528,7 +1602,6 @@ impl SetDisks { let read_length = till_offset.saturating_sub(read_offset); let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); - let metrics_path = get_codec_streaming_metrics_path(); let reader_stage_metrics = stage_metrics_enabled.then_some(BitrotReaderStageMetrics { path: metrics_path, reader_construction_stage: GET_STAGE_READER_TASK_READER_CONSTRUCTION, @@ -1751,6 +1824,7 @@ struct LazyCodecPartContext { skip_verify_bitrot: bool, metrics_object_class: &'static str, metrics_size_bucket: &'static str, + metrics_path: &'static str, } type LazyPartBuildHandle = tokio::task::JoinHandle>; @@ -4594,6 +4668,154 @@ mod tests { assert_eq!(body, part_data); } + #[tokio::test] + #[serial_test::serial] + async fn mid_size_reader_restores_plain_objects_without_duplex() { + for size in [256_usize * 1024, 1024_usize * 1024] { + let payload = (0..size) + .map(|index| u8::try_from(index % 251).expect("pattern byte fits u8")) + .collect::>(); + let erasure = coding::Erasure::new(4, 2, 1024 * 1024); + let mut fi = codec_streaming_test_fileinfo(i64::try_from(size).expect("test size fits i64"), 1); + fi.erasure.block_size = erasure.block_size; + fi.erasure.distribution = (1..=erasure.total_shard_count()).collect(); + let files = codec_streaming_inline_files(&erasure, &payload).await; + let (_dirs, disks) = local_test_disks(files.len(), CODEC_STREAMING_TEST_BUCKET).await; + + let outcome = SetDisks::get_object_mid_size_reader_with_fileinfo( + CODEC_STREAMING_TEST_BUCKET, + CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), + &fi, + &files, + &disks, + 0, + 0, + false, + "plain_single_part", + "le_1mib", + false, + ) + .await + .expect("mid-size reader setup should succeed"); + let GetCodecStreamingReaderBuildOutcome::Reader(mut reader) = outcome else { + panic!("mid-size plain object should use streaming reader"); + }; + + let mut body = Vec::new(); + reader + .read_to_end(&mut body) + .await + .expect("mid-size reader should restore full body"); + assert_eq!(body, payload, "mid-size reader must preserve every payload byte for object size {size}"); + } + } + + #[tokio::test] + #[serial_test::serial] + async fn mid_size_reader_restores_plain_objects_from_external_part_files() { + for size in [256_usize * 1024, 1024_usize * 1024] { + let payload = (0..size) + .map(|index| u8::try_from(index % 251).expect("pattern byte fits u8")) + .collect::>(); + let erasure = coding::Erasure::new(4, 2, 1024 * 1024); + let mut fi = codec_streaming_test_fileinfo(i64::try_from(size).expect("test size fits i64"), 1); + fi.erasure.block_size = erasure.block_size; + fi.erasure.distribution = (1..=erasure.total_shard_count()).collect(); + let (_dirs, disks) = local_test_disks(erasure.total_shard_count(), CODEC_STREAMING_TEST_BUCKET).await; + let (data_dir, files) = codec_streaming_external_files(&erasure, &payload, &disks).await; + fi.data_dir = Some(data_dir); + + let outcome = SetDisks::get_object_mid_size_reader_with_fileinfo( + CODEC_STREAMING_TEST_BUCKET, + CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), + &fi, + &files, + &disks, + 0, + 0, + false, + "plain_single_part", + "le_1mib", + false, + ) + .await + .expect("external mid-size reader setup should succeed"); + let GetCodecStreamingReaderBuildOutcome::Reader(mut reader) = outcome else { + panic!("external plain object should use streaming reader"); + }; + + let mut body = Vec::new(); + reader + .read_to_end(&mut body) + .await + .expect("external mid-size reader should restore full body"); + assert_eq!( + body, payload, + "external part files must preserve every payload byte for object size {size}" + ); + } + } + + #[test] + fn minio_large_object_metadata_remains_reader_compatible() { + let raw = rustfs_filemeta::test_data::create_minio_large_object_xlmeta().expect("load MinIO large-object fixture"); + let file_info = rustfs_filemeta::FileMeta::load(&raw) + .expect("MinIO xl.meta should decode") + .into_fileinfo("interop", "large.bin", "", false, false, true) + .expect("MinIO xl.meta should materialize FileInfo"); + + file_info + .validate_for_metadata_read() + .expect("MinIO metadata should satisfy the reader validation contract"); + assert_eq!(file_info.size, 300_000); + assert_eq!(file_info.parts.len(), 1); + assert!(file_info.data_dir.is_some()); + assert!(!file_info.inline_data()); + } + + #[tokio::test] + #[serial_test::serial] + async fn mid_size_reader_reuses_open_readers_for_degraded_quorum() { + let size = 256 * 1024; + let payload = vec![0x5a; size]; + let erasure = coding::Erasure::new(4, 2, 1024 * 1024); + let mut fi = codec_streaming_test_fileinfo(i64::try_from(size).expect("test size fits i64"), 1); + fi.erasure.block_size = erasure.block_size; + fi.erasure.distribution = (1..=erasure.total_shard_count()).collect(); + let mut files = codec_streaming_inline_files(&erasure, &payload).await; + files[0].data = None; + let (_dirs, disks) = local_test_disks(files.len(), CODEC_STREAMING_TEST_BUCKET).await; + + let outcome = SetDisks::get_object_mid_size_reader_with_fileinfo( + CODEC_STREAMING_TEST_BUCKET, + CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), + &fi, + &files, + &disks, + 0, + 0, + false, + "plain_single_part", + "le_1mib", + false, + ) + .await + .expect("degraded mid-size reader setup should retain read quorum"); + let GetCodecStreamingReaderBuildOutcome::Reader(mut reader) = outcome else { + panic!("degraded quorum should use in-place legacy fallback reader"); + }; + + let mut body = Vec::new(); + reader + .read_to_end(&mut body) + .await + .expect("degraded reader should reconstruct full body"); + assert_eq!(body, payload); + } + #[tokio::test] async fn get_object_with_fileinfo_restores_missing_inline_data_shard_and_submits_repair() { let part_data = b"abcdefgh"; @@ -4654,6 +4876,7 @@ mod tests { "test-size-bucket", false, false, + get_codec_streaming_metrics_path(), ) .await; assert!(oversized.is_err(), "part_length > part_size must be rejected"); @@ -4674,6 +4897,7 @@ mod tests { "test-size-bucket", false, false, + get_codec_streaming_metrics_path(), ) .await; assert!(missing_quorum.is_err(), "reader setup must fail closed when no shard can answer"); @@ -5208,7 +5432,7 @@ mod tests { Ok(decoded) } - async fn codec_streaming_inline_files(erasure: &coding::Erasure, part_data: &'static [u8]) -> Vec { + async fn codec_streaming_inline_files(erasure: &coding::Erasure, part_data: &[u8]) -> Vec { let shards = erasure.encode_data(part_data).expect("test part should encode"); let distribution = (1..=erasure.total_shard_count()).collect::>(); let mut files = Vec::with_capacity(shards.len()); @@ -5231,6 +5455,27 @@ mod tests { files } + async fn codec_streaming_external_files( + erasure: &coding::Erasure, + part_data: &[u8], + disks: &[Option], + ) -> (Uuid, Vec) { + let data_dir = Uuid::from_u128(0x2060_0000_0000_0000_0000_0000_0000_0001); + let mut files = codec_streaming_inline_files(erasure, part_data).await; + for (index, file) in files.iter_mut().enumerate() { + let shard = file.data.take().expect("external fixture shard should be encoded"); + file.data_dir = Some(data_dir); + let path = format!("{CODEC_STREAMING_TEST_OBJECT}/{data_dir}/part.1"); + disks[index] + .as_ref() + .expect("external fixture disk should be online") + .write_all(CODEC_STREAMING_TEST_BUCKET, &path, shard) + .await + .expect("external fixture shard should be written"); + } + (data_dir, files) + } + #[tokio::test] async fn bitrot_reader_setup_stops_at_read_quorum() { let setup = setup_inline_bitrot_readers( diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index 809425b82..34d8f7f14 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -180,6 +180,14 @@ const GET_OBJECT_STAGE_PATH_S3_HANDLER: &str = "s3_handler"; const GET_OBJECT_STAGE_REQUEST_INGRESS_TO_CONTEXT: &str = "request_ingress_to_context"; +const GET_OBJECT_STAGE_REQUEST_SHAPE: &str = "request_shape"; + +const GET_OBJECT_STAGE_REQUEST_VALIDATION: &str = "request_validation"; + +const GET_OBJECT_STAGE_BUCKET_VALIDATION: &str = "bucket_validation"; + +const GET_OBJECT_STAGE_RESPONSE_FINALIZE: &str = "response_finalize"; + const GET_OBJECT_STAGE_OUTPUT_STRATEGY: &str = "output_strategy"; const GET_OBJECT_STAGE_BODY_BUILD: &str = "body_build"; @@ -3817,6 +3825,8 @@ impl DefaultObjectUsecase { let _ = context.object_store(); } + let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); + let request_shape_start = stage_metrics_enabled.then(std::time::Instant::now); let inbound_request_context = req.extensions.get::(); let request_id = inbound_request_context .map(|ctx| ctx.request_id.clone()) @@ -3831,6 +3841,7 @@ impl DefaultObjectUsecase { ); } let bootstrap = self.init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id)?; + record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_SHAPE, request_shape_start); let timeout_config = bootstrap.timeout_config; let wrapper = bootstrap.wrapper; let request_start = bootstrap.request_start; @@ -3842,6 +3853,7 @@ impl DefaultObjectUsecase { // Cheap request-shape validations run first so invalid requests keep // their InvalidArgument precedence over bucket existence. + let request_validation_start = stage_metrics_enabled.then(std::time::Instant::now); let validated = match Self::validate_get_object_request(&req) { Ok(validated) => validated, Err(err) => { @@ -3849,6 +3861,7 @@ impl DefaultObjectUsecase { return Err(err); } }; + record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_VALIDATION, request_validation_start); // SF05: Store lookup next (5s-TTL bucket-validation cache). Bucket // existence is established before any bucket-metadata work, so requests @@ -3859,24 +3872,26 @@ impl DefaultObjectUsecase { let object_metadata_progress = object_traffic_health .as_deref() .and_then(ObjectTrafficHealth::track_read_metadata); - let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); + let store_lookup_start = stage_metrics_enabled.then(std::time::Instant::now); let Some(store) = self.object_store() else { lifecycle.finish_err(); return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await { - lifecycle.finish_err(); - return Err(err); - } if let Some(store_lookup_start) = store_lookup_start { rustfs_io_metrics::record_get_object_stage_duration( - "s3_handler", + GET_OBJECT_STAGE_PATH_S3_HANDLER, "store_lookup", store_lookup_start.elapsed().as_secs_f64(), ); } + let bucket_validation_start = stage_metrics_enabled.then(std::time::Instant::now); + if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await { + lifecycle.finish_err(); + return Err(err); + } + record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BUCKET_VALIDATION, bucket_validation_start); - let request_context_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); + let request_context_start = stage_metrics_enabled.then(std::time::Instant::now); let request_context = match Self::prepare_get_object_request_context(validated, &req.headers).await { Ok(request_context) => request_context, Err(err) => { @@ -4051,7 +4066,8 @@ impl DefaultObjectUsecase { optimal_buffer_size, ); - Self::finalize_get_object_response( + let response_finalize_start = stage_metrics_enabled.then(std::time::Instant::now); + let response = Self::finalize_get_object_response( helper, &bucket, &req.method, @@ -4061,7 +4077,9 @@ impl DefaultObjectUsecase { output, extra_checksum_headers, ) - .await + .await; + record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_RESPONSE_FINALIZE, response_finalize_start); + response } pub async fn execute_get_object_attributes( diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 386089cc1..d57cd111c 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -23,8 +23,18 @@ const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 102 const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES"; +/// Maximum body size materialized by the ordinary eager PUT path. +/// +/// Bodies above this boundary stay streaming so a 1 MiB request does not +/// reserve a full request-sized buffer while the EC writer is consuming it. +/// The environment override keeps the boundary reversible for workload A/B +/// tests and for deployments whose measured workload favors eager ingestion. +const ENV_SMALL_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_SMALL_EAGER_PUT_MAX_SIZE_BYTES"; + const DEFAULT_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024; +const DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES: usize = 512 * 1024; + const PUT_EAGER_STATUS_ELIGIBLE: &str = "eligible"; const PUT_EAGER_STATUS_EXTRACT: &str = "extract"; @@ -43,6 +53,8 @@ const PUT_EAGER_STATUS_AWS_CHUNKED_MISSING_DECODED_LENGTH: &str = "aws_chunked_m static CACHED_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: std::sync::OnceLock = std::sync::OnceLock::new(); +static CACHED_SMALL_EAGER_PUT_MAX_SIZE_BYTES: std::sync::OnceLock = std::sync::OnceLock::new(); + const EVENT_PUT_OBJECT_STORE_INFLIGHT_SLOW: &str = "put_object_store_inflight_slow"; const EVENT_PUT_OBJECT_STORE_RETURNED: &str = "put_object_store_returned"; @@ -483,13 +495,29 @@ fn should_use_small_eager_put_path( should_compress: bool, is_extract: bool, ) -> bool { - const SMALL_EAGER_PUT_MAX_SIZE: i64 = 1024 * 1024; + should_use_small_eager_put_path_with_max_size( + size, + headers, + server_side_encryption_requested, + should_compress, + is_extract, + small_eager_put_max_size_bytes(), + ) +} +fn should_use_small_eager_put_path_with_max_size( + size: i64, + headers: &HeaderMap, + server_side_encryption_requested: bool, + should_compress: bool, + is_extract: bool, + max_size: i64, +) -> bool { if is_extract || should_compress || server_side_encryption_requested { return false; } - if size <= 0 || size > SMALL_EAGER_PUT_MAX_SIZE { + if size <= 0 || size > max_size { return false; } @@ -504,6 +532,42 @@ fn should_use_small_eager_put_path( true } +fn small_eager_put_max_size_bytes() -> i64 { + let configured = *CACHED_SMALL_EAGER_PUT_MAX_SIZE_BYTES + .get_or_init(|| rustfs_utils::get_env_usize(ENV_SMALL_EAGER_PUT_MAX_SIZE_BYTES, DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES)); + i64::try_from(configured).unwrap_or(i64::MAX) +} + +fn select_put_path( + size: i64, + headers: &HeaderMap, + server_side_encryption_requested: bool, + should_compress: bool, + is_extract: bool, +) -> (&'static str, &'static str, bool, bool) { + let use_empty_or_small_eager_put_path = size == 0 + || should_use_small_eager_put_path(size, headers, server_side_encryption_requested, should_compress, is_extract); + let zero_copy_eager_put_path_status = + zero_copy_eager_put_path_status(size, headers, server_side_encryption_requested, should_compress, is_extract); + let use_zero_copy_eager_put_path = zero_copy_eager_put_path_status == PUT_EAGER_STATUS_ELIGIBLE; + let put_path = if should_compress { + "stream_compressed" + } else if use_zero_copy_eager_put_path { + "zero_copy_eager" + } else if use_empty_or_small_eager_put_path { + "small_eager" + } else { + "streaming" + }; + + ( + put_path, + zero_copy_eager_put_path_status, + use_zero_copy_eager_put_path, + use_empty_or_small_eager_put_path, + ) +} + /// Objects at or below this size bypass BytesPool and use direct allocation. /// This avoids Small-tier Mutex contention under high concurrency for tiny objects /// where the allocation cost is negligible (≤4KiB memcpy). @@ -800,12 +864,26 @@ impl DefaultObjectUsecase { async fn execute_put_object_inner(&self, _fs: &FS, req: S3Request) -> S3Result> { let start_time = std::time::Instant::now(); + let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled(); let mut req = req; + let request_shape_stage_start = put_stage_metrics_enabled.then(Instant::now); + if let Some(context) = &self.context { let _ = context.object_store(); } + // Authentication and header parsing happen in the S3 middleware before + // this use case runs. Attribute that already-paid request prefix from + // the request context without adding per-request work when stage + // metrics are disabled. + if put_stage_metrics_enabled && let Some(context) = req.extensions.get::() { + rustfs_io_metrics::record_put_object_stage_duration( + "request_ingress_to_context", + context.start_time.elapsed().as_secs_f64() * 1000.0, + ); + } + let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req); let max_content_length = parse_presigned_put_max_content_length( &req.headers, @@ -893,6 +971,7 @@ impl DefaultObjectUsecase { req.headers.get("content-type").and_then(|value| value.to_str().ok()), req.headers.get("content-encoding").and_then(|value| value.to_str().ok()), )?; + rustfs_io_metrics::record_put_object_stage_duration_from("app_request_shape", request_shape_stage_start); let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; @@ -929,6 +1008,7 @@ impl DefaultObjectUsecase { // The app check preserves the existing S3 error contract; the storage // commit path reserves the exact net logical growth under its locks. + let quota_stage_start = put_stage_metrics_enabled.then(Instant::now); let quota_check = self .check_bucket_quota( &bucket, @@ -936,6 +1016,7 @@ impl DefaultObjectUsecase { u64::try_from(size).map_err(|_| S3Error::new(S3ErrorCode::UnexpectedContent))?, ) .await?; + rustfs_io_metrics::record_put_object_stage_duration_from("app_quota_check", quota_stage_start); let quota_enabled = quota_check.as_ref().is_some_and(|result| result.quota_limit.is_some()); if quota_enabled && ciphertext_passthrough { return Err(S3Error::with_message( @@ -944,7 +1025,6 @@ impl DefaultObjectUsecase { )); } - let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled(); let ingress_stage_start = put_stage_metrics_enabled.then(Instant::now); let should_compress = is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough; @@ -954,11 +1034,13 @@ impl DefaultObjectUsecase { // Resolve the store through the request-bound server context // (backlog#1052 S6), not the process-global handle, so an embedded // second server never writes into the first server's store. + let store_lookup_stage_start = put_stage_metrics_enabled.then(Instant::now); let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now); validate_bucket_exists(&store, &bucket).await?; + rustfs_io_metrics::record_put_object_stage_duration_from("app_store_lookup", store_lookup_stage_start); rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start); let put_admission = match get_concurrency_manager() @@ -1006,24 +1088,12 @@ impl DefaultObjectUsecase { debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key); } - let use_empty_or_small_eager_put_path = size == 0 - || should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false); - let zero_copy_eager_put_path_status = - zero_copy_eager_put_path_status(size, &req.headers, server_side_encryption_requested, should_compress, false); - let use_zero_copy_eager_put_path = zero_copy_eager_put_path_status == PUT_EAGER_STATUS_ELIGIBLE; + let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) = + select_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false); if use_zero_copy_eager_put_path { counter!(buffered_write::ATTEMPTS_TOTAL).increment(1); histogram!(buffered_write::ATTEMPT_SIZE_BYTES).record(size as f64); } - let put_path = if should_compress { - "stream_compressed" - } else if use_zero_copy_eager_put_path { - "zero_copy_eager" - } else if use_empty_or_small_eager_put_path { - "small_eager" - } else { - "streaming" - }; rustfs_io_metrics::record_put_object_diagnostics( put_path, zero_copy_eager_put_path_status, @@ -1518,6 +1588,7 @@ impl DefaultObjectUsecase { Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned }) } }); + let commit_stage_start = put_stage_metrics_enabled.then(Instant::now); let put_commit_result = if let Some(cancellation) = eager_put_commit_cancellation { EagerPutCommitOwner::new(put_commit, cancellation, EAGER_PUT_COMMIT_CANCELLATION_GRACE) .join() @@ -1525,6 +1596,7 @@ impl DefaultObjectUsecase { } else { put_commit.await }; + rustfs_io_metrics::record_put_object_stage_duration_from("store_commit", commit_stage_start); let PutObjectCommitResult { obj_info, put_versioned } = match put_commit_result { Ok(Ok(result)) => result, Ok(Err(err)) => { @@ -1589,10 +1661,12 @@ impl DefaultObjectUsecase { // For browser-based POST uploads (multipart/form-data), response status/body handling // is decided by s3s PostObject serializer (success_action_status / redirect semantics). + let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now); let mut response = S3Response::new(output); // Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed // field for (#1256). inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers); + rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start); let result = Ok(response); let _ = helper.complete(&result); @@ -2230,12 +2304,51 @@ mod tests { } #[test] - fn should_use_small_eager_put_path_allows_up_to_1mb() { + fn should_use_small_eager_put_path_keeps_small_objects_eager() { let headers = HeaderMap::new(); assert!(should_use_small_eager_put_path(1024, &headers, false, false, false)); - assert!(should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false)); - assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false)); + assert!(should_use_small_eager_put_path(128 * 1024, &headers, false, false, false)); + assert!(should_use_small_eager_put_path(512 * 1024, &headers, false, false, false)); + assert!(!should_use_small_eager_put_path(512 * 1024 + 1, &headers, false, false, false)); + assert!(!should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false)); + } + + #[test] + fn select_put_path_switches_at_small_eager_boundary() { + let headers = HeaderMap::new(); + + let (small_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024, &headers, false, false, false); + assert_eq!(small_path, "small_eager"); + assert!(!use_zero_copy); + assert!(use_small_eager); + + let (streaming_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024 + 1, &headers, false, false, false); + assert_eq!(streaming_path, "streaming"); + assert!(!use_zero_copy); + assert!(!use_small_eager); + } + + #[test] + fn should_use_small_eager_put_path_allows_a_b_override_at_1mb() { + let headers = HeaderMap::new(); + + assert!(should_use_small_eager_put_path_with_max_size( + 1024 * 1024, + &headers, + false, + false, + false, + 1024 * 1024, + )); + assert!(!should_use_small_eager_put_path_with_max_size( + 1024 * 1024 + 1, + &headers, + false, + false, + false, + 1024 * 1024, + )); } #[test] @@ -2456,6 +2569,35 @@ mod tests { assert_eq!(extra.code(), &S3ErrorCode::UnexpectedContent); } + #[tokio::test] + async fn streaming_put_hash_reader_rejects_extra_byte_at_eager_boundary() { + use tokio::io::AsyncReadExt; + + let declared_size = 512 * 1024; + let declared_size_i64 = i64::try_from(declared_size).expect("test size should fit i64"); + let mut reader = HashReader::from_stream( + std::io::Cursor::new(vec![0x5a; declared_size + 1]), + declared_size_i64, + declared_size_i64, + None, + None, + false, + ) + .expect("streaming PUT hash reader should be constructed"); + let mut body = Vec::new(); + + let err = reader + .read_to_end(&mut body) + .await + .expect_err("streaming PUT must reject a body larger than Content-Length"); + + assert!( + err.to_string().contains("more bytes than specified"), + "unexpected extra-body error: {err}" + ); + assert_eq!(body.len(), declared_size); + } + #[tokio::test] async fn read_small_put_body_exact_direct_handles_empty_body_boundary() { let empty = read_small_put_body_exact_direct(std::io::Cursor::new(Vec::::new()), 0)