stream: always invoke end callback

Ensure that the callback passed into end() is always invoke in
order to avoid bug such as deadlock the user.

PR-URL: https://github.com/nodejs/node/pull/29747
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

1 files changed, 30 insertions, 4 deletions

diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index 9b4036e476..c7a3047dc7 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -611,11 +611,11 @@ Writable.prototype.end = function(chunk, encoding, cb) {
   }
 
   // Ignore unnecessary end() calls.
-  if (!state.ending)
+  if (!state.ending) {
     endWritable(this, state, cb);
-  else if (typeof cb === 'function') {
+  } else if (typeof cb === 'function') {
     if (!state.finished) {
-      this.once('finish', cb);
+      onFinished(this, state, cb);
     } else {
       cb(new ERR_STREAM_ALREADY_FINISHED('end'));
     }
@@ -695,7 +695,7 @@ function endWritable(stream, state, cb) {
     if (state.finished)
       process.nextTick(cb);
     else
-      stream.once('finish', cb);
+      onFinished(stream, state, cb);
   }
   state.ended = true;
   stream.writable = false;
@@ -715,6 +715,32 @@ function onCorkedFinish(corkReq, state, err) {
   state.corkedRequestsFree.next = corkReq;
 }
 
+function onFinished(stream, state, cb) {
+  if (state.destroyed && state.errorEmitted) {
+    // TODO(ronag): Backwards compat. Should be moved to end() without
+    // errorEmitted check and with errorOrDestroy.
+    const err = new ERR_STREAM_DESTROYED('end');
+    process.nextTick(cb, err);
+    return;
+  }
+
+  function onerror(err) {
+    stream.removeListener('finish', onfinish);
+    stream.removeListener('error', onerror);
+    cb(err);
+    if (stream.listenerCount('error') === 0) {
+      stream.emit('error', err);
+    }
+  }
+  function onfinish() {
+    stream.removeListener('finish', onfinish);
+    stream.removeListener('error', onerror);
+    cb();
+  }
+  stream.on('finish', onfinish);
+  stream.prependListener('error', onerror);
+}
+
 Object.defineProperty(Writable.prototype, 'destroyed', {
   // Making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in