zlib: fix zstd reset

doc/api/zlib.md

@@ -2158,7 +2158,8 @@ added: v0.5.8
 -->
 
 * `kind` **Default:** `zlib.constants.Z_FULL_FLUSH` for zlib-based streams,
-  `zlib.constants.BROTLI_OPERATION_FLUSH` for Brotli-based streams.
+  `zlib.constants.BROTLI_OPERATION_FLUSH` for Brotli-based streams, and
+  `zlib.constants.ZSTD_e_flush` for Zstd-based streams.
 * `callback` {Function}
 
 Flush pending data. Don't call this frivolously, premature flushes negatively
@@ -2190,8 +2191,14 @@ Only applicable to deflate algorithm.
 added: v0.7.0
 -->
 
-Reset the compressor/decompressor to factory defaults. Only applicable to
-the inflate and deflate algorithms.
+For inflate and deflate streams, reset the compressor/decompressor to factory
+defaults.
+
+For Zstd streams, cancel the current frame and start a new session while
+preserving the configured parameters and dictionary. If `pledgedSrcSize` was
+configured for a Zstd compressor, it applies again to the next frame.
+
+Calling `reset()` while a write is in progress throws an `Error`.
 
 ## Class: `ZstdOptions`
 
@@ -2223,6 +2230,9 @@ Each Zstd-based class takes an `options` object. All options are optional.
 * `finishFlush` {integer} **Default:** `zlib.constants.ZSTD_e_end`
 * `chunkSize` {integer} **Default:** `16 * 1024`
 * `params` {Object} Key-value object containing indexed [Zstd parameters][].
+* `pledgedSrcSize` {number} Expected total size of the uncompressed input. It
+  must be a non-negative safe integer and must match the input size when
+  compression finishes. Only applicable to Zstd compressors.
 * `maxOutputLength` {integer} Limits output size when using
   [convenience methods][]. **Default:** [`buffer.kMaxLength`][]
 * `info` {boolean} If `true`, returns an object with `buffer` and `engine`. **Default:** `false`
@@ -3062,6 +3072,8 @@ Compress a chunk of data with [`ZstdCompress`][].
 
 ### `zlib.zstdDecompress(buffer[, options], callback)`
 
+> Stability: 1 - Experimental
+
 <!-- YAML
 added:
   - v23.8.0

lib/zlib.js

@@ -907,7 +907,11 @@ class Zstd extends ZlibBase {
       if (isAnyArrayBuffer(dictionary)) {
         dictionary = new Uint8Array(dictionary);
       } else {
-        dictionary = undefined;
+        throw new ERR_INVALID_ARG_TYPE(
+          'options.dictionary',
+          ['Buffer', 'TypedArray', 'DataView', 'ArrayBuffer'],
+          dictionary,
+        );
       }
     }
 

src/node_zlib.cc

@@ -1703,7 +1703,29 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size,
 }
 
 CompressionError ZstdCompressContext::ResetStream() {
-  return Init(pledged_src_size_);
+  size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only);
+  if (ZSTD_isError(result)) {
+    const ZSTD_ErrorCode error = ZSTD_getErrorCode(result);
+    return CompressionError(
+        ZSTD_getErrorString(error), ZstdStrerror(error), error);
+  }
+
+  result = ZSTD_CCtx_setPledgedSrcSize(cctx_.get(), pledged_src_size_);
+  if (ZSTD_isError(result)) {
+    const ZSTD_ErrorCode error = ZSTD_getErrorCode(result);
+    return CompressionError(
+        ZSTD_getErrorString(error), ZstdStrerror(error), error);
+  }
+
+  if (pledged_src_size_ == ZSTD_CONTENTSIZE_UNKNOWN) {
+    consumed_src_size_.reset();
+  } else {
+    consumed_src_size_ = 0;
+  }
+  error_ = ZSTD_error_no_error;
+  error_string_.clear();
+  error_code_string_.clear();
+  return {};
 }
 
 void ZstdCompressContext::DoThreadPoolWork() {
@@ -1777,9 +1799,18 @@ CompressionError ZstdDecompressContext::Init(uint64_t pledged_src_size,
 }
 
 CompressionError ZstdDecompressContext::ResetStream() {
-  // We pass ZSTD_CONTENTSIZE_UNKNOWN because the argument is ignored for
-  // decompression.
-  return Init(ZSTD_CONTENTSIZE_UNKNOWN);
+  const size_t result = ZSTD_DCtx_reset(dctx_.get(), ZSTD_reset_session_only);
+  if (ZSTD_isError(result)) {
+    const ZSTD_ErrorCode error = ZSTD_getErrorCode(result);
+    return CompressionError(
+        ZSTD_getErrorString(error), ZstdStrerror(error), error);
+  }
+
+  frame_complete_ = false;
+  error_ = ZSTD_error_no_error;
+  error_string_.clear();
+  error_code_string_.clear();
+  return {};
 }
 
 void ZstdDecompressContext::DoThreadPoolWork() {

test/parallel/test-zlib-zstd-dictionary.js

@@ -40,3 +40,24 @@ for (const dict of [arrayBuffer, uint8, dataView]) {
   const decompressed = zlib.zstdDecompressSync(compressed, { dictionary: dict });
   assert.strictEqual(decompressed.toString(), input.toString());
 }
+
+for (const dictionary of [null, 'string', 123, true, {}, [1, 2, 3]]) {
+  const options = { dictionary };
+  const expected = {
+    code: 'ERR_INVALID_ARG_TYPE',
+    name: 'TypeError',
+  };
+
+  assert.throws(() => zlib.createZstdCompress(options), expected);
+  assert.throws(() => zlib.createZstdDecompress(options), expected);
+  assert.throws(() => zlib.zstdCompressSync(input, options), expected);
+  assert.throws(() => zlib.zstdDecompressSync(input, options), expected);
+  assert.throws(
+    () => zlib.zstdCompress(input, options, common.mustNotCall()),
+    expected,
+  );
+  assert.throws(
+    () => zlib.zstdDecompress(input, options, common.mustNotCall()),
+    expected,
+  );
+}

test/parallel/test-zlib-zstd-reset.js

@@ -0,0 +1,68 @@
+'use strict';
+
+require('../common');
+const assert = require('assert');
+const { finished } = require('stream/promises');
+const test = require('node:test');
+const zlib = require('zlib');
+
+const dictionary = Buffer.from(
+  'Lorem ipsum dolor sit amet, consectetur adipiscing elit. ' +
+  'Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.',
+);
+const input = Buffer.from(
+  'Lorem ipsum dolor sit amet, consectetur adipiscing elit. '.repeat(100),
+);
+
+async function collect(stream, ...data) {
+  const chunks = [];
+  stream.on('data', (chunk) => chunks.push(chunk));
+  for (let i = 0; i < data.length - 1; i++) {
+    stream.write(data[i]);
+  }
+  stream.end(data[data.length - 1]);
+  await finished(stream);
+  return Buffer.concat(chunks);
+}
+
+test('ZstdCompress reset preserves its initial options', async () => {
+  const options = {
+    dictionary,
+    pledgedSrcSize: input.length,
+    params: {
+      [zlib.constants.ZSTD_c_compressionLevel]: 19,
+      [zlib.constants.ZSTD_c_checksumFlag]: 1,
+    },
+  };
+  const expected = await collect(zlib.createZstdCompress(options), input);
+  const reset = zlib.createZstdCompress(options);
+  reset.reset();
+
+  assert.deepStrictEqual(await collect(reset, input), expected);
+});
+
+test('ZstdDecompress reset preserves its dictionary', async () => {
+  const compressed = zlib.zstdCompressSync(input, { dictionary });
+  const decompress = zlib.createZstdDecompress({ dictionary });
+  decompress.reset();
+
+  assert.deepStrictEqual(await collect(decompress, compressed), input);
+});
+
+test('ZstdDecompress reset preserves its parameters', async () => {
+  const compressed = await collect(zlib.createZstdCompress({
+    params: {
+      [zlib.constants.ZSTD_c_windowLog]: 11,
+    },
+  }), Buffer.alloc(2048), Buffer.alloc(2048));
+  const decompress = zlib.createZstdDecompress({
+    params: {
+      [zlib.constants.ZSTD_d_windowLogMax]: 10,
+    },
+  });
+  decompress.reset();
+
+  await assert.rejects(collect(decompress, compressed), {
+    code: 'ZSTD_error_frameParameter_windowTooLarge',
+  });
+});