stream: finish pipeline if dst closes before src

If the destination stream is closed before the source has completed the pipeline should finnish with premature close. Fixes: #43682 PR-URL: #43701 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Luigi Pinca <[email protected]>

Author: ronag

lib/internal/streams/pipeline.js

@@ -20,6 +20,7 @@ const{
 ERR_INVALID_RETURN_VALUE,
 ERR_MISSING_ARGS,
 ERR_STREAM_DESTROYED,
+ERR_STREAM_PREMATURE_CLOSE,
 },
  AbortError,
 }=require('internal/errors');
@@ -344,13 +345,24 @@ function pipelineImpl(streams, callback, opts){
 }
 
 functionpipe(src,dst,finish,{ end }){
+letended=false;
+dst.on('close',()=>{
+if(!ended){
+// Finish if the destination closes before the source has completed.
+finish(newERR_STREAM_PREMATURE_CLOSE());
+}
+});
+
 src.pipe(dst,{ end });
 
 if(end){
 // Compat. Before node v10.12.0 stdio used to throw an error so
 // pipe() did/does not end() stdio destinations.
 // Now they allow it but "secretly" don't close the underlying fd.
-src.once('end',()=>dst.end());
+src.once('end',()=>{
+ended=true;
+dst.end();
+});
 }else{
 finish();
 }

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

@@ -0,0 +1,21 @@
+'use strict';
+
+constcommon=require('../common');
+const{ pipeline, Duplex, PassThrough }=require('stream');
+constassert=require('assert');
+
+constremote=newPassThrough();
+constlocal=newDuplex({
+read(){},
+write(chunk,enc,callback){
+callback();
+}
+});
+
+pipeline(remote,local,remote,common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+}));
+
+setImmediate(()=>{
+remote.end();
+});