stream: fix addAbortSignal() for web streams

doc/api/stream.md

@@ -3540,6 +3540,11 @@ readable.getReader().read().then((result) => {
 <!-- YAML
 added: v15.4.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/62450
+    description: For web streams, `addAbortSignal()` now runs `cancel` steps on
+                 `ReadableStream`, and aborts `controller.signal` and runs
+                 `abort` steps on `WritableStream`.
   - version:
     - v19.7.0
     - v18.16.0
@@ -3557,7 +3562,10 @@ control stream destruction using an `AbortController`.
 
 Calling `abort` on the `AbortController` corresponding to the passed
 `AbortSignal` will behave the same way as calling `.destroy(new AbortError())`
-on the stream, and `controller.error(new AbortError())` for webstreams.
+on a Node.js stream. For web streams, it will error the stream with an
+`AbortError`. For `ReadableStream`, it will then run the stream's cancel steps.
+For `WritableStream`, it will abort `controller.signal` and then run the
+stream's abort steps.
 
 ```js
 const fs = require('node:fs');

lib/internal/streams/add-abort-signal.js

@@ -13,8 +13,9 @@ const {
 
 const {
   isNodeStream,
-  isWebStream,
-  kControllerErrorFunction,
+  isReadableStream,
+  isWritableStream,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const { eos } = require('internal/streams/end-of-stream');
@@ -32,7 +33,9 @@ const validateAbortSignal = (signal, name) => {
 
 module.exports.addAbortSignal = function addAbortSignal(signal, stream) {
   validateAbortSignal(signal, 'signal');
-  if (!isNodeStream(stream) && !isWebStream(stream)) {
+  if (!isNodeStream(stream) &&
+      !isReadableStream(stream) &&
+      !isWritableStream(stream)) {
     throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
   }
   return module.exports.addAbortSignalNoValidate(signal, stream);
@@ -47,7 +50,7 @@ module.exports.addAbortSignalNoValidate = function(signal, stream) {
       stream.destroy(new AbortError(undefined, { cause: signal.reason }));
     } :
     () => {
-      stream[kControllerErrorFunction](new AbortError(undefined, { cause: signal.reason }));
+      stream[kControllerAbortFunction](new AbortError(undefined, { cause: signal.reason }));
     };
   if (signal.aborted) {
     onAbort();

lib/internal/streams/utils.js

@@ -20,7 +20,7 @@ const kIsDisturbed = SymbolFor('nodejs.stream.disturbed');
 const kOnConstructed = Symbol('kOnConstructed');
 
 const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
-const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');
+const kControllerAbortFunction = SymbolFor('nodejs.webstream.controllerAbortFunction');
 
 const kState = Symbol('kState');
 const kObjectMode = 1 << 0;
@@ -326,7 +326,7 @@ module.exports = {
   isReadable,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
   kIsWritable,
   isClosed,
   isDuplexNodeStream,

lib/internal/webstreams/readablestream.js

@@ -93,7 +93,7 @@ const {
   kIsErrored,
   kIsReadable,
   kIsClosedPromise,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -287,6 +287,13 @@ class ReadableStream {
     }
   }
 
+  [kControllerAbortFunction](error) {
+    if (!isReadableStream(this))
+      throw new ERR_INVALID_THIS('ReadableStream');
+    if (this[kState].state === 'readable')
+      setPromiseHandled(readableStreamCancel(this, error, 'errored'));
+  }
+
   get [kIsDisturbed]() {
     return this[kState].disturbed;
   }
@@ -299,15 +306,6 @@ class ReadableStream {
     return this[kState].state === 'readable';
   }
 
-  [kControllerErrorFunction](error) {
-    // Used by the internal stream interop (addAbortSignal). Historically
-    // only default controllers were wired here; byte stream controllers
-    // keep the previous no-op behavior.
-    const controller = this[kState].controller;
-    if (isReadableStreamDefaultController(controller))
-      controller.error(error);
-  }
-
   // Used by the internal stream interop (end-of-stream). Materialized
   // lazily since its settlement is derivable from the stream state; the
   // settle sites in readableStreamClose/Error only touch the cache.
@@ -2184,22 +2182,26 @@ function isReadableStreamLocked(stream) {
   return stream[kState].reader !== undefined;
 }
 
-function readableStreamCancel(stream, reason) {
+function readableStreamCancel(stream, reason, finalState = 'closed') {
   stream[kState].disturbed = true;
   switch (stream[kState].state) {
     case 'closed':
       return PromiseResolve();
     case 'errored':
       return PromiseReject(stream[kState].storedError);
   }
-  readableStreamClose(stream);
-  const {
-    reader,
-  } = stream[kState];
-  if (reader !== undefined && readableStreamHasBYOBReader(stream)) {
-    const readIntoRequests = reader[kState].readIntoRequests;
-    while (readIntoRequests.length)
-      readIntoRequests.shift()[kClose]();
+  if (finalState === 'errored') {
+    readableStreamError(stream, reason);
+  } else {
+    readableStreamClose(stream);
+    const {
+      reader,
+    } = stream[kState];
+    if (reader !== undefined && readableStreamHasBYOBReader(stream)) {
+      const readIntoRequests = reader[kState].readIntoRequests;
+      while (readIntoRequests.length)
+        readIntoRequests.shift()[kClose]();
+    }
   }
 
   return PromisePrototypeThen(

lib/internal/webstreams/writablestream.js

@@ -84,7 +84,7 @@ const {
   kIsClosedPromise,
   kIsErrored,
   kIsWritable,
-  kControllerErrorFunction,
+  kControllerAbortFunction,
 } = require('internal/streams/utils');
 
 const {
@@ -201,6 +201,13 @@ class WritableStream {
       size);
   }
 
+  [kControllerAbortFunction](reason) {
+    if (!isWritableStream(this))
+      throw new ERR_INVALID_THIS('WritableStream');
+    if (this[kState].state === 'writable')
+      setPromiseHandled(writableStreamAbort(this, reason));
+  }
+
   get [kIsErrored]() {
     return this[kState].state === 'errored';
   }
@@ -209,11 +216,6 @@ class WritableStream {
     return this[kState].state === 'writable';
   }
 
-  [kControllerErrorFunction](error) {
-    // Used by the internal stream interop (addAbortSignal).
-    this[kState].controller.error(error);
-  }
-
   // Used by the internal stream interop (end-of-stream). Materialized
   // lazily since its settlement is derivable from the stream state.
   get [kIsClosedPromise]() {

test/parallel/test-stream-add-abort-signal.js

@@ -4,6 +4,7 @@
 require('../common');
 const assert = require('assert');
 const { addAbortSignal, Readable } = require('stream');
+const { TransformStream } = require('stream/web');
 const {
   addAbortSignalNoValidate,
 } = require('internal/streams/add-abort-signal');
@@ -17,6 +18,10 @@ const {
   assert.throws(() => {
     addAbortSignal(ac.signal, 'INVALID_STREAM');
   }, /ERR_INVALID_ARG_TYPE/);
+
+  assert.throws(() => {
+    addAbortSignal(ac.signal, new TransformStream());
+  }, /ERR_INVALID_ARG_TYPE/);
 }
 
 {