diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 8ca905cf12f4..e3d99cc06c0a 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/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::().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; } diff --git a/test/js/node/zlib/zlib.test.js b/test/js/node/zlib/zlib.test.js index 2c0e172128a5..43df8054569f 100644 --- a/test/js/node/zlib/zlib.test.js +++ b/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)) { diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 08eed3af0e21..adca372f0813 100644 --- a/test/js/web/streams/compression.test.ts +++ b/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", () => {