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)); + } + }); + } + }); +});