From d36392003b381411e8f46259efc93266c45a5d39 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 13:08:32 +0200
Subject: [PATCH 1/6] stream: add isDisturbed helper

Adds a helper util used to determine whether a stream has been
disturbed (read or cancelled).

Refs: https://github.com/nodejs/node/issues/39627
---
 lib/internal/streams/utils.js             | 13 +++++++++++++
 lib/internal/webstreams/readablestream.js |  9 +++++++++
 lib/stream.js                             |  1 +
 3 files changed, 23 insertions(+)

diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 01396b5113340f..0b7f80ea1b93b6 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -7,6 +7,7 @@ const {
 } = primordials;
 
 const kDestroyed = Symbol('kDestroyed');
+const kIsDisturbed = Symbol('kIsDisturbed');
 
 function isReadableNodeStream(obj) {
   return !!(
@@ -195,8 +196,20 @@ function willEmitClose(stream) {
   );
 }
 
+function isDisturbed(stream) {
+  const state = stream && stream._readableState;
+  return stream && (
+    stream.readableDidRead ||
+    isDestroyed(stream) ||
+    stream[kIsDisturbed] ||
+    (state && state.endEmitted)
+  );
+}
+
 module.exports = {
   kDestroyed,
+  isDisturbed,
+  kIsDisturbed,
   isClosed,
   isDestroyed,
   isFinished,
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index 62da68ff28086f..efe890b7ebedaf 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -80,6 +80,10 @@ const {
   queueMicrotask,
 } = require('internal/process/task_queues');
 
+const {
+  kIsDisturbed,
+} = require('internal/stream/utils');
+
 const {
   ArrayBufferViewGetBuffer,
   ArrayBufferViewGetByteLength,
@@ -200,6 +204,7 @@ class ReadableStream {
         promise: undefined,
       }
     };
+
     // The spec requires handling of the strategy first
     // here. Specifically, if getting the size and
     // highWaterMark from the strategy fail, that has
@@ -232,6 +237,10 @@ class ReadableStream {
     return makeTransferable(this);
   }
 
+  get [kIsDisturbed]() {
+    return this[kState].disturbed;
+  }
+
   /**
    * @readonly
    * @type {boolean}
diff --git a/lib/stream.js b/lib/stream.js
index b84efb0fd8862d..43f59788f62bc8 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -38,6 +38,7 @@ const internalBuffer = require('internal/buffer');
 const promises = require('stream/promises');
 
 const Stream = module.exports = require('internal/streams/legacy').Stream;
+Stream.isDisturbed = require('internal/streams/utils').isDisturbed;
 Stream.Readable = require('internal/streams/readable');
 Stream.Writable = require('internal/streams/writable');
 Stream.Duplex = require('internal/streams/duplex');

From ac0a1df2de4cc7c495ecd88932441ea9a8cf33c7 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 13:41:17 +0200
Subject: [PATCH 2/6] fixup

---
 doc/api/stream.md                             | 12 +++++
 lib/internal/streams/readable.js              |  2 +-
 lib/internal/streams/utils.js                 |  4 +-
 lib/internal/webstreams/readablestream.js     |  2 +-
 test/parallel/test-stream-readable-didRead.js |  4 +-
 test/parallel/test-whatwg-readablestream.js   | 50 +++++++++++++++++++
 6 files changed, 69 insertions(+), 5 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 312f200e61c4d5..3e5ee464f32682 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2046,6 +2046,18 @@ added: REPLACEME
   * `signal` {AbortSignal}
 * Returns: {stream.Readable}
 
+### `stream.Readable.isDisturbed(stream)`
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `stream` {stream.Readable|ReadableStream}
+* Returns: `boolean`
+
+Returns whether the stream has been read from.
+
 ### `stream.Readable.toWeb(streamReadable)`
 <!-- YAML
 added: REPLACEME
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index a6a1913961bb41..c720e7fdca3d14 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1185,7 +1185,7 @@ ObjectDefineProperties(Readable.prototype, {
   readableDidRead: {
     enumerable: false,
     get: function() {
-      return (
+      return !!(
         this._readableState.dataEmitted ||
         this._readableState.endEmitted ||
         this._readableState.errorEmitted ||
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 0b7f80ea1b93b6..f396872270a8de 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -198,12 +198,12 @@ function willEmitClose(stream) {
 
 function isDisturbed(stream) {
   const state = stream && stream._readableState;
-  return stream && (
+  return !!(stream && (
     stream.readableDidRead ||
     isDestroyed(stream) ||
     stream[kIsDisturbed] ||
     (state && state.endEmitted)
-  );
+  ));
 }
 
 module.exports = {
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index efe890b7ebedaf..fe9b26b991f04e 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -82,7 +82,7 @@ const {
 
 const {
   kIsDisturbed,
-} = require('internal/stream/utils');
+} = require('internal/streams/utils');
 
 const {
   ArrayBufferViewGetBuffer,
diff --git a/test/parallel/test-stream-readable-didRead.js b/test/parallel/test-stream-readable-didRead.js
index 7508dd76899013..1bc76829d65e12 100644
--- a/test/parallel/test-stream-readable-didRead.js
+++ b/test/parallel/test-stream-readable-didRead.js
@@ -1,12 +1,13 @@
 'use strict';
 const common = require('../common');
 const assert = require('assert');
-const Readable = require('stream').Readable;
+const { isDisturbed, Readable } = require('stream');
 
 function noop() {}
 
 function check(readable, data, fn) {
   assert.strictEqual(readable.readableDidRead, false);
+  assert.strictEqual(isDisturbed(readable), false);
   if (data === -1) {
     readable.on('error', common.mustCall());
     readable.on('data', common.mustNotCall());
@@ -28,6 +29,7 @@ function check(readable, data, fn) {
   fn();
   setImmediate(() => {
     assert.strictEqual(readable.readableDidRead, true);
+    assert.strictEqual(isDisturbed(readable), true);
   });
 }
 
diff --git a/test/parallel/test-whatwg-readablestream.js b/test/parallel/test-whatwg-readablestream.js
index 1c18efeec41963..c8b82fa9823786 100644
--- a/test/parallel/test-whatwg-readablestream.js
+++ b/test/parallel/test-whatwg-readablestream.js
@@ -2,6 +2,7 @@
 'use strict';
 
 const common = require('../common');
+const { isDisturbed } = require('stream');
 const assert = require('assert');
 const {
   isPromise,
@@ -1520,3 +1521,52 @@ class Source {
   readableByteStreamControllerClose(controller);
   readableByteStreamControllerEnqueue(controller);
 }
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue('a');
+      controller.close();
+    },
+    pull: common.mustNotCall(),
+  });
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.close();
+    },
+    pull: common.mustNotCall(),
+  });
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}
+
+{
+  const stream = new ReadableStream({
+    start(controller) {
+    },
+    pull: common.mustNotCall(),
+  });
+  stream.cancel();
+
+  const reader = stream.getReader();
+  (async () => {
+    isDisturbed(stream, false);
+    await reader.read();
+    isDisturbed(stream, true);
+  })().then(common.mustCall());
+}

From 8673e33685470102ece6e651b1be206fbc8a16df Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 09:23:07 +0200
Subject: [PATCH 3/6] fixup

---
 doc/api/stream.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 3e5ee464f32682..77a981772b6a65 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2056,7 +2056,7 @@ added: REPLACEME
 * `stream` {stream.Readable|ReadableStream}
 * Returns: `boolean`
 
-Returns whether the stream has been read from.
+Returns whether the stream has been read from or cancelled.
 
 ### `stream.Readable.toWeb(streamReadable)`
 <!-- YAML

From caa8e3b6a1d3bc404a502631593c419f67da0780 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 09:34:24 +0200
Subject: [PATCH 4/6] fixup: readableDidRead readableAborted

---
 doc/api/stream.md                             | 17 +++++-
 lib/internal/streams/readable.js              | 23 +++++---
 lib/internal/streams/utils.js                 |  6 +-
 test/parallel/test-stream-readable-aborted.js | 55 +++++++++++++++++++
 4 files changed, 87 insertions(+), 14 deletions(-)
 create mode 100644 test/parallel/test-stream-readable-aborted.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 77a981772b6a65..e7ef3aaa40ac1a 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1261,16 +1261,27 @@ added: v11.4.0
 Is `true` if it is safe to call [`readable.read()`][stream-read], which means
 the stream has not been destroyed or emitted `'error'` or `'end'`.
 
+##### `readable.readableAborted`
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* {boolean}
+
+Returns whether the stream was destroyed before `'end'` has been emitted.
+
 ##### `readable.readableDidRead`
 <!-- YAML
 added: REPLACEME
 -->
 
+> Stability: 1 - Experimental
+
 * {boolean}
 
-Allows determining if the stream has been or is about to be read.
-Returns true if `'data'`, `'end'`, `'error'` or `'close'` has been
-emitted.
+Returns whether `'data'` has been emitted.
 
 ##### `readable.readableEncoding`
 <!-- YAML
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index c720e7fdca3d14..1ee1fcdbed764d 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -173,6 +173,8 @@ function ReadableState(options, stream, isDuplex) {
 
   this.dataEmitted = false;
 
+  this.aborted = false;
+
   this.decoder = null;
   this.encoding = null;
   if (options && options.encoding) {
@@ -215,7 +217,12 @@ function Readable(options) {
   });
 }
 
-Readable.prototype.destroy = destroyImpl.destroy;
+Readable.prototype.destroy = function(err) {
+  if (!this._readableState.endEmitted) {
+    this._readableState.aborted = true;
+  }
+  destroyImpl.destroy.call(this, err);
+};
 Readable.prototype._undestroy = destroyImpl.undestroy;
 Readable.prototype._destroy = function(err, cb) {
   cb(err);
@@ -1185,12 +1192,14 @@ ObjectDefineProperties(Readable.prototype, {
   readableDidRead: {
     enumerable: false,
     get: function() {
-      return !!(
-        this._readableState.dataEmitted ||
-        this._readableState.endEmitted ||
-        this._readableState.errorEmitted ||
-        this._readableState.closeEmitted
-      );
+      return this._readableState.dataEmitted;
+    }
+  },
+
+  readableAborted: {
+    enumerable: false,
+    get: function() {
+      return this._readableState.aborted;
     }
   },
 
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index f396872270a8de..a7ecf419de3baa 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -197,12 +197,10 @@ function willEmitClose(stream) {
 }
 
 function isDisturbed(stream) {
-  const state = stream && stream._readableState;
   return !!(stream && (
     stream.readableDidRead ||
-    isDestroyed(stream) ||
-    stream[kIsDisturbed] ||
-    (state && state.endEmitted)
+    stream.readableAborted ||
+    stream[kIsDisturbed]
   ));
 }
 
diff --git a/test/parallel/test-stream-readable-aborted.js b/test/parallel/test-stream-readable-aborted.js
new file mode 100644
index 00000000000000..059959d20fc646
--- /dev/null
+++ b/test/parallel/test-stream-readable-aborted.js
@@ -0,0 +1,55 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Readable } = require('stream');
+
+{
+  const readable = new Readable({
+    read() {
+    }
+  });
+  assert.strictEqual(readable.readableAborted, false);
+  readable.destroy();
+  assert.strictEqual(readable.readableAborted, true);
+}
+
+{
+  const readable = new Readable({
+    read() {
+    }
+  });
+  assert.strictEqual(readable.readableAborted, false);
+  readable.push(null);
+  readable.destroy();
+  assert.strictEqual(readable.readableAborted, true);
+}
+
+{
+  const readable = new Readable({
+    read() {
+    }
+  });
+  assert.strictEqual(readable.readableAborted, false);
+  readable.push('asd');
+  readable.destroy();
+  assert.strictEqual(readable.readableAborted, true);
+}
+
+{
+  const readable = new Readable({
+    read() {
+    }
+  });
+  assert.strictEqual(readable.readableAborted, false);
+  readable.push('asd');
+  readable.push(null);
+  readable.on('end', common.mustCall(() => {
+    assert.strictEqual(readable.readableAborted, false);
+    readable.destroy();
+    assert.strictEqual(readable.readableAborted, false);
+    queueMicrotask(() => {
+      assert.strictEqual(readable.readableAborted, false);
+    });
+  }));
+}

From 13123c46cb91e7e8aaa9d6210409bcaffe13f661 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 09:52:02 +0200
Subject: [PATCH 5/6] fixup

---
 doc/api/stream.md                |  2 +-
 lib/internal/streams/readable.js | 12 +++---------
 2 files changed, 4 insertions(+), 10 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index e7ef3aaa40ac1a..1a1bebb9328715 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1270,7 +1270,7 @@ added: REPLACEME
 
 * {boolean}
 
-Returns whether the stream was destroyed before `'end'` has been emitted.
+Returns whether the stream was destroyed or errored before emitting `'end'`.
 
 ##### `readable.readableDidRead`
 <!-- YAML
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 1ee1fcdbed764d..a487af17e360e8 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -173,8 +173,6 @@ function ReadableState(options, stream, isDuplex) {
 
   this.dataEmitted = false;
 
-  this.aborted = false;
-
   this.decoder = null;
   this.encoding = null;
   if (options && options.encoding) {
@@ -217,12 +215,7 @@ function Readable(options) {
   });
 }
 
-Readable.prototype.destroy = function(err) {
-  if (!this._readableState.endEmitted) {
-    this._readableState.aborted = true;
-  }
-  destroyImpl.destroy.call(this, err);
-};
+Readable.prototype.destroy = destroyImpl.destroy;
 Readable.prototype._undestroy = destroyImpl.undestroy;
 Readable.prototype._destroy = function(err, cb) {
   cb(err);
@@ -1199,7 +1192,8 @@ ObjectDefineProperties(Readable.prototype, {
   readableAborted: {
     enumerable: false,
     get: function() {
-      return this._readableState.aborted;
+      return (this._readableState.destroyed || this._readableState.errored) &&
+        !this._readableState.endEmitted;
     }
   },
 

From 0dda118df1cfa1dfee5acf27ea6ef72a42ab54e2 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 10:37:40 +0200
Subject: [PATCH 6/6] fixup

---
 lib/internal/streams/readable.js              | 2 +-
 test/parallel/test-stream-readable-aborted.js | 2 ++
 test/parallel/test-stream-readable-didRead.js | 6 ++++--
 3 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index a487af17e360e8..e46a9cb3ce9b16 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1192,7 +1192,7 @@ ObjectDefineProperties(Readable.prototype, {
   readableAborted: {
     enumerable: false,
     get: function() {
-      return (this._readableState.destroyed || this._readableState.errored) &&
+      return !!(this._readableState.destroyed || this._readableState.errored) &&
         !this._readableState.endEmitted;
     }
   },
diff --git a/test/parallel/test-stream-readable-aborted.js b/test/parallel/test-stream-readable-aborted.js
index 059959d20fc646..8915e07de45373 100644
--- a/test/parallel/test-stream-readable-aborted.js
+++ b/test/parallel/test-stream-readable-aborted.js
@@ -44,6 +44,7 @@ const { Readable } = require('stream');
   assert.strictEqual(readable.readableAborted, false);
   readable.push('asd');
   readable.push(null);
+  assert.strictEqual(readable.readableAborted, false);
   readable.on('end', common.mustCall(() => {
     assert.strictEqual(readable.readableAborted, false);
     readable.destroy();
@@ -52,4 +53,5 @@ const { Readable } = require('stream');
       assert.strictEqual(readable.readableAborted, false);
     });
   }));
+  readable.resume();
 }
diff --git a/test/parallel/test-stream-readable-didRead.js b/test/parallel/test-stream-readable-didRead.js
index 1bc76829d65e12..c67289e65442c5 100644
--- a/test/parallel/test-stream-readable-didRead.js
+++ b/test/parallel/test-stream-readable-didRead.js
@@ -28,8 +28,10 @@ function check(readable, data, fn) {
   readable.on('close', common.mustCall());
   fn();
   setImmediate(() => {
-    assert.strictEqual(readable.readableDidRead, true);
-    assert.strictEqual(isDisturbed(readable), true);
+    assert.strictEqual(readable.readableDidRead, data > 0);
+    if (data > 0) {
+      assert.strictEqual(isDisturbed(readable), true);
+    }
   });
 }