DecompressionStream("zstd"): decode concatenated frames instead of rejecting as trailing junk

src/runtime/node/zlib/NativeZstd.rs

@@ -491,6 +491,29 @@ mod _impl {
             self.flush = flush;
         }
 
+        const ZSTD_MAGICNUMBER: [u8; 4] = 0xFD2FB528u32.to_le_bytes();
+        const ZSTD_MAGIC_SKIPPABLE: [u8; 4] = 0x184D2A50u32.to_le_bytes();
+
+        /// True when the unconsumed input begins a zstd/skippable frame magic (or a prefix of one).
+        fn next_input_is_frame(&self) -> bool {
+            let n = (self.input.size - self.input.pos).min(4);
+            if n == 0 {
+                return false;
+            }
+            let mut head = [0u8; 4];
+            // SAFETY: input.src[pos..pos+n] lies within the slice installed by set_buffers.
+            unsafe {
+                ptr::copy_nonoverlapping(
+                    self.input.src.cast::<u8>().add(self.input.pos),
+                    head.as_mut_ptr(),
+                    n,
+                );
+            }
+            head[..n] == Self::ZSTD_MAGICNUMBER[..n]
+                || (head[0] & 0xF0 == Self::ZSTD_MAGIC_SKIPPABLE[0]
+                    && head[1..n] == Self::ZSTD_MAGIC_SKIPPABLE[1..n])
+        }
+
         pub fn do_work(&mut self) {
             // A handle driven before `init()` has no CCtx/DCtx; zstd
             // dereferences the context pointer unconditionally.
@@ -508,14 +531,31 @@ mod _impl {
                         self.flush as c_uint,
                     )
                 },
-                // SAFETY: state is a valid DCtx.
-                NodeMode::ZSTD_DECOMPRESS => unsafe {
-                    c::ZSTD_decompressStream(
-                        self.state_ptr().cast(),
-                        &raw mut self.output,
-                        &raw mut self.input,
-                    )
-                },
+                NodeMode::ZSTD_DECOMPRESS => {
+                    // SAFETY: state is a valid DCtx.
+                    let mut ret = unsafe {
+                        c::ZSTD_decompressStream(
+                            self.state_ptr().cast(),
+                            &raw mut self.output,
+                            &raw mut self.input,
+                        )
+                    };
+                    // ret == 0 is frame-complete; mirrors NativeZlib::do_work_inflate's GUNZIP loop.
+                    while ret == 0
+                        && self.output.pos < self.output.size
+                        && self.next_input_is_frame()
+                    {
+                        // SAFETY: state is a valid DCtx; input/output point to caller-kept-alive buffers.
+                        ret = unsafe {
+                            c::ZSTD_decompressStream(
+                                self.state_ptr().cast(),
+                                &raw mut self.output,
+                                &raw mut self.input,
+                            )
+                        };
+                    }
+                    ret
+                }
                 _ => unreachable!(),
             } as u64;
         }

test/js/node/zlib/zlib.test.js

@@ -566,6 +566,12 @@ describe("zlib.zstd", () => {
     expect(roundtrip.toString()).toEqual(inputString);
   });
 
+  it("zstdDecompressSync decodes concatenated frames", () => {
+    const f1 = zlib.zstdCompressSync(Buffer.from("first\n"));
+    const f2 = zlib.zstdCompressSync(Buffer.from("second\n"));
+    expect(zlib.zstdDecompressSync(Buffer.concat([f1, f2])).toString()).toBe("first\nsecond\n");
+  });
+
   it("can compress streaming", async () => {
     const encoder = zlib.createZstdCompress();
     for (const chunk of window(inputString, 55)) {

test/js/web/streams/compression.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, test } from "bun:test";
+import zlib from "node:zlib";
 
 describe("CompressionStream and DecompressionStream", () => {
   describe("brotli", () => {
@@ -217,6 +218,78 @@ describe("CompressionStream and DecompressionStream", () => {
         expect(output).toBe(input);
       }
     });
+
+    // RFC 8878 §3.1: a zstd stream is one or more concatenated frames; a
+    // decoder MUST decode each in order. pzstd output is multi-frame by
+    // construction, as is `cat a.zst b.zst`.
+    test("decompresses a multi-frame zstd stream", async () => {
+      const f1 = zlib.zstdCompressSync(Buffer.from("first frame\n"));
+      const f2 = zlib.zstdCompressSync(Buffer.from("second frame\n"));
+      const cat = Buffer.concat([f1, f2]);
+
+      const out = await new Response(new Blob([cat]).stream().pipeThrough(new DecompressionStream("zstd"))).text();
+      expect(out).toBe("first frame\nsecond frame\n");
+    });
+
+    // Skippable frame: magic 0x184D2A50..5F, 4-byte LE size, then size bytes.
+    const skippable = Buffer.concat([
+      Buffer.from([0x55, 0x2a, 0x4d, 0x18]), // magic (variant 5)
+      Buffer.from([0x03, 0x00, 0x00, 0x00]), // size = 3
+      Buffer.from([0xaa, 0xbb, 0xcc]),
+    ]);
+
+    test.each([
+      ["zstd frame", zlib.zstdCompressSync(Buffer.from("second frame\n")), "first frame\nsecond frame\n"],
+      [
+        "skippable frame",
+        Buffer.concat([skippable, zlib.zstdCompressSync(Buffer.from("second frame\n"))]),
+        "first frame\nsecond frame\n",
+      ],
+    ] as const)("decompresses a multi-frame zstd stream split across writes (next = %s)", async (_, next, expected) => {
+      const f1 = zlib.zstdCompressSync(Buffer.from("first frame\n"));
+      const cat = Buffer.concat([f1, next]);
+
+      for (const offset of [1, 2, 3]) {
+        const ds = new DecompressionStream("zstd");
+        const writer = ds.writable.getWriter();
+        const read = new Response(ds.readable).text();
+        const split = f1.length + offset;
+        await writer.write(cat.subarray(0, split));
+        await writer.write(cat.subarray(split));
+        await writer.close();
+        expect(await read).toBe(expected);
+      }
+    });
+
+    test("decompresses many concatenated zstd frames larger than one output chunk", async () => {
+      const piece = Buffer.alloc(4096, "Z");
+      const frame = zlib.zstdCompressSync(piece);
+      const frames: Buffer[] = [];
+      for (let i = 0; i < 64; i++) frames.push(frame);
+      const cat = Buffer.concat(frames);
+
+      const out = Buffer.from(
+        await new Response(new Blob([cat]).stream().pipeThrough(new DecompressionStream("zstd"))).arrayBuffer(),
+      );
+      expect(out.length).toBe(piece.length * 64);
+      expect(out.equals(Buffer.alloc(piece.length * 64, "Z"))).toBe(true);
+    });
+
+    test("decompresses a zstd stream with a leading skippable frame", async () => {
+      const frame = zlib.zstdCompressSync(Buffer.from("payload"));
+      const cat = Buffer.concat([skippable, frame, skippable]);
+
+      const out = await new Response(new Blob([cat]).stream().pipeThrough(new DecompressionStream("zstd"))).text();
+      expect(out).toBe("payload");
+    });
+
+    test("rejects trailing garbage after a zstd frame", async () => {
+      const frame = zlib.zstdCompressSync(Buffer.from("hello"));
+      const withJunk = Buffer.concat([frame, Buffer.from([0xde, 0xad, 0xbe, 0xef])]);
+
+      const read = new Response(new Blob([withJunk]).stream().pipeThrough(new DecompressionStream("zstd"))).text();
+      await expect(read).rejects.toMatchObject({ code: "ERR_TRAILING_JUNK_AFTER_STREAM_END" });
+    });
   });
 
   describe("all formats", () => {