stream: always invoke callback before emitting error

Ensure the callback is always invoked before emitting
the error in both sync and async case.

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

4 files changed, 81 insertions, 2 deletions

diff --git a/test/parallel/test-http2-reset-flood.js b/test/parallel/test-http2-reset-flood.js
index a6553401fb..9977bfd1a3 100644
--- a/test/parallel/test-http2-reset-flood.js
+++ b/test/parallel/test-http2-reset-flood.js
@@ -67,7 +67,10 @@ const worker = new Worker(__filename).on('message', common.mustCall((port) => {
       h2header.writeIntBE(1, 0, 3);  // Length: 1
       h2header.writeIntBE(i, 5, 4);  // Stream ID
       // 0x88 = :status: 200
-      conn.write(Buffer.concat([h2header, Buffer.from([0x88])]));
+      if (!conn.write(Buffer.concat([h2header, Buffer.from([0x88])]))) {
+        process.nextTick(writeRequests);
+        break;
+      }
     }
   }
 
diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
index ac107ecbb7..c4a96788ab 100644
--- a/test/parallel/test-stream-writable-destroy.js
+++ b/test/parallel/test-stream-writable-destroy.js
@@ -18,6 +18,20 @@ const assert = require('assert');
 
 {
   const write = new Writable({
+    write(chunk, enc, cb) {
+      this.destroy(new Error('asd'));
+      cb();
+    }
+  });
+
+  write.on('error', common.mustCall());
+  write.on('finish', common.mustNotCall());
+  write.end('asd');
+  assert.strictEqual(write.destroyed, true);
+}
+
+{
+  const write = new Writable({
     write(chunk, enc, cb) { cb(); }
   });
 
diff --git a/test/parallel/test-stream-writable-write-cb-error.js b/test/parallel/test-stream-writable-write-cb-error.js
new file mode 100644
index 0000000000..72db1b7e3f
--- /dev/null
+++ b/test/parallel/test-stream-writable-write-cb-error.js
@@ -0,0 +1,58 @@
+'use strict';
+const common = require('../common');
+const { Writable } = require('stream');
+const assert = require('assert');
+
+// Ensure callback is always invoked before
+// error is emitted. Regardless if error was
+// sync or async.
+
+{
+  let callbackCalled = false;
+  // Sync Error
+  const writable = new Writable({
+    write: common.mustCall((buf, enc, cb) => {
+      cb(new Error());
+    })
+  });
+  writable.on('error', common.mustCall(() => {
+    assert.strictEqual(callbackCalled, true);
+  }));
+  writable.write('hi', common.mustCall(() => {
+    callbackCalled = true;
+  }));
+}
+
+{
+  let callbackCalled = false;
+  // Async Error
+  const writable = new Writable({
+    write: common.mustCall((buf, enc, cb) => {
+      process.nextTick(cb, new Error());
+    })
+  });
+  writable.on('error', common.mustCall(() => {
+    assert.strictEqual(callbackCalled, true);
+  }));
+  writable.write('hi', common.mustCall(() => {
+    callbackCalled = true;
+  }));
+}
+
+{
+  // Sync Error
+  const writable = new Writable({
+    write: common.mustCall((buf, enc, cb) => {
+      cb(new Error());
+    })
+  });
+
+  writable.on('error', common.mustCall());
+
+  let cnt = 0;
+  // Ensure we don't live lock on sync error
+  while (writable.write('a'))
+    cnt++;
+
+  assert.strictEqual(cnt, 0);
+}
diff --git a/test/parallel/test-wrap-js-stream-exceptions.js b/test/parallel/test-wrap-js-stream-exceptions.js
index eeab26f525..2cc592a760 100644
--- a/test/parallel/test-wrap-js-stream-exceptions.js
+++ b/test/parallel/test-wrap-js-stream-exceptions.js
@@ -16,4 +16,8 @@ const socket = new JSStreamWrap(new Duplex({
   })
 }));
 
-assert.throws(() => socket.end('foo'), /Error: write EPROTO/);
+socket.end('foo');
+socket.on('error', common.expectsError({
+  type: Error,
+  message: 'write EPROTO'
+}));