stream: do not swallow errors with async iterators and pipeline

mcollina · MylesBorins · commit 0a005521229b · 2020-03-11T14:21:05.000-04: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. Co-Authored-By: Robert Nagy <ronagy@icloud.com> PR-URL: #32051 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -109,6 +109,7 @@ async function pump(iterable, writable, finish) {
 if (!EE) {
 EE = require('events');
 }
+ let error;
 try {
 for await (const chunk of iterable) {
 if (!writable.write(chunk)) {
@@ -118,7 +119,9 @@ async function pump(iterable, writable, finish) {
 }
 writable.end();
 } catch (err) {
- finish(err);
+ error = err;
+ } finally {
+ finish(error);
 }
 }
 
@@ -135,36 +138,37 @@ function pipeline(...streams) {
 let value;
 const destroys = [];
 
- function finish(err, final) {
- if (!error && err) {
+ let finishCount = 0;
+
+ function finish(err) {
+ const final = --finishCount === 0;
+
+ if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE')) {
 error = err;
 }
 
- if (error || final) {
- for (const destroy of destroys) {
- destroy(error);
- }
+ if (!error && !final) {
+ return;
+ }
+
+ while (destroys.length) {
+ destroys.shift()(error);
 }
 
 if (final) {
 callback(error, value);
 }
 }
 
- function wrap(stream, reading, writing, final) {
- destroys.push(destroyer(stream, reading, writing, final, (err) => {
- finish(err, final);
- }));
- }
-
 let ret;
 for (let i = 0; i < streams.length; i++) {
 const stream = streams[i];
 const reading = i < streams.length - 1;
 const writing = i > 0;
 
 if (isStream(stream)) {
- wrap(stream, reading, writing, !reading);
+ finishCount++;
+ destroys.push(destroyer(stream, reading, writing, !reading, finish));
 }
 
 if (i === 0) {
@@ -210,20 +214,25 @@ function pipeline(...streams) {
 pt.destroy(err);
 });
 } else if (isIterable(ret, true)) {
+ finishCount++;
 pump(ret, pt, finish);
 } else {
 throw new ERR_INVALID_RETURN_VALUE(
 'AsyncIterable or Promise', 'destination', ret);
 }
 
 ret = pt;
- wrap(ret, false, true, true);
+
+ finishCount++;
+ destroys.push(destroyer(ret, false, true, true, finish));
 }
 } else if (isStream(stream)) {
 if (isReadable(ret)) {
 ret.pipe(stream);
 } else {
 ret = makeAsyncIterable(ret);
+
+ finishCount++;
 pump(ret, stream, finish);
 }
 ret = stream;
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -968,3 +968,30 @@ const { promisify } = require('util');
 }));
 src.end();
 }
+
+{
+ 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, '');
+ }));
+}

-Original file line number
+Diff line change
 if(!EE){
 EE=require('events');
+}
 +leterror;
 try{
 forawait(constchunkofiterable){
 if(!writable.write(chunk)){
+}
 writable.end();
 }catch(err){
 -finish(err);
 +error=err;
 +}finally{
 +finish(error);
+}
+}
 letvalue;
 constdestroys=[];
 -functionfinish(err,final){
 -if(!error&&err){
 +letfinishCount=0;
++
 +functionfinish(err){
 +constfinal=--finishCount===0;
++
 +if(err&&(!error||error.code==='ERR_STREAM_PREMATURE_CLOSE')){
 error=err;
+}
 -if(error||final){
 -for(constdestroyofdestroys){
 -destroy(error);
 -}
 +if(!error&&!final){
 +return;
 +}
++
 +while(destroys.length){
 +destroys.shift()(error);
+}
 if(final){
 callback(error,value);
+}
+}
 -functionwrap(stream,reading,writing,final){
 -destroys.push(destroyer(stream,reading,writing,final,(err)=>{
 -finish(err,final);
 -}));
 -}
+-
 letret;
 for(leti=0;i<streams.length;i++){
 conststream=streams[i];
 constreading=i<streams.length-1;
 constwriting=i>0;
 if(isStream(stream)){
 -wrap(stream,reading,writing,!reading);
 +finishCount++;
 +destroys.push(destroyer(stream,reading,writing,!reading,finish));
+}
 if(i===0){
 pt.destroy(err);
 });
 }elseif(isIterable(ret,true)){
 +finishCount++;
 pump(ret,pt,finish);
 }else{
 thrownewERR_INVALID_RETURN_VALUE(
 'AsyncIterable or Promise','destination',ret);
+}
 ret=pt;
 -wrap(ret,false,true,true);
++
 +finishCount++;
 +destroys.push(destroyer(ret,false,true,true,finish));
+}
 }elseif(isStream(stream)){
 if(isReadable(ret)){
 ret.pipe(stream);
 }else{
 ret=makeAsyncIterable(ret);
++
 +finishCount++;
 pump(ret,stream,finish);
+}
 ret=stream;
-Original file line number
+Diff line change
 }));
 src.end();
+}
++
 +{
 +letres='';
 +constrs=newReadable({
 +read(){
 +setImmediate(()=>{
 +rs.push('hello');
 +});
 +}
 +});
 +constws=newWritable({
 +write: common.mustNotCall()
 +});
 +pipeline(rs,asyncfunction*(stream){
 +/* eslint no-unused-vars: off */
 +forawait(constchunkofstream){
 +thrownewError('kaboom');
 +}
 +},asyncfunction*(source){
 +forawait(constchunkofsource){
 +res+=chunk;
 +}
 +},ws,common.mustCall((err)=>{
 +assert.strictEqual(err.message,'kaboom');
 +assert.strictEqual(res,'');
 +}));
 +}