stream: finish must always follow error

When _write completes with an Error, 'finish' was emitted before
'error' if the callback was asynchronous. This commit restore the
previous behavior.
The logic is still less then ideal, because we call the write()
callback before emitting error if asynchronous, but after if
synchronous. This commit do not try to change the behavior.
This commit fixes a regression introduced by:
https://github.com/nodejs/node/pull/13195.

Fixes: https://github.com/nodejs/node/issues/13812
PR-URL: https://github.com/nodejs/node/pull/13850
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Calvin Metcalf <calvin.metcalf@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>

1 files changed, 156 insertions, 0 deletions

diff --git a/test/parallel/test-stream-writable-write-writev-finish.js b/test/parallel/test-stream-writable-write-writev-finish.js
new file mode 100644
index 0000000000..c4aaa36606
--- /dev/null
+++ b/test/parallel/test-stream-writable-write-writev-finish.js
@@ -0,0 +1,156 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const stream = require('stream');
+
+// ensure consistency between the finish event when using cork()
+// and writev and when not using them
+
+{
+  const writable = new stream.Writable();
+
+  writable._write = (chunks, encoding, cb) => {
+    cb(new Error('write test error'));
+  };
+
+  let firstError = false;
+  writable.on('finish', common.mustCall(function() {
+    assert.strictEqual(firstError, true);
+  }));
+
+  writable.on('prefinish', common.mustCall());
+
+  writable.on('error', common.mustCall((er) => {
+    assert.strictEqual(er.message, 'write test error');
+    firstError = true;
+  }));
+
+  writable.end('test');
+}
+
+{
+  const writable = new stream.Writable();
+
+  writable._write = (chunks, encoding, cb) => {
+    setImmediate(cb, new Error('write test error'));
+  };
+
+  let firstError = false;
+  writable.on('finish', common.mustCall(function() {
+    assert.strictEqual(firstError, true);
+  }));
+
+  writable.on('prefinish', common.mustCall());
+
+  writable.on('error', common.mustCall((er) => {
+    assert.strictEqual(er.message, 'write test error');
+    firstError = true;
+  }));
+
+  writable.end('test');
+}
+
+{
+  const writable = new stream.Writable();
+
+  writable._write = (chunks, encoding, cb) => {
+    cb(new Error('write test error'));
+  };
+
+  writable._writev = (chunks, cb) => {
+    cb(new Error('writev test error'));
+  };
+
+  let firstError = false;
+  writable.on('finish', common.mustCall(function() {
+    assert.strictEqual(firstError, true);
+  }));
+
+  writable.on('prefinish', common.mustCall());
+
+  writable.on('error', common.mustCall((er) => {
+    assert.strictEqual(er.message, 'writev test error');
+    firstError = true;
+  }));
+
+  writable.cork();
+  writable.write('test');
+
+  setImmediate(function() {
+    writable.end('test');
+  });
+}
+
+{
+  const writable = new stream.Writable();
+
+  writable._write = (chunks, encoding, cb) => {
+    setImmediate(cb, new Error('write test error'));
+  };
+
+  writable._writev = (chunks, cb) => {
+    setImmediate(cb, new Error('writev test error'));
+  };
+
+  let firstError = false;
+  writable.on('finish', common.mustCall(function() {
+    assert.strictEqual(firstError, true);
+  }));
+
+  writable.on('prefinish', common.mustCall());
+
+  writable.on('error', common.mustCall((er) => {
+    assert.strictEqual(er.message, 'writev test error');
+    firstError = true;
+  }));
+
+  writable.cork();
+  writable.write('test');
+
+  setImmediate(function() {
+    writable.end('test');
+  });
+}
+
+// Regression test for
+// https://github.com/nodejs/node/issues/13812
+
+{
+  const rs = new stream.Readable();
+  rs.push('ok');
+  rs.push(null);
+  rs._read = () => {};
+
+  const ws = new stream.Writable();
+  let firstError = false;
+
+  ws.on('finish', common.mustCall(function() {
+    assert.strictEqual(firstError, true);
+  }));
+  ws.on('error', common.mustCall(function() {
+    firstError = true;
+  }));
+
+  ws._write = (chunk, encoding, done) => {
+    setImmediate(done, new Error());
+  };
+  rs.pipe(ws);
+}
+
+{
+  const rs = new stream.Readable();
+  rs.push('ok');
+  rs.push(null);
+  rs._read = () => {};
+
+  const ws = new stream.Writable();
+
+  ws.on('finish', common.mustNotCall());
+  ws.on('error', common.mustCall());
+
+  ws._write = (chunk, encoding, done) => {
+    done(new Error());
+  };
+  rs.pipe(ws);
+}