From f8088e722e1bfe719e16c09f275ca503a2114e95 Mon Sep 17 00:00:00 2001 From: Lazizbek Ergashev Date: Sun, 26 Jul 2026 07:00:54 +0500 Subject: [PATCH] zlib: decode concatenated zstd frames ZstdDecompressContext::DoThreadPoolWork decoded a single frame per call, so a stream that received several concatenated frames in one write kept only the first frame and dropped the rest. Other codecs that support concatenation, such as gunzip, already loop over the remaining input. Keep decoding while a frame ends with input still pending and output space available, and thread rejectGarbageAfterEnd through to the zstd context so trailing frames are still rejected when that option is set. Signed-off-by: Lazizbek Ergashev --- doc/api/zlib.md | 8 +++- lib/zlib.js | 1 + src/node_zlib.cc | 41 +++++++++++----- .../test-zlib-from-concatenated-zstd.js | 47 +++++++++++++++++++ .../test-zlib-reject-garbage-after-end.js | 2 +- 5 files changed, 85 insertions(+), 14 deletions(-) create mode 100644 test/parallel/test-zlib-from-concatenated-zstd.js diff --git a/doc/api/zlib.md b/doc/api/zlib.md index 46b8eef1ad9f45..da863820957795 100644 --- a/doc/api/zlib.md +++ b/doc/api/zlib.md @@ -1100,6 +1100,9 @@ added: - v23.8.0 - v22.15.0 changes: + - version: REPLACEME + pr-url: https://github.com/nodejs/node/pull/64748 + description: Multiple concatenated zstd frames are decoded now. - version: REPLACEME pr-url: https://github.com/nodejs/node/pull/64599 description: The `dictionary` option can be a `TypedArray`, `DataView`, or @@ -1124,7 +1127,10 @@ Each Zstd-based class takes an `options` object. All options are optional. to improve compression efficiency when compressing or decompressing data that shares common patterns with the dictionary. * `rejectGarbageAfterEnd` {boolean} If `true`, decompression fails when - input remains after the first complete compressed stream. **Default:** `false` + trailing input is detected after the end of the compressed stream. This + includes unreadable bytes and additional zstd frames following the first + frame, which are otherwise decoded as part of the same stream. + **Default:** `false` For example: diff --git a/lib/zlib.js b/lib/zlib.js index e6c5c420d55182..0757e0a6b1cffd 100644 --- a/lib/zlib.js +++ b/lib/zlib.js @@ -949,6 +949,7 @@ class Zstd extends ZlibBase { writeState, processCallback, dictionary, + opts?.rejectGarbageAfterEnd === true, ); super(opts, mode, handle, zstdDefaultOpts); diff --git a/src/node_zlib.cc b/src/node_zlib.cc index 7d2950aed45a2d..8181802266da95 100644 --- a/src/node_zlib.cc +++ b/src/node_zlib.cc @@ -307,6 +307,7 @@ class ZstdContext : public MemoryRetainer { // Streaming-related, should be available for all compression libraries: void SetBuffers(const char* in, uint32_t in_len, char* out, uint32_t out_len); void SetFlush(int flush); + void SetRejectGarbageAfterEnd(bool reject_garbage_after_end); void GetAfterWriteOffsets(uint32_t* avail_in, uint32_t* avail_out) const; CompressionError GetErrorInfo() const; @@ -315,6 +316,7 @@ class ZstdContext : public MemoryRetainer { protected: ZSTD_EndDirective flush_ = ZSTD_e_continue; + bool reject_garbage_after_end_ = false; ZSTD_inBuffer input_ = {nullptr, 0, 0}; ZSTD_outBuffer output_ = {nullptr, 0, 0}; @@ -941,9 +943,9 @@ class ZstdStream final : public CompressionStream { } static void Init(const FunctionCallbackInfo& args) { - CHECK((args.Length() == 4 || args.Length() == 5) && - "init(params, pledgedSrcSize, writeResult, writeCallback[, " - "dictionary])"); + CHECK((args.Length() == 5 || args.Length() == 6) && + "init(params, pledgedSrcSize, writeResult, writeCallback, " + "dictionary[, rejectGarbageAfterEnd])"); ZstdStream* wrap; ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This()); @@ -980,7 +982,7 @@ class ZstdStream final : public CompressionStream { AllocScope alloc_scope(wrap); std::string_view dictionary; ArrayBufferViewContents contents; - if (args.Length() == 5 && !args[4]->IsUndefined()) { + if (!args[4]->IsUndefined()) { if (!args[4]->IsArrayBufferView()) { THROW_ERR_INVALID_ARG_TYPE( wrap->env(), "dictionary must be an ArrayBufferView if provided"); @@ -997,6 +999,11 @@ class ZstdStream final : public CompressionStream { return; } + if (args.Length() == 6) { + CHECK(args[5]->IsBoolean()); + wrap->context()->SetRejectGarbageAfterEnd(args[5]->IsTrue()); + } + CHECK(args[0]->IsUint32Array()); const uint32_t* data = reinterpret_cast(Buffer::Data(args[0])); size_t len = args[0].As()->Length(); @@ -1624,6 +1631,10 @@ void ZstdContext::SetFlush(int flush) { flush_ = static_cast(flush); } +void ZstdContext::SetRejectGarbageAfterEnd(bool reject_garbage_after_end) { + reject_garbage_after_end_ = reject_garbage_after_end; +} + void ZstdContext::GetAfterWriteOffsets(uint32_t* avail_in, uint32_t* avail_out) const { *avail_in = input_.size - input_.pos; @@ -1765,15 +1776,21 @@ void ZstdDecompressContext::DoThreadPoolWork() { return; } - size_t const ret = ZSTD_decompressStream(dctx_.get(), &output_, &input_); - if (ZSTD_isError(ret)) { - frame_complete_ = false; - error_ = ZSTD_getErrorCode(ret); - error_code_string_ = ZstdStrerror(error_); - error_string_ = ZSTD_getErrorString(error_); - } else { + do { + size_t const ret = ZSTD_decompressStream(dctx_.get(), &output_, &input_); + if (ZSTD_isError(ret)) { + frame_complete_ = false; + error_ = ZSTD_getErrorCode(ret); + error_code_string_ = ZstdStrerror(error_); + error_string_ = ZSTD_getErrorString(error_); + return; + } frame_complete_ = ret == 0; - } + // A single input buffer may hold several concatenated frames. Keep decoding + // while one frame ends with input still pending and output space to fill, + // unless trailing input is meant to be rejected as garbage. + } while (frame_complete_ && !reject_garbage_after_end_ && + input_.pos < input_.size && output_.pos < output_.size); } CompressionError ZstdDecompressContext::GetErrorInfo() const { diff --git a/test/parallel/test-zlib-from-concatenated-zstd.js b/test/parallel/test-zlib-from-concatenated-zstd.js new file mode 100644 index 00000000000000..5c6b25b011f0bf --- /dev/null +++ b/test/parallel/test-zlib-from-concatenated-zstd.js @@ -0,0 +1,47 @@ +'use strict'; +// Test decompressing a zstd stream that contains multiple concatenated frames + +const common = require('../common'); +const assert = require('assert'); +const zlib = require('zlib'); + +const abc = 'abc'; +const def = 'def'; + +const abcEncoded = zlib.zstdCompressSync(abc); +const defEncoded = zlib.zstdCompressSync(def); + +const data = Buffer.concat([ + abcEncoded, + defEncoded, +]); + +assert.strictEqual(zlib.zstdDecompressSync(data).toString(), (abc + def)); + +zlib.zstdDecompress(data, common.mustSucceed((result) => { + assert.strictEqual(result.toString(), (abc + def)); +})); + +// Test that the next zstd frame can wrap around the input buffer boundary +[0, 1, 2, 3, 4, defEncoded.length].forEach((offset) => { + const resultBuffers = []; + + const decompress = zlib.createZstdDecompress() + .on('error', common.mustNotCall()) + .on('data', (data) => resultBuffers.push(data)) + .on('finish', common.mustCall(() => { + assert.strictEqual( + Buffer.concat(resultBuffers).toString(), + 'abcdef', + `result should match original input (offset = ${offset})` + ); + })); + + // First write: write "abc" + the first bytes of "def" + decompress.write(Buffer.concat([ + abcEncoded, defEncoded.slice(0, offset), + ])); + + // Write remaining bytes of "def" + decompress.end(defEncoded.slice(offset)); +}); diff --git a/test/parallel/test-zlib-reject-garbage-after-end.js b/test/parallel/test-zlib-reject-garbage-after-end.js index 8039865f5f1193..8d1abc118b37b4 100644 --- a/test/parallel/test-zlib-reject-garbage-after-end.js +++ b/test/parallel/test-zlib-reject-garbage-after-end.js @@ -78,7 +78,7 @@ const cases = [ decompress: zlib.zstdDecompress, decompressSync: zlib.zstdDecompressSync, createDecompress: zlib.createZstdDecompress, - defaultOutput: 'a', + defaultOutput: 'aa', }, ];