stream: don't emit 'data' after 'error' or 'close'

As per doc we should not emit further events after 'close' and
only 'close' after 'error'.

Fixes: #39630

PR-URL: #39639
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>
Reviewed-By: Antoine du Hamel <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>

lib/internal/streams/readable.js

@@ -276,6 +276,8 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
     if (addToFront) {
       if (state.endEmitted)
         errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
+      else if (state.destroyed || state.errored)
+        return false;
       else
         addChunk(stream, state, chunk, true);
     } else if (state.ended) {
@@ -316,6 +318,7 @@ function addChunk(stream, state, chunk, addToFront) {
     } else {
       state.awaitDrainWriters = null;
     }
+
     state.dataEmitted = true;
     stream.emit('data', chunk);
   } else {
@@ -542,7 +545,7 @@ Readable.prototype.read = function(n) {
       endReadable(this);
   }
 
-  if (ret !== null) {
+  if (ret !== null && !state.errorEmitted && !state.closeEmitted) {
     state.dataEmitted = true;
     this.emit('data', ret);
   }

test/parallel/test-stream-readable-destroy.js

@@ -319,3 +319,85 @@ const assert = require('assert');
   })(), /AbortError/);
   setTimeout(() => controller.abort(), 0);
 }
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.on('data', common.mustNotCall());
+  read.on('error', common.mustCall((e) => {
+    read.push('asd');
+    read.read();
+  }));
+  read.on('close', common.mustCall((e) => {
+    read.push('asd');
+    read.read();
+  }));
+  read.destroy(new Error('asd'));
+}
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.on('data', common.mustNotCall());
+  read.on('close', common.mustCall((e) => {
+    read.push('asd');
+    read.read();
+  }));
+  read.destroy();
+}
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.on('data', common.mustNotCall());
+  read.on('close', common.mustCall((e) => {
+    read.push('asd');
+    read.unshift('asd');
+  }));
+  read.destroy();
+}
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.on('data', common.mustNotCall());
+  read.destroy();
+  read.unshift('asd');
+}
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.resume();
+  read.on('data', common.mustNotCall());
+  read.on('close', common.mustCall((e) => {
+    read.push('asd');
+  }));
+  read.destroy();
+}
+
+{
+  const read = new Readable({
+    read() {
+    },
+  });
+
+  read.on('data', common.mustNotCall());
+  read.destroy();
+  read.push('asd');
+}