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. Backport-PR-URL: #31975 PR-URL: #31940 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Ruben Bridgewater <[email protected]> Reviewed-By: Luigi Pinca <[email protected]>
nodejs · Mar 1, 2020 · 8a2b62e · 8a2b62e
1 parent 313ecaa
commit 8a2b62e
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 13 deletions.
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -38,27 +38,20 @@ function destroyStream(stream, err) {
 
 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;
+    destroyStream(stream, err);
+    callback(err);
   });
 
-  let destroyed = false;
   return (err) => {
-    if (closed) return;
     if (destroyed) return;
     destroyed = true;
-
     destroyStream(stream, err);
-
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };
 }

diff --git a/test/parallel/test-stream-pipeline.js b/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();
+}