Fix: zstd decompression is truncated to the first frame in the stream

crates/async-compression/Cargo.toml

@@ -72,6 +72,7 @@ tokio = { version = "1.38.2", default-features = false, features = [
     "macros",
     "rt-multi-thread",
     "io-std",
+    "fs",
 ] }
 tokio-util = { version = "0.7", default-features = false, features = ["io"] }
 
@@ -131,6 +132,10 @@ required-features = ["zstd", "tokio"]
 name = "zstd-window-size"
 required-features = ["zstd", "tokio"]
 
+[[test]]
+name = "zstd-bufread"
+required-features = ["zstd", "tokio"]
+
 [[example]]
 name = "zlib_tokio_write"
 required-features = ["zlib", "tokio"]

crates/async-compression/tests/zstd-bufread.rs

@@ -0,0 +1,13 @@
+use async_compression::tokio::bufread::ZstdDecoder;
+
+#[tokio::test]
+async fn multiple_frames() {
+    let llvm_as = tokio::fs::File::open("tests/artifacts/llvm-as.zstd")
+        .await
+        .unwrap();
+    let buffered = tokio::io::BufReader::new(llvm_as);
+    let mut decoder = ZstdDecoder::new(buffered);
+    let mut out = Vec::new();
+    tokio::io::copy(&mut decoder, &mut out).await.unwrap();
+    assert_eq!(out.len(), 4401672)
+}

crates/compression-codecs/src/zstd/decoder.rs

@@ -61,7 +61,12 @@ impl Decode for ZstdDecoder {
             .run_on_buffers(input.unwritten(), output.unwritten_mut())?;
         input.advance(status.bytes_read);
         output.advance(status.bytes_written);
-        Ok(status.remaining == 0)
+
+        // See docs on remaining field
+        let finished_frame = status.remaining == 0;
+        // ...                    && "no more data"
+        let done = finished_frame && status.bytes_read == 0;
+        Ok(done)
     }
 
     fn flush(