stream: error state cleanup

Clean up end simplify errored state.

- errorEmitted should be set in the same tick as 'error' is emitted.
- errored should be set as soon as an error occurs.
- errored should exist on Readable as well.
- refactor destroy logic and make it easier to follow.

PR-URL: https://github.com/nodejs/node/pull/30851
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

2 files changed, 82 insertions, 9 deletions

diff --git a/test/parallel/test-stream-readable-destroy.js b/test/parallel/test-stream-readable-destroy.js
index 7687ea90cc8..d446dd36117 100644
--- a/test/parallel/test-stream-readable-destroy.js
+++ b/test/parallel/test-stream-readable-destroy.js
@@ -129,13 +129,20 @@ const assert = require('assert');
     cb(expected);
   });
 
+  let ticked = false;
   read.on('end', common.mustNotCall('no end event'));
   read.on('error', common.mustCall((err) => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(read._readableState.errorEmitted, true);
+    assert.strictEqual(read._readableState.errored, true);
     assert.strictEqual(err, expected);
   }));
 
   read.destroy();
+  assert.strictEqual(read._readableState.errorEmitted, false);
+  assert.strictEqual(read._readableState.errored, true);
   assert.strictEqual(read.destroyed, true);
+  ticked = true;
 }
 
 {
@@ -174,10 +181,58 @@ const assert = require('assert');
 
   const expected = new Error('kaboom');
 
-  read.on('close', common.mustCall());
+  let ticked = false;
+  read.on('close', common.mustCall(() => {
+    assert.strictEqual(read._readableState.errorEmitted, false);
+    assert.strictEqual(ticked, true);
+  }));
+  // 'error' should not be emitted since a callback is passed to
+  // destroy(err, callback);
+  read.on('error', common.mustNotCall());
+
+  assert.strictEqual(read._readableState.errored, false);
+  assert.strictEqual(read._readableState.errorEmitted, false);
+
   read.destroy(expected, common.mustCall(function(err) {
+    assert.strictEqual(read._readableState.errored, true);
     assert.strictEqual(err, expected);
   }));
+  assert.strictEqual(read._readableState.errorEmitted, false);
+  assert.strictEqual(read._readableState.errored, true);
+  ticked = true;
+}
+
+{
+  const readable = new Readable({
+    destroy: common.mustCall(function(err, cb) {
+      process.nextTick(cb, new Error('kaboom 1'));
+    }),
+    read() {}
+  });
+
+  let ticked = false;
+  readable.on('close', common.mustCall(() => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(readable._readableState.errorEmitted, true);
+  }));
+  readable.on('error', common.mustCall((err) => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(err.message, 'kaboom 2');
+    assert.strictEqual(readable._readableState.errorEmitted, true);
+  }));
+
+  readable.destroy();
+  assert.strictEqual(readable.destroyed, true);
+  assert.strictEqual(readable._readableState.errored, false);
+  assert.strictEqual(readable._readableState.errorEmitted, false);
+
+  // Test case where `readable.destroy()` is called again with an error before
+  // the `_destroy()` callback is called.
+  readable.destroy(new Error('kaboom 2'));
+  assert.strictEqual(readable._readableState.errorEmitted, false);
+  assert.strictEqual(readable._readableState.errored, true);
+
+  ticked = true;
 }
 
 {
diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
index 1186c618634..d321d808199 100644
--- a/test/parallel/test-stream-writable-destroy.js
+++ b/test/parallel/test-stream-writable-destroy.js
@@ -157,13 +157,22 @@ const assert = require('assert');
     write(chunk, enc, cb) { cb(); }
   });
 
-  write.on('close', common.mustCall());
-  write.on('error', common.mustCall());
+  let ticked = false;
+  write.on('close', common.mustCall(() => {
+    assert.strictEqual(ticked, true);
+  }));
+  write.on('error', common.mustCall((err) => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(err.message, 'kaboom 1');
+    assert.strictEqual(write._writableState.errorEmitted, true);
+  }));
 
   write.destroy(new Error('kaboom 1'));
   write.destroy(new Error('kaboom 2'));
-  assert.strictEqual(write._writableState.errorEmitted, true);
+  assert.strictEqual(write._writableState.errored, true);
+  assert.strictEqual(write._writableState.errorEmitted, false);
   assert.strictEqual(write.destroyed, true);
+  ticked = true;
 }
 
 {
@@ -176,20 +185,29 @@ const assert = require('assert');
     }
   });
 
-  writable.on('close', common.mustCall());
-  writable.on('error', common.expectsError({
-    type: Error,
-    message: 'kaboom 2'
+  let ticked = false;
+  writable.on('close', common.mustCall(() => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(writable._writableState.errorEmitted, true);
+  }));
+  writable.on('error', common.mustCall((err) => {
+    assert.strictEqual(ticked, true);
+    assert.strictEqual(err.message, 'kaboom 2');
+    assert.strictEqual(writable._writableState.errorEmitted, true);
   }));
 
   writable.destroy();
   assert.strictEqual(writable.destroyed, true);
+  assert.strictEqual(writable._writableState.errored, false);
   assert.strictEqual(writable._writableState.errorEmitted, false);
 
   // Test case where `writable.destroy()` is called again with an error before
   // the `_destroy()` callback is called.
   writable.destroy(new Error('kaboom 2'));
-  assert.strictEqual(writable._writableState.errorEmitted, true);
+  assert.strictEqual(writable._writableState.errorEmitted, false);
+  assert.strictEqual(writable._writableState.errored, true);
+
+  ticked = true;
 }
 
 {