Repository: nodejs/node
Pull request: stream: pipeline wait for close before calling the callback

> // Use the pipeline API to easily pipe a series of streams
// together and get notified when the pipeline is fully done.

> stream.pipeline() closes all the streams when an error is raised

pipeline doc: [here](https://nodejs.org/api/stream.html#streampipelinesource-transforms-destination-callback)

The pipeline should wait for close event to finish before calling the callback as that should be the proper "fully done" state.

The `finishCount` should not below 0 when calling `finishImpl` function (should not be negative).

Fixes: https://github.com/nodejs/node/issues/51540

<!--
Before submitting a pull request, please read:

- the CONTRIBUTING guide at https://github.com/nodejs/node/blob/HEAD/CONTRIBUTING.md
- the commit message formatting guidelines at
  https://github.com/nodejs/node/blob/HEAD/doc/contributing/pull-requests.md#commit-message-guidelines

For code changes:
1. Include tests for any bug fixes or new features.
2. Update documentation if relevant.
3. Ensure that `make -j4 test` (UNIX), or `vcbuild test` (Windows) passes.

If you believe this PR should be highlighted in the Node.js CHANGELOG
please add the `notable-change` label.

Developer's Certificate of Origin 1.1

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.
-->


Changed files:
filename: lib/internal/streams/pipeline.js
patch: @@ -225,6 +225,10 @@ function pipelineImpl(streams, callback, opts) {
     finishImpl(err, --finishCount === 0);
   }
 
+  function finishOnlyHandleError(err) {
+    finishImpl(err, false);
+  }
+
   function finishImpl(err, final) {
     if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE')) {
       error = err;
@@ -279,7 +283,7 @@ function pipelineImpl(streams, callback, opts) {
           err.name !== 'AbortError' &&
           err.code !== 'ERR_STREAM_PREMATURE_CLOSE'
         ) {
-          finish(err);
+          finishOnlyHandleError(err);
         }
       }
       stream.on('error', onError);
@@ -372,7 +376,7 @@ function pipelineImpl(streams, callback, opts) {
     } else if (isNodeStream(stream)) {
       if (isReadableNodeStream(ret)) {
         finishCount += 2;
-        const cleanup = pipe(ret, stream, finish, { end });
+        const cleanup = pipe(ret, stream, finish, finishOnlyHandleError, { end });
         if (isReadable(stream) && isLastStream) {
           lastStreamCleanup.push(cleanup);
         }
@@ -415,12 +419,12 @@ function pipelineImpl(streams, callback, opts) {
   return ret;
 }
 
-function pipe(src, dst, finish, { end }) {
+function pipe(src, dst, finish, finishOnlyHandleError, { 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());
+      finishOnlyHandleError(new ERR_STREAM_PREMATURE_CLOSE());
     }
   });
 

filename: test/parallel/test-stream-pipeline.js
patch: @@ -1677,5 +1677,47 @@ tmpdir.refresh();
   pipeline(r, w, common.mustCall((err) => {
     assert.strictEqual(err, undefined);
   }));
+}
+
+{
+  // See https://github.com/nodejs/node/issues/51540 for the following 2 tests
+  const src = new Readable();
+  const dst = new Writable({
+    destroy(error, cb) {
+      // Takes a while to destroy
+      setImmediate(cb);
+    },
+  });
+
+  pipeline(src, dst, (err) => {
+    assert.strictEqual(src.closed, true);
+    assert.strictEqual(dst.closed, true);
+    assert.strictEqual(err.message, 'problem');
+  });
+  src.destroy(new Error('problem'));
+}
 
+{
+  const src = new Readable();
+  const dst = new Writable({
+    destroy(error, cb) {
+      // Takes a while to destroy
+      setImmediate(cb);
+    },
+  });
+  const passThroughs = [];
+  for (let i = 0; i < 10; i++) {
+    passThroughs.push(new PassThrough());
+  }
+
+  pipeline(src, ...passThroughs, dst, (err) => {
+    assert.strictEqual(src.closed, true);
+    assert.strictEqual(dst.closed, true);
+    assert.strictEqual(err.message, 'problem');
+
+    for (let i = 0; i < passThroughs.length; i++) {
+      assert.strictEqual(passThroughs[i].closed, true);
+    }
+  });
+  src.destroy(new Error('problem'));
 }