stream: ensure pipeline always destroys streams

There was an edge case where an incorrect assumption was made
in regardos whether eos/finished means that the stream is
actually destroyed or not.

Author: ronag

lib/internal/streams/pipeline.js

@@ -27,27 +27,20 @@ let createReadableStreamAsyncIterator;
 
 function destroyer(stream, reading, writing, callback) {
   callback = once(callback);
-
-  let closed = false;
-  stream.on('close', () => {
-    closed = true;
-  });
+  let destroyed = false;
 
   if (eos === undefined) eos = require('internal/streams/end-of-stream');
   eos(stream, { readable: reading, writable: writing }, (err) => {
-    if (err) return callback(err);
-    closed = true;
-    callback();
+    if (destroyed) return;
+    destroyed = true;
+    destroyImpl.destroyer(stream, err);
+    callback(err);
   });
 
-  let destroyed = false;
   return (err) => {
-    if (closed) return;
     if (destroyed) return;
     destroyed = true;
-
     destroyImpl.destroyer(stream, err);
-
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };
 }

test/parallel/test-stream-pipeline.js

@@ -763,7 +763,10 @@ const { promisify } = require('util');
     s.emit('data', 'asd');
     s.emit('end');
   });
-  s.close = common.mustCall();
+  // 'destroyer' can be called multiple times,
+  // once from stream wrapper and
+  // once from iterator wrapper.
+  s.close = common.mustCallAtLeast(1);
   let ret = '';
   pipeline(s, async function(source) {
     for await (const chunk of source) {
@@ -909,3 +912,13 @@ const { promisify } = require('util');
     assert.strictEqual(err.message, 'kaboom');
   }));
 }
+
+{
+  const src = new PassThrough({ autoDestroy: false });
+  const dst = new PassThrough({ autoDestroy: false });
+  pipeline(src, dst, common.mustCall(() => {
+    assert.strictEqual(src.destroyed, true);
+    assert.strictEqual(dst.destroyed, true);
+  }));
+  src.end();
+}