diff --git a/doc/api/zlib.md b/doc/api/zlib.md index 656331d9b08e..be7e8bad8599 100644 --- a/doc/api/zlib.md +++ b/doc/api/zlib.md @@ -2200,6 +2200,12 @@ configured for a Zstd compressor, it applies again to the next frame. Calling `reset()` while a write is in progress throws an `Error`. +Calling `reset()` on a Zstd compressor while a frame is still in progress throws +an `Error` with the code `ERR_ZLIB_INCOMPLETE_FRAME`. Resetting at that point +would discard the frame state while the bytes already written out remain at the +start of the output stream, leaving it undecodable. Call `.flush()` and +`.end()`, or start over with a new stream, instead. + ## Class: `ZstdOptions` > Stability: 1 - Experimental diff --git a/src/node_errors.h b/src/node_errors.h index cab1dc76dcc7..2e65f8ef6d1e 100644 --- a/src/node_errors.h +++ b/src/node_errors.h @@ -133,6 +133,7 @@ void OOMErrorHandler(const char* location, const v8::OOMDetails& details); V(ERR_VM_MODULE_CACHED_DATA_REJECTED, Error) \ V(ERR_VM_MODULE_LINK_FAILURE, Error) \ V(ERR_WASI_NOT_STARTED, Error) \ + V(ERR_ZLIB_INCOMPLETE_FRAME, Error) \ V(ERR_ZLIB_INITIALIZATION_FAILED, Error) \ V(ERR_WORKER_INIT_FAILED, Error) \ V(ERR_PROTO_ACCESS, Error) diff --git a/src/node_zlib.cc b/src/node_zlib.cc index 810df1c779fd..93db4ce11fe1 100644 --- a/src/node_zlib.cc +++ b/src/node_zlib.cc @@ -352,6 +352,13 @@ class ZstdCompressContext final : public ZstdContext { uint64_t pledged_src_size_ = ZSTD_CONTENTSIZE_UNKNOWN; std::optional consumed_src_size_; + + // Tracks whether the current frame has been fully flushed. A frame is only + // complete once ZSTD_compressStream2() has been called with ZSTD_e_end and + // has returned 0. Resetting while a frame is still in progress silently + // discards the frame state, so any bytes already written out remain as an + // unusable fragment at the start of the output stream. + bool frame_complete_ = true; }; class ZstdDecompressContext final : public ZstdContext { @@ -1678,6 +1685,7 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size, std::string_view dictionary, bool) { pledged_src_size_ = pledged_src_size; + frame_complete_ = true; if (pledged_src_size == ZSTD_CONTENTSIZE_UNKNOWN) { consumed_src_size_.reset(); } else { @@ -1718,6 +1726,20 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size, } CompressionError ZstdCompressContext::ResetStream() { + // Resetting drops the state of the frame currently being compressed. If that + // frame was already partially written out (for example by an earlier + // flush()), those bytes cannot be discarded and the next frame will be + // appended to an incomplete frame, producing an unreadable stream. zstd + // requires internal buffers to be fully flushed before a new compression job + // starts, so refuse instead of silently corrupting the output. + if (!frame_complete_) { + return CompressionError( + "Cannot reset a zstd stream with an incomplete frame; end the frame " + "or discard the output produced so far", + "ERR_ZLIB_INCOMPLETE_FRAME", + ZSTD_error_stage_wrong); + } + size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only); if (ZSTD_isError(result)) { const ZSTD_ErrorCode error = ZSTD_getErrorCode(result); @@ -1755,15 +1777,20 @@ void ZstdCompressContext::DoThreadPoolWork() { error_ = ZSTD_getErrorCode(remaining); error_code_string_ = ZstdStrerror(error_); error_string_ = ZSTD_getErrorString(error_); - } else if (remaining == 0 && flush_ == ZSTD_e_end && - consumed_src_size_.has_value()) { - uint64_t const consumed_src_size = *consumed_src_size_; - consumed_src_size_.reset(); - if (consumed_src_size != pledged_src_size_) { - error_ = ZSTD_error_srcSize_wrong; - error_code_string_ = ZstdStrerror(error_); - error_string_ = ZSTD_getErrorString(error_); + frame_complete_ = false; + } else if (remaining == 0 && flush_ == ZSTD_e_end) { + frame_complete_ = true; + if (consumed_src_size_.has_value()) { + uint64_t const consumed_src_size = *consumed_src_size_; + consumed_src_size_.reset(); + if (consumed_src_size != pledged_src_size_) { + error_ = ZSTD_error_srcSize_wrong; + error_code_string_ = ZstdStrerror(error_); + error_string_ = ZSTD_getErrorString(error_); + } } + } else { + frame_complete_ = false; } } diff --git a/test/parallel/test-zlib-zstd-reset-incomplete-frame.js b/test/parallel/test-zlib-zstd-reset-incomplete-frame.js new file mode 100644 index 000000000000..182c9e64da1f --- /dev/null +++ b/test/parallel/test-zlib-zstd-reset-incomplete-frame.js @@ -0,0 +1,66 @@ +'use strict'; + +// Tests that reset() refuses to run while a zstd frame is still in progress. +// +// A frame is only complete once ZSTD_compressStream2() has been called with +// ZSTD_e_end and returned 0. Resetting earlier dropped the frame state, and any +// bytes already written out (for example by flush()) stayed in the output +// stream as an unusable fragment. The next frame was then appended to that +// fragment, so the resulting stream could not be decompressed. + +require('../common'); +const assert = require('assert'); +const { finished } = require('stream/promises'); +const test = require('node:test'); +const zlib = require('zlib'); + +test('ZstdCompress reset throws when a frame is incomplete', async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + + // The frame started by write() is still open, so reset() must refuse + // instead of silently producing a stream that cannot be decoded. + stream.reset(); + stream.end(Buffer.from('world')); + + await assert.rejects(finished(stream), { + code: 'ERR_ZLIB_INCOMPLETE_FRAME', + }); +}); + +test('ZstdCompress flush followed by end still produces a valid stream', + async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + stream.write(Buffer.from('hello')); + await new Promise((resolve) => stream.flush(resolve)); + stream.end(Buffer.from('world')); + await finished(stream); + + assert.strictEqual( + zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(), + 'helloworld', + ); + }); + +test('ZstdCompress reset before any write still works', async () => { + const stream = zlib.createZstdCompress(); + const chunks = []; + stream.on('data', (chunk) => chunks.push(chunk)); + + // No frame has been started yet, so reset() is allowed. + stream.reset(); + stream.end(Buffer.from('hello')); + await finished(stream); + + assert.strictEqual( + zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(), + 'hello', + ); +});