stream: fix nested compose error propagation

PR-URL: #62556Fixes: #60083
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: mcollina

lib/internal/streams/duplexify.js

@@ -239,7 +239,17 @@ function fromAsyncGen(fn) {
 _resolve({__proto__: null,done: true, cb });
 },
 destroy(err,cb){
-ac.abort();
+ac.abort(err);
+
+// If the source async iterator is waiting for the next write/final
+// signal, unblock it so the readable side can observe the abort and
+// finish destroying.
+if(resolve!==null){
+const_resolve=resolve;
+resolve=null;
+_resolve({__proto__: null,done: true,cb(){}});
+}
+
 cb(err);
 },
 };

lib/internal/streams/from.js

@@ -8,9 +8,12 @@ const {
 const{ Buffer }=require('buffer');
 
 const{
-ERR_INVALID_ARG_TYPE,
-ERR_STREAM_NULL_VALUES,
-}=require('internal/errors').codes;
+ aggregateTwoErrors,
+codes: {
+ERR_INVALID_ARG_TYPE,
+ERR_STREAM_NULL_VALUES,
+},
+}=require('internal/errors');
 
 functionfrom(Readable,iterable,opts){
 letiterator;
@@ -43,6 +46,7 @@ function from(Readable, iterable, opts) {
 // TODO(ronag): What options should be allowed?
  ...opts,
 });
+constoriginalDestroy=readable._destroy;
 
 // Flag to protect against _read
 // being called before last iteration completion.
@@ -64,11 +68,18 @@ function from(Readable, iterable, opts) {
 };
 
 readable._destroy=function(error,cb){
-PromisePrototypeThen(
-close(error),
-()=>process.nextTick(cb,error),// nextTick is here in case cb throws
-(e)=>process.nextTick(cb,e||error),
-);
+originalDestroy.call(this,error,(destroyError)=>{
+constcombinedError=destroyError||error;
+PromisePrototypeThen(
+close(combinedError),
+// nextTick is here in case cb throws
+()=>process.nextTick(cb,combinedError),
+(closeError)=>process.nextTick(
+cb,
+aggregateTwoErrors(combinedError,closeError),
+),
+);
+});
 };
 
 asyncfunctionclose(error){

test/parallel/test-stream-readable-compose.js

@@ -116,6 +116,24 @@ const assert = require('assert');
 ).then(common.mustCall());
 }
 
+{
+// Errors from nested `.compose()` calls should propagate instead of hanging.
+conststream=Readable.from(['hello'])
+.compose(asyncfunction*(source){// eslint-disable-line require-yield
+forawait(constchunkofsource){
+thrownewError(`boom: ${chunk}`);
+}
+})
+.compose(asyncfunction*(source){
+yield*source;
+});
+
+assert.rejects(
+stream.toArray(),
+/boom:hello/,
+).then(common.mustCall());
+}
+
 {
 // AbortSignal
 constac=newAbortController();