stream: pipeline should only destroy un-finished streams.

This PR logically reverts nodejs#31940 which
has caused lots of unnecessary breakage in the ecosystem.

This PR also aligns better with the actual documented behavior:

`stream.pipeline()` will call `stream.destroy(err)` on all streams except:
  * `Readable` streams which have emitted `'end'` or `'close'`.
  * `Writable` streams which have emitted `'finish'` or `'close'`.

The behavior introduced in nodejs#31940
was much more aggressive in terms of destroying streams. This was
good for avoiding potential resources leaks however breaks some
common assumputions in legacy streams.

Furthermore, it makes the code simpler and removes some hacks.

Fixes: nodejs#32954
Fixes: nodejs#32955

lib/internal/streams/pipeline.js

@@ -25,43 +25,18 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
-function isIncoming(stream) {
-  return (
-    stream.socket &&
-    typeof stream.complete === 'boolean' &&
-    ArrayIsArray(stream.rawTrailers) &&
-    ArrayIsArray(stream.rawHeaders)
-  );
-}
-
-function isOutgoing(stream) {
-  return (
-    stream.socket &&
-    typeof stream.setHeader === 'function'
-  );
-}
-
-function destroyer(stream, reading, writing, final, callback) {
-  const _destroy = once((err) => {
-    if (!err && (isIncoming(stream) || isOutgoing(stream))) {
-      // http/1 request objects have a coupling to their response and should
-      // not be prematurely destroyed. Assume they will handle their own
-      // lifecycle.
-      return callback();
-    }
-
-    if (!err && reading && !writing && stream.writable) {
-      return callback();
-    }
+function destroyer(stream, reading, writing, callback) {
+  callback = once(callback);
 
-    if (err || !final || !stream.readable) {
-      destroyImpl.destroyer(stream, err);
-    }
-    callback(err);
+  let finished = false;
+  stream.on('close', () => {
+    finished = true;
   });
 
   if (eos === undefined) eos = require('internal/streams/end-of-stream');
   eos(stream, { readable: reading, writable: writing }, (err) => {
+    finished = !err;
+
     const rState = stream._readableState;
     if (
       err &&
@@ -78,14 +53,19 @@ function destroyer(stream, reading, writing, final, callback) {
       // eos will only fail with premature close on the reading side for
       // duplex streams.
       stream
-        .once('end', _destroy)
-        .once('error', _destroy);
+        .once('end', callback)
+        .once('error', callback);
     } else {
-      _destroy(err);
+      callback(err);
     }
   });
 
-  return (err) => _destroy(err || new ERR_STREAM_DESTROYED('pipe'));
+  return once((err) => {
+    if (!finished) {
+      destroyImpl.destroyer(stream, err);
+    }
+    callback(err || new ERR_STREAM_DESTROYED('pipe'));
+  });
 }
 
 function popCallback(streams) {
@@ -204,7 +184,7 @@ function pipeline(...streams) {
 
     if (isStream(stream)) {
       finishCount++;
-      destroys.push(destroyer(stream, reading, writing, !reading, finish));
+      destroys.push(destroyer(stream, reading, writing, finish));
     }
 
     if (i === 0) {
@@ -262,7 +242,7 @@ function pipeline(...streams) {
         ret = pt;
 
         finishCount++;
-        destroys.push(destroyer(ret, false, true, true, finish));
+        destroys.push(destroyer(ret, false, true, finish));
       }
     } else if (isStream(stream)) {
       if (isReadable(ret)) {

test/parallel/test-stream-pipeline.js

@@ -916,7 +916,7 @@ const { promisify } = require('util');
   const src = new PassThrough({ autoDestroy: false });
   const dst = new PassThrough({ autoDestroy: false });
   pipeline(src, dst, common.mustCall(() => {
-    assert.strictEqual(src.destroyed, true);
+    assert.strictEqual(src.destroyed, false);
     assert.strictEqual(dst.destroyed, false);
   }));
   src.end();