From 565cbdffedd80c98689cea2075cd24fb4e97d33e Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 5 May 2026 20:58:26 +0800 Subject: [PATCH] fix(getobject): prevent large-download memory buffering (#2809) --- crates/config/src/constants/runtime.rs | 9 +- rustfs/src/app/object_usecase.rs | 180 +++++++++++++++++++++++-- 2 files changed, 175 insertions(+), 14 deletions(-) diff --git a/crates/config/src/constants/runtime.rs b/crates/config/src/constants/runtime.rs index e11a25c05..b164a03b3 100644 --- a/crates/config/src/constants/runtime.rs +++ b/crates/config/src/constants/runtime.rs @@ -77,10 +77,13 @@ pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_WRITE_ENABLE: bool = false; pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_READ_ENABLE: bool = false; pub const DEFAULT_OBJECT_FILE_CACHE_RECLAIM_THRESHOLD: usize = 4 * 1024 * 1024; -/// Threshold for small object seek support in megabytes. +/// Threshold for small object seek support in bytes. /// -/// When an object is smaller than this size, rustfs will provide seek support. +/// When an object response is smaller than this size, rustfs may provide +/// in-memory seek support. Runtime GET logic also enforces a hard safety cap +/// (`64 MiB`) to prevent large-download memory spikes even if this threshold +/// is configured higher. /// -/// Default is set to 10MB. +/// Default is set to 10 MiB. pub const ENV_OBJECT_SEEK_SUPPORT_THRESHOLD: &str = "RUSTFS_OBJECT_SEEK_SUPPORT_THRESHOLD"; pub const DEFAULT_OBJECT_SEEK_SUPPORT_THRESHOLD: usize = 10 * 1024 * 1024; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index a1691cdeb..4543cd934 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -119,6 +119,7 @@ use std::collections::HashMap; use std::ops::Add; use std::path::Path; use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; @@ -132,6 +133,8 @@ use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; const ACCEPT_RANGES_BYTES: &str = "bytes"; +const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024; +static GET_OBJECT_BUFFER_THRESHOLD_WARNED: AtomicBool = AtomicBool::new(false); struct DeadlockRequestGuard { deadlock_detector: Arc, @@ -394,6 +397,47 @@ fn object_seek_support_threshold() -> usize { }) } +fn should_buffer_get_object_in_memory( + info: &ObjectInfo, + response_content_length: i64, + part_number: Option, + has_range: bool, +) -> bool { + let configured_threshold = object_seek_support_threshold() as i64; + should_buffer_get_object_in_memory_with_threshold(info, response_content_length, part_number, has_range, configured_threshold) +} + +fn should_buffer_get_object_in_memory_with_threshold( + _info: &ObjectInfo, + response_content_length: i64, + part_number: Option, + has_range: bool, + configured_threshold: i64, +) -> bool { + if part_number.is_some() || has_range || response_content_length <= 0 || configured_threshold <= 0 { + return false; + } + + let effective_threshold = configured_threshold.min(MAX_GET_OBJECT_MEMORY_BUFFER_BYTES); + if configured_threshold > MAX_GET_OBJECT_MEMORY_BUFFER_BYTES + && GET_OBJECT_BUFFER_THRESHOLD_WARNED + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + warn!( + configured_threshold_bytes = configured_threshold, + hard_limit_bytes = MAX_GET_OBJECT_MEMORY_BUFFER_BYTES, + "RUSTFS_OBJECT_SEEK_SUPPORT_THRESHOLD exceeds safety cap; using capped in-memory buffer threshold" + ); + } + + if response_content_length > effective_threshold { + return false; + } + + true +} + #[cfg(test)] mod deadlock_request_guard_tests { use super::DeadlockRequestGuard; @@ -1518,7 +1562,7 @@ impl DefaultObjectUsecase { #[allow(clippy::too_many_arguments)] async fn build_get_object_body( mut final_stream: R, - _info: &ObjectInfo, + info: &ObjectInfo, response_content_length: i64, optimal_buffer_size: usize, part_number: Option, @@ -1529,11 +1573,8 @@ impl DefaultObjectUsecase { R: AsyncRead + Send + Sync + Unpin + 'static, { if encryption_applied { - let seekable_object_size_threshold = object_seek_support_threshold(); - let should_buffer_encrypted_object = response_content_length > 0 - && response_content_length <= seekable_object_size_threshold as i64 - && part_number.is_none() - && !has_range; + let should_buffer_encrypted_object = + should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range); if should_buffer_encrypted_object { let mut buf = Vec::with_capacity(response_content_length as usize); @@ -1560,11 +1601,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_reader_blob(final_stream, response_content_length, optimal_buffer_size)); } - let seekable_object_size_threshold = object_seek_support_threshold(); - let should_provide_seek_support = response_content_length > 0 - && response_content_length <= seekable_object_size_threshold as i64 - && part_number.is_none() - && !has_range; + let should_provide_seek_support = + should_buffer_get_object_in_memory(info, response_content_length, part_number, has_range); if should_provide_seek_support { let mut buf = Vec::with_capacity(response_content_length as usize); @@ -4439,6 +4477,11 @@ mod tests { DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, }; + use std::pin::Pin; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + use std::task::{Context, Poll}; + use tokio::io::{AsyncRead, ReadBuf}; fn build_request(input: T, method: Method) -> S3Request { S3Request { @@ -4524,6 +4567,121 @@ mod tests { assert!(!should_use_zero_copy(2 * 1024 * 1024, &headers)); } + #[test] + fn should_buffer_get_object_in_memory_respects_hard_safety_cap() { + let info = ObjectInfo::default(); + let configured_threshold = 20_i64 * 1024 * 1024 * 1024; + let response_len = 80_i64 * 1024 * 1024; + let should_buffer = + should_buffer_get_object_in_memory_with_threshold(&info, response_len, None, false, configured_threshold); + + assert!( + !should_buffer, + "64MiB hard cap must force streaming when response exceeds cap even if configured threshold is much higher" + ); + } + + #[test] + fn should_buffer_get_object_in_memory_allows_small_non_range_requests() { + let info = ObjectInfo::default(); + let configured_threshold = 10_i64 * 1024 * 1024; + + assert!(should_buffer_get_object_in_memory_with_threshold( + &info, + 1024 * 1024, + None, + false, + configured_threshold + )); + assert!(!should_buffer_get_object_in_memory_with_threshold( + &info, + 1024 * 1024, + Some(1), + false, + configured_threshold + )); + assert!(!should_buffer_get_object_in_memory_with_threshold( + &info, + 1024 * 1024, + None, + true, + configured_threshold + )); + } + + struct ReadProbeReader { + reads: Arc, + } + + impl AsyncRead for ReadProbeReader { + fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll> { + self.reads.fetch_add(1, AtomicOrdering::Relaxed); + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn build_get_object_body_keeps_large_objects_on_streaming_path_without_preread() { + let reads = Arc::new(AtomicUsize::new(0)); + let reader = ReadProbeReader { + reads: Arc::clone(&reads), + }; + let info = ObjectInfo { + size: 18_i64 * 1024 * 1024 * 1024, + ..Default::default() + }; + + let body = DefaultObjectUsecase::build_get_object_body( + reader, + &info, + 18_i64 * 1024 * 1024 * 1024, + 128 * 1024, + None, + false, + false, + ) + .await + .expect("build_get_object_body should succeed for streaming path"); + + assert!(body.is_some()); + assert_eq!( + reads.load(AtomicOrdering::Relaxed), + 0, + "large-object response construction should not pre-read object data" + ); + } + + #[tokio::test] + async fn build_get_object_body_keeps_large_encrypted_objects_on_streaming_path_without_preread() { + let reads = Arc::new(AtomicUsize::new(0)); + let reader = ReadProbeReader { + reads: Arc::clone(&reads), + }; + let info = ObjectInfo { + size: 18_i64 * 1024 * 1024 * 1024, + ..Default::default() + }; + + let body = DefaultObjectUsecase::build_get_object_body( + reader, + &info, + 18_i64 * 1024 * 1024 * 1024, + 128 * 1024, + None, + false, + true, + ) + .await + .expect("build_get_object_body should succeed for encrypted streaming path"); + + assert!(body.is_some()); + assert_eq!( + reads.load(AtomicOrdering::Relaxed), + 0, + "large encrypted object response construction should not pre-read object data" + ); + } + #[test] fn should_use_zero_copy_rejects_encrypted_requests_with_sse_customer_algorithm() { let mut headers = HeaderMap::new();