stream: pipe should not swallow error

PR-URL: https://github.com/nodejs/node/pull/30993
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

lib/_stream_readable.js

@@ -798,7 +798,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     unpipe();
     dest.removeListener('error', onerror);
     if (EE.listenerCount(dest, 'error') === 0) {
-      if (!dest.destroyed) {
+      const s = dest._writableState || dest._readableState;
+      if (s && !s.errorEmitted) {
+        // User incorrectly emitted 'error' directly on the stream.
         errorOrDestroy(dest, er);
       } else {
         dest.emit('error', er);

test/parallel/test-stream2-finish-pipe-error.js

@@ -0,0 +1,20 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+
+process.on('uncaughtException', common.mustCall());
+
+const r = new stream.Readable();
+r._read = function(size) {
+  r.push(Buffer.allocUnsafe(size));
+};
+
+const w = new stream.Writable();
+w._write = function(data, encoding, cb) {
+  cb(null);
+};
+
+r.pipe(w);
+
+// end() after pipe should cause unhandled exception
+w.end();

test/parallel/test-stream2-finish-pipe.js

@@ -35,7 +35,10 @@ w._write = function(data, encoding, cb) {
 
 r.pipe(w);
 
-// This might sound unrealistic, but it happens in net.js. When
-// `socket.allowHalfOpen === false`, EOF will cause `.destroySoon()` call which
-// ends the writable side of net.Socket.
-w.end();
+// end() must be called in nextTick or a WRITE_AFTER_END error occurs.
+process.nextTick(() => {
+  // This might sound unrealistic, but it happens in net.js. When
+  // socket.allowHalfOpen === false, EOF will cause .destroySoon() call which
+  // ends the writable side of net.Socket.
+  w.end();
+});