stream: fix destroy() behavior

Ensure errorEmitted is always set. Only emit 'error' once.

PR-URL: https://github.com/nodejs/node/pull/29058
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Franziska Hinkelmann <franziska.hinkelmann@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

6 files changed, 87 insertions, 35 deletions

diff --git a/test/parallel/test-net-connect-buffer.js b/test/parallel/test-net-connect-buffer.js
index 8c95400fb6..41df6a0666 100644
--- a/test/parallel/test-net-connect-buffer.js
+++ b/test/parallel/test-net-connect-buffer.js
@@ -69,12 +69,14 @@ tcp.listen(0, common.mustCall(function() {
     [],
     {}
   ].forEach((value) => {
-    common.expectsError(() => socket.write(value), {
+    // We need to check the callback since 'error' will only
+    // be emitted once per instance.
+    socket.write(value, common.expectsError({
       code: 'ERR_INVALID_ARG_TYPE',
       type: TypeError,
       message: 'The "chunk" argument must be one of type string or Buffer. ' +
                `Received type ${typeof value}`
-    });
+    }));
   });
 
   // Write a string that contains a multi-byte character sequence to test that
diff --git a/test/parallel/test-stream-error-once.js b/test/parallel/test-stream-error-once.js
new file mode 100644
index 0000000000..71f268cfa4
--- /dev/null
+++ b/test/parallel/test-stream-error-once.js
@@ -0,0 +1,19 @@
+'use strict';
+const common = require('../common');
+const { Writable, Readable } = require('stream');
+
+{
+  const writable = new Writable();
+  writable.on('error', common.mustCall());
+  writable.end();
+  writable.write('h');
+  writable.write('h');
+}
+
+{
+  const readable = new Readable();
+  readable.on('error', common.mustCall());
+  readable.push(null);
+  readable.push('h');
+  readable.push('h');
+}
diff --git a/test/parallel/test-stream-readable-invalid-chunk.js b/test/parallel/test-stream-readable-invalid-chunk.js
index fcd7414bb6..3e147c065f 100644
--- a/test/parallel/test-stream-readable-invalid-chunk.js
+++ b/test/parallel/test-stream-readable-invalid-chunk.js
@@ -3,17 +3,32 @@
 const common = require('../common');
 const stream = require('stream');
 
-const readable = new stream.Readable({
-  read: () => {}
-});
-
-function checkError(fn) {
-  common.expectsError(fn, {
+function testPushArg(val) {
+  const readable = new stream.Readable({
+    read: () => {}
+  });
+  readable.on('error', common.expectsError({
     code: 'ERR_INVALID_ARG_TYPE',
     type: TypeError
+  }));
+  readable.push(val);
+}
+
+testPushArg([]);
+testPushArg({});
+testPushArg(0);
+
+function testUnshiftArg(val) {
+  const readable = new stream.Readable({
+    read: () => {}
   });
+  readable.on('error', common.expectsError({
+    code: 'ERR_INVALID_ARG_TYPE',
+    type: TypeError
+  }));
+  readable.unshift(val);
 }
 
-checkError(() => readable.push([]));
-checkError(() => readable.push({}));
-checkError(() => readable.push(0));
+testUnshiftArg([]);
+testUnshiftArg({});
+testUnshiftArg(0);
diff --git a/test/parallel/test-stream-readable-unshift.js b/test/parallel/test-stream-readable-unshift.js
index d574b7d046..6eefb55d73 100644
--- a/test/parallel/test-stream-readable-unshift.js
+++ b/test/parallel/test-stream-readable-unshift.js
@@ -113,23 +113,6 @@ const { Readable } = require('stream');
 }
 
 {
-  // Check that error is thrown for invalid chunks
-
-  const readable = new Readable({ read() {} });
-  function checkError(fn) {
-    common.expectsError(fn, {
-      code: 'ERR_INVALID_ARG_TYPE',
-      type: TypeError
-    });
-  }
-
-  checkError(() => readable.unshift([]));
-  checkError(() => readable.unshift({}));
-  checkError(() => readable.unshift(0));
-
-}
-
-{
   // Check that ObjectMode works
   const readable = new Readable({ objectMode: true, read() {} });
 
diff --git a/test/parallel/test-stream-unshift-read-race.js b/test/parallel/test-stream-unshift-read-race.js
index f7b633338c..562e10776e 100644
--- a/test/parallel/test-stream-unshift-read-race.js
+++ b/test/parallel/test-stream-unshift-read-race.js
@@ -86,13 +86,7 @@ w._write = function(chunk, encoding, cb) {
 };
 
 r.on('end', common.mustCall(function() {
-  common.expectsError(function() {
-    r.unshift(Buffer.allocUnsafe(1));
-  }, {
-    code: 'ERR_STREAM_UNSHIFT_AFTER_END_EVENT',
-    type: Error,
-    message: 'stream.unshift() after end event'
-  });
+  r.unshift(Buffer.allocUnsafe(1));
   w.end();
 }));
 
diff --git a/test/parallel/test-stream2-writable.js b/test/parallel/test-stream2-writable.js
index 262606d906..b20f5d3f18 100644
--- a/test/parallel/test-stream2-writable.js
+++ b/test/parallel/test-stream2-writable.js
@@ -402,3 +402,42 @@ const helloWorldBuffer = Buffer.from('hello world');
   w.write(Buffer.allocUnsafe(1));
   w.end(Buffer.allocUnsafe(0));
 }
+
+{
+  // Verify that error is only emitted once when failing in _finish.
+  const w = new W();
+
+  w._final = common.mustCall(function(cb) {
+    cb(new Error('test'));
+  });
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.message, 'test');
+    w.on('error', common.mustNotCall());
+    w.destroy(new Error());
+  }));
+  w.end();
+}
+
+{
+  // Verify that error is only emitted once when failing in write.
+  const w = new W();
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.code, 'ERR_STREAM_NULL_VALUES');
+  }));
+  w.write(null);
+  w.destroy(new Error());
+}
+
+{
+  // Verify that error is only emitted once when failing in write after end.
+  const w = new W();
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+  w.end();
+  w.write('hello');
+  w.destroy(new Error());
+}