From 52f6ddede4162d74f36a27e4fcc2cc40a81e26d7 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 27 Sep 2026 16:21:38 +0000 Subject: [PATCH] node:zlib: zstd decoder follows Node v26.10.0 across writes and at the end of the input The zstd decoder keeps Node's frame state between writes, and reads it in two places. After a complete frame, input that cannot start a frame ends the stream. The decoder ignores that input and all later input. Before, this was true only when the input came in the same write as the end of the frame and the output chunk had room. In a later write the same input raised "Unknown frame descriptor" (ZSTD_error_prefix_unknown). When a decoder finishes with ZSTD_e_end and the input stops inside a frame, the decoder reports "unexpected end of file" (Z_BUF_ERROR, errno -5). Before, it reported success, and the caller got a short or an empty result. Node does this since v26.6.0 and v24.20.0. finishFlush ZSTD_e_flush or ZSTD_e_continue turns the check off. Fewer than 4 bytes after a complete frame do not show that a frame began, so they are not an error. The second part is the opposite of a maintainer's request of 2026-07-06 on #33413 ("Scope the strictness to DecompressionStream"). Node did not have the check then. Both parts are a port of ZstdDecompressContext in Node v26.10.0 (DoThreadPoolWork and GetErrorInfo), without rejectGarbageAfterEnd. The state fits the tail padding of Context, which stays 88 bytes. init() clears it, so reset() clears it too. test/js/node/zlib/zstd-frames.test.ts uses node:test and node:assert. The same file passes under Node v26.10.0. test-zlib-truncated.js and test-stream-iter-transform-sync.js are the copies from Node v26.10.0. --- src/runtime/node/zlib/NativeZstd.rs | 159 ++++- src/zstd/lib.rs | 1 + .../test-stream-iter-transform-sync.js | 11 + .../node/test/parallel/test-zlib-truncated.js | 78 ++- test/js/node/zlib/zlib.test.js | 5 +- test/js/node/zlib/zstd-frames.test.ts | 570 ++++++++++++++++++ 6 files changed, 771 insertions(+), 53 deletions(-) create mode 100644 test/js/node/zlib/zstd-frames.test.ts diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 8a56248734c2..e13c6de4495f 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -300,6 +300,16 @@ mod _impl { } } + bitflags::bitflags! { + /// The fields of the same name in node's `ZstdDecompressContext`. + #[derive(Clone, Copy, Default, PartialEq, Eq)] + pub(crate) struct DecodeState: u8 { + const FRAME_COMPLETE = 1 << 0; + const DECODING_FRAME_AFTER_COMPLETE = 1 << 1; + const IGNORING_TRAILING_INPUT = 1 << 2; + } + } + pub(crate) struct Context { pub(crate) mode: NodeMode, // LIFETIMES.tsv: FFI → Option<*mut c_void> (ZSTD_createCCtx/DCtx; freed in deinit_state) @@ -309,8 +319,16 @@ mod _impl { pub(crate) output: c::ZSTD_outBuffer, pub(crate) pledged_src_size: u64, pub(crate) remaining: u64, + pub(crate) decode: DecodeState, + /// Magic-number bytes seen of the frame after a complete one, counted across writes up to 4. + pub(crate) frame_prefix_size: u8, + /// `ZSTD_FRAME | SKIPPABLE_FRAME`, less each type that those bytes rule out. + pub(crate) possible_frame_types: u8, } + // The decode state fits the tail padding of `Context`, so a handle does not grow. + const _: () = assert!(core::mem::size_of::() == 88); + impl Default for Context { fn default() -> Self { Self { @@ -329,6 +347,9 @@ mod _impl { }, pledged_src_size: u64::MAX, remaining: 0, + decode: DecodeState::empty(), + frame_prefix_size: 0, + possible_frame_types: 0, } } } @@ -360,6 +381,9 @@ mod _impl { if self.state.is_some() { self.deinit_state(); } + self.decode = DecodeState::empty(); + self.frame_prefix_size = 0; + self.possible_frame_types = 0; match self.mode { NodeMode::ZSTD_COMPRESS => { self.pledged_src_size = pledged_src_size; @@ -505,12 +529,15 @@ mod _impl { const ZSTD_MAGICNUMBER: [u8; 4] = 0xFD2FB528u32.to_le_bytes(); const ZSTD_MAGIC_SKIPPABLE: [u8; 4] = 0x184D2A50u32.to_le_bytes(); + const ZSTD_FRAME: u8 = 1 << 0; + const SKIPPABLE_FRAME: u8 = 1 << 1; - /// 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); + /// Reads up to 4 bytes of the input as the next bytes of a frame magic number, and consumes none. + fn classify_frame_prefix(&mut self) { + let seen = usize::from(self.frame_prefix_size); + let n = (self.input.size - self.input.pos).min(4 - seen); if n == 0 { - return false; + return; } let mut head = [0u8; 4]; // SAFETY: input.src[pos..pos+n] lies within the slice installed by set_buffers. @@ -521,9 +548,70 @@ mod _impl { n, ); } - head[..n] == Self::ZSTD_MAGICNUMBER[..n] - || (head[0] & 0xF0 == Self::ZSTD_MAGIC_SKIPPABLE[0] - && head[1..n] == Self::ZSTD_MAGIC_SKIPPABLE[1..n]) + for (index, &byte) in (seen..).zip(&head[..n]) { + if byte != Self::ZSTD_MAGICNUMBER[index] { + self.possible_frame_types &= !Self::ZSTD_FRAME; + } + // The low 4 bits of the first byte of a skippable magic number are free. + let skippable = if index == 0 { + byte & 0xF0 == Self::ZSTD_MAGIC_SKIPPABLE[0] + } else { + byte == Self::ZSTD_MAGIC_SKIPPABLE[index] + }; + if !skippable { + self.possible_frame_types &= !Self::SKIPPABLE_FRAME; + } + } + self.frame_prefix_size += n as u8; + } + + /// Node's `ZstdDecompressContext::DoThreadPoolWork` without `rejectGarbageAfterEnd`: https://github.com/nodejs/node/blob/v26.10.0/src/node_zlib.cc#L1841-L1911 + fn do_work_decompress(&mut self) -> usize { + if self.decode.contains(DecodeState::IGNORING_TRAILING_INPUT) { + return 0; + } + // The JS drivers write again with no input after a full output buffer, and that write starts no frame. + if self.decode.contains(DecodeState::FRAME_COMPLETE) && self.input.size == 0 { + return 0; + } + loop { + if self.decode.contains(DecodeState::FRAME_COMPLETE) { + self.decode + .insert(DecodeState::DECODING_FRAME_AFTER_COMPLETE); + self.frame_prefix_size = 0; + self.possible_frame_types = Self::ZSTD_FRAME | Self::SKIPPABLE_FRAME; + } + if self + .decode + .contains(DecodeState::DECODING_FRAME_AFTER_COMPLETE) + && self.frame_prefix_size < 4 + { + self.classify_frame_prefix(); + if self.possible_frame_types == 0 { + // The stream ended with the last frame: this input and all later input stay unconsumed. + self.decode = + DecodeState::FRAME_COMPLETE | DecodeState::IGNORING_TRAILING_INPUT; + return 0; + } + } + // SAFETY: state is a valid DCtx; input/output point to caller-kept-alive buffers (set_buffers). + let ret = unsafe { + c::ZSTD_decompressStream( + self.state_ptr().cast(), + &raw mut self.output, + &raw mut self.input, + ) + }; + // 0 is frame-complete. A size hint and an error code are both nonzero. + if ret != 0 { + self.decode.remove(DecodeState::FRAME_COMPLETE); + return ret; + } + self.decode = DecodeState::FRAME_COMPLETE; + if self.input.pos == self.input.size || self.output.pos == self.output.size { + return 0; + } + } } pub(crate) fn do_work(&mut self) { @@ -543,31 +631,7 @@ mod _impl { self.flush as c_uint, ) }, - 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 - } + NodeMode::ZSTD_DECOMPRESS => self.do_work_decompress(), _ => unreachable!(), } as u64; } @@ -577,11 +641,40 @@ mod _impl { *avail_out = u32::try_from(self.output.size - self.output.pos).expect("int cast"); } + /// Node's `ZstdDecompressContext::GetErrorInfo` without `rejectGarbageAfterEnd`: https://github.com/nodejs/node/blob/v26.10.0/src/node_zlib.cc#L1913-L1937 + fn input_ended_inside_frame(&self) -> bool { + // A compressor keeps no frame state, and a handle that `init()` did not set up decoded nothing. + if self.mode != NodeMode::ZSTD_DECOMPRESS || self.state.is_none() { + return false; + } + // With a full output buffer the caller writes again, and that write gets the result. + if self.flush != c::ZSTD_e_end as c_int + || self.decode.contains(DecodeState::FRAME_COMPLETE) + || self.input.pos != self.input.size + || self.output.pos == self.output.size + { + return false; + } + // Fewer than 4 bytes after a complete frame do not show that a frame began, so node ignores them. + !(self + .decode + .contains(DecodeState::DECODING_FRAME_AFTER_COMPLETE) + && self.frame_prefix_size < 4) + } + pub(crate) fn get_error_info(&mut self) -> Error { // Compute result, then clear `remaining`, then return. let err = c::ZSTD_getErrorCode(self.remaining as usize); let result = if err == 0 { - Error::OK + if self.input_ended_inside_frame() { + Error::init( + c"unexpected end of file".as_ptr(), + bun_zlib::ReturnCode::BufError as c_int, + c"Z_BUF_ERROR".as_ptr(), + ) + } else { + Error::OK + } } else { Error { err: err as c_int, diff --git a/src/zstd/lib.rs b/src/zstd/lib.rs index 974a545f386d..e3fc88631b75 100644 --- a/src/zstd/lib.rs +++ b/src/zstd/lib.rs @@ -33,6 +33,7 @@ pub mod c { // ZSTD_EndDirective pub const ZSTD_e_continue: ZSTD_EndDirective = 0; + pub const ZSTD_e_end: ZSTD_EndDirective = 2; // ZSTD_cParameter pub const ZSTD_c_compressionLevel: ZSTD_cParameter = 100; diff --git a/test/js/node/test/parallel/test-stream-iter-transform-sync.js b/test/js/node/test/parallel/test-stream-iter-transform-sync.js index d674c26cca10..991dba2748ed 100644 --- a/test/js/node/test/parallel/test-stream-iter-transform-sync.js +++ b/test/js/node/test/parallel/test-stream-iter-transform-sync.js @@ -19,6 +19,7 @@ const { decompressBrotliSync, decompressZstdSync, } = require('zlib/iter'); +const { zstdCompressSync } = require('zlib'); // ============================================================================= // Helper: sync compress then decompress, verify round-trip equality @@ -118,6 +119,15 @@ function testZstdLargeData() { assert.strictEqual(result, input); } +function testZstdConcatenatedFrames() { + const input = Buffer.concat([ + zstdCompressSync('a'), + zstdCompressSync('b'), + ]); + const result = bytesSync(pullSync(fromSync(input), decompressZstdSync())); + assert.strictEqual(Buffer.from(result).toString(), 'ab'); +} + // ============================================================================= // Cross-algorithm: compress async-compatible, decompress sync (and vice versa) // The sync transforms should produce output compatible with the standard format @@ -218,6 +228,7 @@ testBrotliRoundTrip(); testBrotliLargeData(); testZstdRoundTrip(); testZstdLargeData(); +testZstdConcatenatedFrames(); testGzipWithOptions(); testBrotliWithOptions(); testMixedStatelessAndStateful(); diff --git a/test/js/node/test/parallel/test-zlib-truncated.js b/test/js/node/test/parallel/test-zlib-truncated.js index 94bc0e21cb7a..0f9ce776b1db 100644 --- a/test/js/node/test/parallel/test-zlib-truncated.js +++ b/test/js/node/test/parallel/test-zlib-truncated.js @@ -1,7 +1,7 @@ 'use strict'; // Tests zlib streams with truncated compressed input -require('../common'); +const common = require('../common'); const assert = require('assert'); const zlib = require('zlib'); @@ -22,9 +22,14 @@ const errMessage = /unexpected end of file/; { comp: 'gzip', decomp: 'unzip', decompSync: 'unzipSync' }, { comp: 'deflate', decomp: 'inflate', decompSync: 'inflateSync' }, { comp: 'deflateRaw', decomp: 'inflateRaw', decompSync: 'inflateRawSync' }, + { + comp: 'zstdCompress', + decomp: 'zstdDecompress', + decompSync: 'zstdDecompressSync', + partialFlush: zlib.constants.ZSTD_e_flush, + }, ].forEach(function(methods) { - zlib[methods.comp](inputString, function(err, compressed) { - assert.ifError(err); + zlib[methods.comp](inputString, common.mustSucceed((compressed) => { const truncated = compressed.slice(0, compressed.length / 2); const toUTF8 = (buffer) => buffer.toString('utf-8'); @@ -33,10 +38,9 @@ const errMessage = /unexpected end of file/; assert.strictEqual(toUTF8(decompressed), inputString); // async sanity - zlib[methods.decomp](compressed, function(err, result) { - assert.ifError(err); + zlib[methods.decomp](compressed, common.mustSucceed((result) => { assert.strictEqual(toUTF8(result), inputString); - }); + })); // Sync truncated input test assert.throws(function() { @@ -44,21 +48,63 @@ const errMessage = /unexpected end of file/; }, errMessage); // Async truncated input test - zlib[methods.decomp](truncated, function(err, result) { + zlib[methods.decomp](truncated, common.mustCall((err) => { assert.match(err.message, errMessage); - }); + })); - const syncFlushOpt = { finishFlush: zlib.constants.Z_SYNC_FLUSH }; + const partialFlushOpt = { + finishFlush: methods.partialFlush ?? zlib.constants.Z_SYNC_FLUSH, + }; - // Sync truncated input test, finishFlush = Z_SYNC_FLUSH - const result = toUTF8(zlib[methods.decompSync](truncated, syncFlushOpt)); + // Sync truncated input test with a non-finalizing finish flush. + const result = toUTF8(zlib[methods.decompSync](truncated, partialFlushOpt)); assert.strictEqual(result, inputString.slice(0, result.length)); - // Async truncated input test, finishFlush = Z_SYNC_FLUSH - zlib[methods.decomp](truncated, syncFlushOpt, function(err, decompressed) { - assert.ifError(err); + // Async truncated input test with a non-finalizing finish flush. + zlib[methods.decomp](truncated, partialFlushOpt, common.mustSucceed((decompressed) => { const result = toUTF8(decompressed); assert.strictEqual(result, inputString.slice(0, result.length)); - }); - }); + })); + })); }); + +// A non-zero return from ZSTD_decompressStream() can also mean that the +// output buffer is full. Make sure that is drained before treating the return +// value as truncated input. +{ + const input = Buffer.alloc(zlib.constants.Z_DEFAULT_CHUNK * 2, 0x61); + const compressed = zlib.zstdCompressSync(input); + const decompressed = zlib.zstdDecompressSync(compressed, { + chunkSize: zlib.constants.Z_MIN_CHUNK, + }); + assert.deepStrictEqual(decompressed, input); +} + +// Ending a stream after a previous write completed a frame must not be +// mistaken for an empty, truncated frame. +{ + const input = Buffer.from(inputString); + const compressed = zlib.zstdCompressSync(input); + const decompressor = zlib.createZstdDecompress(); + const output = []; + + decompressor.on('data', (chunk) => output.push(chunk)); + decompressor.on('end', common.mustCall(() => { + assert.deepStrictEqual(Buffer.concat(output), input); + })); + decompressor.write(compressed, common.mustCall(() => decompressor.end())); +} + +// Conversely, ending after a previous write supplied only part of a frame +// must report that the frame is incomplete. +{ + const compressed = zlib.zstdCompressSync(inputString); + const truncated = compressed.subarray(0, compressed.length / 2); + const decompressor = zlib.createZstdDecompress(); + + decompressor.on('error', common.mustCall((error) => { + assert.match(error.message, errMessage); + })); + decompressor.write(truncated, common.mustCall(() => decompressor.end())); + decompressor.resume(); +} diff --git a/test/js/node/zlib/zlib.test.js b/test/js/node/zlib/zlib.test.js index 7555ee802170..3a6887e22c37 100644 --- a/test/js/node/zlib/zlib.test.js +++ b/test/js/node/zlib/zlib.test.js @@ -502,11 +502,8 @@ for (const [compress, decompressor] of [ }, ]; for (const i in variants) { - let should_skip = false; - if (decompressor === zlib.createZstdDecompress && i == 1) should_skip = true; // fails in node too - if (decompressor === zlib.createZstdDecompress && i == 2) should_skip = true; // fails in node too // prettier-ignore - it.skipIf(should_skip)(`premature end handles bytesWritten properly: ${compress.name} + ${decompressor.name}: variant ${i}`, async () => { + it(`premature end handles bytesWritten properly: ${compress.name} + ${decompressor.name}: variant ${i}`, async () => { const variant = variants[i]; const { promise, resolve, reject } = Promise.withResolvers(); let output = ""; diff --git a/test/js/node/zlib/zstd-frames.test.ts b/test/js/node/zlib/zstd-frames.test.ts new file mode 100644 index 000000000000..8d2d89dea8d6 --- /dev/null +++ b/test/js/node/zlib/zstd-frames.test.ts @@ -0,0 +1,570 @@ +/** + * The zstd decoder of node:zlib follows ZstdDecompressContext of Node v26.10.0: + * https://github.com/nodejs/node/blob/v26.10.0/src/node_zlib.cc#L1841-L1937 + * + * This file uses node:test and node:assert, so that the identical file also + * runs under `node --test` on Node v26.10.0. Each expected value is what that + * run gives. For the same reason the file does not import from "harness". + */ +import assert from "node:assert"; +import { execFile } from "node:child_process"; +import { describe, test } from "node:test"; +import { promisify } from "node:util"; +import zlib from "node:zlib"; + +const { ZSTD_e_end } = zlib.constants; + +const first = zlib.zstdCompressSync("first\n"); +const second = zlib.zstdCompressSync("second\n"); +const junk = Buffer.from("not valid compressed data"); +const magic = Buffer.from([0x28, 0xb5, 0x2f, 0xfd]); +// The first byte of a skippable magic number is 0x50 to 0x5f. A length of 4 and 4 bytes to skip follow. +const skippable = Buffer.from([0x5a, 0x2a, 0x4d, 0x18, 4, 0, 0, 0, 1, 2, 3, 4]); +const bytewise = (buffer: Buffer) => Array.from(buffer, byte => Buffer.of(byte)); + +const describeError = (err: any) => ({ + name: err.constructor.name, + message: err.message, + code: err.code, + errno: err.errno, + keys: Object.keys(err).sort(), +}); +const zstdError = (message: string, code: string, errno: number) => ({ + name: "Error", + message, + code, + errno, + keys: ["code", "errno"], +}); + +function thrownBy(fn: () => unknown) { + try { + fn(); + } catch (err) { + return describeError(err); + } +} +const rejectionOf = (promise: Promise) => promise.then(() => undefined, describeError); + +type Decoder = ReturnType; +type Operation = Buffer | ((decoder: Decoder) => unknown); + +// Runs the operations on one stream, where a buffer is a write(), and resolves at 'close'. +async function decode(operations: Operation[], options?: zlib.ZstdOptions) { + const decoder = zlib.createZstdDecompress(options); + const output: Buffer[] = []; + const events: unknown[] = []; + const closed = new Promise(resolve => decoder.on("close", resolve)); + decoder.on("data", chunk => output.push(chunk)); + decoder.on("end", () => events.push("end")); + decoder.on("error", err => events.push(describeError(err))); + for (const operation of operations) { + if (typeof operation === "function") await operation(decoder); + else decoder.write(operation); + } + await closed; + return { output: Buffer.concat(output).toString("latin1"), events, bytesWritten: decoder.bytesWritten }; +} +const end = (decoder: Decoder) => decoder.end(); +const reset = (decoder: Decoder) => decoder.reset(); +// A write() that waits for its callback, or for 'close' when the stream stops first. +const written = (chunk: Buffer) => (decoder: Decoder) => + new Promise(resolve => { + decoder.once("close", resolve); + decoder.write(chunk, () => resolve()); + }); +const ended = (output: string, bytesWritten: number) => ({ output, events: ["end"], bytesWritten }); + +// _processChunk on a handle that stays open, which is how minizlib drives the decoder. +function processChunks(steps: ([Buffer, number] | "reset")[]) { + const decoder = new zlib.ZstdDecompress() as any; + const handle = decoder._handle; + const close = handle.close; + handle.close = () => {}; + let output = ""; + try { + for (const step of steps) { + if (step === "reset") handle.reset(); + else output += Buffer.from(decoder._processChunk(step[0], step[1])).toString("latin1"); + decoder._handle = handle; + } + return { output }; + } catch (err) { + return { output, error: describeError(err) }; + } finally { + close.call(handle); + } +} + +// node:zlib/iter needs a flag, so its cases run in one child of the same runtime. The file waits +// for the child before its tests start, so that no test has to wait for a process. The deadline +// ends a child that does not exit, so that this file fails and does not stall the run. +// Each result is that of decompressZstd and then that of decompressZstdSync. +const iterChild = await promisify(execFile)( + process.execPath, + [ + "--experimental-stream-iter", + "-e", + ` + const { bytes, bytesSync, from, fromSync, pull, pullSync } = require("node:stream/iter"); + const { decompressZstd, decompressZstdSync } = require("node:zlib/iter"); + const zlib = require("node:zlib"); + const a = zlib.zstdCompressSync("a"); + const b = zlib.zstdCompressSync("b"); + const junk = Buffer.from("junk"); + const wrongChecksum = zlib.zstdCompressSync("b", { params: { [zlib.constants.ZSTD_c_checksumFlag]: 1 } }); + wrongChecksum[wrongChecksum.length - 1] ^= 1; + const cases = { + "after a complete frame": { + "frame, tail, frame": [a, junk, b], + "frame, frame, tail": [a, b, junk], + "frame and tail in one chunk": [Buffer.concat([a, junk])], + "frame, frame": [a, b], + "frame, frame with a wrong checksum": [a, wrongChecksum], + }, + "the end of the input": { + "no chunk": [], + "an empty chunk": [Buffer.alloc(0)], + "a frame without its last byte": [a.subarray(0, -1)], + "a frame, a frame without its last byte": [a, b.subarray(0, -1)], + "a frame, 4 bytes of a magic number": [a, b.subarray(0, 4)], + "a frame, 3 bytes of a magic number": [a, b.subarray(0, 3)], + }, + }; + const result = decode => decode().then( + output => Buffer.from(output).toString("latin1"), + err => ({ name: err.constructor.name, message: err.message, code: err.code, errno: err.errno }), + ); + (async () => { + const results = {}; + for (const [group, rows] of Object.entries(cases)) { + results[group] = {}; + for (const [name, chunks] of Object.entries(rows)) { + results[group][name] = [ + await result(async () => bytes(pull(from(chunks), decompressZstd()))), + await result(async () => bytesSync(pullSync(fromSync(chunks), decompressZstdSync()))), + ]; + } + } + console.log(JSON.stringify(results)); + })(); + `, + ], + { env: { ...process.env, BUN_DEBUG_QUIET_LOGS: "1" }, timeout: 60_000, killSignal: "SIGKILL" }, +); +const iterResults: Record> = JSON.parse(iterChild.stdout); + +describe("zstd: input after a complete frame", () => { + // The last number is how many bytes of the tail a magic number can begin with. + const tails: [string, Buffer, number][] = [ + ["text", junk, 0], + ["1 byte of the zstd magic number and then text", Buffer.from("(hello world"), 1], + ["2 bytes of the zstd magic number and then another byte", Buffer.from([0x28, 0xb5, 0x58]), 2], + ["3 bytes of the zstd magic number and then another byte", Buffer.from([0x28, 0xb5, 0x2f, 0x58]), 3], + ["1 byte of a skippable magic number and then text", Buffer.from("PK\x03\x04"), 1], + ["2 bytes of a skippable magic number and then another byte", Buffer.from([0x50, 0x2a, 0x58]), 2], + ["3 bytes of a skippable magic number and then another byte", Buffer.from([0x5f, 0x2a, 0x4d, 0x19]), 3], + ["a first byte above the skippable magic numbers", Buffer.from([0x60, 0x2a, 0x4d, 0x18]), 0], + ["a first byte below the skippable magic numbers", Buffer.from([0x4f, 0x2a, 0x4d, 0x18]), 0], + ["a first byte below the zstd magic number", Buffer.from([0x27, 0xb5, 0x2f, 0xfd]), 0], + ]; + for (const [name, tail, accepted] of tails) { + test(`${name} ends the stream`, async () => { + const input = Buffer.concat([first, tail]); + assert.strictEqual(zlib.zstdDecompressSync(input).toString(), "first\n"); + assert.strictEqual((await promisify(zlib.zstdDecompress)(input)).toString(), "first\n"); + assert.deepStrictEqual(await decode([d => d.end(input)]), ended("first\n", first.length)); + assert.deepStrictEqual(await decode([first, tail, second, end]), ended("first\n", first.length)); + assert.deepStrictEqual(await decode([input, second, end]), ended("first\n", first.length)); + // zstd takes each byte that a magic number can begin with, so those count as written. + assert.deepStrictEqual( + await decode([first, ...bytewise(tail), second, end]), + ended("first\n", first.length + accepted), + ); + }); + } + + test("a tail after two frames, and after a skippable frame, ends the stream", async () => { + const both = first.length + second.length; + assert.strictEqual(zlib.zstdDecompressSync(Buffer.concat([first, second, junk])).toString(), "first\nsecond\n"); + assert.deepStrictEqual(await decode([first, second, junk, end]), ended("first\nsecond\n", both)); + assert.deepStrictEqual( + await decode([first, skippable, junk, second, end]), + ended("first\n", first.length + skippable.length), + ); + }); + + test("a tail ends the stream when the frame fills the output chunk", () => { + // The sync driver calls the handle again when the chunk is full, with only the tail as the input. + const exact = Buffer.alloc(zlib.constants.Z_DEFAULT_CHUNK, "a"); + const input = Buffer.concat([zlib.zstdCompressSync(exact), Buffer.from("junkjunk")]); + assert.deepStrictEqual(zlib.zstdDecompressSync(input), exact); + }); + + describe("reset()", () => { + test("after a tail, the stream decodes again, and its output comes after 'end'", async () => { + const { events, ...rest } = await decode([written(first), written(junk), reset, written(second), end]); + assert.deepStrictEqual(rest, { output: "first\n", bytesWritten: first.length + second.length }); + assert.deepStrictEqual( + events.map((event: any) => event.code ?? event), + ["end", "ERR_STREAM_PUSH_AFTER_EOF"], + ); + }); + + test("after a tail, a handle that stays open decodes again", () => { + assert.deepStrictEqual( + processChunks([[first, 0], [junk, 0], "reset", [second, 0], [Buffer.alloc(0), ZSTD_e_end]]), + { output: "first\nsecond\n" }, + ); + }); + + test("in the middle of a magic number, the next frame starts at its first byte", async () => { + assert.deepStrictEqual( + await decode([written(first), written(magic.subarray(0, 2)), reset, written(second), end]), + ended("first\nsecond\n", first.length + 2 + second.length), + ); + }); + }); + + // The cases below follow the five zstd tests of test/parallel/test-zlib-reject-garbage-after-end.js + // of Node v26.10.0 (lines 133-259), with the default options. That file is not vendored, because + // its other tests need options.rejectGarbageAfterEnd. + describe("concatenated frames", () => { + for (const [name, frames] of [ + ["two frames", [first, second]], + ["two frames with a skippable frame between them", [first, skippable, second]], + ] as [string, Buffer[]][]) { + test(`${name} decode at each split of the input`, async () => { + const input = Buffer.concat(frames); + assert.strictEqual(zlib.zstdDecompressSync(input).toString(), "first\nsecond\n"); + assert.strictEqual((await promisify(zlib.zstdDecompress)(input)).toString(), "first\nsecond\n"); + for (let split = 0; split <= input.length; split++) { + assert.deepStrictEqual( + await decode([input.subarray(0, split), input.subarray(split), end]), + ended("first\nsecond\n", input.length), + `split at byte ${split}`, + ); + } + assert.deepStrictEqual(await decode([...bytewise(input), end]), ended("first\nsecond\n", input.length)); + }); + } + + test("a skippable frame alone decodes to nothing", async () => { + assert.deepStrictEqual(zlib.zstdDecompressSync(skippable), Buffer.alloc(0)); + assert.deepStrictEqual(await decode([skippable, end]), ended("", skippable.length)); + }); + + test("1 to 3 bytes of a frame after a frame are a tail", async () => { + for (let length = 1; length <= 3; length++) { + const tail = second.subarray(0, length); + assert.strictEqual(zlib.zstdDecompressSync(Buffer.concat([first, tail])).toString(), "first\n"); + assert.deepStrictEqual(await decode([first, tail, end]), ended("first\n", first.length + length)); + } + }); + + test("two frames decode across many output chunks", async () => { + const payloads = [Buffer.alloc(1024, "first "), Buffer.alloc(1024, "second ")]; + const input = Buffer.concat(payloads.map(payload => zlib.zstdCompressSync(payload))); + const options = { chunkSize: zlib.constants.Z_MIN_CHUNK }; + assert.deepStrictEqual(zlib.zstdDecompressSync(input, options), Buffer.concat(payloads)); + assert.deepStrictEqual( + await decode([input, end], options), + ended(Buffer.concat(payloads).toString("latin1"), input.length), + ); + }); + + // Each frame begins with a valid magic number, so it is a frame and not a tail. + const checksummed = zlib.zstdCompressSync("second\n", { params: { [zlib.constants.ZSTD_c_checksumFlag]: 1 } }); + checksummed[checksummed.length - 1] ^= 1; + const badFrames: [string, Buffer, ReturnType][] = [ + [ + "a wrong checksum", + checksummed, + zstdError("Restored data doesn't match checksum", "ZSTD_error_checksum_wrong", 22), + ], + [ + "a block of the reserved type", + Buffer.from([...magic, 0x20, 0x01, 0x0f, 0x00, 0x00, 0x00]), + zstdError("Data corruption detected", "ZSTD_error_corruption_detected", 20), + ], + [ + "the reserved bit in its header", + Buffer.from([...magic, 0x28, 0x01, 0x01, 0x00, 0x00, 0x00]), + zstdError("Unsupported frame parameter", "ZSTD_error_frameParameter_unsupported", 14), + ], + ]; + for (const [name, bad, error] of badFrames) { + test(`a later frame with ${name} is an error`, async () => { + const input = Buffer.concat([first, bad]); + assert.deepStrictEqual( + thrownBy(() => zlib.zstdDecompressSync(input)), + error, + ); + assert.deepStrictEqual(await rejectionOf(promisify(zlib.zstdDecompress)(input)), error); + // A write that fails gives none of its output, also not that of the frames before the bad one. + assert.deepStrictEqual(await decode([input, end]), { output: "", events: [error], bytesWritten: 0 }); + for (const [writes, bytesWritten] of [ + [[first, bad], first.length], + [[Buffer.concat([first, bad.subarray(0, 2)]), bad.subarray(2)], first.length + 2], + [[first, skippable, bad], first.length + skippable.length], + ] as [Buffer[], number][]) { + assert.deepStrictEqual(await decode([...writes, end]), { output: "first\n", events: [error], bytesWritten }); + } + }); + } + }); + + test("input that is not a frame at the start of the stream is an error, not a tail", async () => { + const error = zstdError("Unknown frame descriptor", "ZSTD_error_prefix_unknown", 10); + assert.deepStrictEqual( + thrownBy(() => zlib.zstdDecompressSync(junk)), + error, + ); + assert.deepStrictEqual(await decode([junk, end]), { output: "", events: [error], bytesWritten: 0 }); + }); + + test("zlib/iter stops at a chunk that cannot start a frame", () => { + const checksumWrong = { + name: "Error", + message: "Restored data doesn't match checksum", + code: "ZSTD_error_checksum_wrong", + errno: 22, + }; + assert.deepStrictEqual(iterResults["after a complete frame"], { + "frame, tail, frame": ["a", "a"], + "frame, frame, tail": ["ab", "ab"], + "frame and tail in one chunk": ["a", "a"], + "frame, frame": ["ab", "ab"], + "frame, frame with a wrong checksum": [checksumWrong, checksumWrong], + }); + }); +}); + +// With finishFlush ZSTD_e_end, which is the default, the input must end where a frame ends. +describe("zstd: the end of the input", () => { + const { ZSTD_e_continue, ZSTD_e_flush } = zlib.constants; + const unexpectedEnd = zstdError("unexpected end of file", "Z_BUF_ERROR", -5); + const payload = Buffer.alloc(192, "hello world "); + const text = payload.toString("latin1"); + const frame = zlib.zstdCompressSync(payload); + const truncated = frame.subarray(0, -10); + const checksummed = zlib.zstdCompressSync(payload, { params: { [zlib.constants.ZSTD_c_checksumFlag]: 1 } }); + const withoutChecksumByte = checksummed.subarray(0, -1); + const withoutChecksum = checksummed.subarray(0, -4); + // A compressor that got a flush and no end gives whole blocks, and not the last block of the frame. + const flushed = zlib.zstdCompressSync(payload, { finishFlush: ZSTD_e_flush }); + const empty = Buffer.alloc(0); + const flushEnd = (decoder: Decoder) => decoder.flush(ZSTD_e_end); + const failed = (payloads: number, bytesWritten: number) => ({ + output: text.repeat(payloads), + events: [unexpectedEnd], + bytesWritten, + }); + + describe("inside a frame", () => { + // After the input, each row has the number of payloads that the input gives before it ends: + // with the input in one write, and with each byte in a write of its own. Then it has that number + // and bytesWritten for a write of the input that has ZSTD_e_end. A write that fails gives no + // output and counts no bytes. + const inputs: [string, Buffer, number, number, [number, number]][] = [ + ["no input", empty, 0, 0, [0, 0]], + ["a frame without its last 10 bytes", truncated, 0, 0, [0, 0]], + ["2 bytes of a magic number", magic.subarray(0, 2), 0, 0, [0, 0]], + ["a skippable frame without its last 2 bytes", skippable.subarray(0, -2), 0, 0, [0, 0]], + ["a frame without the last byte of its checksum", withoutChecksumByte, 1, 1, [1, withoutChecksumByte.length]], + ["a frame without its checksum", withoutChecksum, 1, 0, [1, withoutChecksum.length]], + ["the output of a compressor that did not end", flushed, 1, 0, [1, flushed.length]], + ["a frame and a frame without its last 10 bytes", Buffer.concat([frame, truncated]), 1, 1, [1, frame.length]], + ["a frame and 4 bytes of a magic number", Buffer.concat([frame, magic]), 1, 1, [1, frame.length]], + [ + "two frames and 4 bytes of a skippable magic number", + Buffer.concat([frame, frame, skippable.subarray(0, 4)]), + 2, + 2, + [2, 2 * frame.length], + ], + ]; + for (const [name, input, payloads, payloadsBytewise, finishing] of inputs) { + test(`${name} is an error`, async () => { + assert.deepStrictEqual( + thrownBy(() => zlib.zstdDecompressSync(input)), + unexpectedEnd, + ); + assert.deepStrictEqual(await rejectionOf(promisify(zlib.zstdDecompress)(input)), unexpectedEnd); + assert.deepStrictEqual(processChunks([[input, ZSTD_e_end]]), { output: "", error: unexpectedEnd }); + assert.deepStrictEqual( + processChunks([ + [input, ZSTD_e_continue], + [empty, ZSTD_e_end], + ]), + { output: text.repeat(payloads), error: unexpectedEnd }, + ); + assert.deepStrictEqual(await decode([d => d.end(input)]), failed(payloads, input.length)); + assert.deepStrictEqual( + await decode([...bytewise(input), end]), + failed(payloadsBytewise, Math.max(input.length - 1, 0)), + ); + for (const chunkSize of [zlib.constants.Z_MIN_CHUNK, payload.length]) { + assert.deepStrictEqual(await decode([input, end], { flush: ZSTD_e_end, chunkSize }), failed(...finishing)); + } + }); + + test(`${name} is not an error with another finishFlush`, async () => { + for (const finishFlush of [ZSTD_e_flush, ZSTD_e_continue]) { + const options = { finishFlush, chunkSize: zlib.constants.Z_MIN_CHUNK }; + assert.strictEqual(zlib.zstdDecompressSync(input, options).toString("latin1"), text.repeat(payloads)); + assert.strictEqual( + (await promisify(zlib.zstdDecompress)(input, options)).toString("latin1"), + text.repeat(payloads), + ); + assert.deepStrictEqual( + await decode([d => d.end(input)], options), + ended(text.repeat(payloads), input.length), + ); + } + }); + } + + const orders: [string, Operation[], zlib.ZstdOptions | undefined, number, number][] = [ + ["end() with no write", [end], undefined, 0, 0], + ["flush(ZSTD_e_end) before the first write", [flushEnd, end], undefined, 0, 0], + ["flush(ZSTD_e_end) after the input", [truncated, flushEnd, end], undefined, 0, truncated.length], + ["options.flush = ZSTD_e_end", [truncated.subarray(0, 10), end], { flush: ZSTD_e_end }, 0, 0], + ["a write that waits for its callback", [written(truncated), end], undefined, 0, truncated.length], + ["a frame and the input in one tick", [frame, truncated, end], undefined, 1, frame.length], + [ + "a magic number in two writes after a frame", + [frame, magic.subarray(0, 2), magic.subarray(2), end], + undefined, + 1, + frame.length + 2, + ], + ["reset() after a frame", [written(frame), reset, end], undefined, 1, frame.length], + ]; + for (const [name, operations, options, payloads, bytesWritten] of orders) { + test(`${name}: the stream gives the error`, async () => { + assert.deepStrictEqual(await decode(operations, options), failed(payloads, bytesWritten)); + }); + } + + test("after a frame, the output of the input comes before the error", async () => { + const options = { chunkSize: payload.length }; + for (const input of [withoutChecksumByte, withoutChecksum, flushed]) { + assert.deepStrictEqual(await decode([frame, input, end], options), failed(2, frame.length + input.length)); + } + assert.deepStrictEqual(await decode([frame, truncated, end], options), failed(1, frame.length)); + }); + + test("the blocks that are complete come out before the error", async () => { + const block = 128 * 1024; + const big = zlib.zstdCompressSync(Buffer.alloc(8 * block, "hello world ")); + const cut = big.subarray(0, -10); + // The shortest input that gives the first block ends where that block ends. + const lengths = Array.from(big.keys(), index => index + 1); + const firstBlockEnd = lengths.find( + length => zlib.zstdDecompressSync(big.subarray(0, length), { finishFlush: ZSTD_e_flush }).length === block, + ); + assert.ok(firstBlockEnd, "no prefix of the frame gives exactly its first block"); + const oneBlock = big.subarray(0, firstBlockEnd); + const sizes = ({ output, ...rest }: Awaited>) => ({ ...rest, output: output.length }); + for (const [input, output] of [ + [cut, 7 * block], + [oneBlock, block], + ] as [Buffer, number][]) { + const expected = { events: [unexpectedEnd], bytesWritten: input.length, output }; + assert.deepStrictEqual(sizes(await decode([input, end])), expected); + // In the next two orders the write of the input has ZSTD_e_end, and it fills output chunks before it ends. + assert.deepStrictEqual(sizes(await decode([empty, input, end])), expected); + assert.deepStrictEqual(sizes(await decode([d => d.end(input)], { flush: ZSTD_e_end })), expected); + } + }); + + test("a stream that stops inside a frame gives no error", async () => { + const stopped = { output: "", events: [], bytesWritten: truncated.length }; + assert.deepStrictEqual(await decode([written(truncated), d => d.destroy()]), stopped); + assert.deepStrictEqual(await decode([written(truncated), d => d.close()]), stopped); + }); + + test("the finishFlush of the stream at the time of end() decides", async () => { + const lenient = (decoder: any) => (decoder._finishFlushFlag = ZSTD_e_flush); + assert.deepStrictEqual(await decode([lenient, d => d.end(truncated)]), ended("", truncated.length)); + }); + + // Node aborts on a write before init(), so this test is for bun only. + test("a handle that init() did not set up reports nothing", { skip: typeof Bun === "undefined" }, () => { + const Handle = (zlib.createZstdDecompress() as any)._handle.constructor; + const handle = new Handle(zlib.constants.ZSTD_DECOMPRESS); + const errors: string[] = []; + handle.onerror = (message: string, errno: number, code: string) => errors.push(code); + handle.writeSync(ZSTD_e_end, null, 0, 0, new Uint8Array(64), 0, 64); + assert.deepStrictEqual(errors, []); + }); + + test("zlib/iter gives the error", () => { + const { keys, ...error } = unexpectedEnd; + assert.deepStrictEqual(iterResults["the end of the input"], { + "no chunk": [error, error], + "an empty chunk": [error, error], + "a frame without its last byte": [error, error], + "a frame, a frame without its last byte": [error, error], + "a frame, 4 bytes of a magic number": [error, error], + "a frame, 3 bytes of a magic number": ["a", "a"], + }); + }); + }); + + describe("where a frame ends", () => { + const twice = text.repeat(2); + // Fewer than 4 bytes after a frame do not show that a frame began. + const tails: [string, Buffer][] = [ + ["nothing", empty], + ["1 byte of a magic number", magic.subarray(0, 1)], + ["2 bytes of a magic number", magic.subarray(0, 2)], + ["3 bytes of a magic number", magic.subarray(0, 3)], + ["1 byte of a skippable magic number", skippable.subarray(0, 1)], + ["2 bytes of a skippable magic number", skippable.subarray(0, 2)], + ["3 bytes of a skippable magic number", skippable.subarray(0, 3)], + ]; + for (const [name, tail] of tails) { + test(`${name} after the last frame is not an error`, async () => { + for (const [frames, output] of [ + [[frame], text], + [[frame, frame], twice], + [[frame, skippable], text], + ] as [Buffer[], string][]) { + const input = Buffer.concat([...frames, tail]); + for (const chunkSize of [zlib.constants.Z_DEFAULT_CHUNK, zlib.constants.Z_MIN_CHUNK, payload.length]) { + const options = { chunkSize }; + assert.strictEqual(zlib.zstdDecompressSync(input, options).toString("latin1"), output); + assert.deepStrictEqual(processChunks([[input, ZSTD_e_end]]), { output }); + assert.deepStrictEqual(await decode([d => d.end(input)], options), ended(output, input.length)); + assert.deepStrictEqual(await decode([...bytewise(input), end], options), ended(output, input.length)); + assert.deepStrictEqual( + await decode([input, end], { ...options, flush: ZSTD_e_end }), + ended(output, input.length), + ); + } + } + }); + } + + const orders: [string, Operation[], string, number][] = [ + ["flush(ZSTD_e_end) between two frames", [frame, flushEnd, frame, end], twice, 2 * frame.length], + ["20 times flush(ZSTD_e_end)", [frame, ...Array(20).fill(flushEnd), end], text, frame.length], + ["20 empty writes", [frame, ...Array(20).fill(empty), end], text, frame.length], + [ + "reset() inside a frame, and then a frame", + [written(truncated), reset, frame, end], + text, + truncated.length + frame.length, + ], + ]; + for (const [name, operations, output, bytesWritten] of orders) { + test(`${name} is not an error`, async () => { + for (const chunkSize of [zlib.constants.Z_DEFAULT_CHUNK, zlib.constants.Z_MIN_CHUNK]) { + assert.deepStrictEqual(await decode(operations, { chunkSize }), ended(output, bytesWritten)); + } + }); + } + }); +});