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: nodejs#43682

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) {
 }
 
 function pipe(src, dst, finish, { end }) {
+  let ended = false
+  dst.on('close', () => {
+    if (!ended) {
+      // Finish if the destination closes before the source has completed.
+      finish(new ERR_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';
+
+const common = require('../common');
+const { pipeline, Duplex, PassThrough } = require('stream');
+const assert = require('assert');
+
+const remote = new PassThrough();
+const local = new Duplex({
+  read () {},
+  write (chunk, enc, callback) {
+    callback()
+  }
+});
+
+pipeline(remote, local, remote, common.mustCall((err) => {
+  assert.equal(err.code, 'ERR_STREAM_PREMATURE_CLOSE')
+}));
+
+setImmediate(() => {
+  remote.end();
+});