fix(s3): bound stalled UploadPart request bodies

Cargo.lock

@@ -11205,7 +11205,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
 [[package]]
 name = "s3s"
 version = "0.15.0"
-source = "git+https://github.com/rustfs/s3s.git?rev=bdcb6259339c41369f9f1c60e3a42b5ab8da607b#bdcb6259339c41369f9f1c60e3a42b5ab8da607b"
+source = "git+https://github.com/s3s-project/s3s.git?rev=f3e17541f366696bf0cbaf380fcbd8b44c17eba4#f3e17541f366696bf0cbaf380fcbd8b44c17eba4"
 dependencies = [
  "arc-swap",
  "arrayvec",
@@ -11263,7 +11263,7 @@ dependencies = [
 [[package]]
 name = "s3s-rfc2047"
 version = "0.16.0-alpha.1"
-source = "git+https://github.com/rustfs/s3s.git?rev=bdcb6259339c41369f9f1c60e3a42b5ab8da607b#bdcb6259339c41369f9f1c60e3a42b5ab8da607b"
+source = "git+https://github.com/s3s-project/s3s.git?rev=f3e17541f366696bf0cbaf380fcbd8b44c17eba4#f3e17541f366696bf0cbaf380fcbd8b44c17eba4"
 dependencies = [
  "base64-simd",
  "thiserror 2.0.20",
@@ -11272,7 +11272,7 @@ dependencies = [
 [[package]]
 name = "s3s-sigv2"
 version = "0.16.0-alpha.1"
-source = "git+https://github.com/rustfs/s3s.git?rev=bdcb6259339c41369f9f1c60e3a42b5ab8da607b#bdcb6259339c41369f9f1c60e3a42b5ab8da607b"
+source = "git+https://github.com/s3s-project/s3s.git?rev=f3e17541f366696bf0cbaf380fcbd8b44c17eba4#f3e17541f366696bf0cbaf380fcbd8b44c17eba4"
 dependencies = [
  "base64-simd",
  "hmac 0.13.0",
@@ -11285,7 +11285,7 @@ dependencies = [
 [[package]]
 name = "s3s-sigv4"
 version = "0.16.0-alpha.1"
-source = "git+https://github.com/rustfs/s3s.git?rev=bdcb6259339c41369f9f1c60e3a42b5ab8da607b#bdcb6259339c41369f9f1c60e3a42b5ab8da607b"
+source = "git+https://github.com/s3s-project/s3s.git?rev=f3e17541f366696bf0cbaf380fcbd8b44c17eba4#f3e17541f366696bf0cbaf380fcbd8b44c17eba4"
 dependencies = [
  "arrayvec",
  "base64-simd",

Cargo.toml

@@ -312,7 +312,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 = "bdcb6259339c41369f9f1c60e3a42b5ab8da607b", version = "0.15.0", features = ["minio"] }
+s3s = { git = "https://github.com/s3s-project/s3s.git", rev = "f3e17541f366696bf0cbaf380fcbd8b44c17eba4", version = "0.15.0", features = ["minio"] }
 serial_test = "4.0.1"
 shadow-rs = { default-features = false, version = "2.0.0" }
 siphasher = "1.0.3"

crates/ecstore/src/erasure/coding/encode.rs

@@ -1626,6 +1626,72 @@ mod tests {
         aborting_full_queue_settles_pending_send().await;
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn delayed_reader_error_keeps_source_and_drops_every_encode_path() {
+        #[derive(Debug, thiserror::Error)]
+        #[error("injected request body inactivity")]
+        struct BodyInactivity;
+
+        #[derive(Debug)]
+        struct StalledReader {
+            data: Cursor<Vec<u8>>,
+            timer: Option<Pin<Box<tokio::time::Sleep>>>,
+            dropped: Arc<std::sync::atomic::AtomicBool>,
+        }
+
+        impl AsyncRead for StalledReader {
+            fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+                if self.data.position() < self.data.get_ref().len() as u64 {
+                    return Pin::new(&mut self.data).poll_read(cx, buf);
+                }
+                let timer = self
+                    .timer
+                    .get_or_insert_with(|| Box::pin(tokio::time::sleep(Duration::from_secs(300))));
+                std::task::ready!(timer.as_mut().poll(cx));
+                Poll::Ready(Err(std::io::Error::other(BodyInactivity)))
+            }
+        }
+
+        impl Drop for StalledReader {
+            fn drop(&mut self) {
+                self.dropped.store(true, std::sync::atomic::Ordering::Release);
+            }
+        }
+
+        // Explicit entry points select the paths; environment caches and input
+        // size heuristics cannot silently turn this into repeated Vec coverage.
+        for path in ["direct", "vec", "bytesmut", "batched"] {
+            let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
+            let reader = StalledReader {
+                data: Cursor::new(vec![7; 64]),
+                timer: None,
+                dropped: Arc::clone(&dropped),
+            };
+            let committed = Arc::new(Mutex::new(Vec::new()));
+            let mut writers = (0..4)
+                .map(|_| Some(bitrot_writer(DeferredCommitWriter::new(Arc::clone(&committed)), 32)))
+                .collect::<Vec<_>>();
+            let erasure = Arc::new(Erasure::new(2, 2, 64));
+            let result = match path {
+                "direct" => erasure.encode_single_block_non_inline(reader, &mut writers, 2).await,
+                "vec" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, false).await,
+                "bytesmut" => erasure.encode_with_ingest_mode(reader, &mut writers, 2, true).await,
+                "batched" => erasure.encode_batched(reader, &mut writers, 2).await,
+                _ => unreachable!(),
+            };
+            let error = result.expect_err("stalled input must fail before shard commit");
+            assert!(error.get_ref().is_some_and(|source| source.is::<BodyInactivity>()), "{path}: {error:?}");
+            assert!(
+                dropped.load(std::sync::atomic::Ordering::Acquire),
+                "{path} must release its reader/producer before returning"
+            );
+            assert!(
+                committed.lock().expect("committed bytes").is_empty(),
+                "{path} must not commit partial shards"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn helper_writers_cover_flush_and_shutdown_paths() {
         let mut failing_write = FailingWriteWriter;

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -5767,6 +5767,61 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn capped_staging_queue_does_not_poll_the_part_reader() {
+        use futures::StreamExt;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-staging-body-demand";
+        let object = "object";
+        make_bucket_on_all(&disks, bucket).await;
+        let mut options = ObjectOptions::default();
+        insert_str(&mut options.user_defined, "max-total-object-size", "1024".to_owned());
+        let upload = set_disks
+            .new_multipart_upload(bucket, object, &options)
+            .await
+            .expect("capped upload");
+        let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
+        let semaphore = capped_multipart_staging_semaphore(&upload_path);
+        let held = Arc::clone(&semaphore).acquire_owned().await.expect("hold staging permit");
+        let owners = Arc::strong_count(&semaphore);
+        let polls = Arc::new(AtomicUsize::new(0));
+        let body_polls = Arc::clone(&polls);
+        let stream = futures::stream::iter([Ok::<Bytes, std::io::Error>(Bytes::from(vec![7; 512]))]).inspect(move |_| {
+            body_polls.fetch_add(1, Ordering::Relaxed);
+        });
+        let input = tokio_util::io::StreamReader::new(stream);
+        let mut reader = PutObjReader::new(HashReader::from_stream(input, 512, 512, None, None, false).expect("part reader"));
+        let task = tokio::spawn(async move {
+            set_disks
+                .put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while Arc::strong_count(&semaphore) == owners {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("part must reach the actual staging semaphore");
+        tokio::time::pause();
+        tokio::time::advance(Duration::from_secs(600)).await;
+        tokio::time::resume();
+        assert_eq!(polls.load(Ordering::Relaxed), 0, "staging admission must not create read demand");
+        assert!(!task.is_finished());
+        drop(held);
+        let part = tokio::time::timeout(Duration::from_secs(10), task)
+            .await
+            .expect("staging permit released")
+            .expect("part task")
+            .expect("queued part");
+        assert_eq!(part.size, 512);
+        assert_eq!(polls.load(Ordering::Relaxed), 1);
+        drop(semaphore);
+        remove_capped_multipart_staging_semaphore(&upload_path);
+    }
+
     #[tokio::test]
     async fn put_object_part_recovers_transaction_with_one_faulty_disk_at_write_quorum() {
         use tokio::io::AsyncReadExt as _;

docs/operations/reverse-proxy.md

@@ -102,10 +102,16 @@ Cloudflare's proxy may buffer the entire request body before forwarding and can
 1. Bypass the proxy. Send the failing request to `http://<host>:9000` directly. Success confirms the fault is in the proxy/CDN path.
 2. Bypass the CDN, keep the proxy. Point the proxy straight at the origin (Cloudflare grey cloud / direct DNS). If it now works, the CDN was buffering or re-chunking the body.
 3. Check idle reuse. Intermittent failures that correlate with upload size are almost always the keep-alive mismatch. Lower the proxy keepalive (or disable it) and retry.
-4. Check for a truncated body. If the upload hangs indefinitely rather than resetting, the proxy is forwarding a partial body and then going silent without closing the connection. RustFS bounds this wait with `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT` (`DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT`, 300; `0` disables) and on timeout logs `put_object_body_read_stalled` with the received/expected byte counts.
+4. Check for a stalled body. A client or intermediary can stop forwarding data without closing the connection. `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT` defaults to 300 seconds. `PutObject` logs `put_object_body_read_stalled` when its body-read guard expires. `UploadPart` logs `upload_part_body_read_stalled` and returns `RequestTimeout` (HTTP 400); its log records `raw_bytes_received`, `expected_decoded_bytes`, `timeout_secs`, bucket, key, and request ID. The event identifies missing input progress, without attributing the cause to a particular proxy.
 5. Compare bytes. Confirm the proxy forwards exactly `Content-Length` body bytes with no compression or transformation.
 6. Confirm signed headers survive. `Host` and `x-amz-*` must reach RustFS unchanged; a `SignatureDoesNotMatch` (rather than a hang) points here.
 
+For HTTP `UploadPart`, the inactivity budget counts time waiting for raw request-body bytes while storage is requesting input. Positive raw bytes reset the budget, including fragments of a signed AWS chunk that has not yet finished decoding. Foreground admission, capped-session staging, and storage backpressure do not consume the budget. Finishing the declared payload does not bypass the signed terminator, required trailers, or final body validation. This is an inactivity limit, so an upload making progress can take longer than 300 seconds overall.
+
+Setting the timeout to `0` disables it for ordinary uploads. Multipart sessions with an explicit total-object-size cap retain a minimum 300-second timeout, including when the configured value is `0`. After a body-stall timeout, HTTP/1 uses the existing raw-body drain and closes the connection; HTTP/2 releases the affected stream and keeps the connection usable.
+
+`UploadPart` requires a known logical byte length. RustFS uses the length normalized by S3S after authentication and decoding, with an exact logical stream length as a fallback. A bare `x-amz-decoded-content-length` or `Content-Encoding: aws-chunked` declaration cannot supply this length by itself. Requests reaching an ordinary upload session without a known length return `MissingContentLength` (HTTP 411) before body ingestion; capped sessions retain their `UnexpectedContent` rejection. Preserve the client's framing and signed headers through the proxy. The 5 GiB limit applies to each part request, not the combined size of an ordinary multipart upload.
+
 ## Known failure signatures
 
 | Symptom | Forwarding fault | Issue |

rustfs/src/app/multipart_usecase.rs

@@ -90,6 +90,7 @@ use crate::auth::{
 use crate::capacity::record_capacity_write;
 use crate::error::ApiError;
 use crate::table_catalog;
+#[cfg(test)]
 use bytes::Bytes;
 use futures::StreamExt;
 use http::{HeaderMap, HeaderValue, Uri};
@@ -104,7 +105,7 @@ use rustfs_utils::http::{
     SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICATION_GENERATION,
     SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP,
     SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header, get_source_scheme,
-    headers::{AMZ_CHECKSUM_TYPE, AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
+    headers::{AMZ_CHECKSUM_TYPE, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
     insert_str,
 };
 use s3s::dto::{
@@ -114,6 +115,7 @@ use s3s::dto::{
     ServerSideEncryption, StreamingBlob, Timestamp, UploadPartCopyInput, UploadPartCopyOutput, UploadPartInput, UploadPartOutput,
 };
 use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
+use s3s::stream::ByteStream;
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use std::collections::{HashMap, HashSet};
 use std::str::FromStr;
@@ -377,42 +379,31 @@ fn extract_request_host(headers: &HeaderMap, uri: &Uri) -> Option<String> {
         .or_else(|| uri.authority().map(|authority| authority.as_str().to_string()))
 }
 
-fn decoded_content_length_from_headers(headers: &HeaderMap) -> S3Result<Option<i64>> {
-    let Some(val) = headers.get(AMZ_DECODED_CONTENT_LENGTH) else {
-        return Ok(None);
-    };
+fn resolve_upload_part_size(content_length: Option<i64>, body: Option<&StreamingBlob>) -> S3Result<Option<i64>> {
+    if let Some(length) = content_length {
+        return Ok(Some(length));
+    }
+    body.and_then(|body| body.remaining_length().exact())
+        .map(i64::try_from)
+        .transpose()
+        .map_err(|_| s3_error!(UnexpectedContent))
+}
 
-    match atoi::atoi::<i64>(val.as_bytes()) {
-        Some(x) => Ok(Some(x)),
-        None => Err(s3_error!(UnexpectedContent)),
+fn require_upload_part_size(size: Option<i64>, capped: bool) -> S3Result<i64> {
+    match size {
+        Some(size) if size >= 0 => Ok(size),
+        Some(_) => Err(s3_error!(UnexpectedContent)),
+        None if capped => Err(s3_error!(UnexpectedContent)),
+        None => Err(s3_error!(MissingContentLength)),
     }
 }
 
-fn request_uses_aws_chunked(headers: &HeaderMap) -> bool {
-    let has_aws_chunked = |header_name: &str| {
-        headers
-            .get(header_name)
-            .and_then(|value| value.to_str().ok())
-            .is_some_and(|value| value.split(',').any(|part| part.trim().eq_ignore_ascii_case("aws-chunked")))
-    };
-
-    has_aws_chunked("content-encoding") || has_aws_chunked("transfer-encoding")
-}
-
-fn resolve_upload_part_size(headers: &HeaderMap, content_length: Option<i64>) -> S3Result<Option<i64>> {
-    let decoded_content_length = decoded_content_length_from_headers(headers)?;
-    let size = match (request_uses_aws_chunked(headers), decoded_content_length, content_length) {
-        (true, Some(decoded), _) => Some(decoded),
-        (_, _, Some(length)) => Some(length),
-        (_, Some(decoded), None) => Some(decoded),
-        _ => None,
-    };
-
-    if size == Some(-1) {
-        return Err(s3_error!(UnexpectedContent));
+fn upload_part_body_read_timeout(configured: Duration, capped: bool) -> Duration {
+    if capped {
+        configured.max(Duration::from_secs(rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT))
+    } else {
+        configured
     }
-
-    Ok(size)
 }
 
 fn build_complete_multipart_location(headers: &HeaderMap, uri: &Uri, bucket: &str, key: &str) -> String {
@@ -1168,7 +1159,7 @@ impl DefaultMultipartUsecase {
 
         validate_table_catalog_object_mutation(&bucket, &key).await?;
 
-        let mut size = resolve_upload_part_size(&req.headers, content_length)?;
+        let size = resolve_upload_part_size(content_length, body.as_ref())?;
         if let Some(size) = size {
             reject_oversize_single_upload(size)?;
         }
@@ -1181,20 +1172,15 @@ impl DefaultMultipartUsecase {
             .await
             .map_err(ApiError::from)?;
         let max_total_object_size = multipart_max_total_object_size(&fi.user_defined)?;
-        if max_total_object_size.is_some() && size.is_some_and(|size| size < 0) {
-            return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
-        }
-        if max_total_object_size.is_some() && size.is_none() {
-            return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
-        }
-        if let (Some(limit), Some(size)) = (max_total_object_size, size)
+        let mut size = require_upload_part_size(size, max_total_object_size.is_some())?;
+        if let Some(limit) = max_total_object_size
             && u64::try_from(size).is_ok_and(|size| size > limit)
         {
             return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
         }
         let upload_part_admission = match self
             .concurrency_manager()
-            .admit_multipart_part(size.unwrap_or(-1))
+            .admit_multipart_part(size)
             .await
             .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "foreground write admission closed"))?
         {
@@ -1211,42 +1197,30 @@ impl DefaultMultipartUsecase {
                 ));
             }
         };
-        if max_total_object_size.is_some() {
-            let request_id = req
-                .extensions
-                .get::<super::storage_api::multipart_usecase::request_context::RequestContext>()
-                .map(|ctx| ctx.request_id.clone())
-                .unwrap_or_default();
-            body_stream = guard_put_object_body_read_timeout(
-                body_stream,
+        let request_id = req
+            .extensions
+            .get::<super::storage_api::multipart_usecase::request_context::RequestContext>()
+            .map(|ctx| ctx.request_id.as_str())
+            .unwrap_or_default();
+        let timeout = upload_part_body_read_timeout(put_object_body_read_timeout(), max_total_object_size.is_some());
+        let raw_control = req.extensions.get::<super::object::request_body::BodyReadControl>().cloned();
+        let observe_read_demand = if let Some(control) = &raw_control {
+            control.activate(
+                timeout,
                 &bucket,
                 &key,
-                &request_id,
-                content_length,
-                put_object_body_read_timeout().max(Duration::from_secs(rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT)),
-            );
-        }
-
-        if size.is_none() {
-            let mut total = 0i64;
-            let mut buffer = bytes::BytesMut::new();
-            while let Some(chunk) = body_stream.next().await {
-                let chunk = chunk.map_err(|e| ApiError::from(s3s_body_error_to_io(e)))?;
-                total += chunk.len() as i64;
-                buffer.extend_from_slice(&chunk);
+                request_id,
+                u64::try_from(size).map_err(|_| s3_error!(UnexpectedContent))?,
+            )
+        } else {
+            // Direct protocol callers have no raw HTTP body. Retain their
+            // existing capped-session guard without inventing a client cause.
+            if max_total_object_size.is_some() {
+                body_stream = guard_put_object_body_read_timeout(body_stream, &bucket, &key, request_id, Some(size), timeout);
             }
+            false
+        };
 
-            if total <= 0 {
-                return Err(s3_error!(UnexpectedContent));
-            }
-
-            size = Some(total);
-            let combined = buffer.freeze();
-            let stream = futures::stream::once(async move { Ok::<Bytes, std::io::Error>(combined) });
-            body_stream = StreamingBlob::wrap(stream);
-        }
-
-        let mut size = size.ok_or_else(|| s3_error!(UnexpectedContent))?;
         let ingress_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(std::time::Instant::now);
 
         // Apply adaptive buffer sizing based on part size for optimal streaming performance.
@@ -1391,6 +1365,11 @@ impl DefaultMultipartUsecase {
         };
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
+        if observe_read_demand && let Some(control) = raw_control {
+            use rustfs_rio::HashReaderMut;
+            let inner = reader.take_inner();
+            reader.inner = rustfs_rio::boxed_reader(super::object::request_body::DemandReader::new(inner, control));
+        }
 
         let mut reader = PutObjReader::new(reader);
 
@@ -1935,6 +1914,8 @@ fn passthrough_part_actual_size(headers: &HeaderMap) -> Option<i64> {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    mod body_read_tests;
     use http::{Extensions, HeaderMap, Method, Uri, header::HeaderValue};
     use rustfs_filemeta::ObjectPartInfo;
     use rustfs_utils::http::{
@@ -2139,23 +2120,45 @@ mod tests {
     }
 
     #[test]
-    fn resolve_upload_part_size_uses_decoded_length_for_aws_chunked() {
-        let mut headers = HeaderMap::new();
-        headers.insert("content-encoding", HeaderValue::from_static("aws-chunked"));
-        headers.insert(AMZ_DECODED_CONTENT_LENGTH, HeaderValue::from_static("5242880"));
-
-        let size = resolve_upload_part_size(&headers, Some(5242962)).expect("decoded size should parse");
-
-        assert_eq!(size, Some(5242880));
+    fn resolve_upload_part_size_uses_normalized_logical_length() {
+        assert_eq!(resolve_upload_part_size(Some(5242880), None).expect("DTO length"), Some(5242880));
+        assert_eq!(resolve_upload_part_size(None, None).expect("unknown length"), None);
+        let body = StreamingBlob::from(Bytes::from_static(b"abc"));
+        assert_eq!(resolve_upload_part_size(None, Some(&body)).expect("exact bytes"), Some(3));
+        assert_eq!(resolve_upload_part_size(Some(0), Some(&body)).expect("explicit length wins"), Some(0));
+        let body = StreamingBlob::wrap(futures::stream::iter([Ok::<_, std::io::Error>(Bytes::from_static(b"abc"))]));
+        assert_eq!(futures::Stream::size_hint(&body), (1, Some(1)), "the stream knows its item count");
+        assert_eq!(resolve_upload_part_size(None, Some(&body)).expect("unknown byte length"), None);
     }
 
     #[test]
-    fn resolve_upload_part_size_preserves_regular_content_length() {
-        let headers = HeaderMap::new();
+    fn upload_part_length_contract_rejects_unknown_and_all_negative_lengths() {
+        assert_eq!(
+            require_upload_part_size(None, false).expect_err("ordinary unknown").code(),
+            &S3ErrorCode::MissingContentLength
+        );
+        assert_eq!(
+            require_upload_part_size(None, true).expect_err("capped unknown").code(),
+            &S3ErrorCode::UnexpectedContent
+        );
+        for capped in [false, true] {
+            for size in [-1, -2, i64::MIN] {
+                assert_eq!(
+                    require_upload_part_size(Some(size), capped).expect_err("negative").code(),
+                    &S3ErrorCode::UnexpectedContent
+                );
+            }
+            assert_eq!(require_upload_part_size(Some(0), capped).expect("zero is valid"), 0);
+        }
+    }
 
-        let size = resolve_upload_part_size(&headers, Some(5242880)).expect("regular size should parse");
-
-        assert_eq!(size, Some(5242880));
+    #[test]
+    fn upload_part_timeout_policy_preserves_disabled_and_capped_floor() {
+        for seconds in [0, 1, 299, 300, 601] {
+            let configured = Duration::from_secs(seconds);
+            assert_eq!(upload_part_body_read_timeout(configured, false), configured);
+            assert_eq!(upload_part_body_read_timeout(configured, true), Duration::from_secs(seconds.max(300)));
+        }
     }
 
     #[test]
@@ -3260,6 +3263,85 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_upload_part_rejects_unknown_length_before_admission_or_body_polling() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
+
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+        let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+        let bucket = format!("upload-part-length-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket");
+        let concurrency_manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(
+            true,
+            1,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+            Duration::ZERO,
+        ));
+        let held = concurrency_manager
+            .admit_multipart_part(1024)
+            .await
+            .expect("hold the only permit");
+        let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), concurrency_manager);
+
+        for capped in [false, true] {
+            let mut options = ObjectOptions::default();
+            if capped {
+                insert_str(&mut options.user_defined, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "1024".to_owned());
+            }
+            let upload = store
+                .new_multipart_upload(&bucket, "object", &options)
+                .await
+                .expect("upload session");
+            for content_length in [None, Some(-1)] {
+                for declared_chunk_encoding in [false, true] {
+                    let (body, polls) = crate::app::object::PollCountingBody::streaming_blob();
+                    let input = UploadPartInput::builder()
+                        .bucket(bucket.clone())
+                        .key("object".to_owned())
+                        .upload_id(upload.upload_id.clone())
+                        .part_number(1)
+                        .content_length(content_length)
+                        .body(Some(body))
+                        .build()
+                        .expect("part request");
+                    let mut request = build_request(input, Method::PUT);
+                    request
+                        .headers
+                        .insert("x-amz-decoded-content-length", HeaderValue::from_static("1024"));
+                    if declared_chunk_encoding {
+                        request
+                            .headers
+                            .insert("content-encoding", HeaderValue::from_static("aws-chunked"));
+                    }
+                    let error = usecase
+                        .execute_upload_part(request)
+                        .await
+                        .expect_err("unknown or negative logical size");
+                    assert_eq!(
+                        error.code(),
+                        &if capped || content_length.is_some() {
+                            S3ErrorCode::UnexpectedContent
+                        } else {
+                            S3ErrorCode::MissingContentLength
+                        }
+                    );
+                    assert_eq!(polls.load(std::sync::atomic::Ordering::Relaxed), 0);
+                }
+            }
+            let parts = store
+                .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+                .await
+                .expect("list rejected session");
+            assert!(parts.parts.is_empty());
+        }
+        drop(held);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn execute_upload_part_rejects_when_foreground_write_admission_is_full() {

rustfs/src/app/multipart_usecase/tests/body_read_tests.rs

@@ -0,0 +1,309 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::app::object::request_body::{BodyReadControl, ObservedBody};
+use crate::app::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
+use crate::app::storage_api::test::contract::object::ObjectIO;
+use http_body::Frame;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use tokio::sync::{mpsc, oneshot};
+
+fn part_request(bucket: &str, upload: &str, body: StreamingBlob, size: i64) -> S3Request<UploadPartInput> {
+    build_request(
+        UploadPartInput::builder()
+            .bucket(bucket.to_owned())
+            .key("object".to_owned())
+            .upload_id(upload.to_owned())
+            .part_number(1)
+            .content_length(Some(size))
+            .body(Some(body))
+            .build()
+            .expect("part input"),
+        Method::PUT,
+    )
+}
+
+type BodySender = mpsc::UnboundedSender<Result<Frame<Bytes>, std::io::Error>>;
+
+fn observed_request(
+    bucket: &str,
+    upload: &str,
+    size: usize,
+) -> (S3Request<UploadPartInput>, BodySender, oneshot::Receiver<()>, Arc<AtomicUsize>) {
+    let (sender, mut receiver) = mpsc::unbounded_channel();
+    let (started, waiting) = oneshot::channel();
+    let mut started = Some(started);
+    let polls = Arc::new(AtomicUsize::new(0));
+    let body_polls = Arc::clone(&polls);
+    let stream = futures::stream::poll_fn(move |cx| {
+        body_polls.fetch_add(1, Ordering::Relaxed);
+        let result = receiver.poll_recv(cx);
+        if result.is_pending()
+            && let Some(started) = started.take()
+        {
+            let _ = started.send(());
+        }
+        result
+    });
+    let control = BodyReadControl::default();
+    let body = ObservedBody::new(http_body_util::StreamBody::new(stream), control.clone());
+    let mut request = part_request(bucket, upload, StreamingBlob::from(s3s::Body::http_body_unsync(body)), size as i64);
+    request.extensions.insert(control);
+    (request, sender, waiting, polls)
+}
+
+async fn temporary_entries(disks: &[std::path::PathBuf]) -> std::collections::BTreeSet<std::path::PathBuf> {
+    let mut entries = std::collections::BTreeSet::new();
+    for disk in disks {
+        let path = disk.join(".rustfs.sys/tmp");
+        let mut directory = match tokio::fs::read_dir(path).await {
+            Ok(directory) => directory,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => panic!("temporary directory: {error}"),
+        };
+        while let Some(entry) = directory.next_entry().await.expect("temporary entry") {
+            if entry.file_name() != ".trash" {
+                entries.insert(entry.path());
+            }
+        }
+    }
+    entries
+}
+
+#[test]
+#[serial_test::serial]
+fn upload_part_body_timeout_cleans_storage_preserves_old_part_and_releases_permits() {
+    crate::app::gating_test_env::run_large_stack_test("upload-part-timeout-direct", || async {
+        assert_body_timeout_storage_lifecycle(4096, 512).await;
+    });
+}
+
+#[test]
+#[serial_test::serial]
+fn upload_part_body_timeout_pipeline_storage_lifecycle() {
+    // Fresh processes with the existing ingest and batching settings cover
+    // Vec, BytesMut, and batched pipelines independently.
+    crate::app::gating_test_env::run_large_stack_test("upload-part-timeout-pipeline", || async {
+        assert_body_timeout_storage_lifecycle(3 * 1024 * 1024, 2 * 1024 * 1024 + 512).await;
+    });
+}
+
+async fn assert_body_timeout_storage_lifecycle(part_size: usize, partial_size: usize) {
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+    struct RestoreMetrics(bool);
+    impl Drop for RestoreMetrics {
+        fn drop(&mut self) {
+            rustfs_io_metrics::set_put_stage_metrics_enabled(self.0);
+        }
+    }
+    let _restore = RestoreMetrics(rustfs_io_metrics::put_stage_metrics_enabled());
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    rustfs_io_metrics::set_put_stage_metrics_enabled(true);
+    let path = if part_size == 4096 {
+        "multipart_write_single_block_non_inline"
+    } else if part_size >= rustfs_utils::get_env_usize("RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES", 128 * 1024 * 1024) {
+        "multipart_write_pipeline_batched_large"
+    } else {
+        "multipart_write_pipeline"
+    };
+    let path_count = || {
+        snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .filter_map(|(key, _, _, value)| {
+                if key.key().name() == "rustfs_s3_put_object_path_total"
+                    && key.key().labels().any(|label| label.key() == "path" && label.value() == path)
+                    && let DebugValue::Counter(count) = value
+                {
+                    Some(count)
+                } else {
+                    None
+                }
+            })
+            .sum::<u64>()
+    };
+    let (disks, store) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await;
+    let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+    let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+    let manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(
+        true,
+        1,
+        rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+        Duration::ZERO,
+    ));
+    let usecase = Arc::new(DefaultMultipartUsecase::with_context_and_concurrency_manager(
+        Some(context),
+        Arc::clone(&manager),
+    ));
+    let bucket = format!("body-stall-{}", Uuid::new_v4().simple());
+    store
+        .make_bucket(&bucket, &MakeBucketOptions::default())
+        .await
+        .expect("bucket");
+
+    for capped in [false, true] {
+        let mut options = ObjectOptions::default();
+        if capped {
+            insert_str(&mut options.user_defined, SUFFIX_MAX_TOTAL_OBJECT_SIZE, (2 * part_size).to_string());
+        }
+        let upload = store
+            .new_multipart_upload(&bucket, "object", &options)
+            .await
+            .expect("upload session");
+        let mut old_etag = None;
+        for (replacement, retry_after_failure) in [(false, true), (true, true), (true, false)] {
+            let baseline = temporary_entries(&disks).await;
+            let _ = path_count();
+            let (request, sender, waiting, _) = observed_request(&bucket, &upload.upload_id, part_size);
+            sender
+                .send(Ok(Frame::data(Bytes::from(vec![7; partial_size]))))
+                .expect("partial body");
+            let mut upload_future = Box::pin(usecase.execute_upload_part(request));
+            tokio::time::timeout(Duration::from_secs(10), async {
+                tokio::select! {
+                    result = &mut upload_future => panic!("upload finished before storage requested raw input: {:?}", result.err().map(|error| error.code().clone())),
+                    result = waiting => result.expect("raw reader"),
+                }
+            })
+                .await
+                .expect("storage must request raw input");
+            // Advance only after actual storage demand; filesystem setup and
+            // cleanup run on a real clock and cannot race auto-advance.
+            tokio::time::pause();
+            tokio::time::advance(Duration::from_secs(300)).await;
+            tokio::time::resume();
+            let error = tokio::time::timeout(Duration::from_secs(10), upload_future)
+                .await
+                .expect("inline cleanup must complete")
+                .expect_err("stalled part");
+            assert_eq!(error.code(), &S3ErrorCode::RequestTimeout);
+            assert_eq!(path_count(), 1, "failure must exercise {path}");
+            assert!(sender.is_closed(), "producer must release the failed raw body");
+            assert_eq!(
+                temporary_entries(&disks).await,
+                baseline,
+                "failed part must clean temporary shards inline"
+            );
+            let parts = store
+                .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+                .await
+                .expect("list parts after error");
+            assert_eq!(parts.parts.len(), usize::from(replacement));
+            if replacement {
+                assert_eq!(parts.parts[0].etag, old_etag, "failed overwrite must preserve the committed part");
+            }
+            if !retry_after_failure {
+                continue;
+            }
+            let payload = vec![if replacement { 9 } else { 8 }; part_size];
+            let request = part_request(&bucket, &upload.upload_id, StreamingBlob::from(Bytes::from(payload)), part_size as i64);
+            let retry = tokio::time::timeout(Duration::from_secs(10), usecase.execute_upload_part(request))
+                .await
+                .expect("foreground and capped staging permits must be released")
+                .expect("same-number retry");
+            old_etag = retry.output.e_tag.map(|etag| etag.value().to_owned());
+        }
+        let parts = store
+            .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+            .await
+            .expect("successful retry");
+        assert_eq!(parts.parts.len(), 1);
+        assert_eq!(parts.parts[0].etag, old_etag);
+        assert_eq!(parts.parts[0].size, part_size);
+        store
+            .clone()
+            .complete_multipart_upload(
+                &bucket,
+                "object",
+                &upload.upload_id,
+                vec![CompletePart {
+                    part_num: 1,
+                    etag: old_etag,
+                    ..CompletePart::default()
+                }],
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("failed replacement must leave the prior part completable");
+        let mut object = store
+            .get_object_reader(&bucket, "object", None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("completed old part remains readable");
+        let mut restored = Vec::new();
+        object.stream.read_to_end(&mut restored).await.expect("read all old bytes");
+        assert_eq!(restored, vec![9; part_size]);
+    }
+    eprintln!(
+        "verified storage lifecycle: path={path}, bytesmut={}",
+        rustfs_utils::get_env_bool("RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST", true)
+    );
+}
+
+#[test]
+#[serial_test::serial]
+fn upload_part_foreground_queue_does_not_consume_body_timeout() {
+    crate::app::gating_test_env::run_large_stack_test("upload-part-timeout-queue", assert_foreground_queue);
+}
+
+async fn assert_foreground_queue() {
+    let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+    let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+    let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+    let manager = Arc::new(ConcurrencyManager::with_multipart_admission_queue_for_test(
+        1,
+        Duration::from_secs(1200),
+        1,
+    ));
+    let held = manager.admit_multipart_part(4096).await.expect("hold foreground permit");
+    let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::clone(&manager));
+    let bucket = format!("body-queue-{}", Uuid::new_v4().simple());
+    store
+        .make_bucket(&bucket, &MakeBucketOptions::default())
+        .await
+        .expect("bucket");
+    let upload = store
+        .new_multipart_upload(&bucket, "object", &ObjectOptions::default())
+        .await
+        .expect("upload session");
+    let (request, sender, waiting, polls) = observed_request(&bucket, &upload.upload_id, 4096);
+    let task = tokio::spawn(async move { usecase.execute_upload_part(request).await });
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while manager.put_object_admission_snapshot().queued != Some(1) {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("request must enter the actual foreground queue");
+    tokio::time::pause();
+    tokio::time::advance(Duration::from_secs(600)).await;
+    tokio::time::resume();
+    assert_eq!(polls.load(Ordering::Relaxed), 0, "queued requests must not poll the raw body");
+    assert!(!task.is_finished());
+    drop(held);
+    waiting.await.expect("read begins after admission");
+    sender
+        .send(Ok(Frame::data(Bytes::from(vec![7; 4096]))))
+        .expect("body after admission");
+    drop(sender);
+    tokio::time::timeout(Duration::from_secs(10), task)
+        .await
+        .expect("queued upload completes")
+        .expect("upload task")
+        .expect("queue time is not client inactivity");
+}

rustfs/src/app/object/mod.rs

@@ -211,6 +211,7 @@ mod head;
 mod internal_put;
 mod on_demand_migration_put;
 mod put;
+pub(crate) mod request_body;
 mod restore;
 pub(crate) mod shared;
 #[cfg(test)]

rustfs/src/app/object/request_body.rs

@@ -0,0 +1,364 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Client inactivity is observed before decoding, but charged only while the
+//! final, transformed reader is waiting. Compression can return buffered output
+//! after its input returned Pending, so the transport cannot infer read demand.
+
+use super::{LOG_COMPONENT_APP, LOG_SUBSYSTEM_OBJECT};
+use crate::error::ClientBodyReadTimeout;
+use bytes::Bytes;
+use http_body::{Body, Frame, SizeHint};
+use parking_lot::Mutex;
+use rustfs_rio::{DynReader, EtagResolvable, HashReaderDetector, HashReaderMut, Index, TryGetIndex};
+use std::error::Error;
+use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::task::{Context, Poll};
+use std::time::Duration;
+use tokio::io::{AsyncRead, ReadBuf};
+use tokio::time::{Instant, Sleep};
+
+const EVENT_UPLOAD_PART_BODY_READ_STALLED: &str = "upload_part_body_read_stalled";
+
+struct ReadPolicy {
+    timeout: Duration,
+    bucket: String,
+    key: String,
+    request_id: String,
+    expected_decoded_bytes: u64,
+}
+
+#[derive(Default)]
+struct ReadBudget {
+    policy: Option<ReadPolicy>,
+    demand: bool,
+    waiting_since: Option<Instant>,
+    waited: Duration,
+    raw_bytes_received: u64,
+    finished: bool,
+}
+
+impl ReadBudget {
+    fn pause(&mut self) {
+        if let Some(start) = self.waiting_since.take() {
+            self.waited = self.waited.saturating_add(start.elapsed());
+        }
+        self.demand = false;
+    }
+}
+
+/// Server-owned extension shared by the raw Body and the storage-facing reader.
+#[derive(Clone, Default)]
+pub(crate) struct BodyReadControl(Arc<SharedBudget>);
+
+#[derive(Default)]
+struct SharedBudget {
+    active: AtomicBool,
+    budget: Mutex<ReadBudget>,
+}
+
+impl BodyReadControl {
+    pub(crate) fn activate(
+        &self,
+        timeout: Duration,
+        bucket: &str,
+        key: &str,
+        request_id: &str,
+        expected_decoded_bytes: u64,
+    ) -> bool {
+        if timeout.is_zero() {
+            return false;
+        }
+        let mut state = self.0.budget.lock();
+        if state.finished {
+            return false;
+        }
+        state.policy = Some(ReadPolicy {
+            timeout,
+            bucket: bucket.to_owned(),
+            key: key.to_owned(),
+            request_id: request_id.to_owned(),
+            expected_decoded_bytes,
+        });
+        self.0.active.store(true, Ordering::Release);
+        true
+    }
+
+    fn begin_read(&self) {
+        let mut state = self.0.budget.lock();
+        if !state.finished {
+            state.demand = true;
+        }
+    }
+
+    fn pause_read(&self) {
+        self.0.budget.lock().pause();
+    }
+
+    fn progress(&self, bytes: usize) {
+        // Other HTTP operations never activate this UploadPart policy.
+        if !self.0.active.load(Ordering::Acquire) {
+            return;
+        }
+        let mut state = self.0.budget.lock();
+        state.raw_bytes_received = state
+            .raw_bytes_received
+            .saturating_add(u64::try_from(bytes).unwrap_or(u64::MAX));
+        state.waited = Duration::ZERO;
+        state.waiting_since = None;
+    }
+
+    fn finish(&self) {
+        let mut state = self.0.budget.lock();
+        self.0.active.store(false, Ordering::Release);
+        state.finished = true;
+        state.policy = None;
+        state.waiting_since = None;
+        state.demand = false;
+    }
+
+    fn waiting_deadline(&self) -> Option<Instant> {
+        if !self.0.active.load(Ordering::Acquire) {
+            return None;
+        }
+        let mut state = self.0.budget.lock();
+        let timeout = state.policy.as_ref()?.timeout;
+        if !state.demand || state.finished {
+            return None;
+        }
+        let remaining = timeout.saturating_sub(state.waited);
+        let start = *state.waiting_since.get_or_insert_with(Instant::now);
+        // A timeout beyond the clock's representable range cannot elapse.
+        start.checked_add(remaining)
+    }
+
+    fn expire(&self) -> Option<ClientBodyReadTimeout> {
+        let (policy, raw_bytes_received) = {
+            let mut state = self.0.budget.lock();
+            let timeout = state.policy.as_ref()?.timeout;
+            if !state.demand || state.waited.saturating_add(state.waiting_since?.elapsed()) < timeout {
+                return None;
+            }
+            state.finished = true;
+            self.0.active.store(false, Ordering::Release);
+            state.demand = false;
+            state.waiting_since = None;
+            (state.policy.take()?, state.raw_bytes_received)
+        };
+        tracing::error!(
+            event = EVENT_UPLOAD_PART_BODY_READ_STALLED,
+            component = LOG_COMPONENT_APP,
+            subsystem = LOG_SUBSYSTEM_OBJECT,
+            state = "stall_timeout",
+            operation = "UploadPart",
+            request_id = %policy.request_id,
+            bucket = %policy.bucket,
+            key = %policy.key,
+            raw_bytes_received,
+            expected_decoded_bytes = policy.expected_decoded_bytes,
+            timeout_secs = policy.timeout.as_secs(),
+            "UploadPart request body read stalled"
+        );
+        Some(ClientBodyReadTimeout {
+            timeout: policy.timeout,
+            raw_bytes_received,
+        })
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum ObservedBodyError<E> {
+    Transport(E),
+    Inactivity(ClientBodyReadTimeout),
+}
+
+impl<E: fmt::Display> fmt::Display for ObservedBodyError<E> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Transport(error) => error.fmt(f),
+            Self::Inactivity(error) => error.fmt(f),
+        }
+    }
+}
+
+impl<E: Error + 'static> Error for ObservedBodyError<E> {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        Some(match self {
+            Self::Transport(error) => error,
+            Self::Inactivity(error) => error,
+        })
+    }
+}
+
+/// Wraps the retained raw HTTP body, so a synthesized error never marks the
+/// underlying transport complete or prevents HTTP/1 early-response draining.
+pub(crate) struct ObservedBody<B> {
+    inner: B,
+    control: BodyReadControl,
+    timer: Option<Pin<Box<Sleep>>>,
+    ended: bool,
+}
+
+impl<B> ObservedBody<B> {
+    pub(crate) fn new(inner: B, control: BodyReadControl) -> Self {
+        Self {
+            inner,
+            control,
+            timer: None,
+            ended: false,
+        }
+    }
+
+    fn poll_timeout(&mut self, cx: &mut Context<'_>) -> Option<ClientBodyReadTimeout> {
+        let deadline = self.control.waiting_deadline()?;
+        let timer = self.timer.get_or_insert_with(|| Box::pin(tokio::time::sleep_until(deadline)));
+        if timer.deadline() != deadline {
+            timer.as_mut().reset(deadline);
+        }
+        if timer.as_mut().poll(cx).is_ready() {
+            return self.control.expire();
+        }
+        None
+    }
+}
+
+impl<B: Body<Data = Bytes> + Unpin> Body for ObservedBody<B> {
+    type Data = Bytes;
+    type Error = ObservedBodyError<B::Error>;
+
+    fn poll_frame(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, Self::Error>>> {
+        if self.ended {
+            return Poll::Ready(None);
+        }
+        // Ignore empty data without resetting the budget. Bound work per poll
+        // even if a body repeatedly returns immediately-ready empty frames.
+        for _ in 0..32 {
+            match Pin::new(&mut self.inner).poll_frame(cx) {
+                Poll::Ready(Some(Ok(frame))) => {
+                    if let Some(data) = frame.data_ref() {
+                        if data.is_empty() {
+                            if let Some(error) = self.poll_timeout(cx) {
+                                self.ended = true;
+                                return Poll::Ready(Some(Err(ObservedBodyError::Inactivity(error))));
+                            }
+                            continue;
+                        }
+                        self.control.progress(data.len());
+                    }
+                    return Poll::Ready(Some(Ok(frame)));
+                }
+                Poll::Ready(Some(Err(error))) => {
+                    self.ended = true;
+                    self.control.finish();
+                    return Poll::Ready(Some(Err(ObservedBodyError::Transport(error))));
+                }
+                Poll::Ready(None) => {
+                    self.ended = true;
+                    self.control.finish();
+                    return Poll::Ready(None);
+                }
+                Poll::Pending => {
+                    if let Some(error) = self.poll_timeout(cx) {
+                        self.ended = true;
+                        return Poll::Ready(Some(Err(ObservedBodyError::Inactivity(error))));
+                    }
+                    return Poll::Pending;
+                }
+            }
+        }
+        cx.waker().wake_by_ref();
+        Poll::Pending
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.ended || self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        if self.ended {
+            SizeHint::with_exact(0)
+        } else {
+            self.inner.size_hint()
+        }
+    }
+}
+
+impl<B> Drop for ObservedBody<B> {
+    fn drop(&mut self) {
+        self.control.finish();
+    }
+}
+
+/// Must wrap the final HashReader's inner reader after all write transforms.
+/// Current erasure readers are owned by their read future/producer: canceling
+/// that owner drops this reader. A future retained-reader cancellation path
+/// must explicitly pause its read demand before retaining the reader.
+pub(crate) struct DemandReader {
+    inner: DynReader,
+    control: BodyReadControl,
+}
+
+impl DemandReader {
+    pub(crate) fn new(inner: DynReader, control: BodyReadControl) -> Self {
+        Self { inner, control }
+    }
+}
+
+impl AsyncRead for DemandReader {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        self.control.begin_read();
+        let result = Pin::new(&mut self.inner).poll_read(cx, buf);
+        if result.is_ready() {
+            self.control.pause_read();
+        }
+        result
+    }
+}
+
+impl Drop for DemandReader {
+    fn drop(&mut self) {
+        self.control.finish();
+    }
+}
+
+impl EtagResolvable for DemandReader {
+    fn is_etag_reader(&self) -> bool {
+        self.inner.is_etag_reader()
+    }
+    fn try_resolve_etag(&mut self) -> Option<String> {
+        self.inner.try_resolve_etag()
+    }
+}
+
+impl HashReaderDetector for DemandReader {
+    fn is_hash_reader(&self) -> bool {
+        self.inner.is_hash_reader()
+    }
+    fn as_hash_reader_mut(&mut self) -> Option<&mut dyn HashReaderMut> {
+        self.inner.as_hash_reader_mut()
+    }
+}
+
+impl TryGetIndex for DemandReader {
+    fn try_get_index(&self) -> Option<&Index> {
+        self.inner.try_get_index()
+    }
+}
+
+#[cfg(test)]
+mod tests;

rustfs/src/app/object/request_body/tests.rs

@@ -0,0 +1,267 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::error::ApiError;
+use futures::{StreamExt, poll};
+use http_body_util::StreamBody;
+use s3s::S3ErrorCode;
+use s3s::dto::StreamingBlob;
+use std::io;
+use tokio::io::AsyncReadExt;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::io::StreamReader;
+
+mod protocol;
+
+type FrameSender = mpsc::UnboundedSender<Result<Frame<Bytes>, io::Error>>;
+
+fn raw_reader(timeout: Duration) -> (FrameSender, DynReader, BodyReadControl) {
+    let (sender, receiver) = mpsc::unbounded_channel();
+    let control = BodyReadControl::default();
+    control.activate(timeout, "bucket", "object", "request", 65536);
+    let body = ObservedBody::new(StreamBody::new(UnboundedReceiverStream::new(receiver)), control.clone());
+    let stream = StreamingBlob::from(s3s::Body::http_body_unsync(body));
+    let reader = rustfs_rio::wrap_reader(StreamReader::new(stream.map(|item| item.map_err(io::Error::other))));
+    (sender, reader, control)
+}
+
+#[tokio::test(start_paused = true)]
+async fn body_stall_survives_s3s_and_io_wrapping() {
+    let (_sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let mut reader = DemandReader::new(inner, control);
+    let mut output = Vec::new();
+    let mut read = Box::pin(reader.read_to_end(&mut output));
+    assert!(poll!(read.as_mut()).is_pending());
+    tokio::time::advance(Duration::from_secs(299)).await;
+    assert!(poll!(read.as_mut()).is_pending());
+    tokio::time::advance(Duration::from_secs(1)).await;
+    let error = read.await.expect_err("a body that remains open must time out");
+    let api = ApiError::from(error);
+    assert_eq!(api.code, S3ErrorCode::RequestTimeout);
+    let s3_error = s3s::S3Error::from(api);
+    assert_eq!(s3_error.status_code(), Some(http::StatusCode::BAD_REQUEST));
+}
+
+#[tokio::test(start_paused = true)]
+async fn positive_raw_progress_can_outlast_the_inactivity_timeout() {
+    let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let mut reader = DemandReader::new(inner, control);
+    let mut output = Vec::new();
+    let mut read = Box::pin(reader.read_to_end(&mut output));
+    assert!(poll!(read.as_mut()).is_pending());
+    let start = Instant::now();
+    for _ in 0..8 {
+        tokio::time::advance(Duration::from_secs(60)).await;
+        sender
+            .send(Ok(Frame::data(Bytes::from(vec![7; 8192]))))
+            .expect("body receiver");
+        assert!(poll!(read.as_mut()).is_pending());
+    }
+    drop(sender);
+    assert_eq!(read.await.expect("progressing upload"), 65536);
+    assert_eq!(start.elapsed(), Duration::from_secs(480));
+    assert_eq!(output, vec![7; 65536]);
+}
+
+#[tokio::test(start_paused = true)]
+async fn empty_frames_do_not_extend_the_inactivity_budget() {
+    let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let mut reader = DemandReader::new(inner, control);
+    let mut output = Vec::new();
+    let mut read = Box::pin(reader.read_to_end(&mut output));
+    assert!(poll!(read.as_mut()).is_pending());
+    for _ in 0..4 {
+        tokio::time::advance(Duration::from_secs(60)).await;
+        sender.send(Ok(Frame::data(Bytes::new()))).expect("body receiver");
+        assert!(poll!(read.as_mut()).is_pending());
+    }
+    tokio::time::advance(Duration::from_secs(60)).await;
+    sender.send(Ok(Frame::data(Bytes::new()))).expect("body receiver");
+    assert_eq!(
+        ApiError::from(read.await.expect_err("empty frames are not progress")).code,
+        S3ErrorCode::RequestTimeout
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn disabled_or_not_yet_read_body_has_no_inactivity_deadline() {
+    for timeout in [Duration::ZERO, Duration::from_secs(300)] {
+        let (sender, inner, control) = raw_reader(timeout);
+        // Covers both foreground admission and staging admission, before the
+        // owner ever asks its final reader for input.
+        tokio::time::advance(Duration::from_secs(1000)).await;
+        let mut reader = DemandReader::new(inner, control);
+        let mut output = Vec::new();
+        let mut read = Box::pin(reader.read_to_end(&mut output));
+        assert!(poll!(read.as_mut()).is_pending());
+        if timeout.is_zero() {
+            tokio::time::advance(Duration::from_secs(1000)).await;
+            assert!(poll!(read.as_mut()).is_pending());
+        }
+        sender
+            .send(Ok(Frame::data(Bytes::from_static(b"ok"))))
+            .expect("body receiver");
+        drop(sender);
+        assert_eq!(read.await.expect("queued or disabled body"), 2);
+        assert_eq!(output, b"ok");
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn compressed_output_pauses_raw_wait_during_storage_backpressure() {
+    use rustfs_utils::compress::CompressionAlgorithm;
+
+    let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let compressed = rustfs_rio::CompressReader::with_block_size(inner, 8192, CompressionAlgorithm::default());
+    let mut reader = DemandReader::new(rustfs_rio::boxed_reader(compressed), control);
+    sender
+        .send(Ok(Frame::data(Bytes::from(vec![3; 1024]))))
+        .expect("body receiver");
+    let mut buffer = vec![0; 16384];
+    // CompressReader sees the partial input and then Pending, yet can return
+    // a complete compressed block to the storage writer.
+    let first = reader.read(&mut buffer).await.expect("buffered compressed block");
+    assert!(first > 0);
+    let mut compressed_bytes = buffer[..first].to_vec();
+    tokio::time::advance(Duration::from_secs(600)).await;
+
+    let mut read = Box::pin(reader.read(&mut buffer));
+    assert!(poll!(read.as_mut()).is_pending(), "storage backpressure is not a client stall");
+    tokio::time::advance(Duration::from_secs(299)).await;
+    assert!(poll!(read.as_mut()).is_pending());
+    sender
+        .send(Ok(Frame::data(Bytes::from(vec![4; 1024]))))
+        .expect("body receiver");
+    drop(sender);
+    let next = read.await.expect("input after storage backpressure");
+    compressed_bytes.extend_from_slice(&buffer[..next]);
+    reader
+        .read_to_end(&mut compressed_bytes)
+        .await
+        .expect("remaining compressed data");
+
+    let mut restored = Vec::new();
+    rustfs_rio::DecompressReader::new(std::io::Cursor::new(compressed_bytes), CompressionAlgorithm::default())
+        .read_to_end(&mut restored)
+        .await
+        .expect("roundtrip after backpressure");
+    assert_eq!(restored, [vec![3; 1024], vec![4; 1024]].concat());
+}
+
+#[tokio::test(start_paused = true)]
+async fn read_owner_cancellation_releases_the_raw_body() {
+    let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let mut reader = DemandReader::new(inner, control);
+    let owner = tokio::spawn(async move { reader.read_to_end(&mut Vec::new()).await });
+    tokio::task::yield_now().await;
+    owner.abort();
+    assert!(owner.await.expect_err("read owner was canceled").is_cancelled());
+    assert!(sender.is_closed(), "the canceled producer must release the body receiver");
+}
+
+#[tokio::test(start_paused = true)]
+async fn buffered_output_pauses_but_does_not_reset_elapsed_inactivity() {
+    struct BufferedOutput {
+        inner: DynReader,
+        ready: Arc<AtomicBool>,
+    }
+    impl AsyncRead for BufferedOutput {
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+            if self.ready.swap(false, Ordering::AcqRel) {
+                buf.put_slice(b"x");
+                return Poll::Ready(Ok(()));
+            }
+            Pin::new(&mut self.inner).poll_read(cx, buf)
+        }
+    }
+    let (_sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let ready = Arc::new(AtomicBool::new(false));
+    let transform = BufferedOutput {
+        inner,
+        ready: Arc::clone(&ready),
+    };
+    let mut reader = DemandReader::new(rustfs_rio::wrap_reader(transform), control);
+    let mut buffer = [0; 1];
+    let mut read = Box::pin(reader.read(&mut buffer));
+    assert!(poll!(read.as_mut()).is_pending());
+    tokio::time::advance(Duration::from_secs(200)).await;
+    assert!(poll!(read.as_mut()).is_pending());
+    ready.store(true, Ordering::Release);
+    assert_eq!(read.await.expect("transform releases buffered output"), 1);
+    tokio::time::advance(Duration::from_secs(600)).await;
+    let mut read = Box::pin(reader.read(&mut buffer));
+    assert!(poll!(read.as_mut()).is_pending());
+    tokio::time::advance(Duration::from_secs(99)).await;
+    assert!(poll!(read.as_mut()).is_pending());
+    tokio::time::advance(Duration::from_secs(1)).await;
+    let result = poll!(read.as_mut());
+    let Poll::Ready(Err(error)) = result else {
+        panic!("remaining inactivity budget must expire immediately at 100 seconds");
+    };
+    assert_eq!(ApiError::from(error).code, S3ErrorCode::RequestTimeout);
+}
+
+#[tokio::test(start_paused = true)]
+async fn final_write_reader_preserves_checksums_through_compression_and_sse() {
+    use crate::app::storage_api::multipart_usecase::io::{HashReader, WriteEncryption, WritePlan};
+    use rustfs_rio::{Checksum, ChecksumType};
+    use rustfs_utils::CompressionAlgorithm;
+
+    let payload = vec![7; 65536];
+    for plan in [
+        WritePlan::new().with_compression(CompressionAlgorithm::default()),
+        WritePlan::new().with_encryption(WriteEncryption::multipart([5; 32], [9; 12], 1)),
+        WritePlan::new()
+            .with_compression(CompressionAlgorithm::default())
+            .with_encryption(WriteEncryption::multipart([5; 32], [9; 12], 1)),
+    ] {
+        let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+        let checksum = Checksum::new_from_data(ChecksumType::CRC32, &payload).expect("plaintext checksum");
+        let mut plaintext = HashReader::from_reader(inner, 65536, 65536, None, None, false).expect("plaintext reader");
+        plaintext
+            .add_non_trailing_checksum(Some(checksum.clone()), false)
+            .expect("attach checksum");
+        let mut reader = plan.apply(plaintext, 65536).expect("write plan");
+        let inner = reader.take_inner();
+        reader.inner = rustfs_rio::boxed_reader(DemandReader::new(inner, control));
+        let mut output = Vec::new();
+        let mut read = Box::pin(reader.read_to_end(&mut output));
+        assert!(poll!(read.as_mut()).is_pending());
+        for chunk in payload.chunks(8192) {
+            tokio::time::advance(Duration::from_secs(60)).await;
+            sender
+                .send(Ok(Frame::data(Bytes::copy_from_slice(chunk))))
+                .expect("raw body progress");
+            assert!(poll!(read.as_mut()).is_pending());
+        }
+        drop(sender);
+        read.await.expect("transformed reader completes beyond one inactivity period");
+        assert!(!output.is_empty());
+        assert_eq!(reader.content_crc_type(), Some(ChecksumType::CRC32));
+        assert_eq!(reader.content_crc().get("CRC32"), Some(&checksum.encoded));
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn native_body_errors_preserve_their_original_source() {
+    let (sender, inner, control) = raw_reader(Duration::from_secs(300));
+    let mut reader = DemandReader::new(inner, control);
+    sender
+        .send(Err(io::Error::new(io::ErrorKind::TimedOut, "disk timeout")))
+        .expect("body receiver");
+    let error = reader.read_to_end(&mut Vec::new()).await.expect_err("native error");
+    assert_eq!(ApiError::from(error).code, S3ErrorCode::InternalError);
+}

rustfs/src/app/object/request_body/tests/protocol.rs

@@ -0,0 +1,280 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use http_body_util::BodyExt;
+use s3s::config::{S3Config, StaticConfigProvider};
+use s3s::dto::{UploadPartInput, UploadPartOutput};
+use s3s::service::{S3Service, S3ServiceBuilder};
+use s3s::{S3, S3Request, S3Response, S3Result};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[derive(Clone, Default)]
+struct Consumer {
+    received: Arc<AtomicUsize>,
+    committed: Arc<Mutex<Option<Vec<u8>>>>,
+}
+
+#[async_trait::async_trait]
+impl S3 for Consumer {
+    async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
+        let expected = req.input.content_length.expect("S3S must normalize the decoded length");
+        let control = req
+            .extensions
+            .get::<BodyReadControl>()
+            .expect("HTTP control extension")
+            .clone();
+        control.activate(Duration::from_secs(300), "test-bucket", "test-key", "request", expected as u64);
+        let stream = req.input.body.expect("body");
+        let inner = rustfs_rio::wrap_reader(StreamReader::new(stream.map(|item| item.map_err(io::Error::other))));
+        let mut reader =
+            rustfs_rio::HashReader::from_stream(DemandReader::new(inner, control), expected, expected, None, None, false)
+                .expect("logical body reader");
+        reader
+            .add_checksum_from_s3s(&req.headers, req.trailing_headers, false)
+            .expect("request checksum context");
+        let mut output = Vec::new();
+        let mut buffer = [0; 8192];
+        loop {
+            let count = reader
+                .read(&mut buffer)
+                .await
+                .map_err(|error| s3s::S3Error::from(ApiError::from(error)))?;
+            if count == 0 {
+                break;
+            }
+            self.received.fetch_add(count, Ordering::Relaxed);
+            output.extend_from_slice(&buffer[..count]);
+        }
+        assert_eq!(output.len() as i64, expected, "wire length must not reach the business DTO");
+        *self.committed.lock() = Some(output);
+        Ok(S3Response::new(UploadPartOutput {
+            checksum_crc32: reader.content_crc().get("CRC32").cloned(),
+            ..UploadPartOutput::default()
+        }))
+    }
+}
+
+fn service(consumer: Consumer) -> S3Service {
+    let mut builder = S3ServiceBuilder::new(consumer);
+    builder.set_auth(s3s::auth::SimpleAuth::from_single("test-access", "test-secret"));
+    let mut config = S3Config::default();
+    config.presigned_url_max_skew_time_secs = u32::MAX;
+    builder.set_config(Arc::new(StaticConfigProvider::new(Arc::new(config))));
+    builder.build()
+}
+
+struct SignedRequest {
+    request: http::Request<s3s::Body>,
+    sender: FrameSender,
+    prefix: Bytes,
+    suffix: Bytes,
+}
+
+/// Constructs a real SigV4 fixture using the production crypto primitives.
+/// S3S verifies both the request authorization and every signed chunk.
+fn signed_request(payload: &[u8], unsigned_trailer: bool) -> SignedRequest {
+    use rustfs_utils::{hex_sha256, hmac_sha256};
+
+    const DATE: &str = "20130524T000000Z";
+    const SCOPE: &str = "20130524/us-east-1/s3/aws4_request";
+    let mode = if unsigned_trailer {
+        "STREAMING-UNSIGNED-PAYLOAD-TRAILER"
+    } else {
+        "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
+    };
+    let signed_headers = "host;x-amz-content-sha256;x-amz-date;x-amz-decoded-content-length";
+    let headers = format!(
+        "host:s3.amazonaws.com\nx-amz-content-sha256:{mode}\nx-amz-date:{DATE}\nx-amz-decoded-content-length:{}\n",
+        payload.len()
+    );
+    let canonical = format!("PUT\n/test-bucket/test-key\npartNumber=1&uploadId=test-upload\n{headers}\n{signed_headers}\n{mode}");
+    let key = hmac_sha256("AWS4test-secret", "20130524");
+    let key = hmac_sha256(key, "us-east-1");
+    let key = hmac_sha256(key, "s3");
+    let key = hmac_sha256(key, "aws4_request");
+    let digest = |data: &[u8]| hex_sha256(data, str::to_owned);
+    let encode = |data: [u8; 32]| hex_simd::encode_to_string(data, hex_simd::AsciiCase::Lower);
+    let seed = encode(hmac_sha256(
+        key,
+        format!("AWS4-HMAC-SHA256\n{DATE}\n{SCOPE}\n{}", digest(canonical.as_bytes())),
+    ));
+    let chunk_signature = |previous: &str, data: &[u8]| {
+        encode(hmac_sha256(
+            key,
+            format!("AWS4-HMAC-SHA256-PAYLOAD\n{DATE}\n{SCOPE}\n{previous}\n{}\n{}", digest(b""), digest(data)),
+        ))
+    };
+    let (prefix, suffix) = if unsigned_trailer {
+        (
+            format!("{:x}\r\n", payload.len()),
+            "\r\n0\r\nx-amz-checksum-crc32:y/Q5Jg==\r\n\r\n".to_owned(),
+        )
+    } else {
+        let signature = chunk_signature(&seed, payload);
+        (
+            format!("{:x};chunk-signature={signature}\r\n", payload.len()),
+            format!("\r\n0;chunk-signature={}\r\n\r\n", chunk_signature(&signature, b"")),
+        )
+    };
+    let (sender, receiver) = mpsc::unbounded_channel();
+    let control = BodyReadControl::default();
+    let body = ObservedBody::new(StreamBody::new(UnboundedReceiverStream::new(receiver)), control.clone());
+    let mut builder = http::Request::builder()
+        .method("PUT")
+        .uri("https://s3.amazonaws.com/test-bucket/test-key?partNumber=1&uploadId=test-upload")
+        .header("host", "s3.amazonaws.com")
+        .header("content-encoding", "aws-chunked")
+        .header("content-length", prefix.len() + payload.len() + suffix.len())
+        .header("x-amz-content-sha256", mode)
+        .header("x-amz-date", DATE)
+        .header("x-amz-decoded-content-length", payload.len())
+        .header(
+            "authorization",
+            format!("AWS4-HMAC-SHA256 Credential=test-access/{SCOPE}, SignedHeaders={signed_headers}, Signature={seed}"),
+        );
+    if unsigned_trailer {
+        builder = builder.header("x-amz-trailer", "x-amz-checksum-crc32");
+    }
+    let mut request = builder.body(s3s::Body::http_body_unsync(body)).expect("signed request");
+    request.extensions_mut().insert(control);
+    SignedRequest {
+        request,
+        sender,
+        prefix: Bytes::from(prefix),
+        suffix: Bytes::from(suffix),
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn signed_chunk_with_raw_progress_survives_eight_minutes_without_decoded_output() {
+    let payload = vec![7; 65536];
+    let SignedRequest {
+        request,
+        sender,
+        prefix,
+        suffix,
+    } = signed_request(&payload, false);
+    let consumer = Consumer::default();
+    let service = service(consumer.clone());
+    let mut call = Box::pin(service.call(request));
+    sender.send(Ok(Frame::data(prefix))).expect("body receiver");
+    assert!(poll!(call.as_mut()).is_pending());
+    let start = Instant::now();
+    for chunk in payload.chunks(8192) {
+        assert_eq!(
+            consumer.received.load(Ordering::Relaxed),
+            0,
+            "incomplete chunks must not escape signature validation"
+        );
+        tokio::time::advance(Duration::from_secs(60)).await;
+        sender
+            .send(Ok(Frame::data(Bytes::copy_from_slice(chunk))))
+            .expect("body receiver");
+        assert!(poll!(call.as_mut()).is_pending());
+    }
+    sender.send(Ok(Frame::data(suffix))).expect("body receiver");
+    drop(sender);
+    let response = call.await.expect("S3 response");
+    assert_eq!(response.status(), http::StatusCode::OK);
+    assert_eq!(start.elapsed(), Duration::from_secs(480));
+    assert_eq!(*consumer.committed.lock(), Some(payload));
+}
+
+#[tokio::test(start_paused = true)]
+async fn decoded_length_does_not_end_waiting_for_terminator_trailer_or_raw_eof() {
+    for (unsigned_trailer, send_suffix) in [(false, false), (false, true), (true, false), (true, true)] {
+        let payload = b"123456789";
+        let SignedRequest {
+            request,
+            sender,
+            prefix,
+            suffix,
+        } = signed_request(payload, unsigned_trailer);
+        let consumer = Consumer::default();
+        let service = service(consumer.clone());
+        sender.send(Ok(Frame::data(prefix))).expect("prefix");
+        sender.send(Ok(Frame::data(Bytes::from_static(payload)))).expect("payload");
+        sender
+            .send(Ok(Frame::data(if send_suffix { suffix } else { Bytes::from_static(b"\r\n") })))
+            .expect("suffix");
+        let mut call = Box::pin(service.call(request));
+        assert!(poll!(call.as_mut()).is_pending());
+        assert_eq!(consumer.received.load(Ordering::Relaxed), payload.len());
+        tokio::time::advance(Duration::from_secs(300)).await;
+        let response = call.await.expect("S3 error response");
+        assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
+        let xml = BodyExt::collect(response.into_body()).await.expect("error XML").to_bytes();
+        assert!(String::from_utf8_lossy(&xml).contains("<Code>RequestTimeout</Code>"));
+        assert!(consumer.committed.lock().is_none());
+    }
+}
+
+#[tokio::test]
+async fn unsigned_trailer_normalizes_length_and_signed_corruption_cannot_commit() {
+    for corrupt in [false, true] {
+        let payload = b"123456789";
+        let SignedRequest {
+            request,
+            sender,
+            prefix,
+            suffix,
+        } = signed_request(payload, !corrupt);
+        let consumer = Consumer::default();
+        let service = service(consumer.clone());
+        sender.send(Ok(Frame::data(prefix))).expect("prefix");
+        sender
+            .send(Ok(Frame::data(Bytes::from_static(if corrupt { b"923456789" } else { payload }))))
+            .expect("payload");
+        sender.send(Ok(Frame::data(suffix))).expect("suffix");
+        drop(sender);
+        let response = service.call(request).await.expect("S3 response");
+        if corrupt {
+            assert_ne!(response.status(), http::StatusCode::OK);
+            assert_eq!(consumer.received.load(Ordering::Relaxed), 0);
+            assert!(consumer.committed.lock().is_none());
+        } else {
+            assert_eq!(response.status(), http::StatusCode::OK);
+            assert_eq!(
+                response
+                    .headers()
+                    .get("x-amz-checksum-crc32")
+                    .expect("validated response checksum"),
+                "y/Q5Jg=="
+            );
+            assert_eq!(*consumer.committed.lock(), Some(payload.to_vec()));
+        }
+    }
+}
+
+#[tokio::test]
+async fn unsigned_trailer_with_wrong_checksum_cannot_commit() {
+    let SignedRequest {
+        request, sender, prefix, ..
+    } = signed_request(b"123456789", true);
+    let consumer = Consumer::default();
+    let service = service(consumer.clone());
+    sender.send(Ok(Frame::data(prefix))).expect("prefix");
+    sender
+        .send(Ok(Frame::data(Bytes::from_static(
+            b"123456789\r\n0\r\nx-amz-checksum-crc32:AAAAAA==\r\n\r\n",
+        ))))
+        .expect("invalid trailer");
+    drop(sender);
+    let response = service.call(request).await.expect("S3 response");
+    assert_eq!(response.status(), http::StatusCode::BAD_REQUEST);
+    let xml = BodyExt::collect(response.into_body()).await.expect("error XML").to_bytes();
+    assert!(String::from_utf8_lossy(&xml).contains("<Code>BadDigest</Code>"));
+    assert!(consumer.committed.lock().is_none());
+}

rustfs/src/error.rs

@@ -126,6 +126,26 @@ impl std::fmt::Display for UploadLimitExceeded {
 
 impl std::error::Error for UploadLimitExceeded {}
 
+/// Identifies inactivity of an external client body, rather than a storage timeout.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ClientBodyReadTimeout {
+    pub timeout: std::time::Duration,
+    pub raw_bytes_received: u64,
+}
+
+impl std::fmt::Display for ClientBodyReadTimeout {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "request body made no progress for {} seconds after {} raw bytes",
+            self.timeout.as_secs(),
+            self.raw_bytes_received
+        )
+    }
+}
+
+impl std::error::Error for ClientBodyReadTimeout {}
+
 /// Marks a server-side object/source reader failure that must not be reported as
 /// a malformed client request body.
 #[derive(Debug)]
@@ -412,25 +432,7 @@ fn error_chain_has_type<T>(err: &(dyn std::error::Error + 'static)) -> bool
 where
     T: std::error::Error + 'static,
 {
-    if err.downcast_ref::<T>().is_some() {
-        return true;
-    }
-
-    if let Some(io_err) = err.downcast_ref::<std::io::Error>()
-        && let Some(inner) = io_err.get_ref()
-        && error_chain_has_type::<T>(inner)
-    {
-        return true;
-    }
-
-    let mut current = Some(err);
-    while let Some(err) = current {
-        if err.downcast_ref::<T>().is_some() {
-            return true;
-        }
-        current = err.source();
-    }
-    false
+    error_chain_find(err, |error| error.is::<T>().then_some(())).is_some()
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -451,67 +453,33 @@ fn classify_s3s_body_stream_error_display(err: &(dyn std::error::Error + 'static
 }
 
 fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) -> Option<S3sBodyStreamError> {
-    if let Some(classified) = classify_s3s_body_stream_error_display(err) {
-        return Some(classified);
-    }
+    error_chain_find(err, classify_s3s_body_stream_error_display)
+}
 
-    if let Some(io_err) = err.downcast_ref::<std::io::Error>()
-        && let Some(inner) = io_err.get_ref()
-        && let Some(classified) = error_chain_s3s_body_stream_error(inner)
-    {
-        return Some(classified);
-    }
-
-    let mut current = err.source();
-    while let Some(err) = current {
-        if let Some(classified) = classify_s3s_body_stream_error_display(err) {
+/// `io::Error::source` skips its custom payload itself. Visit that payload
+/// explicitly at every level, then follow its source exactly once. The bound
+/// also makes cyclic or excessively deep foreign error chains safe.
+fn error_chain_find<T>(
+    err: &(dyn std::error::Error + 'static),
+    mut classify: impl FnMut(&(dyn std::error::Error + 'static)) -> Option<T>,
+) -> Option<T> {
+    let mut current = Some(err);
+    for _ in 0..64 {
+        let error = current?;
+        if let Some(classified) = classify(error) {
             return Some(classified);
         }
-        if let Some(io_err) = err.downcast_ref::<std::io::Error>()
-            && let Some(inner) = io_err.get_ref()
-            && let Some(classified) = error_chain_s3s_body_stream_error(inner)
-        {
-            return Some(classified);
-        }
-        current = err.source();
+        current = error
+            .downcast_ref::<std::io::Error>()
+            .and_then(std::io::Error::get_ref)
+            .map(|inner| inner as &(dyn std::error::Error + 'static))
+            .or_else(|| error.source());
     }
     None
 }
 
-/// Walk an error chain (including `io::Error` custom payloads) and return
-/// whether any link satisfies `pred`.
-fn error_chain_any(err: &(dyn std::error::Error + 'static), pred: &dyn Fn(&(dyn std::error::Error + 'static)) -> bool) -> bool {
-    if pred(err) {
-        return true;
-    }
-    if let Some(io_err) = err.downcast_ref::<std::io::Error>()
-        && let Some(inner) = io_err.get_ref()
-        && error_chain_any(inner, pred)
-    {
-        return true;
-    }
-    let mut current = err.source();
-    while let Some(err) = current {
-        if error_chain_any(err, pred) {
-            return true;
-        }
-        current = err.source();
-    }
-    false
-}
-
-/// s3s raises `BodySizeLimitExceeded` when the streaming-body budget
-/// (`put_object_max_size`) runs out mid-stream. The type lives in s3s's
-/// private `http` module, so it is recognised by its `Display` form
-/// (`body size {size} exceeds limit {limit}`), like the other s3s body-stream
-/// errors above. Switch to a typed downcast once s3s re-exports the type.
-fn is_body_size_limit_exceeded_display(err: &(dyn std::error::Error + 'static)) -> bool {
-    let text = err.to_string();
-    text.starts_with("body size ") && text.contains(" exceeds limit ")
-}
-
 fn error_chain_has_body_size_limit_exceeded(err: &(dyn std::error::Error + 'static)) -> bool {
-    error_chain_any(err, &is_body_size_limit_exceeded_display)
+    error_chain_has_type::<s3s::BodySizeLimitExceeded>(err)
 }
 
 /// hyper reports a request body whose connection hit EOF before
@@ -526,7 +494,7 @@ fn is_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
 }
 
 fn error_chain_has_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
-    error_chain_any(err, &is_hyper_body_eof)
+    error_chain_find(err, |error| is_hyper_body_eof(error).then_some(())).is_some()
 }
 
 impl From<ApiError> for S3Error {
@@ -586,6 +554,14 @@ impl From<StorageError> for ApiError {
                 };
             }
 
+            if error_chain_has_type::<ClientBodyReadTimeout>(inner) {
+                return ApiError {
+                    code: S3ErrorCode::RequestTimeout,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout),
+                    source: Some(Box::new(err)),
+                };
+            }
+
             if error_chain_has_body_size_limit_exceeded(inner) {
                 return ApiError {
                     code: S3ErrorCode::EntityTooLarge,
@@ -745,6 +721,14 @@ impl From<std::io::Error> for ApiError {
                     source: Some(Box::new(err)),
                 };
             }
+            if error_chain_has_type::<ClientBodyReadTimeout>(inner) {
+                return ApiError {
+                    code: S3ErrorCode::RequestTimeout,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout),
+                    source: Some(Box::new(err)),
+                };
+            }
+
             if error_chain_has_body_size_limit_exceeded(inner) {
                 return ApiError {
                     code: S3ErrorCode::EntityTooLarge,
@@ -1040,7 +1024,10 @@ mod tests {
         let nested = || {
             IoError::new(
                 ErrorKind::UnexpectedEof,
-                IoError::other(MockS3sBodyStreamError("body size 16384 exceeds limit 6389")),
+                IoError::other(s3s::BodySizeLimitExceeded {
+                    size: 16384,
+                    limit: 6389,
+                }),
             )
         };
 
@@ -1055,11 +1042,70 @@ mod tests {
         // An unrelated message that merely mentions a limit stays internal.
         let other: ApiError = IoError::other(MockS3sBodyStreamError("limit exceeded for something else")).into();
         assert_eq!(other.code, S3ErrorCode::InternalError);
+        let impostor: ApiError = IoError::other(MockS3sBodyStreamError("body size 16384 exceeds limit 6389")).into();
+        assert_eq!(impostor.code, S3ErrorCode::InternalError);
     }
 
-    /// Trip s3s's real streaming-body budget with a tiny limit so the
-    /// display-based matcher is checked against the pinned dependency's
-    /// actual error, not only the mocked string.
+    #[test]
+    fn client_body_timeout_survives_intermediate_io_and_storage_errors() {
+        #[derive(Debug)]
+        struct DecoderError(IoError);
+        impl std::fmt::Display for DecoderError {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                f.write_str("decoder source failed")
+            }
+        }
+        impl std::error::Error for DecoderError {
+            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+                Some(&self.0)
+            }
+        }
+        let nested = || {
+            IoError::other(DecoderError(IoError::other(IoError::new(
+                ErrorKind::TimedOut,
+                ClientBodyReadTimeout {
+                    timeout: std::time::Duration::from_secs(300),
+                    raw_bytes_received: 8192,
+                },
+            ))))
+        };
+        for error in [ApiError::from(nested()), ApiError::from(StorageError::Io(nested()))] {
+            assert_eq!(error.code, S3ErrorCode::RequestTimeout);
+            assert!(error_chain_has_type::<ClientBodyReadTimeout>(&error));
+            let s3_error = S3Error::from(error);
+            assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST));
+        }
+        for error in [
+            ApiError::from(IoError::new(ErrorKind::TimedOut, "disk read timeout")),
+            ApiError::from(StorageError::Io(IoError::new(ErrorKind::TimedOut, "peer timeout"))),
+        ] {
+            assert_eq!(error.code, S3ErrorCode::InternalError);
+        }
+        // A server-side source wrapper has precedence even if a remote source
+        // has carried its own client-body marker across an I/O boundary.
+        let source = ServerSideSourceReadError::new("CopyObject", nested());
+        assert_eq!(ApiError::from(IoError::other(source)).code, S3ErrorCode::ServiceUnavailable);
+    }
+
+    #[test]
+    fn body_error_classification_bounds_cyclic_source_chains() {
+        #[derive(Debug)]
+        struct Cycle;
+        impl std::fmt::Display for Cycle {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                f.write_str("cycle")
+            }
+        }
+        impl std::error::Error for Cycle {
+            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+                Some(self)
+            }
+        }
+        assert!(!error_chain_has_type::<ClientBodyReadTimeout>(&Cycle));
+        assert_eq!(ApiError::from(IoError::other(Cycle)).code, S3ErrorCode::InternalError);
+    }
+
+    /// Exercise the public error type through the actual body budget.
     #[tokio::test]
     async fn real_s3s_body_size_limit_error_maps_to_entity_too_large() {
         use futures::StreamExt;
@@ -1074,7 +1120,7 @@ mod tests {
         };
 
         let err = real_error().await;
-        assert!(is_body_size_limit_exceeded_display(err.as_ref()), "unexpected display: {err}");
+        assert!(err.is::<s3s::BodySizeLimitExceeded>(), "unexpected body error: {err}");
 
         let err = real_error().await;
         let storage: ApiError = StorageError::Io(IoError::new(ErrorKind::UnexpectedEof, IoError::other(err))).into();

rustfs/src/server/http.rs

@@ -14,6 +14,7 @@
 
 // Import HTTP server components and compression configuration
 use crate::admin;
+use crate::app::object::request_body::{BodyReadControl, ObservedBody};
 use crate::auth::IAMAuth;
 use crate::auth_keystone;
 use crate::config;
@@ -825,13 +826,11 @@ where
 
 impl<S, B, ResBody, ServiceError> Service<HttpRequest<B>> for EarlyResponseBodyService<S>
 where
-    S: Service<HttpRequest<B>, Response = Response<ResBody>, Error = ServiceError>
-        + Service<HttpRequest<EarlyResponseBody<B>>, Response = Response<ResBody>, Error = ServiceError>
+    S: Service<HttpRequest<ObservedBody<EarlyResponseBody<B>>>, Response = Response<ResBody>, Error = ServiceError>
         + Clone
         + Send
         + 'static,
-    <S as Service<HttpRequest<B>>>::Future: Send + 'static,
-    <S as Service<HttpRequest<EarlyResponseBody<B>>>>::Future: Send + 'static,
+    <S as Service<HttpRequest<ObservedBody<EarlyResponseBody<B>>>>>::Future: Send + 'static,
     B: http_body::Body<Data = Bytes> + Send + Unpin + 'static,
     B::Error: std::error::Error + Send + Sync + 'static,
     ResBody: Send + 'static,
@@ -842,32 +841,32 @@ where
     type Future = Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
-        match <S as Service<HttpRequest<B>>>::poll_ready(&mut self.inner, cx)? {
-            Poll::Ready(()) => <S as Service<HttpRequest<EarlyResponseBody<B>>>>::poll_ready(&mut self.inner, cx),
-            Poll::Pending => Poll::Pending,
-        }
+        self.inner.poll_ready(cx)
     }
 
     fn call(&mut self, req: HttpRequest<B>) -> Self::Future {
         let version = req.version();
         let preserve_on_drop = matches!(version, Version::HTTP_10 | Version::HTTP_11) && !req.body().is_end_stream();
         let mut inner = self.inner.clone();
-        if !preserve_on_drop {
-            return Box::pin(async move { <S as Service<HttpRequest<B>>>::call(&mut inner, req).await });
-        }
+        let mut req = req;
+        let control = BodyReadControl::default();
+        req.extensions_mut().insert(control.clone());
 
         let mut drain_context = EarlyResponseBodyDrainContext::from_request(&req, self.idle_timeout);
         let state = Arc::new(EarlyResponseBodyState::default());
         let guarded_req = req.map({
             let state = Arc::clone(&state);
-            move |body| EarlyResponseBody::new(body, state)
+            move |body| ObservedBody::new(EarlyResponseBody::new(body, state), control)
         });
 
         Box::pin(async move {
-            let result = <S as Service<HttpRequest<EarlyResponseBody<B>>>>::call(&mut inner, guarded_req).await;
+            let result = inner.call(guarded_req).await;
             let Some(abandoned) = state.take_abandoned() else {
                 return result;
             };
+            if !preserve_on_drop {
+                return result;
+            }
 
             match result {
                 Ok(mut response) => {
@@ -2458,9 +2457,10 @@ mod tests {
     use crate::storage_api::server::http::ScannerScopedDirtyUsageAckEntry;
     use bytes::Bytes;
     use http::Request as HttpRequest;
+    use http::header::CONTENT_LENGTH;
     use http::{HeaderMap, StatusCode};
     use http_body::Frame;
-    use http_body_util::{Empty, Full};
+    use http_body_util::{BodyExt, Empty, Full};
     use metrics::with_local_recorder;
     use metrics_util::debugging::{DebugValue, DebuggingRecorder};
     use opentelemetry::propagation::Extractor;
@@ -2887,6 +2887,159 @@ mod tests {
         assert_eq!(bytes_polled.load(Ordering::Relaxed), 0);
     }
 
+    #[derive(Clone, Copy)]
+    struct UploadPartTimeoutS3 {
+        timeout: Duration,
+    }
+
+    #[async_trait::async_trait]
+    impl s3s::S3 for UploadPartTimeoutS3 {
+        async fn upload_part(
+            &self,
+            req: s3s::S3Request<s3s::dto::UploadPartInput>,
+        ) -> s3s::S3Result<s3s::S3Response<s3s::dto::UploadPartOutput>> {
+            use futures::StreamExt;
+            use tokio_util::io::StreamReader;
+
+            let control = req
+                .extensions
+                .get::<BodyReadControl>()
+                .expect("HTTP route must install the control")
+                .clone();
+            control.activate(self.timeout, "bucket", "object", "request", 1024);
+            let body = req.input.body.expect("upload body");
+            let inner = rustfs_rio::wrap_reader(StreamReader::new(body.map(|item| item.map_err(std::io::Error::other))));
+            let mut reader = crate::app::object::request_body::DemandReader::new(inner, control);
+            reader
+                .read_to_end(&mut Vec::new())
+                .await
+                .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?;
+            Ok(s3s::S3Response::new(s3s::dto::UploadPartOutput::default()))
+        }
+
+        async fn head_bucket(
+            &self,
+            _req: s3s::S3Request<s3s::dto::HeadBucketInput>,
+        ) -> s3s::S3Result<s3s::S3Response<s3s::dto::HeadBucketOutput>> {
+            Ok(s3s::S3Response::new(s3s::dto::HeadBucketOutput::default()))
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn upload_part_timeout_preserves_http1_drain_and_http2_stream_drop() {
+        for version in [Version::HTTP_11, Version::HTTP_2] {
+            let (sender, body, bytes_polled, dropped) = tracked_request_body();
+            let request = HttpRequest::builder()
+                .version(version)
+                .method(Method::PUT)
+                .uri("/bucket/object?partNumber=1&uploadId=upload")
+                .header(CONTENT_LENGTH, "1024")
+                .body(body)
+                .expect("upload request");
+            let inner = s3s::service::S3ServiceBuilder::new(UploadPartTimeoutS3 {
+                timeout: Duration::from_secs(300),
+            })
+            .build();
+            let mut service = EarlyResponseBodyService::new(inner, Duration::from_secs(30));
+            let mut call = Box::pin(service.call(request));
+            assert!(futures::poll!(call.as_mut()).is_pending());
+            tokio::time::advance(Duration::from_secs(300)).await;
+            let response = call.await.expect("timeout response");
+            assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+            assert_eq!(response.headers().get(CONNECTION).is_some(), version == Version::HTTP_11);
+            let xml = response.into_body().collect().await.expect("timeout XML").to_bytes();
+            assert!(String::from_utf8_lossy(&xml).contains("<Code>RequestTimeout</Code>"));
+            if version == Version::HTTP_11 {
+                assert!(
+                    !dropped.load(Ordering::Acquire),
+                    "synthetic errors must retain the unfinished raw transport"
+                );
+                sender
+                    .send(Bytes::from_static(b"late payload"))
+                    .expect("native drain receiver");
+                drop(sender);
+                tokio::task::yield_now().await;
+                assert_eq!(bytes_polled.load(Ordering::Relaxed), 12);
+            } else {
+                assert!(sender.is_closed(), "HTTP/2 must drop only the failed body");
+                assert_eq!(bytes_polled.load(Ordering::Relaxed), 0);
+            }
+            assert!(dropped.load(Ordering::Acquire));
+        }
+    }
+
+    #[tokio::test]
+    async fn upload_part_timeout_leaves_other_http2_streams_usable() {
+        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
+        let inner = s3s::service::S3ServiceBuilder::new(UploadPartTimeoutS3 {
+            timeout: Duration::from_millis(500),
+        })
+        .build();
+        let service = EarlyResponseBodyService::new(inner, Duration::from_secs(1));
+        let server = tokio::spawn(async move {
+            hyper::server::conn::http2::Builder::new(hyper_util::rt::TokioExecutor::new())
+                .serve_connection(TokioIo::new(server_io), TowerToHyperService::new(service))
+                .await
+        });
+        let (mut client, connection) = hyper::client::conn::http2::handshake::<_, _, TrackedRequestBody>(
+            hyper_util::rt::TokioExecutor::new(),
+            TokioIo::new(client_io),
+        )
+        .await
+        .expect("HTTP/2 handshake");
+        let connection = tokio::spawn(connection);
+        let (_sender, body, _, _) = tracked_request_body();
+        let stalled = client.send_request(
+            HttpRequest::builder()
+                .method(Method::PUT)
+                .uri("http://localhost/bucket/object?partNumber=1&uploadId=upload")
+                .header(CONTENT_LENGTH, "1024")
+                .body(body)
+                .expect("stalled request"),
+        );
+
+        client.ready().await.expect("same connection remains ready");
+        let (sender, body, _, _) = tracked_request_body();
+        drop(sender);
+        let response = client
+            .send_request(
+                HttpRequest::builder()
+                    .method(Method::HEAD)
+                    .uri("http://localhost/bucket")
+                    .body(body)
+                    .expect("healthy stream"),
+            )
+            .await
+            .expect("healthy response");
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(!response.headers().contains_key(CONNECTION));
+        response.into_body().collect().await.expect("healthy stream completes");
+        let response = tokio::time::timeout(Duration::from_secs(5), stalled)
+            .await
+            .expect("stream timeout")
+            .expect("S3 response");
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+        let xml = response.into_body().collect().await.expect("timeout body").to_bytes();
+        assert!(String::from_utf8_lossy(&xml).contains("<Code>RequestTimeout</Code>"));
+        client.ready().await.expect("connection after failed stream");
+        let (sender, body, _, _) = tracked_request_body();
+        drop(sender);
+        let response = client
+            .send_request(
+                HttpRequest::builder()
+                    .method(Method::HEAD)
+                    .uri("http://localhost/bucket")
+                    .body(body)
+                    .expect("later stream"),
+            )
+            .await
+            .expect("later response");
+        assert_eq!(response.status(), StatusCode::OK);
+        drop(client);
+        connection.abort();
+        server.abort();
+    }
+
     #[tokio::test(start_paused = true)]
     async fn early_response_body_drain_releases_stalled_body_after_idle_timeout() {
         let (_sender, body, _bytes_polled, dropped) = tracked_request_body();