From add6453aea3ea9a759648167374674fa662e0496 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 23 Dec 2025 20:27:34 +0800 Subject: [PATCH] feat: add seek support for small objects in rustfs (#1231) Co-authored-by: loverustfs --- crates/config/src/constants/runtime.rs | 7 +++ rustfs/src/storage/ecfs.rs | 72 +++++++++++++++++++++++--- 2 files changed, 73 insertions(+), 6 deletions(-) diff --git a/crates/config/src/constants/runtime.rs b/crates/config/src/constants/runtime.rs index b9fa58629..04afaf844 100644 --- a/crates/config/src/constants/runtime.rs +++ b/crates/config/src/constants/runtime.rs @@ -39,3 +39,10 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024; /// Event polling default (Tokio default 61) pub const DEFAULT_EVENT_INTERVAL: u32 = 61; pub const DEFAULT_RNG_SEED: Option = None; // None means random + +/// Threshold for small object seek support in megabytes. +/// +/// When an object is smaller than this size, rustfs will provide seek support. +/// +/// Default is set to 10MB. +pub const DEFAULT_OBJECT_SEEK_SUPPORT_THRESHOLD: usize = 10 * 1024 * 1024; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 183ccb269..b49494540 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -134,7 +134,10 @@ use std::{ sync::{Arc, LazyLock}, }; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tokio::{io::AsyncRead, sync::mpsc}; +use tokio::{ + io::{AsyncRead, AsyncSeek}, + sync::mpsc, +}; use tokio_stream::wrappers::ReceiverStream; use tokio_tar::Archive; use tokio_util::io::{ReaderStream, StreamReader}; @@ -398,6 +401,19 @@ impl AsyncRead for InMemoryAsyncReader { } } +impl AsyncSeek for InMemoryAsyncReader { + fn start_seek(mut self: std::pin::Pin<&mut Self>, position: std::io::SeekFrom) -> std::io::Result<()> { + // std::io::Cursor natively supports negative SeekCurrent offsets + // It will automatically handle validation and return an error if the final position would be negative + std::io::Seek::seek(&mut self.cursor, position)?; + Ok(()) + } + + fn poll_complete(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { + std::task::Poll::Ready(Ok(self.cursor.position())) + } +} + async fn decrypt_multipart_managed_stream( mut encrypted_stream: Box, parts: &[ObjectPartInfo], @@ -2264,11 +2280,55 @@ impl S3 for FS { ); Some(StreamingBlob::wrap(ReaderStream::with_capacity(final_stream, optimal_buffer_size))) } else { - // Standard streaming path for large objects or range/part requests - Some(StreamingBlob::wrap(bytes_stream( - ReaderStream::with_capacity(final_stream, optimal_buffer_size), - response_content_length as usize, - ))) + let seekable_object_size_threshold = rustfs_config::DEFAULT_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() + && rs.is_none(); + + if should_provide_seek_support { + debug!( + "Reading small object into memory for seek support: key={} size={}", + cache_key, response_content_length + ); + + // Read the stream into memory + let mut buf = Vec::with_capacity(response_content_length as usize); + match tokio::io::AsyncReadExt::read_to_end(&mut final_stream, &mut buf).await { + Ok(_) => { + // Verify we read the expected amount + if buf.len() != response_content_length as usize { + warn!( + "Object size mismatch during seek support read: expected={} actual={}", + response_content_length, + buf.len() + ); + } + + // Create seekable in-memory reader (similar to MinIO SDK's bytes.Reader) + let mem_reader = InMemoryAsyncReader::new(buf); + Some(StreamingBlob::wrap(bytes_stream( + ReaderStream::with_capacity(Box::new(mem_reader), optimal_buffer_size), + response_content_length as usize, + ))) + } + Err(e) => { + error!("Failed to read object into memory for seek support: {}", e); + // Fallback to streaming if read fails + Some(StreamingBlob::wrap(bytes_stream( + ReaderStream::with_capacity(final_stream, optimal_buffer_size), + response_content_length as usize, + ))) + } + } + } else { + // Standard streaming path for large objects or range/part requests + Some(StreamingBlob::wrap(bytes_stream( + ReaderStream::with_capacity(final_stream, optimal_buffer_size), + response_content_length as usize, + ))) + } }; // Extract SSE information from metadata for response