stream: add isDisturbed helper

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

Refs: #39627

Author: ronag

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,18 @@ function willEmitClose(stream) {
   );
 }
 
+function isDisturbed(stream) {
+  return stream && (
+    stream.readableDidRead ||
+    stream.destroyed ||
+    stream[kIsDisturbed]
+  );
+}
+
 module.exports = {
   kDestroyed,
+  isDisturbed,
+  kIsDisturbed,
   isClosed,
   isDestroyed,
   isFinished,

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}

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');