Fix decoding of empty response streams. (#444)

Previously we were returning "zstd stream did not finish", but now we
notice we got no data and transition directly to the done state.

Author: muirdm

crates/async-compression/Cargo.toml

@@ -139,6 +139,10 @@ required-features = ["zlib", "tokio"]
 name = "zstd_gzip"
 required-features = ["zstd", "gzip", "tokio"]
 
+[[test]]
+name = "empty_stream"
+required-features = ["zstd", "tokio"]
+
 [[example]]
 name = "lzma_filters"
 required-features = ["xz", "tokio"]

crates/async-compression/src/generic/bufread/decoder.rs

@@ -17,13 +17,15 @@ enum State {
 pub struct Decoder {
     state: State,
     multiple_members: bool,
+    received_data: bool,
 }
 
 impl Default for Decoder {
     fn default() -> Self {
         Self {
             state: State::Decoding,
             multiple_members: false,
+            received_data: false,
         }
     }
 }
@@ -48,8 +50,16 @@ impl Decoder {
                         // reader has returned EOF.
                         self.multiple_members = false;
 
-                        State::Flushing
+                        // Empty stream (no data received) - return empty output
+                        if !self.received_data {
+                            State::Done
+                        } else {
+                            State::Flushing
+                        }
                     } else {
+                        if !input.unwritten().is_empty() {
+                            self.received_data = true;
+                        }
                         match decoder.decode(input, output) {
                             Ok(true) => State::Flushing,
                             // ignore the first error, occurs when input is empty

crates/async-compression/tests/empty_stream.rs

@@ -0,0 +1,19 @@
+//! Test that bufread decoders handle empty input streams (immediate EOF).
+
+#[macro_use]
+mod utils;
+
+#[tokio::test]
+async fn zstd_empty_stream() {
+    use async_compression::tokio::bufread::ZstdDecoder;
+    use std::io::Cursor;
+    use tokio::io::AsyncReadExt;
+
+    let empty: &[u8] = &[];
+    let mut decoder = ZstdDecoder::new(Cursor::new(empty));
+    let mut output = Vec::new();
+    let result = decoder.read_to_end(&mut output).await;
+    // Empty input should return Ok(0), not error with "zstd stream did not finish"
+    assert!(result.is_ok(), "empty stream failed: {:?}", result);
+    assert!(output.is_empty());
+}