zlib: avoid an extra empty zstd frame

src/node_zlib.cc

@@ -352,6 +352,7 @@ class ZstdCompressContext final : public ZstdContext {
 
   uint64_t pledged_src_size_ = ZSTD_CONTENTSIZE_UNKNOWN;
   std::optional<uint64_t> consumed_src_size_;
+  bool frame_ended_ = false;
 };
 
 class ZstdDecompressContext final : public ZstdContext {
@@ -1678,6 +1679,7 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size,
                                            std::string_view dictionary,
                                            bool) {
   pledged_src_size_ = pledged_src_size;
+  frame_ended_ = false;
   if (pledged_src_size == ZSTD_CONTENTSIZE_UNKNOWN) {
     consumed_src_size_.reset();
   } else {
@@ -1737,20 +1739,26 @@ CompressionError ZstdCompressContext::ResetStream() {
   } else {
     consumed_src_size_ = 0;
   }
+  frame_ended_ = false;
   error_ = ZSTD_error_no_error;
   error_string_.clear();
   error_code_string_.clear();
   return {};
 }
 
 void ZstdCompressContext::DoThreadPoolWork() {
+  // Ending an already ended frame would append an empty one.
+  if (frame_ended_ && input_.size == 0) {
+    return;
+  }
   // Zstd overrides a configured pledge when the first call uses ZSTD_e_end.
   size_t const input_pos = input_.pos;
   size_t const remaining =
       ZSTD_compressStream2(cctx_.get(), &output_, &input_, flush_);
   if (consumed_src_size_.has_value()) {
     *consumed_src_size_ += input_.pos - input_pos;
   }
+  frame_ended_ = remaining == 0 && flush_ == ZSTD_e_end;
   if (ZSTD_isError(remaining)) {
     error_ = ZSTD_getErrorCode(remaining);
     error_code_string_ = ZstdStrerror(error_);

test/parallel/test-zlib-zstd-compress-single-frame.js

@@ -0,0 +1,64 @@
+'use strict';
+
+// Regression test for https://github.com/nodejs/node/issues/66078: ending a
+// ZstdCompress stream with writes still queued appended an empty frame.
+
+const common = require('../common');
+const assert = require('assert');
+const { finished } = require('stream/promises');
+const zlib = require('zlib');
+
+const kMagic = Buffer.from([0x28, 0xb5, 0x2f, 0xfd]);
+
+function countFrames(buffer) {
+  let count = 0;
+  for (let i = buffer.indexOf(kMagic); i !== -1; i = buffer.indexOf(kMagic, i + 1)) {
+    count++;
+  }
+  return count;
+}
+
+async function compress(use) {
+  const stream = zlib.createZstdCompress();
+  const chunks = [];
+  stream.on('data', (chunk) => chunks.push(chunk));
+  await use(stream);
+  stream.end();
+  await finished(stream);
+  return Buffer.concat(chunks);
+}
+
+(async () => {
+  const expected = await compress((stream) => stream.write('hello world'));
+  assert.strictEqual(countFrames(expected), 1);
+
+  assert.deepStrictEqual(await compress((stream) => {
+    stream.write('hello ');
+    stream.write('world');
+  }), expected);
+
+  assert.deepStrictEqual(await compress((stream) => {
+    stream.write('hello world');
+    stream.flush(zlib.constants.ZSTD_e_end);
+  }), expected);
+
+  // The empty flush has to complete before end(), which would otherwise
+  // promote it to ZSTD_e_end.
+  assert.deepStrictEqual(await compress((stream) => {
+    stream.write('hello world');
+    stream.flush(zlib.constants.ZSTD_e_end);
+    return new Promise((resolve) => stream.flush(resolve));
+  }), expected);
+
+  const empty = await compress(() => {});
+  assert.strictEqual(countFrames(empty), 1);
+  assert.strictEqual(zlib.zstdDecompressSync(empty).length, 0);
+
+  const twoFrames = await compress((stream) => {
+    stream.write('hello ');
+    stream.flush(zlib.constants.ZSTD_e_end);
+    stream.write('world');
+  });
+  assert.strictEqual(countFrames(twoFrames), 2);
+  assert.strictEqual(zlib.zstdDecompressSync(twoFrames).toString(), 'hello world');
+})().then(common.mustCall());