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>

3 files changed, 123 insertions, 0 deletions

diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
index c4a96788ab..30e4503c05 100644
--- a/test/parallel/test-stream-writable-destroy.js
+++ b/test/parallel/test-stream-writable-destroy.js
@@ -292,3 +292,55 @@ const assert = require('assert');
   }));
   write.uncork();
 }
+
+{
+  // Call end(cb) after error & destroy
+
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(new Error('asd')); }
+  });
+  write.on('error', common.mustCall(() => {
+    write.destroy();
+    let ticked = false;
+    write.end(common.mustCall((err) => {
+      assert.strictEqual(ticked, true);
+      assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    }));
+    ticked = true;
+  }));
+  write.write('asd');
+}
+
+{
+  // Call end(cb) after finish & destroy
+
+  const write = new Writable({
+    write(chunk, enc, cb) { cb(); }
+  });
+  write.on('finish', common.mustCall(() => {
+    write.destroy();
+    let ticked = false;
+    write.end(common.mustCall((err) => {
+      assert.strictEqual(ticked, false);
+      assert.strictEqual(err.code, 'ERR_STREAM_ALREADY_FINISHED');
+    }));
+    ticked = true;
+  }));
+  write.end();
+}
+
+{
+  // Call end(cb) after error & destroy and don't trigger
+  // unhandled exception.
+
+  const write = new Writable({
+    write(chunk, enc, cb) { process.nextTick(cb); }
+  });
+  write.once('error', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'asd');
+  }));
+  write.end('asd', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'asd');
+  }));
+  write.destroy(new Error('asd'));
+}
diff --git a/test/parallel/test-stream-writable-end-cb-error.js b/test/parallel/test-stream-writable-end-cb-error.js
new file mode 100644
index 0000000000..24989a6d06
--- /dev/null
+++ b/test/parallel/test-stream-writable-end-cb-error.js
@@ -0,0 +1,48 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const stream = require('stream');
+
+{
+  // Invoke end callback on failure.
+  const writable = new stream.Writable();
+
+  writable._write = (chunk, encoding, cb) => {
+    process.nextTick(cb, new Error('kaboom'));
+  };
+
+  writable.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'kaboom');
+  }));
+  writable.write('asd');
+  writable.end(common.mustCall((err) => {
+    assert.strictEqual(err.message, 'kaboom');
+  }));
+  writable.end(common.mustCall((err) => {
+    assert.strictEqual(err.message, 'kaboom');
+  }));
+}
+
+{
+  // Don't invoke end callback twice
+  const writable = new stream.Writable();
+
+  writable._write = (chunk, encoding, cb) => {
+    process.nextTick(cb);
+  };
+
+  let called = false;
+  writable.end('asd', common.mustCall((err) => {
+    called = true;
+    assert.strictEqual(err, undefined);
+  }));
+
+  writable.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.message, 'kaboom');
+  }));
+  writable.on('finish', common.mustCall(() => {
+    assert.strictEqual(called, true);
+    writable.emit('error', new Error('kaboom'));
+  }));
+}
diff --git a/test/parallel/test-stream-writable-end-cb-uncaugth.js b/test/parallel/test-stream-writable-end-cb-uncaugth.js
new file mode 100644
index 0000000000..ab25cac81b
--- /dev/null
+++ b/test/parallel/test-stream-writable-end-cb-uncaugth.js
@@ -0,0 +1,23 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const stream = require('stream');
+
+process.on('uncaughtException', common.mustCall((err) => {
+  assert.strictEqual(err.message, 'kaboom');
+}));
+
+const writable = new stream.Writable();
+
+writable._write = (chunk, encoding, cb) => {
+  cb();
+};
+writable._final = (cb) => {
+  cb(new Error('kaboom'));
+};
+
+writable.write('asd');
+writable.end(common.mustCall((err) => {
+  assert.strictEqual(err.message, 'kaboom');
+}));