stream: handle generator destruction from Duplex.from()

PR-URL: #55096
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Jason Zhang <xzha4350@gmail.com>

Author: matthieusieben

lib/internal/streams/duplexify.js

@@ -83,15 +83,19 @@ module.exports = function duplexify(body, name) {
 }
 
 if(typeofbody==='function'){
-const{ value, write, final, destroy }=fromAsyncGen(body);
+letd;
+
+const{ value, write, final, destroy }=fromAsyncGen(body,()=>{
+destroyer(d);
+});
 
 // Body might be a constructor function instead of an async generator function.
 if(isDuplexNodeStream(value)){
-returnvalue;
+returnd=value;
 }
 
 if(isIterable(value)){
-returnfrom(Duplexify,value,{
+returnd=from(Duplexify,value,{
 // TODO (ronag): highWaterMark?
 objectMode: true,
  write,
@@ -102,12 +106,16 @@ module.exports = function duplexify(body, name) {
 
 constthen=value?.then;
 if(typeofthen==='function'){
-letd;
+letfinalized=false;
 
 constpromise=FunctionPrototypeCall(
 then,
 value,
 (val)=>{
+// The function returned without (fully) consuming the generator.
+if(!finalized){
+destroyer(d);
+}
 if(val!=null){
 thrownewERR_INVALID_RETURN_VALUE('nully','body',val);
 }
@@ -123,6 +131,7 @@ module.exports = function duplexify(body, name) {
 readable: false,
  write,
 final(cb){
+finalized=true;
 final(async()=>{
 try{
 awaitpromise;
@@ -208,11 +217,12 @@ module.exports = function duplexify(body, name) {
 body);
 };
 
-functionfromAsyncGen(fn){
+functionfromAsyncGen(fn,destructor){
 let{ promise, resolve }=PromiseWithResolvers();
 constac=newAbortController();
 constsignal=ac.signal;
-constvalue=fn(asyncfunction*(){
+
+constasyncGenerator=(asyncfunction*(){
 while(true){
 const_promise=promise;
 promise=null;
@@ -222,9 +232,44 @@ function fromAsyncGen(fn) {
 if(signal.aborted)
 thrownewAbortError(undefined,{cause: signal.reason});
 ({ promise, resolve }=PromiseWithResolvers());
+// Next line will "break" the loop if the generator is returned/thrown.
 yieldchunk;
 }
-}(),{ signal });
+})();
+
+constoriginalReturn=asyncGenerator.return;
+asyncGenerator.return=asyncfunction(value){
+try{
+returnawaitoriginalReturn.call(this,value);
+}finally{
+if(promise){
+const_promise=promise;
+promise=null;
+const{ cb }=await_promise;
+process.nextTick(cb);
+
+process.nextTick(destructor);
+}
+}
+};
+
+constoriginalThrow=asyncGenerator.throw;
+asyncGenerator.throw=asyncfunction(err){
+try{
+returnawaitoriginalThrow.call(this,err);
+}finally{
+if(promise){
+const_promise=promise;
+promise=null;
+const{ cb }=await_promise;
+
+// asyncGenerator.throw(undefined) should cause a callback error
+process.nextTick(cb,err??newAbortError());
+}
+}
+};
+
+constvalue=fn(asyncGenerator,{ signal });
 
 return{
  value,

test/parallel/test-stream-duplex-from.js

@@ -5,6 +5,7 @@ const assert = require('assert');
 const{ Duplex, Readable, Writable, pipeline, PassThrough }=require('stream');
 const{ ReadableStream, WritableStream }=require('stream/web');
 const{ Blob }=require('buffer');
+constsleep=require('util').promisify(setTimeout);
 
 {
 constd=Duplex.from({
@@ -401,3 +402,193 @@ function makeATestWritableStream(writeFunc) {
 assert.strictEqual(d.writable,false);
 }));
 }
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+constvalues=awaitArray.fromAsync(asyncGenerator);
+assert.deepStrictEqual(values,['foo','bar','baz']);
+
+awaitasyncGenerator.return();
+awaitasyncGenerator.return();
+awaitasyncGenerator.return();
+}),
+common.mustSucceed(()=>{
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+// eslint-disable-next-line no-unused-vars
+forawait(const_ofasyncGenerator)break;
+}),
+common.mustSucceed(()=>{
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+consta=awaitasyncGenerator.next();
+assert.strictEqual(a.done,false);
+assert.strictEqual(a.value.toString(),'foo');
+constb=awaitasyncGenerator.return();
+assert.strictEqual(b.done,true);
+}),
+common.mustSucceed(()=>{
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+// Note: the generator is not even started at this point
+awaitasyncGenerator.return();
+}),
+common.mustSucceed(()=>{
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+// Same as before, with a delay
+awaitsleep(100);
+awaitasyncGenerator.return();
+}),
+common.mustSucceed(()=>{
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){}),
+common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+awaitsleep(100);
+}),
+common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar','baz']);
+constd=Duplex.from(asyncfunction(asyncGenerator){
+while(!(awaitasyncGenerator.next()).done)awaitsleep(100);
+});
+
+setTimeout(()=>d.destroy(),150);
+
+pipeline(
+r,
+d,
+common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Duplex.from(asyncfunction*(){
+for(constvalueof['foo','bar','baz']){
+awaitsleep(50);
+yieldvalue;
+}
+});
+constd=Duplex.from(asyncfunction(asyncGenerator){
+while(!(awaitasyncGenerator.next()).done);
+});
+
+setTimeout(()=>r.destroy(),75);
+
+pipeline(
+r,
+d,
+common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+assert.strictEqual(r.destroyed,true);
+assert.strictEqual(d.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+awaitasyncGenerator.throw(newError('my error'));
+}),
+common.mustCall((err)=>{
+assert.strictEqual(err.message,'my error');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+awaitasyncGenerator.next();
+awaitasyncGenerator.throw(newError('my error'));
+}),
+common.mustCall((err)=>{
+assert.strictEqual(err.message,'my error');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}
+
+{
+constr=Readable.from(['foo','bar']);
+pipeline(
+r,
+Duplex.from(asyncfunction(asyncGenerator){
+awaitasyncGenerator.next();
+awaitasyncGenerator.throw();
+}),
+common.mustCall((err)=>{
+assert.strictEqual(err.code,'ABORT_ERR');
+assert.strictEqual(r.destroyed,true);
+})
+);
+}