node:zlib: reset a zstd stream in place, keep its dictionary and parameters

patches/zstd/mt-clear-job-ready.patch

@@ -0,0 +1,35 @@
+Clear jobReady when the job table is erased.
+
+With nbWorkers >= 1, ZSTDMT_createCompressionJob() sets mtctx->jobReady when
+it has prepared a job and no worker is free to take it.
+ZSTDMT_releaseAllJobResources() erases every job description, the prepared one
+too, and leaves the flag set. ZSTDMT_initCStream_internal() does not clear it
+either.
+
+The next frame on that context then skips the preparation and posts the
+erased description. The worker calls ZSTDMT_getCCtx(NULL) and the process gets
+SIGSEGV (zstdmt_compress.c:697).
+
+The caller does not have to reset anything to get there. zstd erases the table
+and starts a new session by itself when a job fails, and when
+ZSTDMT_compressStream_generic() returns stage_wrong. A frame on the same
+context after one of those, while a job waited for a worker, crashes.
+
+ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) in an open frame is one more
+way in. reset() of a node:zlib zstd stream makes that call, and Node v26.10.0
+exits with SIGSEGV there.
+
+facebook/zstd has the same code on its dev branch, and no issue there reports
+it (checked 2026-09-29).
+
+--- a/lib/compress/zstdmt_compress.c
++++ b/lib/compress/zstdmt_compress.c
+@@ -1023,6 +1023,8 @@
+     }
+     mtctx->inBuff.buffer = g_nullBuffer;
+     mtctx->inBuff.filled = 0;
++    /* Bun: the loop above erased the job that jobReady refers to. */
++    mtctx->jobReady = 0;
+     mtctx->allJobsCompleted = 1;
+ }
+

scripts/build/deps/zstd.ts

@@ -41,10 +41,17 @@ export const zstd: Dependency = {
     commit: ZSTD_COMMIT,
   }),
 
-  // x64 targets nehalem, so zstd picks its BMI2 kernels at run time and
-  // probes CPUID in every CCtx/DCtx init. CPUID is a VM exit under a
-  // hypervisor (about 2 us each, two per init). Probe once instead.
-  patches: ["patches/zstd/bmi2-probe-once.patch"],
+  patches: [
+    // x64 targets nehalem, so zstd picks its BMI2 kernels at run time and
+    // probes CPUID in every CCtx/DCtx init. CPUID is a VM exit under a
+    // hypervisor (about 2 us each, two per init). Probe once instead.
+    "patches/zstd/bmi2-probe-once.patch",
+    // With nbWorkers >= 1, zstd keeps its jobReady flag when it erases the
+    // job table, so the next frame on that context posts an erased job to a
+    // worker (SIGSEGV). zstd gets there by itself after a failed job, and
+    // node:zlib reset() gets there too. Not reported upstream yet.
+    "patches/zstd/mt-clear-job-ready.patch",
+  ],
 
   build: cfg => {
     const sources = [...SOURCES];

src/runtime/node/node_zlib_binding.rs

@@ -731,7 +731,7 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
         global_this: &JSGlobalObject,
         callframe: &CallFrame,
     ) -> JsResult<JSValue> {
-        // reset() destroys and re-creates the brotli/zstd encoder state (or
+        // reset() re-creates the brotli encoder state (or resets the zstd session, or
         // mutates the z_stream). Doing so while an async write is running on
         // the threadpool would be a use-after-free / data race, so node throws
         // a plain Error here rather than touching live state.

src/runtime/node/zlib/NativeZstd.rs

@@ -458,11 +458,48 @@ mod _impl {
             }
         }
 
+        /// Keeps the dictionary and parameters, as node does since v26.10.0 (nodejs/node#65867).
         pub(crate) fn reset(&mut self) -> Error {
-            // Matches node's `ZstdContext::ResetStream()`, which calls `Init()`
-            // with its default (empty) dictionary — a reset drops the dictionary.
-            // `init` frees the previous context itself.
-            self.init(self.pledged_src_size, None)
+            // No `..`: a field added to `Context` must be kept or cleared here to compile.
+            let Self {
+                mode,
+                state,
+                pledged_src_size,
+                flush: _,
+                input: _,
+                output: _,
+                remaining: _,
+            } = *self;
+            // JS can reach this with no context: init() was never called, or it failed.
+            let Some(state) = state else {
+                return Error::OK;
+            };
+            let result = match mode {
+                NodeMode::ZSTD_COMPRESS => {
+                    // SAFETY: state is a valid CCtx set by init().
+                    let result =
+                        unsafe { c::ZSTD_CCtx_reset(state.cast(), c::ZSTD_reset_session_only) };
+                    if c::ZSTD_isError(result) > 0 {
+                        result
+                    } else {
+                        // A session reset clears the pledged size: zstd keeps it for one frame.
+                        // SAFETY: state is a valid CCtx set by init().
+                        unsafe {
+                            c::ZSTD_CCtx_setPledgedSrcSize(state.cast(), pledged_src_size as _)
+                        }
+                    }
+                }
+                // SAFETY: state is a valid DCtx set by init().
+                NodeMode::ZSTD_DECOMPRESS => unsafe {
+                    c::ZSTD_DCtx_reset(state.cast(), c::ZSTD_reset_session_only)
+                },
+                _ => unreachable!(),
+            };
+            if c::ZSTD_isError(result) == 0 {
+                return Error::OK;
+            }
+            self.remaining = result as u64;
+            self.get_error_info()
         }
 
         /// Frees the Zstd encoder/decoder state without changing mode.

src/zstd/lib.rs

@@ -37,6 +37,8 @@ pub mod c {
     // ZSTD_cParameter
     pub const ZSTD_c_compressionLevel: ZSTD_cParameter = 100;
 
+    // ZSTD_ResetDirective
+    pub const ZSTD_reset_session_only: ZSTD_ResetDirective = 1;
     pub const ZSTD_reset_session_and_parameters: ZSTD_ResetDirective = 3;
 
     // ZSTD_ErrorCode (zstd_errors.h) — only the public stable subset.

test/js/node/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',
+  });
+});