From 127b662f3f00f6c985a3e1203faecf90fc7fcf1a Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 18 Aug 2026 19:53:06 +0800 Subject: [PATCH] feat(app): add opt-in small GET body once path (#6216) Use the merged s3s single-chunk StreamingBlob support for exact-length materialized GET bodies when RUSTFS_GET_SMALL_BODY_ONCE_ENABLE is enabled. Keep the default path unchanged and fall back to the guarded MemoryTrackedBytesStream on length mismatch. Co-authored-by: heihutu --- Cargo.lock | 5 +- Cargo.toml | 2 +- rustfs/src/app/object_usecase.rs | 84 +++++++++++++++++++++++++++++++- 3 files changed, 87 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 82f66641e..db75a7992 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10678,7 +10678,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" [[package]] name = "s3s" version = "0.14.1" -source = "git+https://github.com/rustfs/s3s.git?rev=d7028511a53f69d41ed3c69f36899f9b1aede647#d7028511a53f69d41ed3c69f36899f9b1aede647" +source = "git+https://github.com/rustfs/s3s.git?rev=ff8106f46098b60fce8ee88518f34974fbfbec4b#ff8106f46098b60fce8ee88518f34974fbfbec4b" dependencies = [ "arc-swap", "arrayvec", @@ -10706,6 +10706,7 @@ dependencies = [ "numeric_cast", "pin-project-lite", "quick-xml", + "regex", "serde", "serde_json", "serde_urlencoded", @@ -11809,7 +11810,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index ae9b497e0..f2080a15f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -290,7 +290,7 @@ rustify = { version = "0.7", default-features = false } rustix = { version = "1.1.4" } rust-embed = { version = "8.12.0" } rustc-hash = { version = "2.1.3" } -s3s = { git = "https://github.com/rustfs/s3s.git", rev = "d7028511a53f69d41ed3c69f36899f9b1aede647" } +s3s = { git = "https://github.com/rustfs/s3s.git", rev = "ff8106f46098b60fce8ee88518f34974fbfbec4b" } serial_test = "4.0.1" shadow-rs = { default-features = false, version = "2.0.0" } siphasher = "1.0.3" diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 1bb337c69..bcce2b705 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -782,6 +782,7 @@ const MID_BODY_READER_STREAM_BUFFER_THRESHOLD_BYTES: i64 = MI_B as i64; const ENV_RUSTFS_GET_SEEK_BUFFER_ENABLE: &str = "RUSTFS_GET_SEEK_BUFFER_ENABLE"; const ENV_RUSTFS_GET_READER_STREAM_BUFFER_SIZE: &str = "RUSTFS_GET_READER_STREAM_BUFFER_SIZE"; const ENV_RUSTFS_GET_OUTPUT_HANDOFF_ATTRIBUTION_ENABLE: &str = "RUSTFS_GET_OUTPUT_HANDOFF_ATTRIBUTION_ENABLE"; +const ENV_RUSTFS_GET_SMALL_BODY_ONCE_ENABLE: &str = "RUSTFS_GET_SMALL_BODY_ONCE_ENABLE"; const GET_READER_STREAM_BUFFER_SOURCE_SELECTED: &str = "selected"; const GET_READER_STREAM_BUFFER_SOURCE_ENV_OVERRIDE: &str = "env_override"; const GET_READER_STREAM_POLL_PENDING: &str = "pending"; @@ -813,6 +814,18 @@ fn is_get_output_handoff_attribution_enabled() -> bool { *ENABLED.get_or_init(|| rustfs_utils::get_env_bool(ENV_RUSTFS_GET_OUTPUT_HANDOFF_ATTRIBUTION_ENABLE, false)) } +fn is_get_small_body_once_enabled() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool(ENV_RUSTFS_GET_SMALL_BODY_ONCE_ENABLE, false) + } + #[cfg(not(test))] + { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(|| rustfs_utils::get_env_bool(ENV_RUSTFS_GET_SMALL_BODY_ONCE_ENABLE, false)) + } +} + fn is_get_seek_buffer_enabled() -> bool { static ENABLED: OnceLock = OnceLock::new(); *ENABLED.get_or_init(|| rustfs_utils::get_env_bool(ENV_RUSTFS_GET_SEEK_BUFFER_ENABLE, false)) @@ -932,6 +945,30 @@ struct MemoryTrackedBytesStream { lifecycle: GetObjectBodyLifecycle, } +struct MemoryOnceBodyOwner { + bytes: Bytes, + _guard: Option, + // Body::Once has no poll hook, so this opt-in path only holds the request + // guard until the bytes are dropped; the result status remains unknown. + _lifecycle: GetObjectBodyLifecycle, +} + +impl MemoryOnceBodyOwner { + fn new(bytes: Bytes, guard: Option, lifecycle: GetObjectBodyLifecycle) -> Self { + Self { + bytes, + _guard: guard, + _lifecycle: lifecycle, + } + } +} + +impl AsRef<[u8]> for MemoryOnceBodyOwner { + fn as_ref(&self) -> &[u8] { + self.bytes.as_ref() + } +} + #[derive(Default)] struct GetObjectBodyLifecycle { request_guard: Option, @@ -4160,7 +4197,12 @@ impl DefaultObjectUsecase { let bytes_len = bytes.len(); let guard = rustfs_io_metrics::track_get_object_buffered_bytes(bytes_len); let remaining = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX); - let blob = StreamingBlob::new(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle)); + let blob = if is_get_small_body_once_enabled() && bytes_len == remaining { + let owner = MemoryOnceBodyOwner::new(bytes, guard, lifecycle); + StreamingBlob::from_bytes(Bytes::from_owner(owner)) + } else { + StreamingBlob::new(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle)) + }; if let Some(handoff_start) = handoff_start { rustfs_io_metrics::record_get_object_response_handoff( "single_chunk", @@ -12955,6 +12997,46 @@ mod tests { assert_eq!(blob.remaining_length().exact(), Some(5)); } + #[test] + #[serial_test::serial] + fn memory_blob_once_fast_path_holds_guard_until_bytes_drop() { + temp_env::with_var(ENV_RUSTFS_GET_SMALL_BODY_ONCE_ENABLE, Some("true"), || { + let initial = GetObjectGuard::concurrent_count(); + let guard = GetObjectGuard::new(); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + + let blob = DefaultObjectUsecase::build_memory_bytes_blob( + Bytes::from_static(b"hello"), + 5, + GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + GetObjectBodyLifecycle::tracked(guard), + ); + let mut body = s3s::Body::from(blob); + let bytes = body.take_bytes().expect("opt-in exact memory body should stay on Body::Once"); + + assert_eq!(bytes, Bytes::from_static(b"hello")); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + drop(bytes); + assert_eq!(GetObjectGuard::concurrent_count(), initial); + }); + } + + #[test] + #[serial_test::serial] + fn memory_blob_once_fast_path_rejects_length_mismatch() { + temp_env::with_var(ENV_RUSTFS_GET_SMALL_BODY_ONCE_ENABLE, Some("true"), || { + let blob = DefaultObjectUsecase::build_memory_bytes_blob( + Bytes::from_static(b"test"), + 5, + GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + GetObjectBodyLifecycle::disabled(), + ); + let mut body = s3s::Body::from(blob); + + assert!(body.take_bytes().is_none(), "mismatched memory body must keep the guarded stream path"); + }); + } + #[tokio::test] async fn get_object_streaming_reader_times_out_when_body_stalls() { let reader = GetObjectStreamingReader::new(