fix(ecstore): preserve CopyObject producer errors (#6090)

* fix(ecstore): preserve CopyObject producer errors

* fix(app): resume preserved relocation I/O errors

* fix(copy): preserve transformed source errors

crates/ecstore/src/object_api/mod.rs

@@ -22,7 +22,7 @@ use crate::bucket::replication::{
 use crate::bucket::versioning::VersioningApi as _;
 use crate::config::storageclass;
 use crate::error::{Error, Result};
-use crate::io_support::rio::{HashReader, LimitReader};
+use crate::io_support::rio::{HardLimitReader, HashReader};
 use crate::storage_api_contracts::{
     lifecycle::{ExpirationOptions, TransitionedObject},
     range::HTTPRangeSpec,

crates/ecstore/src/object_api/readers.rs

@@ -808,7 +808,7 @@ impl ReadPlan {
                         }
                     }
                 } else {
-                    Box::new(LimitReader::new(dec_reader, total_plaintext_size))
+                    Box::new(HardLimitReader::new(dec_reader, decompressed_length))
                 };
 
                 let mut object_info = oi.clone();
@@ -900,7 +900,7 @@ impl ReadPlan {
                             )?;
                             Box::new(ranged_reader)
                         } else {
-                            Box::new(LimitReader::new(decompressed_reader, total_plaintext_size))
+                            Box::new(HardLimitReader::new(decompressed_reader, total_plaintext_size_i64))
                         }
                     } else if plaintext_offset > 0 || plaintext_length != total_plaintext_size_i64 {
                         Box::new(RangedDecompressReader::new(
@@ -910,7 +910,7 @@ impl ReadPlan {
                             total_plaintext_size,
                         )?)
                     } else {
-                        Box::new(LimitReader::new(decrypted_reader, total_plaintext_size))
+                        Box::new(HardLimitReader::new(decrypted_reader, total_plaintext_size_i64))
                     };
 
                 let mut object_info = oi.clone();

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

@@ -315,6 +315,41 @@ async fn get_object_reader_with_context(
     GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
 }
 
+async fn get_legacy_object_reader_with_context<R>(
+    ctx: &InstanceContext,
+    reader: R,
+    terminal: tokio::sync::oneshot::Receiver<Result<()>>,
+    range: Option<HTTPRangeSpec>,
+    object_info: &ObjectInfo,
+    opts: &ObjectOptions,
+    headers: &HeaderMap<HeaderValue>,
+) -> Result<(GetObjectReader, usize, i64)>
+where
+    R: AsyncRead + Unpin + Send + Sync + 'static,
+{
+    // ReadPlan validates this size below; failure here only keeps the terminal
+    // guard inside the transform until that validation returns its typed error.
+    let full_plaintext_size = object_info.get_actual_size().ok();
+    let whole_object = opts.part_number.is_none()
+        && match (&range, full_plaintext_size) {
+            (None, _) => true,
+            (Some(range), Some(size)) => range
+                .get_offset_length(size)
+                .is_ok_and(|(offset, length)| offset == 0 && length == size),
+            (Some(_), None) => false,
+        };
+    let (source, terminal): (Box<dyn AsyncRead + Unpin + Send + Sync>, _) = if whole_object {
+        (Box::new(reader), Some(terminal))
+    } else {
+        (Box::new(LegacyDuplexProducerReader::new(reader, terminal)), None)
+    };
+    let (mut reader, offset, length) = get_object_reader_with_context(ctx, source, range, object_info, opts, headers).await?;
+    if let Some(terminal) = terminal {
+        reader.stream = Box::new(LegacyDuplexProducerReader::new(reader.stream, terminal));
+    }
+    Ok((reader, offset, length))
+}
+
 fn data_read_metadata_early_stop_request_shape_allowed(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions) -> bool {
     range.is_none()
         && opts.part_number.is_none()
@@ -1090,8 +1125,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (rd, wd) = tokio::io::duplex(duplex_buffer_size);
         debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer");
 
+        let (producer_terminal_tx, producer_terminal_rx) = tokio::sync::oneshot::channel();
         let (mut reader, offset, length) =
-            get_object_reader_with_context(&self.ctx, Box::new(rd), range, &object_info, opts, &h).await?;
+            get_legacy_object_reader_with_context(&self.ctx, rd, producer_terminal_rx, range, &object_info, opts, &h).await?;
         // Carry the hook probe result so the app layer skips its now-redundant
         // lookup on the streaming miss path (ODC-16).
         reader.body_source = body_source;
@@ -1111,7 +1147,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout
             // would incorrectly treat downstream backpressure as disk-read latency.
             // Disk read timeouts must be enforced at the actual disk I/O operations.
-            if let Err(e) = Self::get_object_with_fileinfo(
+            let producer_result = Self::get_object_with_fileinfo(
                 &bucket,
                 &object,
                 erasure_cache,
@@ -1129,9 +1165,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 object_class.as_str(),
                 size_bucket,
             )
-            .await
-            {
-                let reason = classify_storage_error(&e);
+            .await;
+            if let Err(e) = &producer_result {
+                let reason = classify_storage_error(e);
                 if reason == GetObjectFailureReason::DownstreamClosed {
                     debug!(
                         event = EVENT_SET_DISK_WRITE,
@@ -1170,6 +1206,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                     );
                 }
             };
+            let _ = producer_terminal_tx.send(producer_result.map(|_| ()));
         });
 
         Ok(reader)
@@ -2558,6 +2595,420 @@ impl<R: AsyncRead + Unpin> AsyncRead for TransitionUploadReader<R> {
     }
 }
 
+struct LegacyDuplexProducerReader<R> {
+    inner: Option<R>,
+    terminal: Option<tokio::sync::oneshot::Receiver<Result<()>>>,
+    inner_eof: bool,
+}
+
+impl<R> LegacyDuplexProducerReader<R> {
+    fn new(inner: R, terminal: tokio::sync::oneshot::Receiver<Result<()>>) -> Self {
+        Self {
+            inner: Some(inner),
+            terminal: Some(terminal),
+            inner_eof: false,
+        }
+    }
+}
+
+impl<R: AsyncRead + Unpin> AsyncRead for LegacyDuplexProducerReader<R> {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+        if !self.inner_eof {
+            let before = buf.filled().len();
+            if let Some(inner) = self.inner.as_mut() {
+                match Pin::new(inner).poll_read(cx, buf) {
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                    Poll::Ready(Ok(())) if buf.filled().len() > before => return Poll::Ready(Ok(())),
+                    Poll::Ready(Ok(())) => {
+                        self.inner_eof = true;
+                        self.inner = None;
+                    }
+                }
+            } else {
+                self.inner_eof = true;
+            }
+        }
+
+        let Some(terminal) = self.terminal.as_mut() else {
+            return Poll::Ready(Ok(()));
+        };
+        match Pin::new(terminal).poll(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Ok(Ok(()))) => {
+                self.terminal = None;
+                Poll::Ready(Ok(()))
+            }
+            Poll::Ready(Ok(Err(err))) => {
+                self.terminal = None;
+                Poll::Ready(Err(std::io::Error::other(err)))
+            }
+            Poll::Ready(Err(_)) => {
+                self.terminal = None;
+                Poll::Ready(Err(std::io::Error::other(StorageError::Unexpected)))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod legacy_duplex_producer_reader_tests {
+    use super::*;
+    use crate::object_api::{EncryptionResolutionError, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionMode};
+    use rustfs_utils::CompressionAlgorithm;
+    use std::collections::HashMap;
+    use std::sync::Arc;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    const TEST_DUPLEX_CAPACITY: usize = 64 * 1024;
+
+    fn storage_error_source(error: &std::io::Error) -> &StorageError {
+        error
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<StorageError>())
+            .expect("legacy duplex terminal error should retain StorageError source")
+    }
+
+    async fn compressed_fixture(plaintext: Vec<u8>, recorded_size: usize) -> (Vec<u8>, ObjectInfo) {
+        let mut compressor = rustfs_rio::CompressReader::new(std::io::Cursor::new(plaintext), CompressionAlgorithm::default());
+        let mut compressed = Vec::new();
+        compressor
+            .read_to_end(&mut compressed)
+            .await
+            .expect("compress test plaintext");
+
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_COMPRESSION,
+            CompressionAlgorithm::default().to_string(),
+        );
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, recorded_size.to_string());
+        let object_info = ObjectInfo {
+            size: i64::try_from(compressed.len()).expect("compressed fixture length should fit in i64"),
+            user_defined: Arc::new(metadata),
+            ..Default::default()
+        };
+        (compressed, object_info)
+    }
+
+    #[tokio::test]
+    async fn legacy_duplex_reader_allows_clean_completion() {
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        writer
+            .write_all(b"complete")
+            .await
+            .expect("duplex write should fit in buffer");
+        drop(writer);
+        terminal_tx.send(Ok(())).expect("terminal receiver should remain installed");
+
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+        let mut out = Vec::new();
+        reader
+            .read_to_end(&mut out)
+            .await
+            .expect("clean producer completion should surface clean EOF");
+
+        assert_eq!(out, b"complete");
+    }
+
+    #[tokio::test]
+    async fn legacy_duplex_reader_ignores_zero_capacity_read_buf() {
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        writer.write_all(b"body").await.expect("duplex write should fit in buffer");
+        drop(writer);
+        terminal_tx
+            .send(Err(StorageError::FileCorrupt))
+            .expect("terminal receiver should remain installed");
+
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+        let mut empty = [];
+        std::future::poll_fn(|cx| {
+            let mut read_buf = ReadBuf::new(&mut empty);
+            Pin::new(&mut reader).poll_read(cx, &mut read_buf)
+        })
+        .await
+        .expect("zero-capacity reads should complete without observing EOF or terminal state");
+        assert!(!reader.inner_eof);
+        assert!(reader.terminal.is_some());
+
+        let mut out = Vec::new();
+        let err = reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("subsequent reads must still receive data and the terminal error");
+        assert_eq!(out, b"body");
+        assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt));
+    }
+
+    #[tokio::test]
+    async fn legacy_duplex_reader_surfaces_terminal_error_after_partial_data() {
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        writer.write_all(b"partial").await.expect("duplex write should fit in buffer");
+        drop(writer);
+        terminal_tx
+            .send(Err(StorageError::FileCorrupt))
+            .expect("terminal receiver should remain installed");
+
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+        let mut out = Vec::new();
+        let err = reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("terminal producer error must not become clean EOF");
+
+        assert_eq!(out, b"partial");
+        assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt));
+    }
+
+    #[tokio::test]
+    async fn legacy_duplex_reader_surfaces_terminal_error_after_declared_length() {
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        writer.write_all(b"exact").await.expect("duplex write should fit in buffer");
+        drop(writer);
+        terminal_tx
+            .send(Err(StorageError::Io(std::io::Error::new(
+                std::io::ErrorKind::ConnectionReset,
+                "remote body reset after final byte",
+            ))))
+            .expect("terminal receiver should remain installed");
+
+        let reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+        let mut reader =
+            HashReader::from_stream(reader, 5, 5, None, None, false).expect("hash reader should accept exact declared length");
+        let mut out = Vec::new();
+        let err = reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("producer terminal error after the declared length must still fail");
+
+        assert_eq!(out, b"exact");
+        assert!(
+            matches!(storage_error_source(&err), StorageError::Io(io_error) if io_error.kind() == std::io::ErrorKind::ConnectionReset)
+        );
+    }
+
+    #[tokio::test]
+    async fn legacy_compressed_reader_surfaces_terminal_error_after_complete_plaintext() {
+        let plaintext = b"compressed terminal result must survive the plaintext limit".repeat(16);
+        let (compressed, object_info) = compressed_fixture(plaintext.clone(), plaintext.len()).await;
+        let full_range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: i64::try_from(plaintext.len()).expect("plaintext fixture length should fit in i64") - 1,
+        };
+        for range in [None, Some(full_range)] {
+            let (mut writer, reader) = tokio::io::duplex(compressed.len().max(1));
+            writer
+                .write_all(&compressed)
+                .await
+                .expect("compressed body should fit in duplex buffer");
+            drop(writer);
+            let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+            terminal_tx
+                .send(Err(StorageError::FileCorrupt))
+                .expect("terminal receiver should remain installed");
+
+            let (mut reader, _, _) = get_legacy_object_reader_with_context(
+                &InstanceContext::new(),
+                reader,
+                terminal_rx,
+                range,
+                &object_info,
+                &ObjectOptions::default(),
+                &HeaderMap::new(),
+            )
+            .await
+            .expect("compressed read plan should build");
+            let mut out = Vec::new();
+            let err = reader
+                .read_to_end(&mut out)
+                .await
+                .expect_err("terminal error after complete decompression must not become clean EOF");
+
+            assert_eq!(out, plaintext);
+            assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt));
+        }
+    }
+
+    #[tokio::test]
+    async fn legacy_exact_reader_rejects_extra_data_without_backpressure_deadlock() {
+        let payload = vec![0x5a; TEST_DUPLEX_CAPACITY * 2];
+        let (mut writer, reader) = tokio::io::duplex(TEST_DUPLEX_CAPACITY);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        let producer = tokio::spawn(async move {
+            let result = writer.write_all(&payload).await;
+            drop(writer);
+            let terminal_result = result
+                .as_ref()
+                .map(|_| ())
+                .map_err(|err| StorageError::Io(std::io::Error::new(err.kind(), err.to_string())));
+            let _ = terminal_tx.send(terminal_result);
+            result
+        });
+        let reader = crate::io_support::rio::HardLimitReader::new(reader, 1);
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+
+        let mut out = Vec::new();
+        tokio::time::timeout(std::time::Duration::from_secs(1), reader.read_to_end(&mut out))
+            .await
+            .expect("extra data beyond the declared size must not deadlock")
+            .expect_err("extra data beyond the declared size must fail closed");
+        assert_eq!(out, [0x5a]);
+        drop(reader);
+        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), producer)
+            .await
+            .expect("producer must unblock after the read fails")
+            .expect("producer task should not panic");
+    }
+
+    #[tokio::test]
+    async fn legacy_terminal_reader_releases_unconsumed_source_before_waiting() {
+        let payload = vec![0x5a; TEST_DUPLEX_CAPACITY * 2];
+        let (mut writer, reader) = tokio::io::duplex(TEST_DUPLEX_CAPACITY);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+        let producer = tokio::spawn(async move {
+            let result = writer.write_all(&payload).await;
+            drop(writer);
+            let terminal_result = result
+                .as_ref()
+                .map(|_| ())
+                .map_err(|err| StorageError::Io(std::io::Error::new(err.kind(), err.to_string())));
+            let _ = terminal_tx.send(terminal_result);
+            result
+        });
+        let reader = rustfs_rio::LimitReader::new(reader, 1);
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+
+        let mut out = Vec::new();
+        let err = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read_to_end(&mut out))
+            .await
+            .expect("terminal wait must not deadlock behind unconsumed source data")
+            .expect_err("unconsumed source data must fail the producer terminal result");
+        assert_eq!(out, [0x5a]);
+        assert!(
+            matches!(storage_error_source(&err), StorageError::Io(io_error) if io_error.kind() == std::io::ErrorKind::BrokenPipe)
+        );
+        producer
+            .await
+            .expect("producer task should not panic")
+            .expect_err("source should close early");
+    }
+
+    struct FixedEncryptionResolver {
+        key_bytes: [u8; 32],
+        base_nonce: [u8; 12],
+    }
+
+    #[async_trait::async_trait]
+    impl ObjectEncryptionResolver for FixedEncryptionResolver {
+        async fn resolve_read_material(
+            &self,
+            _request: crate::object_api::ReadEncryptionRequest<'_>,
+        ) -> std::result::Result<Option<ReadEncryptionMaterial>, EncryptionResolutionError> {
+            Ok(Some(ReadEncryptionMaterial {
+                key_bytes: self.key_bytes,
+                mode: ReadEncryptionMode::Direct {
+                    base_nonce: self.base_nonce,
+                },
+            }))
+        }
+    }
+
+    #[tokio::test]
+    async fn legacy_encrypted_reader_surfaces_terminal_error_after_complete_plaintext() {
+        let plaintext = b"encrypted terminal result must survive the plaintext limit".repeat(16);
+        let key_bytes = [0x31; 32];
+        let base_nonce = [0x42; 12];
+        let mut encryptor = rustfs_rio::EncryptReader::new(std::io::Cursor::new(plaintext.clone()), key_bytes, base_nonce);
+        let mut encrypted = Vec::new();
+        encryptor.read_to_end(&mut encrypted).await.expect("encrypt test plaintext");
+
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "encrypted-object".to_string(),
+            size: i64::try_from(encrypted.len()).expect("encrypted fixture length should fit in i64"),
+            user_defined: Arc::new(HashMap::from([
+                ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()),
+                (
+                    "x-amz-server-side-encryption-customer-original-size".to_string(),
+                    plaintext.len().to_string(),
+                ),
+            ])),
+            ..Default::default()
+        };
+        let ctx = InstanceContext::new();
+        assert!(
+            ctx.set_object_encryption_resolver(Arc::new(FixedEncryptionResolver { key_bytes, base_nonce }))
+                .is_ok(),
+            "fresh context should accept resolver"
+        );
+        let full_range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: i64::try_from(plaintext.len()).expect("plaintext fixture length should fit in i64") - 1,
+        };
+        for range in [None, Some(full_range)] {
+            let (mut writer, reader) = tokio::io::duplex(encrypted.len().max(1));
+            writer
+                .write_all(&encrypted)
+                .await
+                .expect("encrypted body should fit in duplex buffer");
+            drop(writer);
+            let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
+            terminal_tx
+                .send(Err(StorageError::FileCorrupt))
+                .expect("terminal receiver should remain installed");
+
+            let (mut reader, _, _) = get_legacy_object_reader_with_context(
+                &ctx,
+                reader,
+                terminal_rx,
+                range,
+                &object_info,
+                &ObjectOptions::default(),
+                &HeaderMap::new(),
+            )
+            .await
+            .expect("encrypted read plan should build");
+            let mut out = Vec::new();
+            let err = reader
+                .read_to_end(&mut out)
+                .await
+                .expect_err("terminal error after complete decryption must not become clean EOF");
+
+            assert_eq!(out, plaintext);
+            assert!(matches!(storage_error_source(&err), StorageError::FileCorrupt));
+        }
+    }
+
+    #[tokio::test]
+    async fn legacy_duplex_reader_fails_closed_when_terminal_channel_closes() {
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel::<Result<()>>();
+        writer.write_all(b"body").await.expect("duplex write should fit in buffer");
+        drop(writer);
+        drop(terminal_tx);
+
+        let mut reader = LegacyDuplexProducerReader::new(reader, terminal_rx);
+        let mut out = Vec::new();
+        let err = reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("producer disappearance must fail closed");
+
+        assert_eq!(out, b"body");
+        assert!(matches!(storage_error_source(&err), StorageError::Unexpected));
+    }
+}
+
 struct TransitionUploadWriter<W> {
     inner: W,
     produced: u64,

crates/rio/src/hardlimit_reader.rs

@@ -24,12 +24,17 @@ pin_project! {
         #[pin]
         pub inner: R,
         remaining: i64,
+        scratch: Vec<u8>,
     }
 }
 
 impl<R> HardLimitReader<R> {
     pub fn new(inner: R, limit: i64) -> Self {
-        HardLimitReader { inner, remaining: limit }
+        HardLimitReader {
+            inner,
+            remaining: limit,
+            scratch: Vec::new(),
+        }
     }
 }
 
@@ -37,19 +42,21 @@ impl<R> AsyncRead for HardLimitReader<R>
 where
     R: AsyncRead,
 {
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<()>> {
-        if self.remaining < 0 {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<()>> {
+        let mut this = self.project();
+        if *this.remaining < 0 {
             return Poll::Ready(Err(Error::other("input provided more bytes than specified")));
         }
-        let original_filled = buf.filled().len();
-        if self.remaining == 0 {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+        if *this.remaining == 0 {
             let mut discard = [0u8; 8192];
             let mut discard_buf = ReadBuf::new(&mut discard);
-            return match self.as_mut().project().inner.poll_read(cx, &mut discard_buf) {
+            return match this.inner.as_mut().poll_read(cx, &mut discard_buf) {
                 Poll::Pending => Poll::Pending,
                 Poll::Ready(Ok(())) => {
                     if discard_buf.filled().is_empty() {
-                        debug_assert_eq!(buf.filled().len(), original_filled);
                         Poll::Ready(Ok(()))
                     } else {
                         Poll::Ready(Err(Error::other("input provided more bytes than specified")))
@@ -58,30 +65,46 @@ where
                 Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
             };
         }
-        // Save the initial length
-        let before = original_filled;
 
-        // Poll the inner reader
-        let this = self.as_mut().project();
-        let poll = this.inner.poll_read(cx, buf);
-
-        if let Poll::Ready(Ok(())) = &poll {
-            let after = buf.filled().len();
-            let read = (after - before) as i64;
-            if read == 0 && *this.remaining > 0 {
-                return Poll::Ready(Err(Error::new(
-                    std::io::ErrorKind::UnexpectedEof,
-                    IncompleteBody {
-                        remaining: *this.remaining,
-                    },
-                )));
+        let remaining = match usize::try_from(*this.remaining) {
+            Ok(remaining) => remaining,
+            Err(_) => usize::MAX,
+        };
+        let allowed = remaining.min(buf.remaining());
+        let read = if allowed == buf.remaining() {
+            let before = buf.filled().len();
+            match this.inner.as_mut().poll_read(cx, buf) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                Poll::Ready(Ok(())) => buf.filled().len() - before,
             }
-            *this.remaining -= read;
-            if *this.remaining < 0 {
-                return Poll::Ready(Err(Error::other("input provided more bytes than specified")));
+        } else {
+            this.scratch.resize(allowed, 0);
+            let mut scratch_buf = ReadBuf::new(&mut this.scratch[..allowed]);
+            match this.inner.as_mut().poll_read(cx, &mut scratch_buf) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+                Poll::Ready(Ok(())) => {
+                    let read = scratch_buf.filled().len();
+                    buf.put_slice(scratch_buf.filled());
+                    read
+                }
             }
+        };
+        if read == 0 {
+            return Poll::Ready(Err(Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                IncompleteBody {
+                    remaining: *this.remaining,
+                },
+            )));
         }
-        poll
+        let read = match i64::try_from(read) {
+            Ok(read) => read,
+            Err(_) => return Poll::Ready(Err(Error::other("read count exceeds i64::MAX"))),
+        };
+        *this.remaining -= read;
+        Poll::Ready(Ok(()))
     }
 }
 
@@ -140,7 +163,12 @@ mod tests {
         assert!(err.is_some());
 
         let err = err.unwrap();
-        assert_eq!(err.kind(), std::io::ErrorKind::Other);
+        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
+        assert!(
+            err.get_ref()
+                .and_then(|source| source.downcast_ref::<std::io::Error>())
+                .is_some_and(|source| source.to_string().contains("more bytes than specified"))
+        );
     }
 
     #[tokio::test]
@@ -155,6 +183,17 @@ mod tests {
         assert_eq!(&buf, data);
     }
 
+    #[tokio::test]
+    async fn test_hardlimit_reader_zero_capacity_read_does_not_consume_input() {
+        let mut reader = HardLimitReader::new(BufReader::new(&b"abc"[..]), 3);
+        let mut empty = [];
+
+        assert_eq!(reader.read(&mut empty).await.expect("zero-capacity read should succeed"), 0);
+        let mut out = Vec::new();
+        reader.read_to_end(&mut out).await.expect("input should remain readable");
+        assert_eq!(out, b"abc");
+    }
+
     #[tokio::test]
     async fn test_hardlimit_reader_short_input_returns_unexpected_eof() {
         let data = b"abc";
@@ -195,4 +234,18 @@ mod tests {
         assert_eq!(err.kind(), std::io::ErrorKind::Other);
         assert!(err.to_string().contains("more bytes than specified"));
     }
+
+    #[tokio::test]
+    async fn test_hardlimit_reader_caps_each_read_before_reporting_extra_bytes() {
+        let mut reader = HardLimitReader::new(BufReader::new(&b"abcdef"[..]), 3);
+        let mut out = Vec::new();
+
+        let err = reader
+            .read_to_end(&mut out)
+            .await
+            .expect_err("bytes beyond the declared limit must be rejected");
+
+        assert_eq!(out, b"abc");
+        assert!(err.to_string().contains("more bytes than specified"));
+    }
 }

crates/rio/src/http_reader.rs

@@ -138,6 +138,12 @@ impl std::fmt::Display for InternodeHttpErrorKind {
     }
 }
 
+#[derive(thiserror::Error, Debug, Clone, Copy, Eq, PartialEq)]
+#[error("internode body stalled for {timeout:?}")]
+pub struct BodyStalled {
+    pub timeout: Duration,
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct InternodeHttpRequestContext {
     method: String,
@@ -271,6 +277,10 @@ pub fn internode_http_timeout_error(method: &Method, url: &str) -> io::Error {
     internode_kind_error(method, url, internode_rpc_operation(url), InternodeHttpErrorKind::ConnectTimeout)
 }
 
+fn body_stalled_error(stall_timeout: Duration) -> io::Error {
+    Error::new(io::ErrorKind::TimedOut, BodyStalled { timeout: stall_timeout })
+}
+
 /// Clone an internode HTTP I/O error while retaining its structured classification.
 ///
 /// The underlying transport source is intentionally omitted because it is not
@@ -1085,10 +1095,7 @@ impl AsyncRead for HttpReader {
                     );
                     record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation);
                     record_internode_error(*this.track_internode_metrics, *this.internode_operation);
-                    Poll::Ready(Err(Error::new(
-                        io::ErrorKind::TimedOut,
-                        "HttpReader stall timeout: no data received before deadline",
-                    )))
+                    Poll::Ready(Err(body_stalled_error(stall_timeout)))
                 } else {
                     Poll::Pending
                 }
@@ -1217,10 +1224,7 @@ impl ChunkReader for HttpChunkReader {
                         );
                         record_internode_stall_timeout(*this.track_internode_metrics, *this.internode_operation);
                         record_internode_error(*this.track_internode_metrics, *this.internode_operation);
-                        return Poll::Ready(Err(Error::new(
-                            io::ErrorKind::TimedOut,
-                            "HttpReader stall timeout: no data received before deadline",
-                        )));
+                        return Poll::Ready(Err(body_stalled_error(stall_timeout)));
                     }
                     return Poll::Pending;
                 }
@@ -2379,6 +2383,46 @@ mod tests {
             Err(err) => err,
         };
         assert_eq!(err.kind(), io::ErrorKind::TimedOut);
+        let stalled = err
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<BodyStalled>())
+            .expect("stall timeout should retain typed body-stalled source");
+        assert_eq!(stalled.timeout, Duration::from_millis(20));
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn http_chunk_reader_stall_timeout_retains_typed_source() {
+        let state = TestState::default();
+        let Some((base_url, handle)) = start_test_server(state).await else {
+            return;
+        };
+        let url = base_url.replace("/stream", "/stall");
+        let mut reader =
+            HttpChunkReader::new_with_stall_timeout(url, Method::GET, HeaderMap::new(), None, Some(Duration::from_millis(20)))
+                .await
+                .expect("chunk reader should open");
+
+        let first = std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64))
+            .await
+            .expect("initial body chunk should arrive")
+            .expect("initial body chunk should not be EOF");
+        assert_eq!(first, b"hello"[..]);
+
+        let err = tokio::time::timeout(
+            Duration::from_secs(1),
+            std::future::poll_fn(|cx| Pin::new(&mut reader).poll_read_chunk(cx, 64)),
+        )
+        .await
+        .expect("stall timeout should wake chunk reader")
+        .expect_err("chunk reader should return a timeout error");
+        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
+        let stalled = err
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<BodyStalled>())
+            .expect("chunk stall timeout should retain typed body-stalled source");
+        assert_eq!(stalled.timeout, Duration::from_millis(20));
 
         handle.abort();
     }