From 4b33951575fcbcf80c021338b55e2a09ccac0319 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=A0=E4=B8=8B=20=E5=85=8B=E5=BD=A6?= Date: Wed, 29 Jul 2026 07:52:58 +0900 Subject: [PATCH 1/3] stream: use internal abort hooks for web streams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 池下 克彦 --- lib/internal/streams/add-abort-signal.js | 4 +- lib/internal/streams/utils.js | 4 +- lib/internal/webstreams/readablestream.js | 40 +-- lib/internal/webstreams/writablestream.js | 14 +- .../test-webstreams-abort-controller.js | 238 +++++++++++++----- 5 files changed, 213 insertions(+), 87 deletions(-) diff --git a/lib/internal/streams/add-abort-signal.js b/lib/internal/streams/add-abort-signal.js index ccd50dd141e6..e0c6c60a9858 100644 --- a/lib/internal/streams/add-abort-signal.js +++ b/lib/internal/streams/add-abort-signal.js @@ -14,7 +14,7 @@ const { const { isNodeStream, isWebStream, - kControllerErrorFunction, + kControllerAbortFunction, } = require('internal/streams/utils'); const { eos } = require('internal/streams/end-of-stream'); @@ -47,7 +47,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(); diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js index 45f55316104f..1afee846d4ad 100644 --- a/lib/internal/streams/utils.js +++ b/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, diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 9b6dc4b4ad21..5fca5d03be8a 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/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,7 +2182,7 @@ 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': @@ -2192,14 +2190,18 @@ function readableStreamCancel(stream, reason) { 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( diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index f0eb687be129..8890d818fedd 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/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]() { diff --git a/test/parallel/test-webstreams-abort-controller.js b/test/parallel/test-webstreams-abort-controller.js index 64d38af1683e..2be9d1f0dcba 100644 --- a/test/parallel/test-webstreams-abort-controller.js +++ b/test/parallel/test-webstreams-abort-controller.js @@ -4,28 +4,90 @@ const common = require('../common'); const { finished, addAbortSignal } = require('stream'); const { ReadableStream, WritableStream } = require('stream/web'); const assert = require('assert'); +const { setImmediate } = require('timers/promises'); + +const typeErrorPredicate = { name: 'TypeError', code: 'ERR_INVALID_STATE' }; function createTestReadableStream() { - return new ReadableStream({ - start(controller) { + let controller; + const rs = new ReadableStream({ + start(c) { + controller = c; controller.enqueue('a'); controller.enqueue('b'); controller.enqueue('c'); controller.close(); } }); + return [rs, controller]; } function createTestWritableStream(values) { - return new WritableStream({ - write(chunk) { + let controller; + const ws = new WritableStream({ + start(c) { controller = c; }, + write(chunk, c) { values.push(chunk); } }); + return [ ws, controller ]; +} + +/** + * + * @param {ReadableStream} rs + * @param {import('internal/webstreams/readablestream').ReadableStreamReader} reader + * @param {{ + * isByob?: boolean, + * controller?: import('internal/webstreams/readablestream').ReadableStreamController, + * additionalAssertions?: () => void, + * }} options + */ +function assertReadableStreamEventuallyAborted(rs, reader, { + isByob, + controller, + additionalAssertions = common.mustCall() +} = {}) { + finished(rs, { writable: false }, common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + assert.rejects(reader.read(...(isByob ? [new Uint8Array(1)] : [])), /AbortError/).then(common.mustCall()); + assert.rejects(reader.closed, /AbortError/).then(common.mustCall()); + if (controller) { + assert.throws(() => controller.close(), typeErrorPredicate); + assert.throws(() => controller.enqueue(isByob ? new Uint8Array(1) : 'a'), typeErrorPredicate); + controller.error(new Error()); // Never throws + } + additionalAssertions(); + })); +} + +/** + * + * @param {WritableStream} ws + * @param {import('internal/webstreams/writablestream').WritableStreamDefaultWriter} writer + * @param {{ + * controller?: import('internal/webstreams/writablestream').WritableStreamDefaultController, + * additionalAssertions?: () => void, + * }} options + */ +function assertWritableStreamEventuallyAborted(ws, writer, { + controller, + additionalAssertions = common.mustCall(), +} = {}) { + finished(ws, { readable: false }, common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + assert.rejects(writer.write('a'), /AbortError/).then(common.mustCall()); + assert.rejects(writer.closed, /AbortError/).then(common.mustCall()); + if (controller) { + controller.error(new Error()); // Never throws + assert.strictEqual(controller.signal.aborted, true); + } + additionalAssertions(); + })); } { - const rs = createTestReadableStream(); + const [rs, controller] = createTestReadableStream(); const reader = rs.getReader(); @@ -33,11 +95,7 @@ function createTestWritableStream(values) { addAbortSignal(ac.signal, rs); - finished(rs, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(reader.read(), /AbortError/).then(common.mustCall()); - assert.rejects(reader.closed, /AbortError/).then(common.mustCall()); - })); + assertReadableStreamEventuallyAborted(rs, reader, { controller }); reader.read().then(common.mustCall((result) => { assert.strictEqual(result.value, 'a'); @@ -46,7 +104,7 @@ function createTestWritableStream(values) { } { - const rs = createTestReadableStream(); + const [rs] = createTestReadableStream(); const ac = new AbortController(); @@ -62,9 +120,80 @@ function createTestWritableStream(values) { } { - const rs1 = createTestReadableStream(); + const [rs, controller] = createTestReadableStream(); + const reader = rs.getReader(); + const ac = new AbortController(); + + addAbortSignal(ac.signal, rs); + controller.error = common.mustNotCall( + 'addAbortSignal() must not call an overridden controller.error()'); - const rs2 = createTestReadableStream(); + assertReadableStreamEventuallyAborted(rs, reader); + + reader.read().then(common.mustCall(() => { + ac.abort(); + })); +} + +{ + let controller; + const rs = new ReadableStream({ + type: 'bytes', + start(c) { controller = c; }, + }); + const ac = new AbortController(); + addAbortSignal(ac.signal, rs); + + const reader = rs.getReader({ mode: 'byob' }); + assertReadableStreamEventuallyAborted(rs, reader, { controller, isByob: true }); + + ac.abort(); +} + +{ + /** @member {import('internal/webstreams/readablestream').ReadableByteStreamController} */ + let controller; + const rs = new ReadableStream({ + type: 'bytes', + start(c) { controller = c; }, + }); + const ac = new AbortController(); + + addAbortSignal(ac.signal, rs); + controller.error = common.mustNotCall('addAbortSignal() must not call an overridden controller.error()'); + + const reader = rs.getReader({ mode: 'byob' }); + assertReadableStreamEventuallyAborted(rs, reader, { isByob: true }); + + ac.abort(); +} + +{ + /** @member {import('internal/webstreams/readablestream').ReadableStreamDefaultController} */ + let controller; + const pullPromiseWithResolvers = Promise.withResolvers(); + const rs = new ReadableStream({ + start(c) { controller = c; }, + pull() { return pullPromiseWithResolvers.promise; }, + }); + const ac = new AbortController(); + addAbortSignal(ac.signal, rs); + + const reader = rs.getReader(); + assertReadableStreamEventuallyAborted(rs, reader); + + const readPromise = reader.read(); + pullPromiseWithResolvers.resolve(setImmediate().then(() => { + ac.abort(); + controller.enqueue('a'); + })); + assert.rejects(readPromise, /AbortError/).then(common.mustCall()); + assert.rejects(pullPromiseWithResolvers.promise, typeErrorPredicate).then(common.mustCall()); +} + +{ + const [rs1, controller1] = createTestReadableStream(); + const [rs2, controller2] = createTestReadableStream(); const ac = new AbortController(); @@ -74,23 +203,14 @@ function createTestWritableStream(values) { const reader1 = rs1.getReader(); const reader2 = rs2.getReader(); - finished(rs1, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(reader1.read(), /AbortError/).then(common.mustCall()); - assert.rejects(reader1.closed, /AbortError/).then(common.mustCall()); - })); - - finished(rs2, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(reader2.read(), /AbortError/).then(common.mustCall()); - assert.rejects(reader2.closed, /AbortError/).then(common.mustCall()); - })); + assertReadableStreamEventuallyAborted(rs1, reader1, { controller: controller1 }); + assertReadableStreamEventuallyAborted(rs2, reader2, { controller: controller2 }); ac.abort(); } { - const rs = createTestReadableStream(); + const [rs, controller] = createTestReadableStream(); const { 0: rs1, 1: rs2 } = rs.tee(); @@ -101,24 +221,15 @@ function createTestWritableStream(values) { const reader1 = rs1.getReader(); const reader2 = rs2.getReader(); - finished(rs1, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(reader1.read(), /AbortError/).then(common.mustCall()); - assert.rejects(reader1.closed, /AbortError/).then(common.mustCall()); - })); - - finished(rs2, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(reader2.read(), /AbortError/).then(common.mustCall()); - assert.rejects(reader2.closed, /AbortError/).then(common.mustCall()); - })); + assertReadableStreamEventuallyAborted(rs1, reader1, { controller }); + assertReadableStreamEventuallyAborted(rs2, reader2, { controller }); ac.abort(); } { const values = []; - const ws = createTestWritableStream(values); + const [ws, controller] = createTestWritableStream(values); const ac = new AbortController(); @@ -126,23 +237,22 @@ function createTestWritableStream(values) { const writer = ws.getWriter(); - finished(ws, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.deepStrictEqual(values, ['a']); - assert.rejects(writer.write('b'), /AbortError/).then(common.mustCall()); - assert.rejects(writer.closed, /AbortError/).then(common.mustCall()); - })); + assertWritableStreamEventuallyAborted(ws, writer, { + controller, + additionalAssertions: common.mustCall(() => { + assert.deepStrictEqual(values, ['a']); + }), + }); - writer.write('a').then(() => { - ac.abort(); - }).then(common.mustCall()); + writer.write('a') + .then(common.mustCall(() => { ac.abort(); })); } { const values = []; - const ws1 = createTestWritableStream(values); - const ws2 = createTestWritableStream(values); + const [ws1, controller1] = createTestWritableStream(values); + const [ws2, controller2] = createTestWritableStream(values); const ac = new AbortController(); @@ -152,17 +262,29 @@ function createTestWritableStream(values) { const writer1 = ws1.getWriter(); const writer2 = ws2.getWriter(); - finished(ws1, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(writer1.write('a'), /AbortError/).then(common.mustCall()); - assert.rejects(writer1.closed, /AbortError/).then(common.mustCall()); - })); + const additionalAssertions = common.mustCall(() => { + assert.deepStrictEqual(values, []); + }, 2); + assertWritableStreamEventuallyAborted(ws1, writer1, { controller: controller1, additionalAssertions }); + assertWritableStreamEventuallyAborted(ws2, writer2, { controller: controller2, additionalAssertions }); - finished(ws2, common.mustCall((err) => { - assert.strictEqual(err.name, 'AbortError'); - assert.rejects(writer2.write('a'), /AbortError/).then(common.mustCall()); - assert.rejects(writer2.closed, /AbortError/).then(common.mustCall()); - })); + ac.abort(); +} + +{ + /** @member {import('internal/webstreams/writablestream').WritableStreamDefaultController} */ + let controller; + const ws = new WritableStream({ + start(c) { controller = c; }, + }); + const ac = new AbortController(); + addAbortSignal(ac.signal, ws); + + controller.abort = common.mustNotCall('addAbortSignal() must not call an overridden controller.abort()'); + controller.error = common.mustNotCall('addAbortSignal() must not call an overridden controller.error()'); + + const writer = ws.getWriter(); + assertWritableStreamEventuallyAborted(ws, writer); ac.abort(); } From a46e74abbfa5e1152ef3909e7e7c7b6d75ad3950 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=A0=E4=B8=8B=20=E5=85=8B=E5=BD=A6?= Date: Wed, 29 Jul 2026 07:53:01 +0900 Subject: [PATCH 2/3] stream: reject TransformStream in addAbortSignal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 池下 克彦 --- lib/internal/streams/add-abort-signal.js | 7 +++++-- test/parallel/test-stream-add-abort-signal.js | 5 +++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/lib/internal/streams/add-abort-signal.js b/lib/internal/streams/add-abort-signal.js index e0c6c60a9858..c17c8222784a 100644 --- a/lib/internal/streams/add-abort-signal.js +++ b/lib/internal/streams/add-abort-signal.js @@ -13,7 +13,8 @@ const { const { isNodeStream, - isWebStream, + isReadableStream, + isWritableStream, kControllerAbortFunction, } = require('internal/streams/utils'); @@ -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); diff --git a/test/parallel/test-stream-add-abort-signal.js b/test/parallel/test-stream-add-abort-signal.js index 3077137176ff..c81b6a2b29b5 100644 --- a/test/parallel/test-stream-add-abort-signal.js +++ b/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/); } { From 113374f1a6c626a3a3b0b3bef6fa292b088a55de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=A0=E4=B8=8B=20=E5=85=8B=E5=BD=A6?= Date: Wed, 29 Jul 2026 07:53:02 +0900 Subject: [PATCH 3/3] doc: update addAbortSignal webstream behavior MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 池下 克彦 --- doc/api/stream.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/doc/api/stream.md b/doc/api/stream.md index 2f7fd7961a73..6145b04fdc4c 100644 --- a/doc/api/stream.md +++ b/doc/api/stream.md @@ -3540,6 +3540,11 @@ readable.getReader().read().then((result) => {