stream: fix 0 transform hwm backpressure

lib/internal/streams/transform.js

@@ -65,14 +65,16 @@
 
 const {
   ObjectSetPrototypeOf,
-  Symbol
+  Symbol,
+  MathMax
 } = primordials;
 
 module.exports = Transform;
 const {
   ERR_METHOD_NOT_IMPLEMENTED
 } = require('internal/errors').codes;
 const Duplex = require('internal/streams/duplex');
+const { getHighWaterMark } = require('internal/streams/state');
 ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
 ObjectSetPrototypeOf(Transform, Duplex);
 
@@ -82,6 +84,26 @@ function Transform(options) {
   if (!(this instanceof Transform))
     return new Transform(options);
 
+  // TODO (ronag): This should preferably always be
+  // applied but would be semver-major. Or even better
+  // Make transform a Readable with the Writable interface.
+  if (options && options.highWaterMark === 0) {
+    // A Duplex will buffer both on the writable and readable side while
+    // a Transform just wants to buffer hwm number of elements. To avoid
+    // buffering twice we disable buffering on the writable side.
+    const readableHighWaterMark = getHighWaterMark(this, options, 'readableHighWaterMark', true);
+    options = {
+      ...options,
+      highWaterMark: null,
+      readableHighWaterMark: MathMax(1, readableHighWaterMark),
+      // TODO (ronag) This is not optimal since we have
+      // a "bug" where we check needDrain before calling _write and not after.
+      // Refs: https://github.com/nodejs/node/pull/32887
+      // Refs: https://github.com/nodejs/node/pull/35941
+      writableHighWaterMark: 0
+    };
+  }
+
   Duplex.call(this, options);
 
   // We have implemented the _read method, and done the other things
@@ -164,9 +186,7 @@ Transform.prototype._write = function(chunk, encoding, callback) {
     if (
       wState.ended || // Backwards compat.
       length === rState.length || // Backwards compat.
-      rState.length < rState.highWaterMark ||
-      rState.highWaterMark === 0 ||
-      rState.length === 0
+      rState.length < rState.highWaterMark
     ) {
       callback();
     } else {

test/parallel/test-stream-passthrough-drain.js

@@ -1,8 +1,9 @@
 'use strict';
 const common = require('../common');
+const assert = require('assert');
 const { PassThrough } = require('stream');
 
 const pt = new PassThrough({ highWaterMark: 0 });
 pt.on('drain', common.mustCall());
-pt.write('hello');
+assert(!pt.write('hello1'));
 pt.read();

test/parallel/test-stream-transform-hwm0.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Transform } = require('stream');
+
+const t = new Transform({
+  objectMode: true, highWaterMark: 0,
+  transform(chunk, enc, callback) {
+    process.nextTick(() => callback(null, chunk, enc));
+  }
+});
+
+assert.strictEqual(t.write(1), false);
+t.on('drain', common.mustCall(() => {
+  assert.strictEqual(t.write(2), false);
+  t.end();
+}));
+
+t.once('readable', common.mustCall(() => {
+  assert.strictEqual(t.read(), 1);
+  assert.strictEqual(t.read(), null);
+  t.once('readable', common.mustCall(() => {
+    assert.strictEqual(t.read(), 2);
+  }));
+}));

test/parallel/test-stream-transform-split-highwatermark.js

@@ -39,21 +39,6 @@ testTransform(555, 555, {
   writableHighWaterMark: 777,
 });
 
-// Test highWaterMark = 0 overriding
-testTransform(0, 0, {
-  highWaterMark: 0,
-  readableHighWaterMark: 666,
-});
-testTransform(0, 0, {
-  highWaterMark: 0,
-  writableHighWaterMark: 777,
-});
-testTransform(0, 0, {
-  highWaterMark: 0,
-  readableHighWaterMark: 666,
-  writableHighWaterMark: 777,
-});
-
 // Test undefined, null
 [undefined, null].forEach((v) => {
   testTransform(DEFAULT, DEFAULT, { readableHighWaterMark: v });