perf(get): reduce response body chunk overhead (#5897)

rustfs/src/app/object_usecase.rs

@@ -115,7 +115,7 @@ use crate::delete_tail_activity::{DeleteTailActivityGuard, DeleteTailStage};
 use crate::error::ApiError;
 use crate::server::convert_ecstore_object_info;
 use crate::table_catalog;
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use futures::{Stream, StreamExt, TryStreamExt};
 use http::{HeaderMap, HeaderValue, StatusCode};
 use md5::{Digest as Md5Digest, Md5};
@@ -198,7 +198,9 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::io::{AsyncRead, ReadBuf};
 use tokio::sync::{OwnedSemaphorePermit, RwLock};
 use tokio_tar::Archive;
-use tokio_util::io::{ReaderStream, StreamReader};
+#[cfg(test)]
+use tokio_util::io::ReaderStream;
+use tokio_util::io::StreamReader;
 use tracing::{debug, error, instrument, warn};
 use uuid::Uuid;
 
@@ -868,7 +870,9 @@ where
 pin_project! {
     struct GetObjectReaderStream<R> {
         #[pin]
-        inner: ReaderStream<R>,
+        reader: Option<R>,
+        buf: BytesMut,
+        capacity: usize,
         strategy: &'static str,
         buffer_source: &'static str,
         remaining: usize,
@@ -918,7 +922,9 @@ where
             rustfs_io_metrics::record_get_object_reader_stream_buffer_size(strategy, buffer_source, capacity);
         }
         Self {
-            inner: ReaderStream::with_capacity(reader, capacity),
+            reader: Some(reader),
+            buf: BytesMut::with_capacity(capacity.min(remaining)),
+            capacity,
             strategy,
             buffer_source,
             remaining,
@@ -1331,12 +1337,28 @@ where
         }
 
         let remaining_before = *this.remaining;
-        let poll_start = std::time::Instant::now();
+        let attribution_enabled = is_get_output_handoff_attribution_enabled();
-        let result: Poll<Option<Self::Item>> = match this.inner.as_mut().poll_next(cx) {
+        let poll_start = attribution_enabled.then(std::time::Instant::now);
-            Poll::Ready(Some(Ok(mut bytes))) => {
+        let reader = match this.reader.as_mut().as_pin_mut() {
-                if bytes.len() > *this.remaining {
+            Some(reader) => reader,
-                    bytes.truncate(*this.remaining);
+            None => return Poll::Ready(None),
-                }
+        };
+        let read_capacity = (*this.capacity).min(*this.remaining);
+        this.buf.resize(read_capacity, 0);
+
+        let poll_read = {
+            let mut read_buf = ReadBuf::new(&mut this.buf[..read_capacity]);
+            match reader.poll_read(cx, &mut read_buf) {
+                Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())),
+                Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+                Poll::Pending => Poll::Pending,
+            }
+        };
+
+        let result: Poll<Option<Self::Item>> = match poll_read {
+            Poll::Ready(Ok(bytes_read)) if bytes_read > 0 => {
+                let bytes = this.buf.split_to(bytes_read).freeze();
+                this.buf.clear();
                 *this.remaining -= bytes.len();
                 #[cfg(feature = "tracing-chunk-debug")]
                 {
@@ -1354,7 +1376,23 @@ where
                     Poll::Ready(Some(Ok(bytes)))
                 }
             }
-            Poll::Ready(Some(Err(err))) => {
+            Poll::Ready(Ok(_)) => {
+                this.buf.clear();
+                this.reader.set(None);
+                let remaining = i64::try_from(*this.remaining).unwrap_or(i64::MAX);
+                let err = std::io::Error::new(std::io::ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining });
+                #[cfg(feature = "tracing-chunk-debug")]
+                tracing::error!(
+                    emitted = *this.emitted,
+                    expected = *this.expected,
+                    error = %err,
+                    "GetObject ReaderStream ended before expected length"
+                );
+                Poll::Ready(Some(Err(Box::new(err) as S3StdError)))
+            }
+            Poll::Ready(Err(err)) => {
+                this.buf.clear();
+                this.reader.set(None);
                 #[cfg(feature = "tracing-chunk-debug")]
                 tracing::error!(
                     emitted = *this.emitted,
@@ -1364,8 +1402,10 @@ where
                 );
                 Poll::Ready(Some(Err(Box::new(err) as S3StdError)))
             }
-            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => {
-            Poll::Pending => Poll::Pending,
+                this.buf.clear();
+                Poll::Pending
+            }
         };
 
         let emitted_bytes = match &result {
@@ -1378,14 +1418,14 @@ where
             Poll::Ready(Some(Err(_))) => GET_READER_STREAM_POLL_READY_ERROR,
             Poll::Pending => GET_READER_STREAM_POLL_PENDING,
         };
-        if is_get_output_handoff_attribution_enabled() {
+        if attribution_enabled {
             rustfs_io_metrics::record_get_object_reader_stream_poll(
                 this.strategy,
                 this.buffer_source,
                 outcome,
                 remaining_before,
                 emitted_bytes,
-                poll_start.elapsed().as_secs_f64(),
+                poll_start.map_or(0.0, |start| start.elapsed().as_secs_f64()),
             );
         }
 
@@ -1393,7 +1433,11 @@ where
     }
 
     fn size_hint(&self) -> (usize, Option<usize>) {
-        self.inner.size_hint()
+        if self.remaining == 0 || self.reader.is_none() {
+            (0, Some(0))
+        } else {
+            (1, None)
+        }
     }
 }
 
@@ -14782,6 +14826,108 @@ mod tests {
         assert_eq!(body, b"hello");
     }
 
+    #[tokio::test]
+    async fn get_object_reader_stream_bounds_read_buffer_to_remaining() {
+        struct RecordingReader {
+            data: &'static [u8],
+            pos: usize,
+            observed_remaining: Arc<Mutex<Vec<usize>>>,
+        }
+
+        impl AsyncRead for RecordingReader {
+            fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+                let requested = buf.remaining();
+                self.observed_remaining
+                    .lock()
+                    .expect("observed buffer sizes should not poison")
+                    .push(requested);
+                let available = self.data.len().saturating_sub(self.pos);
+                let to_copy = requested.min(available);
+                if to_copy > 0 {
+                    let end = self.pos + to_copy;
+                    buf.put_slice(&self.data[self.pos..end]);
+                    self.pos = end;
+                }
+                Poll::Ready(Ok(()))
+            }
+        }
+
+        let observed_remaining = Arc::new(Mutex::new(Vec::new()));
+        let stream = GetObjectReaderStream::new(
+            RecordingReader {
+                data: b"hello",
+                pos: 0,
+                observed_remaining: Arc::clone(&observed_remaining),
+            },
+            64,
+            5,
+            GetObjectStreamStrategy::Standard.as_str(),
+            GET_READER_STREAM_BUFFER_SOURCE_SELECTED,
+        );
+
+        let chunks = stream
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect("reader stream should read exact payload");
+        assert_eq!(chunks, vec![Bytes::from_static(b"hello")]);
+        assert_eq!(
+            *observed_remaining.lock().expect("observed buffer sizes should not poison"),
+            vec![5],
+            "stream should not ask the reader for more bytes than the response has left"
+        );
+    }
+
+    #[tokio::test]
+    async fn get_object_reader_stream_bounds_multi_chunk_final_read() {
+        let stream = GetObjectReaderStream::new(
+            std::io::Cursor::new(vec![b'a'; 66]),
+            64,
+            65,
+            GetObjectStreamStrategy::Standard.as_str(),
+            GET_READER_STREAM_BUFFER_SOURCE_SELECTED,
+        );
+
+        let chunks = stream
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect("reader stream should ignore bytes past declared length");
+        let chunk_lengths = chunks.iter().map(Bytes::len).collect::<Vec<_>>();
+        let body = chunks.into_iter().fold(Vec::new(), |mut acc, chunk| {
+            acc.extend_from_slice(&chunk);
+            acc
+        });
+
+        assert_eq!(chunk_lengths, vec![64, 1]);
+        assert_eq!(body, vec![b'a'; 65]);
+    }
+
+    #[tokio::test]
+    async fn get_object_reader_stream_errors_on_short_eof() {
+        let stream = GetObjectReaderStream::new(
+            std::io::Cursor::new(b"he".to_vec()),
+            64,
+            5,
+            GetObjectStreamStrategy::Standard.as_str(),
+            GET_READER_STREAM_BUFFER_SOURCE_SELECTED,
+        );
+
+        let err = stream
+            .collect::<Vec<_>>()
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect_err("short reader should fail the streaming body");
+
+        assert_eq!(
+            err.downcast_ref::<std::io::Error>().map(std::io::Error::kind),
+            Some(std::io::ErrorKind::UnexpectedEof)
+        );
+    }
+
     #[tokio::test]
     async fn disk_read_permit_reader_releases_permit_at_eof() {
         use tokio::io::AsyncReadExt;

rustfs/src/server/http.rs

@@ -104,6 +104,9 @@ const METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_http_server_re
 const METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES: &str = "rustfs_http_server_request_body_size_bytes";
 const METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL: &str = "rustfs_http_server_response_body_bytes_total";
 const METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES: &str = "rustfs_http_server_response_body_size_bytes";
+const METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_SIZE_BYTES: &str = "rustfs_http_server_response_body_chunk_size_bytes";
+const METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_LATENCY_SECONDS: &str = "rustfs_http_server_response_body_chunk_latency_seconds";
+const METRIC_HTTP_SERVER_RESPONSE_BODY_STREAM_DURATION_SECONDS: &str = "rustfs_http_server_response_body_stream_duration_seconds";
 const METRIC_HTTP_SERVER_CONNECTION_CAP_SATURATED_TOTAL: &str = "rustfs_http_server_connection_cap_saturated_total";
 
 /// Cached handle for the per-response-body-chunk byte counter. A streamed GET
@@ -112,6 +115,12 @@ const METRIC_HTTP_SERVER_CONNECTION_CAP_SATURATED_TOTAL: &str = "rustfs_http_ser
 /// a registry lookup on every chunk.
 static RESP_BODY_BYTES_COUNTER: std::sync::LazyLock<metrics::Counter> =
     std::sync::LazyLock::new(|| counter!(METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL));
+static RESP_BODY_CHUNK_SIZE_HISTOGRAM: std::sync::LazyLock<metrics::Histogram> =
+    std::sync::LazyLock::new(|| histogram!(METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_SIZE_BYTES));
+static RESP_BODY_CHUNK_LATENCY_HISTOGRAM: std::sync::LazyLock<metrics::Histogram> =
+    std::sync::LazyLock::new(|| histogram!(METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_LATENCY_SECONDS));
+static RESP_BODY_STREAM_DURATION_HISTOGRAM: std::sync::LazyLock<metrics::Histogram> =
+    std::sync::LazyLock::new(|| histogram!(METRIC_HTTP_SERVER_RESPONSE_BODY_STREAM_DURATION_SECONDS));
 const LOG_COMPONENT_SERVER: &str = "server";
 const LOG_SUBSYSTEM_HTTP: &str = "http";
 const LOG_SUBSYSTEM_TRANSPORT: &str = "transport";
@@ -226,11 +235,35 @@ fn status_class_label(status: http::StatusCode) -> &'static str {
     }
 }
 
+#[inline]
+fn usize_to_u64_saturating(value: usize) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
 #[inline]
 fn duration_ms(duration: Duration) -> u64 {
     duration.as_millis().try_into().unwrap_or(u64::MAX)
 }
 
+#[inline]
+fn record_response_body_chunk_observation(chunk_len: usize, latency: Duration) {
+    if !rustfs_io_metrics::metrics_enabled() {
+        return;
+    }
+
+    RESP_BODY_CHUNK_SIZE_HISTOGRAM.record(chunk_len as f64);
+    RESP_BODY_CHUNK_LATENCY_HISTOGRAM.record(latency.as_secs_f64());
+}
+
+#[inline]
+fn record_response_body_stream_duration(stream_duration: Duration) {
+    if !rustfs_io_metrics::metrics_enabled() {
+        return;
+    }
+
+    RESP_BODY_STREAM_DURATION_HISTOGRAM.record(stream_duration.as_secs_f64());
+}
+
 fn log_tls_handshake_failure(peer_addr: &str, kind: TlsHandshakeFailureKind, err: &dyn std::fmt::Display) {
     match kind {
         TlsHandshakeFailureKind::UnexpectedEof => {
@@ -1523,7 +1556,8 @@ fn process_connection(
                         })
                         .on_response(trace_on_response)
                         .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| {
-                            RESP_BODY_BYTES_COUNTER.increment(chunk.len() as u64);
+                            RESP_BODY_BYTES_COUNTER.increment(usize_to_u64_saturating(chunk.len()));
+                            record_response_body_chunk_observation(chunk.len(), latency);
                             #[cfg(feature = "tracing-chunk-debug")]
                             {
                                 let _enter = span.enter();
@@ -1535,6 +1569,7 @@ fn process_connection(
                             }
                         })
                         .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
+                            record_response_body_stream_duration(stream_duration);
                             #[cfg(feature = "tracing-chunk-debug")]
                             {
                                 let _enter = span.enter();
@@ -1618,7 +1653,8 @@ fn process_connection(
                         })
                         .on_response(trace_on_response)
                         .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| {
-                            RESP_BODY_BYTES_COUNTER.increment(chunk.len() as u64);
+                            RESP_BODY_BYTES_COUNTER.increment(usize_to_u64_saturating(chunk.len()));
+                            record_response_body_chunk_observation(chunk.len(), latency);
                             #[cfg(feature = "tracing-chunk-debug")]
                             {
                                 let _enter = span.enter();
@@ -1630,6 +1666,7 @@ fn process_connection(
                             }
                         })
                         .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
+                            record_response_body_stream_duration(stream_duration);
                             #[cfg(feature = "tracing-chunk-debug")]
                             {
                                 let _enter = span.enter();