fix(stream): detect truncated frames; drop per-byte allocations

ZstdInputStream silently returned clean EOF when the underlying stream
ended mid-frame, losing data without error. Track zstd's outstanding-input
hint from decompressStream and throw ZstdException on a non-zero hint at
EOF. Empty input stays a clean EOF.

Also reuse a single-byte scratch buffer in read()/write(int) instead of
allocating new byte[1] per call (streams are already non-thread-safe).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dfa1

zstd/src/main/java/io/github/dfa1/zstd/ZstdInputStream.java

@@ -35,10 +35,14 @@ public final class ZstdInputStream extends InputStream {
     private final ZstdStreamBuffer outBufView = new ZstdStreamBuffer(arena);
     private final byte[] feed;
     private final byte[] hold;
+    private final byte[] single = new byte[1];
     private int holdStart;
     private int holdEnd;
     private boolean inputEof;
     private boolean closed;
+    /// zstd's "bytes still expected" hint from the last decompressStream call;
+    /// non-zero at EOF means the final frame was truncated.
+    private long lastHint;
 
     /// Wraps `in`, decompressing the zstd frame it carries.
     ///
@@ -84,9 +88,8 @@ private void loadDictionary(ZstdDictionary dictionary) {
 
     @Override
     public int read() throws IOException {
-        byte[] one = new byte[1];
-        int n = read(one, 0, 1);
-        return n == -1 ? -1 : (one[0] & 0xFF);
+        int n = read(single, 0, 1);
+        return n == -1 ? -1 : (single[0] & 0xFF);
     }
 
     @Override
@@ -112,13 +115,20 @@ private boolean produce() throws IOException {
                 int r = inputEof ? -1 : in.read(feed);
                 if (r == -1) {
                     inputEof = true;
+                    // The frame boundary is clean only when the last decompressStream
+                    // call reported nothing outstanding; otherwise the stream was cut
+                    // mid-frame and the remaining bytes are lost.
+                    if (lastHint != 0) {
+                        throw new ZstdException("truncated zstd stream: " + lastHint
+                                + " more input byte(s) expected");
+                    }
                     return false;
                 }
                 MemorySegment.copy(feed, 0, inSeg, JAVA_BYTE, 0, r);
                 inBuf.set(inSeg, r, 0);
             }
             outBufView.set(outSeg, outCap, 0);
-            Zstd.call(() -> (long) Bindings.DECOMPRESS_STREAM.invokeExact(
+            lastHint = Zstd.call(() -> (long) Bindings.DECOMPRESS_STREAM.invokeExact(
                     dctx, outBufView.segment(), inBuf.segment()));
             int produced = Math.toIntExact(outBufView.pos());
             if (produced > 0) {
@@ -127,10 +137,17 @@ private boolean produce() throws IOException {
                 holdEnd = produced;
                 return true;
             }
-            // nothing produced: if input is exhausted and the underlying stream is
-            // at EOF, there is no more output to come.
-            if (inputEof && inBuf.pos() == inBuf.size()) {
-                return false;
+            // Nothing produced. If the decoder neither advanced its input nor wants
+            // more, it cannot make progress on this input — stop to avoid spinning.
+            if (inBuf.pos() == inBuf.size()) {
+                if (inputEof) {
+                    if (lastHint != 0) {
+                        throw new ZstdException("truncated zstd stream: " + lastHint
+                                + " more input byte(s) expected");
+                    }
+                    return false;
+                }
+                // input drained but frame wants more: loop to refill from `in`.
             }
         }
     }

zstd/src/main/java/io/github/dfa1/zstd/ZstdOutputStream.java

@@ -41,6 +41,7 @@ public final class ZstdOutputStream extends OutputStream {
     private final ZstdStreamBuffer in = new ZstdStreamBuffer(arena);
     private final ZstdStreamBuffer outBuf = new ZstdStreamBuffer(arena);
     private final byte[] drain;
+    private final byte[] single = new byte[1];
     private boolean closed;
 
     /// Wraps `out`, compressing at the library default level.
@@ -124,7 +125,8 @@ private void loadDictionary(ZstdDictionary dictionary) {
 
     @Override
     public void write(int b) throws IOException {
-        write(new byte[]{(byte) b}, 0, 1);
+        single[0] = (byte) b;
+        write(single, 0, 1);
     }
 
     @Override

zstd/src/test/java/io/github/dfa1/zstd/ZstdStreamTest.java

@@ -12,6 +12,7 @@
 import java.util.Random;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class ZstdStreamTest {
 
@@ -148,6 +149,37 @@ private byte[] record(int i) {
         }
     }
 
+    @Nested
+    class Truncation {
+
+        @ParameterizedTest
+        @ValueSource(ints = {1, 8, 64})
+        void throwsWhenFinalFrameIsCutShort(int bytesDropped) throws IOException {
+            // Given a valid frame with its tail bytes lopped off (random data so the
+            // frame stays large enough to drop bytes from)
+            byte[] original = randomBytes(50_000);
+            byte[] frame = streamCompress(original, 6);
+            byte[] cut = java.util.Arrays.copyOf(frame, frame.length - bytesDropped);
+
+            // When the streaming decompressor drains it
+            // Then it reports the truncation instead of returning a clean EOF
+            try (ZstdInputStream zin = new ZstdInputStream(new ByteArrayInputStream(cut))) {
+                assertThatThrownBy(zin::readAllBytes)
+                        .isInstanceOf(ZstdException.class)
+                        .hasMessageContaining("truncated");
+            }
+        }
+
+        @Test
+        void emptyInputIsCleanEofNotTruncation() throws IOException {
+            // Given no input at all
+            // When read, it is end-of-stream, not an error
+            try (ZstdInputStream zin = new ZstdInputStream(new ByteArrayInputStream(new byte[0]))) {
+                assertThat(zin.read()).isEqualTo(-1);
+            }
+        }
+    }
+
     @Nested
     class PledgedSize {