On error the decoder now emits everything it was able to decode.

Before this we would return the error and whatever we had in our decoded
buffer would be swallowed.

Fixes #453.

Author: orium

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

@@ -2,7 +2,6 @@ use crate::{
     codecs::DecodeV2,
     core::util::{PartialBuffer, WriteBuffer},
 };
-
 use std::{io::Result, ops::ControlFlow};
 
 #[derive(Debug)]
@@ -11,6 +10,7 @@ enum State {
     Flushing,
     Done,
     Next,
+    Error(std::io::Error),
 }
 
 #[derive(Debug)]
@@ -54,7 +54,10 @@ impl Decoder {
                             Ok(true) => State::Flushing,
                             // ignore the first error, occurs when input is empty
                             // but we need to run decode to flush
-                            Err(err) if !first => return ControlFlow::Break(Err(err)),
+                            Err(err) if !first => {
+                                self.state = State::Error(err);
+                                return ControlFlow::Break(Ok(()));
+                            }
                             // poll for more data for the next decode
                             _ => break,
                         }
@@ -66,7 +69,8 @@ impl Decoder {
                         Ok(true) => {
                             if self.multiple_members {
                                 if let Err(err) = decoder.reinit() {
-                                    return ControlFlow::Break(Err(err));
+                                    self.state = State::Error(err);
+                                    return ControlFlow::Break(Ok(()));
                                 }
 
                                 // The decode stage might consume all the input,
@@ -78,7 +82,10 @@ impl Decoder {
                             }
                         }
                         Ok(false) => State::Flushing,
-                        Err(err) => return ControlFlow::Break(Err(err)),
+                        Err(err) => {
+                            self.state = State::Error(err);
+                            return ControlFlow::Break(Ok(()));
+                        }
                     }
                 }
 
@@ -95,6 +102,13 @@ impl Decoder {
                         State::Decoding
                     }
                 }
+
+                State::Error(_) => {
+                    let State::Error(err) = std::mem::replace(&mut self.state, State::Done) else {
+                        unreachable!()
+                    };
+                    return ControlFlow::Break(Err(err));
+                }
             };
 
             if output.has_no_spare_space() {

crates/async-compression/tests/gzip.rs

@@ -51,3 +51,18 @@ fn gzip_bufread_chunks_decompress_with_extra_header() {
 
     assert_eq!(output, &[1, 2, 3, 4, 5, 6][..]);
 }
+
+#[test]
+#[ntest::timeout(1000)]
+#[cfg(feature = "futures-io")]
+fn gzip_bufread_chunks_decompress_without_footer_emits_all_payload() {
+    let mut bytes = compress_with_header(&[1, 2, 3, 4, 5, 6]);
+
+    // Remove the footer.
+    bytes.truncate(bytes.len() - 8);
+
+    let input = InputStream::from(bytes.chunks(2));
+    let output = bufread::decompress(bufread::from(&input));
+
+    assert_eq!(output, &[1, 2, 3, 4, 5, 6][..]);
+}