stream: improve Readable.from error handling

benjamingr · danielleadams · commit 1fea05149a92 · 2021-02-16T09:12:29.000-05:00
PR-URL: #37158 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
diff --git a/lib/internal/streams/from.js b/lib/internal/streams/from.js
@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+ PromisePrototypeThen,
 SymbolAsyncIterator,
 SymbolIterator
 } = primordials;
@@ -42,9 +43,6 @@ function from(Readable, iterable, opts) {
 // being called before last iteration completion.
 let reading = false;
 
- // Flag for when iterator needs to be explicitly closed.
- let needToClose = false;
-
 readable._read = function() {
 if (!reading) {
 reading = true;
@@ -53,18 +51,23 @@ function from(Readable, iterable, opts) {
 };
 
 readable._destroy = function(error, cb) {
- if (needToClose) {
- needToClose = false;
- close().then(
- () => process.nextTick(cb, error),
- (e) => process.nextTick(cb, error || e),
- );
- } else {
- cb(error);
- }
+ PromisePrototypeThen(
+ close(error),
+ () => process.nextTick(cb, error), // nextTick is here in case cb throws
+ (e) => process.nextTick(cb, e || error),
+ );
 };
 
- async function close() {
+ async function close(error) {
+ const hadError = (error !== undefined) && (error !== null);
+ const hasThrow = typeof iterator.throw === 'function';
+ if (hadError && hasThrow) {
+ const { value, done } = await iterator.throw(error);
+ await value;
+ if (done) {
+ return;
+ }
+ }
 if (typeof iterator.return === 'function') {
 const { value } = await iterator.return();
 await value;
@@ -73,13 +76,9 @@ function from(Readable, iterable, opts) {
 
 async function next() {
 try {
- needToClose = false;
 const { value, done } = await iterator.next();
- needToClose = !done;
 if (done) {
 readable.push(null);
- } else if (readable.destroyed) {
- await close();
 } else {
 const res = await value;
 if (res === null) {
diff --git a/test/parallel/test-readable-from.js b/test/parallel/test-readable-from.js
@@ -4,6 +4,7 @@ const { mustCall } = require('../common');
 const { once } = require('events');
 const { Readable } = require('stream');
 const { strictEqual, throws } = require('assert');
+const common = require('../common');
 
 {
 throws(() => {
@@ -187,6 +188,25 @@ async function endWithError() {
 }
 }
 
+async function destroyingStreamWithErrorThrowsInGenerator() {
+ const validateError = common.mustCall((e) => {
+ strictEqual(e, 'Boum');
+ });
+ async function* generate() {
+ try {
+ yield 1;
+ yield 2;
+ yield 3;
+ throw new Error();
+ } catch (e) {
+ validateError(e);
+ }
+ }
+ const stream = Readable.from(generate());
+ stream.read();
+ stream.once('error', common.mustCall());
+ stream.destroy('Boum');
+}
 
 Promise.all([
 toReadableBasicSupport(),
@@ -198,5 +218,6 @@ Promise.all([
 toReadableOnDataNonObject(),
 destroysTheStreamWhenThrowing(),
 asTransformStream(),
- endWithError()
+ endWithError(),
+ destroyingStreamWithErrorThrowsInGenerator(),
 ]).then(mustCall());

-Original file line number
+Diff line change
@@ @@ -1,6 +1,7 @@ @@
 'use strict';
 const{
 + PromisePrototypeThen,
  SymbolAsyncIterator,
  SymbolIterator
 }=primordials;
 // being called before last iteration completion.
 letreading=false;
 -// Flag for when iterator needs to be explicitly closed.
 -letneedToClose=false;
+-
 readable._read=function(){
 if(!reading){
 reading=true;
 };
 readable._destroy=function(error,cb){
 -if(needToClose){
 -needToClose=false;
 -close().then(
 -()=>process.nextTick(cb,error),
 -(e)=>process.nextTick(cb,error||e),
 -);
 -}else{
 -cb(error);
 -}
 +PromisePrototypeThen(
 +close(error),
 +()=>process.nextTick(cb,error),// nextTick is here in case cb throws
 +(e)=>process.nextTick(cb,e||error),
 +);
 };
 -asyncfunctionclose(){
 +asyncfunctionclose(error){
 +consthadError=(error!==undefined)&&(error!==null);
 +consthasThrow=typeofiterator.throw==='function';
 +if(hadError&&hasThrow){
 +const{ value, done }=awaititerator.throw(error);
 +awaitvalue;
 +if(done){
 +return;
 +}
 +}
 if(typeofiterator.return==='function'){
 const{ value }=awaititerator.return();
 awaitvalue;
 asyncfunctionnext(){
 try{
 -needToClose=false;
 const{ value, done }=awaititerator.next();
 -needToClose=!done;
 if(done){
 readable.push(null);
 -}elseif(readable.destroyed){
 -awaitclose();
 }else{
 constres=awaitvalue;
 if(res===null){