Zstd decoding fails after repeated Poll::Pending with compression-codecs ≥ 0.4.40

[bvinc]

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.

[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

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::<usize>())
        .transpose()
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?
        .unwrap_or(32);
    println!("Pending polls between compressed chunks: {pending}");

    let data: Vec<u8> = (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.