Nullus157 / Nullus157/async-compression
Zstd decoding fails after repeated Poll::Pending with compression-codecs ≥ 0.4.40
- Dominant language
- Rust
- Stars
- 666
- Forks
- 118
- Avg merge
- 17h 18m
- Merged PRs (30d)
- 9
Description
Valid Zstd input fails to decode when the input stream returns Poll::Pending repeatedly between chunks:
```
Operation made no progress over multiple calls, due to input being empty
```
Reproduced with async-compression 0.4.46 / compression-codecs 0.4.41.
```toml
[package]
name = "zstd-pending-repro"
version = "0.1.0"
edition = "2021"
[dependencies]
async-compression = { version = "=0.4.46", features = ["tokio", "zstd"] }
compression-codecs = { version = "=0.4.41", features = ["zstd"] }
tokio = { version = "=1.53.1", features = ["rt", "macros", "io-util"] }
tokio-util = { version = "=0.7.19", features = ["io"] }
futures = "0.3"
```
src/main.rs
```rust
use std::{io, task::Poll};
use async_compression::tokio::bufread::{ZstdDecoder, ZstdEncoder};
use futures::stream;
use tokio::io::AsyncReadExt;
use tokio_util::io::StreamReader;
#[tokio::main(flavor = "current_thread")]
async fn main() -> io::Result<()> {
// Use 0 as a passing control; 32 deterministically triggers the bug.
let mut pending = std::env::args()
.nth(1)
.map(|value| value.parse::())
.transpose()
.map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?
.unwrap_or(32);
println!("Pending polls between compressed chunks: {pending}");
let data: Vec = (0u32..4096)
.flat_map(|i| i.wrapping_mul(2654435761).to_le_bytes())
.collect();
let mut encoded = Vec::new();
ZstdEncoder::new(data.as_slice())
.read_to_end(&mut encoded)
.await?;
assert!(encoded.len() > 1024);
let mut first = true;
let mut done = false;
let source = stream::poll_fn(move |cx| {
if first {
first = false;
return Poll::Ready(Some(Ok::<_, io::Error>(io::Cursor::new(
encoded[..1024].to_vec(),
))));
}
if pending > 0 {
pending -= 1;
// Model repeated legal polling while waiting for the next chunk.
cx.waker().wake_by_ref();
return Poll::Pending;
}
if !done {
done = true;
return Poll::Ready(Some(Ok(io::Cursor::new(encoded[1024..].to_vec()))));
}
Poll::Ready(None)
});
let mut decoder = ZstdDecoder::new(StreamReader::new(source));
decoder.multiple_members(true);
let mut output = Vec::new();
tokio::io::copy(&mut decoder, &mut output).await?;
assert_eq!(output, data);
println!("PASS: decoded all {} bytes correctly", output.len());
Ok(())
}
```
Cause and version boundary
The generic decoder calls the codec with empty input at the beginning of each read poll and ignores errors from that initial call. Repeated Pending polls exhaust Zstd’s no-progress budget.
Starting with zstd-safe 8.0.0, decoder errors persist until reset. Once that error occurs, supplying valid input cannot recover the stream.
Holding async-compression at 0.4.43:
compression-codecs 0.4.39 → zstd-safe 7.3.0: passes.
compression-codecs 0.4.40 → zstd-safe 8.0.0: fails.
The dependency upgrade landed in [#481](https://github.com/Nullus157/async-compression/pull/481).
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with the reproduction in src/main.rs and trace the generic decoder behavior described in the report, especially its handling of empty input across repeated Poll::Pending calls. Run the example with the stated dependency versions and varying pending counts; done means valid Zstd data decodes completely and matches the original output without the no-progress error.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 58/100