stream: do not swallow errors with async iterators and pipeline

mcollina · mcollina · commit 89b23682e3ac · 2020-03-03T16:23:20.000+01:00
Before this patch, pipeline() could swallow errors by pre-emptively
producing a ERR_STREAM_PREMATURE_CLOSE that was not really helpful
to the user.
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -6,7 +6,8 @@
 const {
   ArrayIsArray,
   SymbolAsyncIterator,
-  SymbolIterator
+  SymbolIterator,
+  Symbol
 } = primordials;
 
 let eos;
@@ -21,6 +22,8 @@ const {
   ERR_STREAM_DESTROYED
 } = require('internal/errors').codes;
 
+const kSkipPrematureClose = Symbol('kSkipPrematureClose');
+
 let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
@@ -159,9 +162,17 @@ function pipeline(...streams) {
   }
 
   function wrap(stream, reading, writing, final) {
-    destroys.push(destroyer(stream, reading, writing, (err) => {
+    const fn = destroyer(stream, reading, writing, (err) => {
+
+      // We need to skip the premature close error here to avoid swallowing
+      // any errors in the downstream async iterator.
+      if (err && err.code === 'ERR_STREAM_PREMATURE_CLOSE' &&
+          stream[kSkipPrematureClose]) {
+        err = undefined;
+      }
       finish(err, final);
-    }));
+    });
+    destroys.push(fn);
   }
 
   let ret;
@@ -190,6 +201,7 @@ function pipeline(...streams) {
       }
     } else if (typeof stream === 'function') {
       ret = makeAsyncIterable(ret);
+      ret[kSkipPrematureClose] = true;
       ret = stream(ret);
 
       if (reading) {
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -938,3 +938,30 @@ const { promisify } = require('util');
   r.push(null);
   r.emit('close');
 }
+
+{
+  let res = '';
+  const rs = new Readable({
+    read() {
+      setImmediate(() => {
+        rs.push('hello');
+      });
+    }
+  });
+  const ws = new Writable({
+    write: common.mustNotCall()
+  });
+  pipeline(rs, async function*(stream) {
+    /* eslint no-unused-vars: off */
+    for await (const chunk of stream) {
+      throw new Error('kaboom');
+    }
+  }, async function *(source) {
+    for await (const chunk of source) {
+      res += chunk;
+    }
+  }, ws, common.mustCall((err) => {
+    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(res, '');
+  }));
+}
