Fix ZstdDecoder to propagate UnexpectedEof error on truncated streams

The Zstd decoder was silently accepting truncated streams by not
validating stream completion in finish().

This issue was discovered by the generic truncated stream test.

Added state tracking to ZstdDecoder:
- Added stream_ended field to track if remaining == 0 was seen
- Modified decode() to set stream_ended = true when stream completes
- Updated finish() to check stream_ended and return UnexpectedEof if false
- Updated all constructors to initialize stream_ended = false

This matches the behavior of other decoders (bzip2, gzip, lz4, etc.) and
ensures applications cannot accidentally accept corrupted or incomplete
zstd data as valid.

The generic truncated test now passes for Zstd.

Author: claude

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

@@ -13,12 +13,14 @@ use zstd_safe::get_error_name;
 #[derive(Debug)]
 pub struct ZstdDecoder {
     decoder: Unshared<Decoder<'static>>,
+    stream_ended: bool,
 }
 
 impl Default for ZstdDecoder {
     fn default() -> Self {
         Self {
             decoder: Unshared::new(Decoder::new().unwrap()),
+            stream_ended: false,
         }
     }
 }
@@ -35,13 +37,15 @@ impl ZstdDecoder {
         }
         Self {
             decoder: Unshared::new(decoder),
+            stream_ended: false,
         }
     }
 
     pub fn new_with_dict(dictionary: &[u8]) -> io::Result<Self> {
         let decoder = Decoder::with_dictionary(dictionary)?;
         Ok(Self {
             decoder: Unshared::new(decoder),
+            stream_ended: false,
         })
     }
 
@@ -64,6 +68,7 @@ impl ZstdDecoder {
 impl DecodeV2 for ZstdDecoder {
     fn reinit(&mut self) -> Result<()> {
         self.decoder.get_mut().reinit()?;
+        self.stream_ended = false;
         Ok(())
     }
 
@@ -80,15 +85,29 @@ impl DecodeV2 for ZstdDecoder {
             .run_on_buffers(input.unwritten(), output.unwritten_initialized_mut())?;
         input.advance(status.bytes_read);
         output.advance(status.bytes_written);
-        Ok(status.remaining == 0)
+
+        let finished = status.remaining == 0;
+        if finished {
+            self.stream_ended = true;
+        }
+        Ok(finished)
     }
 
     fn flush(&mut self, output: &mut WriteBuffer<'_>) -> Result<bool> {
         self.call_fn_on_out_buffer(output, |decoder, out_buf| decoder.flush(out_buf))
     }
 
     fn finish(&mut self, output: &mut WriteBuffer<'_>) -> Result<bool> {
-        self.call_fn_on_out_buffer(output, |decoder, out_buf| decoder.finish(out_buf, true))
+        self.call_fn_on_out_buffer(output, |decoder, out_buf| decoder.finish(out_buf, true))?;
+
+        if self.stream_ended {
+            Ok(true)
+        } else {
+            Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                "zstd stream did not finish",
+            ))
+        }
     }
 }