stream: add isErrored helper

Refs: nodejs/undici#1134

PR-URL: #41121
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Minwoo Jung <[email protected]>

Author: ronag

doc/api/stream.md

@@ -2154,6 +2154,57 @@ added: v16.8.0
 
 Returns whether the stream has been read from or cancelled.
 
+### `stream.isErrored(stream)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `stream` {Readable|Writable|Duplex|WritableStream|ReadableStream}
+* Returns: {boolean}
+
+Returns whether the stream has encountered an error.
+
+### `stream.Readable.toWeb(streamReadable)`
+
+<!-- YAML
+added: v17.0.0
+-->
+
+> Stability: 1 - Experimental
+
+* `streamReadable` {stream.Readable}
+* Returns: {ReadableStream}
+
+### `stream.Writable.fromWeb(writableStream[, options])`
+
+<!-- YAML
+added: v17.0.0
+-->
+
+> Stability: 1 - Experimental
+
+* `writableStream` {WritableStream}
+* `options` {Object}
+  * `decodeStrings` {boolean}
+  * `highWaterMark` {number}
+  * `objectMode` {boolean}
+  * `signal` {AbortSignal}
+* Returns: {stream.Writable}
+
+### `stream.Writable.toWeb(streamWritable)`
+
+<!-- YAML
+added: v17.0.0
+-->
+
+> Stability: 1 - Experimental
+
+* `streamWritable` {stream.Writable}
+* Returns: {WritableStream}
+
 ### `stream.Duplex.from(src)`
 
 <!-- YAML

lib/internal/streams/utils.js

@@ -6,6 +6,8 @@ const {
   SymbolIterator,
 } = primordials;
 
+const kDestroyed = Symbol('kDestroyed');
+const kIsErrored = Symbol('kIsErrored');
 const kIsDisturbed = Symbol('kIsDisturbed');
 
 function isReadableNodeStream(obj) {
@@ -212,10 +214,30 @@ function willEmitClose(stream) {
   );
 }
 
+function isDisturbed(stream) {
+  return !!(stream && (
+    stream[kIsDisturbed] ??
+    (stream.readableDidRead || stream.readableAborted)
+  ));
+}
+
+function isErrored(stream) {
+  return !!(stream && (
+    stream[kIsErrored] ??
+    stream.readableErrored ??
+    stream.writableErrored ??
+    stream._readableState?.errorEmitted ??
+    stream._writableState?.errorEmitted ??
+    stream._readableState?.errored ??
+    stream._writableState?.errored
+  ));
+}
 
 module.exports = {
   isDisturbed,
+  isErrored,
   kIsDisturbed,
+  kIsErrored,
   isClosed,
   isDestroyed,
   isDuplexNodeStream,

lib/internal/webstreams/readablestream.js

@@ -82,6 +82,7 @@ const {
 
 const {
   kIsDisturbed,
+  kIsErrored,
 } = require('internal/streams/utils');
 
 const {
@@ -241,6 +242,10 @@ class ReadableStream {
     return this[kState].disturbed;
   }
 
+  get [kIsErrored]() {
+    return this[kState].state === 'errored';
+  }
+
   /**
    * @readonly
    * @type {boolean}

lib/stream.js

@@ -36,9 +36,11 @@ const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
 
 const promises = require('stream/promises');
+const utils = require('internal/streams/utils');
 
 const Stream = module.exports = require('internal/streams/legacy').Stream;
-Stream.isDisturbed = require('internal/streams/utils').isDisturbed;
+Stream.isDisturbed = utils.isDisturbed;
+Stream.isErrored = utils.isErrored;
 Stream.Readable = require('internal/streams/readable');
 Stream.Writable = require('internal/streams/writable');
 Stream.Duplex = require('internal/streams/duplex');

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

@@ -1,15 +1,18 @@
 'use strict';
 const common = require('../common');
 const assert = require('assert');
-const { isDisturbed, Readable } = require('stream');
+const { isDisturbed, isErrored, Readable } = require('stream');
 
 function noop() {}
 
 function check(readable, data, fn) {
   assert.strictEqual(readable.readableDidRead, false);
   assert.strictEqual(isDisturbed(readable), false);
+  assert.strictEqual(isErrored(readable), false);
   if (data === -1) {
-    readable.on('error', common.mustCall());
+    readable.on('error', common.mustCall(() => {
+      assert.strictEqual(isErrored(readable), true);
+    }));
     readable.on('data', common.mustNotCall());
     readable.on('end', common.mustNotCall());
   } else {

test/parallel/test-whatwg-readablestream.js

@@ -2,7 +2,7 @@
 'use strict';
 
 const common = require('../common');
-const { isDisturbed } = require('stream');
+const { isDisturbed, isErrored } = require('stream');
 const assert = require('assert');
 const {
   isPromise,
@@ -1572,3 +1572,19 @@ class Source {
     isDisturbed(stream, true);
   })().then(common.mustCall());
 }
+
+
+{
+  const stream = new ReadableStream({
+    pull: common.mustCall((controller) => {
+      controller.error(new Error());
+    }),
+  });
+
+  const reader = stream.getReader();
+  (async () => {
+    isErrored(stream, false);
+    await reader.read().catch(common.mustCall());
+    isErrored(stream, true);
+  })().then(common.mustCall());
+}

tools/doc/type-parser.mjs

@@ -206,6 +206,10 @@ const customTypesMap = {
   'stream.Readable': 'stream.html#class-streamreadable',
   'stream.Transform': 'stream.html#class-streamtransform',
   'stream.Writable': 'stream.html#class-streamwritable',
+  'Duplex': 'stream.html#class-streamduplex',
+  'Readable': 'stream.html#class-streamreadable',
+  'Transform': 'stream.html#class-streamtransform',
+  'Writable': 'stream.html#class-streamwritable',
 
   'Immediate': 'timers.html#class-immediate',
   'Timeout': 'timers.html#class-timeout',