`ResponseDataStream`: `ReaderStream` vs `.bytes`

[johnbchron]

Describe the bug
It looks like the AsyncRead implementation on ResponseDataStream is incorrect. The only datapoint I have at the moment is that I consistently only get about 1/3 the number of bytes back from reading all of tokio_util::io::ReaderStream::new(response_data_stream) vs response_data_stream.bytes.

Converting a impl Stream<Item<Result<bytes::Bytes, _>>> to a impl AsyncRead and then reading it should yield all the same bytes as just reading the stream directly, and I'm guessing it's not an issue with the ReaderStream implementation. I'll do some more testing.

To Reproduce

let req = bucket.get_object_stream(key).await.unwrap();
let stream = req.bytes;

let mut data = Vec::new();
while let Some(Ok(chunk)) = stream.next().await {
    data.extend(&chunk[..]);
}
println!("data.len() = {}", data.len());

// same as before, but now with:
let stream = tokio_util::io::ReaderStream::new(req);

Expected behavior
These two options produce the same number of bytes

Environment

• Rust version: rustc 1.93.0-nightly (f37aa9955 2025-10-26)
• lib version 0.37.0

[dmk978]

I also run into this issue and discovered that definetely some chunks of data in the middle is missing at least in large files (> 1 Gb).

[johnbchron]

Looking at this page of the tokio docs, it looks like dropping data that couldn't fit in the buffer is incorrect behavior. Not certain though, as that might be the reason for the distinction between AsyncRead and AsyncBufRead. Either way I think it's safe to say that it's unexpected behavior for data to be dropped on insufficient buffer space.

• Poll::Ready(Ok(())) means that data was immediately read and placed into the output buffer. The amount of data read can be determined by the increase in the length of the slice returned by ReadBuf::filled. If the difference is 0, either EOF has been reached, or the output buffer had zero capacity (i.e. buf.remaining() == 0).

@durch I recommend instead of implementing AsyncRead directly, switch to providing an into_async_read(self) that wraps the response in a tokio_util::io::StreamReader.

If this is the direction you want to go in, I can provide a PR.