zlib: reject reset while a zstd frame is incomplete

Resetting a ZstdCompress stream while a frame is still in
progress dropped the frame state, but any bytes already
written out stayed at the start of the output stream. The
next frame was then appended to that fragment, so the
resulting stream could not be decompressed. The failure was
silent: the compressor reported no error at all.

zstd requires internal buffers to be fully flushed before a
new compression job starts. Track whether the current frame
has completed and throw ERR_ZLIB_INCOMPLETE_FRAME from
reset() otherwise.

Signed-off-by: bun-unsafe <bun-unsafe@users.noreply.github.com>

Author: bun-unsafe

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

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)

src/node_zlib.cc

@@ -352,6 +352,13 @@ class ZstdCompressContext final : public ZstdContext {
 
   uint64_t pledged_src_size_ = ZSTD_CONTENTSIZE_UNKNOWN;
   std::optional<uint64_t> 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;
   }
 }
 

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