From 17392431e14661ab346373dd8e06e39897048ba6 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:34:48 +0000 Subject: [PATCH 01/13] DecompressionStream(zstd): decode concatenated frames instead of rejecting as trailing junk A zstd stream is one or more concatenated frames (RFC 8878 3.1); pzstd output is multi-frame by construction, as is `cat a.zst b.zst`. DecompressionStream("zstd") threw ERR_TRAILING_JUNK_AFTER_STREAM_END at the second frame because the trailing-junk check added for deflate/brotli treats any input remaining after a frame boundary as junk. When the stream is zstd-decompress and opted in via rejectGarbageAfterEnd (only DecompressionStream does), re-enter the write loop with the remaining input so the native decoder starts the next frame. Actual trailing garbage then surfaces as ZSTD_error_prefix_unknown from zstd itself. The plain node:zlib zstd stream keeps stopping at the first frame to match Node.js. --- src/js/node/zlib.ts | 15 ++++++- test/js/web/streams/compression.test.ts | 59 +++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index fce51dbed631..130a618ad7e7 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -200,6 +200,11 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) { this._maxOutputLength = maxOutputLength; this._rejectGarbageAfterEnd = opts?.rejectGarbageAfterEnd === true; + // A zstd stream is one or more concatenated frames (RFC 8878 §3.1). + // DecompressionStream("zstd") opts in via rejectGarbageAfterEnd and must + // decode every frame; the plain node:zlib zstd stream keeps stopping at + // the first frame to match Node.js. See processCallback. + this._continueOnFrameEnd = this._rejectGarbageAfterEnd && mode === ZSTD_DECOMPRESS; } $toClass(ZlibBase, "ZlibBase", Transform); @@ -456,16 +461,22 @@ function processCallback() { return; } + // Zstd: a frame boundary with input remaining is the start of the next + // frame, not trailing junk. Re-enter the write loop so the native decoder + // starts that frame; actual garbage surfaces as ZSTD_error_prefix_unknown + // on that call. Only DecompressionStream opts in (see _continueOnFrameEnd). + const continueNextFrame = availInAfter > 0 && self._continueOnFrameEnd; + // Exhausted the output buffer, or used all the input create a new one. let chunkSize; - if (availOutAfter === 0 || self._outOffset >= (chunkSize = self._chunkSize)) { + if (availOutAfter === 0 || continueNextFrame || self._outOffset >= (chunkSize = self._chunkSize)) { chunkSize ??= self._chunkSize; handle.availOutBefore = chunkSize; self._outOffset = 0; self._outBuffer = Buffer.allocUnsafe(chunkSize); } - if (availOutAfter === 0) { + if (availOutAfter === 0 || continueNextFrame) { // Not actually done. Need to reprocess. // Also, update the availInBefore to the availInAfter value, // so that if we have to hit it a third (fourth, etc.) time, diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 08eed3af0e21..c4b0d78641f4 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,64 @@ 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"); + }); + + test("decompresses a multi-frame zstd stream split across writes", 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 ds = new DecompressionStream("zstd"); + const writer = ds.writable.getWriter(); + // Split in the middle of the second frame's header so the boundary is + // not a frame boundary. + const split = f1.length + 2; + writer.write(cat.subarray(0, split)); + writer.write(cat.subarray(split)); + writer.close(); + + const out = await new Response(ds.readable).text(); + expect(out).toBe("first frame\nsecond frame\n"); + }); + + 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("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.toThrow(); + }); }); describe("all formats", () => { From cee5fdd9950ec86288f1b629d6eb97d67a9ae556 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:37:10 +0000 Subject: [PATCH 02/13] [autofix.ci] apply automated fixes --- test/js/web/streams/compression.test.ts | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index c4b0d78641f4..da537f7a6c7d 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -227,9 +227,7 @@ describe("CompressionStream and DecompressionStream", () => { 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(); + const out = await new Response(new Blob([cat]).stream().pipeThrough(new DecompressionStream("zstd"))).text(); expect(out).toBe("first frame\nsecond frame\n"); }); @@ -259,9 +257,7 @@ describe("CompressionStream and DecompressionStream", () => { const cat = Buffer.concat(frames); const out = Buffer.from( - await new Response( - new Blob([cat]).stream().pipeThrough(new DecompressionStream("zstd")), - ).arrayBuffer(), + 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); @@ -271,9 +267,7 @@ describe("CompressionStream and DecompressionStream", () => { 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(); + const read = new Response(new Blob([withJunk]).stream().pipeThrough(new DecompressionStream("zstd"))).text(); await expect(read).rejects.toThrow(); }); }); From 16f16c47ec72731115c1369006d6d56b481af97e Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:40:15 +0000 Subject: [PATCH 03/13] trim comments flagged by comment-cop --- src/js/node/zlib.ts | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index 130a618ad7e7..5c5ac1fbc1e5 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -200,10 +200,8 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) { this._maxOutputLength = maxOutputLength; this._rejectGarbageAfterEnd = opts?.rejectGarbageAfterEnd === true; - // A zstd stream is one or more concatenated frames (RFC 8878 §3.1). - // DecompressionStream("zstd") opts in via rejectGarbageAfterEnd and must - // decode every frame; the plain node:zlib zstd stream keeps stopping at - // the first frame to match Node.js. See processCallback. + // zstd streams are one or more concatenated frames (RFC 8878 §3.1); only + // DecompressionStream opts in so plain createZstdDecompress matches Node.js. this._continueOnFrameEnd = this._rejectGarbageAfterEnd && mode === ZSTD_DECOMPRESS; } $toClass(ZlibBase, "ZlibBase", Transform); @@ -461,10 +459,7 @@ function processCallback() { return; } - // Zstd: a frame boundary with input remaining is the start of the next - // frame, not trailing junk. Re-enter the write loop so the native decoder - // starts that frame; actual garbage surfaces as ZSTD_error_prefix_unknown - // on that call. Only DecompressionStream opts in (see _continueOnFrameEnd). + // zstd: input remaining after a frame is the next frame, not trailing junk. const continueNextFrame = availInAfter > 0 && self._continueOnFrameEnd; // Exhausted the output buffer, or used all the input create a new one. From 7dfa011f5cccd6ec75e5a629724b2ffe976e44bb Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:41:46 +0000 Subject: [PATCH 04/13] condense zstd multi-frame comment to one line --- src/js/node/zlib.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index 5c5ac1fbc1e5..04dfbe352713 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -200,8 +200,7 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) { this._maxOutputLength = maxOutputLength; this._rejectGarbageAfterEnd = opts?.rejectGarbageAfterEnd === true; - // zstd streams are one or more concatenated frames (RFC 8878 §3.1); only - // DecompressionStream opts in so plain createZstdDecompress matches Node.js. + // zstd is multi-frame (RFC 8878 §3.1); scoped so createZstdDecompress stays Node-compatible. this._continueOnFrameEnd = this._rejectGarbageAfterEnd && mode === ZSTD_DECOMPRESS; } $toClass(ZlibBase, "ZlibBase", Transform); From 0c375bfeddae37e5a715a3980927f4febabdf670 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:51:17 +0000 Subject: [PATCH 05/13] assert ZSTD_error_prefix_unknown for trailing-garbage test --- test/js/web/streams/compression.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index da537f7a6c7d..74889970da1c 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -268,7 +268,7 @@ describe("CompressionStream and DecompressionStream", () => { 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.toThrow(); + await expect(read).rejects.toMatchObject({ code: "ZSTD_error_prefix_unknown" }); }); }); From 1ad660a66780b7f4c91d07e1d9558337eb42e33d Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 22:13:29 +0000 Subject: [PATCH 06/13] move multi-frame handling to native do_work with magic-byte peek Peek the 4 bytes following a completed frame: the zstd frame magic 0xFD2FB528 or a skippable-frame magic 0x184D2A5? means another frame follows, so call ZSTD_decompressStream again (mirrors the GUNZIP multi-member loop). A 1-3 byte tail that is a prefix of either magic is fed through so a header split across chunks decodes; anything else is left unconsumed so the JS layer's ERR_TRAILING_JUNK_AFTER_STREAM_END check still fires for DecompressionStream and createZstdDecompress keeps silently ignoring trailing data like Node.js. Drops the JS-side _continueOnFrameEnd gate; the peek makes it unnecessary. Adds coverage for a leading/trailing skippable frame. --- src/js/node/zlib.ts | 9 +--- src/runtime/node/zlib/NativeZstd.rs | 69 ++++++++++++++++++++++--- test/js/web/streams/compression.test.ts | 16 +++++- 3 files changed, 79 insertions(+), 15 deletions(-) diff --git a/src/js/node/zlib.ts b/src/js/node/zlib.ts index 04dfbe352713..fce51dbed631 100644 --- a/src/js/node/zlib.ts +++ b/src/js/node/zlib.ts @@ -200,8 +200,6 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) { this._maxOutputLength = maxOutputLength; this._rejectGarbageAfterEnd = opts?.rejectGarbageAfterEnd === true; - // zstd is multi-frame (RFC 8878 §3.1); scoped so createZstdDecompress stays Node-compatible. - this._continueOnFrameEnd = this._rejectGarbageAfterEnd && mode === ZSTD_DECOMPRESS; } $toClass(ZlibBase, "ZlibBase", Transform); @@ -458,19 +456,16 @@ function processCallback() { return; } - // zstd: input remaining after a frame is the next frame, not trailing junk. - const continueNextFrame = availInAfter > 0 && self._continueOnFrameEnd; - // Exhausted the output buffer, or used all the input create a new one. let chunkSize; - if (availOutAfter === 0 || continueNextFrame || self._outOffset >= (chunkSize = self._chunkSize)) { + if (availOutAfter === 0 || self._outOffset >= (chunkSize = self._chunkSize)) { chunkSize ??= self._chunkSize; handle.availOutBefore = chunkSize; self._outOffset = 0; self._outBuffer = Buffer.allocUnsafe(chunkSize); } - if (availOutAfter === 0 || continueNextFrame) { + if (availOutAfter === 0) { // Not actually done. Need to reprocess. // Also, update the availInBefore to the availInAfter value, // so that if we have to hit it a third (fourth, etc.) time, diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 8ca905cf12f4..82b833f5b5ab 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -491,6 +491,42 @@ mod _impl { self.flush = flush; } + const ZSTD_MAGICNUMBER: u32 = 0xFD2FB528; + const ZSTD_MAGIC_SKIPPABLE_START: u32 = 0x184D2A50; + const ZSTD_MAGIC_SKIPPABLE_MASK: u32 = 0xFFFFFFF0; + + /// A zstd stream is one or more concatenated frames (RFC 8878 §3.1). + /// After `ZSTD_decompressStream` returns 0 (frame complete) with input + /// remaining, decide whether those bytes begin another frame so the + /// caller can loop. Bytes that are not a frame/skippable magic (or a + /// prefix of one when fewer than 4 remain) are left unconsumed so the + /// JS layer can treat them as trailing data. + fn next_input_is_frame(&self) -> bool { + let avail = self.input.size - self.input.pos; + if avail == 0 { + return false; + } + // SAFETY: `input.src[..input.size]` is the caller-kept-alive input + // slice installed by `set_buffers`; `input.pos < input.size` here. + let rest = unsafe { + core::slice::from_raw_parts( + self.input.src.cast::().add(self.input.pos), + avail.min(4), + ) + }; + let frame = Self::ZSTD_MAGICNUMBER.to_le_bytes(); + let skippable = Self::ZSTD_MAGIC_SKIPPABLE_START.to_le_bytes(); + if rest.len() < 4 { + // The low nibble of a skippable magic's first byte varies, so + // compare it under the mask. + return rest == &frame[..rest.len()] + || (rest[0] & 0xF0 == skippable[0] && rest[1..] == skippable[1..rest.len()]); + } + let magic = u32::from_le_bytes([rest[0], rest[1], rest[2], rest[3]]); + magic == Self::ZSTD_MAGICNUMBER + || magic & Self::ZSTD_MAGIC_SKIPPABLE_MASK == Self::ZSTD_MAGIC_SKIPPABLE_START + } + pub fn do_work(&mut self) { // A handle driven before `init()` has no CCtx/DCtx; zstd // dereferences the context pointer unconditionally. @@ -508,13 +544,32 @@ 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 GUNZIP multi-member + // handling in NativeZlib::do_work_inflate; libzstd resets + // the DCtx to start-of-frame itself so no explicit reset. + 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/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 74889970da1c..594bdd4a68fa 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -263,12 +263,26 @@ describe("CompressionStream and DecompressionStream", () => { expect(out.equals(Buffer.alloc(piece.length * 64, "Z"))).toBe(true); }); + test("decompresses a zstd stream with a leading skippable frame", async () => { + // 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]), + ]); + 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: "ZSTD_error_prefix_unknown" }); + await expect(read).rejects.toMatchObject({ code: "ERR_TRAILING_JUNK_AFTER_STREAM_END" }); }); }); From dee17862e396010c1e3aa01379368cc61b5e88f0 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 22:15:43 +0000 Subject: [PATCH 07/13] cargo fmt --- src/runtime/node/zlib/NativeZstd.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 82b833f5b5ab..d10c03314d09 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -570,7 +570,7 @@ mod _impl { }; } ret - }, + } _ => unreachable!(), } as u64; } From 5cba1cb299d5a7034a455727d64948eae75492f2 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 22:17:01 +0000 Subject: [PATCH 08/13] trim comments flagged by comment-cop --- src/runtime/node/zlib/NativeZstd.rs | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index d10c03314d09..f10f4e2d6d30 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -495,19 +495,15 @@ mod _impl { const ZSTD_MAGIC_SKIPPABLE_START: u32 = 0x184D2A50; const ZSTD_MAGIC_SKIPPABLE_MASK: u32 = 0xFFFFFFF0; - /// A zstd stream is one or more concatenated frames (RFC 8878 §3.1). - /// After `ZSTD_decompressStream` returns 0 (frame complete) with input - /// remaining, decide whether those bytes begin another frame so the - /// caller can loop. Bytes that are not a frame/skippable magic (or a - /// prefix of one when fewer than 4 remain) are left unconsumed so the - /// JS layer can treat them as trailing data. + /// RFC 8878 §3.1: a zstd stream is one or more concatenated frames. + /// True when the unconsumed input begins a zstd or skippable frame + /// magic (or a prefix of one when fewer than 4 bytes remain). fn next_input_is_frame(&self) -> bool { let avail = self.input.size - self.input.pos; if avail == 0 { return false; } - // SAFETY: `input.src[..input.size]` is the caller-kept-alive input - // slice installed by `set_buffers`; `input.pos < input.size` here. + // SAFETY: input.src[..input.size] is the caller-kept-alive slice from set_buffers; pos < size here. let rest = unsafe { core::slice::from_raw_parts( self.input.src.cast::().add(self.input.pos), @@ -517,8 +513,6 @@ mod _impl { let frame = Self::ZSTD_MAGICNUMBER.to_le_bytes(); let skippable = Self::ZSTD_MAGIC_SKIPPABLE_START.to_le_bytes(); if rest.len() < 4 { - // The low nibble of a skippable magic's first byte varies, so - // compare it under the mask. return rest == &frame[..rest.len()] || (rest[0] & 0xF0 == skippable[0] && rest[1..] == skippable[1..rest.len()]); } @@ -553,9 +547,7 @@ mod _impl { &raw mut self.input, ) }; - // ret == 0 is frame-complete. Mirrors GUNZIP multi-member - // handling in NativeZlib::do_work_inflate; libzstd resets - // the DCtx to start-of-frame itself so no explicit reset. + // 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() From 139e5529e8f51919593f93ac88b1fce5bb5a0445 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 22:18:43 +0000 Subject: [PATCH 09/13] one-line doc comment for next_input_is_frame --- src/runtime/node/zlib/NativeZstd.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index f10f4e2d6d30..0fbcda7f0adb 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -495,9 +495,7 @@ mod _impl { const ZSTD_MAGIC_SKIPPABLE_START: u32 = 0x184D2A50; const ZSTD_MAGIC_SKIPPABLE_MASK: u32 = 0xFFFFFFF0; - /// RFC 8878 §3.1: a zstd stream is one or more concatenated frames. - /// True when the unconsumed input begins a zstd or skippable frame - /// magic (or a prefix of one when fewer than 4 bytes remain). + /// True when the unconsumed input begins a zstd/skippable frame magic (or a prefix of one). fn next_input_is_frame(&self) -> bool { let avail = self.input.size - self.input.pos; if avail == 0 { From 021aa84c8e531c1806be2373e4cb4b388497e12d Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 27 Jul 2026 05:42:04 +0000 Subject: [PATCH 10/13] simplify next_input_is_frame: single copy_nonoverlapping into [u8; 4], one comparison path --- src/runtime/node/zlib/NativeZstd.rs | 33 ++++++++++++----------------- 1 file changed, 14 insertions(+), 19 deletions(-) diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 0fbcda7f0adb..e3d99cc06c0a 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -491,32 +491,27 @@ mod _impl { self.flush = flush; } - const ZSTD_MAGICNUMBER: u32 = 0xFD2FB528; - const ZSTD_MAGIC_SKIPPABLE_START: u32 = 0x184D2A50; - const ZSTD_MAGIC_SKIPPABLE_MASK: u32 = 0xFFFFFFF0; + 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 avail = self.input.size - self.input.pos; - if avail == 0 { + let n = (self.input.size - self.input.pos).min(4); + if n == 0 { return false; } - // SAFETY: input.src[..input.size] is the caller-kept-alive slice from set_buffers; pos < size here. - let rest = unsafe { - core::slice::from_raw_parts( + 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), - avail.min(4), - ) - }; - let frame = Self::ZSTD_MAGICNUMBER.to_le_bytes(); - let skippable = Self::ZSTD_MAGIC_SKIPPABLE_START.to_le_bytes(); - if rest.len() < 4 { - return rest == &frame[..rest.len()] - || (rest[0] & 0xF0 == skippable[0] && rest[1..] == skippable[1..rest.len()]); + head.as_mut_ptr(), + n, + ); } - let magic = u32::from_le_bytes([rest[0], rest[1], rest[2], rest[3]]); - magic == Self::ZSTD_MAGICNUMBER - || magic & Self::ZSTD_MAGIC_SKIPPABLE_MASK == Self::ZSTD_MAGIC_SKIPPABLE_START + 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) { From f62fd6a24115a89d3c03275a4387081a287933ca Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 27 Jul 2026 05:47:33 +0000 Subject: [PATCH 11/13] test: cover skippable-magic prefix at split offsets 1/2/3, await writer ops --- test/js/web/streams/compression.test.ts | 44 +++++++++++++------------ 1 file changed, 23 insertions(+), 21 deletions(-) diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 594bdd4a68fa..8f9081e3d87a 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -231,22 +231,30 @@ describe("CompressionStream and DecompressionStream", () => { expect(out).toBe("first frame\nsecond frame\n"); }); - test("decompresses a multi-frame zstd stream split across writes", async () => { + // 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 f2 = zlib.zstdCompressSync(Buffer.from("second frame\n")); - const cat = Buffer.concat([f1, f2]); - - const ds = new DecompressionStream("zstd"); - const writer = ds.writable.getWriter(); - // Split in the middle of the second frame's header so the boundary is - // not a frame boundary. - const split = f1.length + 2; - writer.write(cat.subarray(0, split)); - writer.write(cat.subarray(split)); - writer.close(); - - const out = await new Response(ds.readable).text(); - expect(out).toBe("first frame\nsecond 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 () => { @@ -264,12 +272,6 @@ describe("CompressionStream and DecompressionStream", () => { }); test("decompresses a zstd stream with a leading skippable frame", async () => { - // 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]), - ]); const frame = zlib.zstdCompressSync(Buffer.from("payload")); const cat = Buffer.concat([skippable, frame, skippable]); From f173c561ec4a2a264ff946afed0a1f75e6b59ee9 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Mon, 27 Jul 2026 05:49:45 +0000 Subject: [PATCH 12/13] [autofix.ci] apply automated fixes --- test/js/web/streams/compression.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 8f9081e3d87a..adca372f0813 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -240,7 +240,11 @@ describe("CompressionStream and DecompressionStream", () => { 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"], + [ + "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]); From f5a1ac96bf36822b45bd31bf0674ccbd1d687b90 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 27 Jul 2026 05:57:18 +0000 Subject: [PATCH 13/13] test: pin zstdDecompressSync multi-frame behavior in node:zlib suite --- test/js/node/zlib/zlib.test.js | 6 ++++++ 1 file changed, 6 insertions(+) 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)) {