diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 84f16dc7b4a..71317f9f817 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -111,6 +111,7 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, kEmptyQueue, kResolvedPromise, kState, @@ -252,6 +253,16 @@ class ReadableStream { */ constructor(source = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + // Empty-argument `new ReadableStream()`: no source, no strategy, and + // no controller. Reads never deliver data, so skip those allocations + // until getReader/cancel/error first need a default controller. + // Subclasses that call those methods after super() materialize the + // controller in the subclass constructor; see + // ensureEmptyDefaultController. + if (source === kEmptyObject && strategy === kEmptyObject) { + this[kState] = createReadableStreamState(); + return; + } validateObject(source, 'source', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); this[kState] = createReadableStreamState(); @@ -301,8 +312,14 @@ class ReadableStream { // only default controllers were wired here; byte stream controllers // keep the previous no-op behavior. const controller = this[kState].controller; + if (controller === undefined) { + if (this[kState].state === 'readable') { + readableStreamError(this, error); + } + return; + } if (isReadableStreamDefaultController(controller)) - controller.error(error); + readableStreamDefaultControllerError(controller, error); } // Used by the internal stream interop (end-of-stream). Materialized @@ -351,6 +368,10 @@ class ReadableStream { return PromiseReject( new ERR_INVALID_STATE.TypeError('ReadableStream is locked')); } + // Only materialize the deferred empty controller when cancel will + // actually run cancel steps. closed/errored streams return immediately. + if (this[kState].state === 'readable') + ensureEmptyDefaultController(this); return readableStreamCancel(this, reason); } @@ -1422,6 +1443,7 @@ function createReadableStreamState() { return { __proto__: null, closedPromise: undefined, + controller: undefined, disturbed: false, reader: undefined, state: 'readable', @@ -2580,6 +2602,7 @@ function setupReadableStreamBYOBReader(reader, stream) { function setupReadableStreamDefaultReader(reader, stream) { if (isReadableStreamLocked(stream)) throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked'); + ensureEmptyDefaultController(stream); readableStreamReaderGenericInitialize(reader, stream); reader[kState].readRequests = kEmptyQueue; } @@ -2729,7 +2752,8 @@ function readableStreamDefaultControllerPull(controller) { // The pull algorithm may be a raw callback (a wrapped user source.pull // returns its result uncoerced; a synchronous throw surfaces here) or an // internal algorithm that always returns a promise; thenAlgorithmResult - // handles both. + // handles both. Non-thenable results react on kResolvedPromise so each + // pull is still separated by a microtask, matching the spec. let result; try { result = controller[kState].pullAlgorithm(controller); @@ -2796,6 +2820,43 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) { readableStreamDefaultControllerPull(controller); } +// Materialize the deferred default controller for `new ReadableStream()`. +// +// started is true immediately: the empty-argument start algorithm is a +// no-op, so there is no initial pull and nothing can observe an unstarted +// controller without first calling getReader/cancel/pipeTo/tee/values, +// all of which come through here. That is also why this still matches +// WPT: those tests either pass a source (leaving this path) or wait for +// start, which is already complete for a no-op start. +// +// Subclasses that call cancel(), getReader(), pipeTo(), tee(), or +// values() in the constructor body after super() will materialize the +// controller before the subclass constructor finishes. Passing a source +// (for example to install start/pull) leaves the empty-argument path +// and creates the controller during super() as usual. +function ensureEmptyDefaultController(stream) { + if (stream[kState].controller !== undefined) + return stream[kState].controller; + const controller = new ReadableStreamDefaultController(kSkipThrow); + controller[kState] = { + cancelAlgorithm: nonOpCancel, + closeRequested: false, + highWaterMark: 1, + pullAgain: false, + pullAlgorithm: nonOpCallback, + pulling: false, + pullFulfilled: undefined, + pullRejected: undefined, + queue: kEmptyQueue, + queueTotalSize: 0, + started: true, + sizeAlgorithm: defaultSizeAlgorithm, + stream, + }; + stream[kState].controller = controller; + return controller; +} + function setupReadableStreamDefaultController( stream, controller, @@ -2824,8 +2885,7 @@ function setupReadableStreamDefaultController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // Non-thenable start result: fulfillment is guaranteed and no .then // lookup on the result is observable, so run the post-start step // directly at the exact microtask position the promise reaction @@ -3708,8 +3768,7 @@ function setupReadableByteStreamController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // See setupReadableStreamDefaultController. queueMicrotask(() => { controller[kState].started = true; diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 535c783a3a3..6cfe15ca7e8 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -123,9 +123,15 @@ class TransformStream { writableStrategy = kEmptyObject, readableStrategy = kEmptyObject) { markTransferMode(this, false, true); - validateObject(transformer, 'transformer', kValidateObjectAllowObjects); - validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull); - validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull); + if (transformer !== kEmptyObject) { + validateObject(transformer, 'transformer', kValidateObjectAllowObjects); + } + if (writableStrategy !== kEmptyObject) { + validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull); + } + if (readableStrategy !== kEmptyObject) { + validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull); + } const readableType = transformer?.readableType; const writableType = transformer?.writableType; const start = transformer?.start; diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 05439a25dcb..8fba954bd52 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -136,6 +136,16 @@ function cloneAsUint8Array(view) { ); } +// True when `value` cannot be a thenable: null, undefined, or a +// non-object non-function primitive. Objects and functions are treated +// as maybe-thenable without looking up `.then` (that lookup is +// observable). Proxies of objects/functions take the maybe-thenable +// path; a Proxy around a primitive is still an object. +function isNonThenable(value) { + return value === null || + (typeof value !== 'object' && typeof value !== 'function'); +} + function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) { return toBuffer !== fromBuffer && !ArrayBufferPrototypeGetDetached(toBuffer) && @@ -333,6 +343,11 @@ function enqueueValueWithSize(controller, value, size) { // each known call-site arity gets its own wrapper. The exact number of // arguments passed through to the user callback is observable and must be // preserved. +// +// Cold algorithms (cancel/close/abort/flush/transform) stay `async` so +// a user thenable is adopted with the same microtask count as before. +// Pull/write use the raw-callback contract instead (see +// createRawCallback*) and route results through thenAlgorithmResult(). function createPromiseCallbackNoParams(name, fn, thisArg) { validateFunction(fn, name); return async () => FunctionPrototypeCall(fn, thisArg); @@ -364,8 +379,7 @@ const kResolvedPromise = PromiseResolve(); // matches the spec's "a promise resolved with" conversion (identity for // native promises). function thenAlgorithmResult(result, onFulfilled, onRejected) { - if (result === null || - (typeof result !== 'object' && typeof result !== 'function')) { + if (isNonThenable(result)) { PromisePrototypeThen(kResolvedPromise, onFulfilled); } else { PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected); @@ -389,7 +403,8 @@ function isPromisePending(promise) { } // Shared shapes for lazily-materialized { promise, resolve, reject } -// records whose settlement is already known. +// records whose settlement is already known. Each call mints a fresh +// promise so public slots (writer.ready / writer.closed) stay distinct. function resolvedRecord() { return { promise: PromiseResolve(), @@ -455,6 +470,7 @@ module.exports = { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, isPromisePending, kEmptyQueue, kResolvedPromise, @@ -465,7 +481,6 @@ module.exports = { nonOpCallback, nonOpCancel, nonOpFlush, - peekQueueValue, rejectedHandledRecord, resetQueue, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 1e9ca02cfe9..c0bb70ca397 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -65,6 +65,7 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, isPromisePending, kEmptyQueue, kState, @@ -183,6 +184,15 @@ class WritableStream { */ constructor(sink = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + if (sink === kEmptyObject && strategy === kEmptyObject) { + this[kState] = createWritableStreamState(); + setupWritableStreamDefaultControllerFromSink( + this, + sink, + 1, + defaultSizeAlgorithm); + return; + } validateObject(sink, 'sink', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); const type = sink?.type; @@ -532,7 +542,7 @@ class WritableStreamDefaultController { get signal() { if (!isWritableStreamDefaultController(this)) throw new ERR_INVALID_THIS('WritableStreamDefaultController'); - return this[kState].abortController.signal; + return (this[kState].abortController ??= new AbortController()).signal; } /** @@ -707,7 +717,9 @@ function writableStreamAbort(stream, reason) { if (state === 'closed' || state === 'errored') return PromiseResolve(); - controller[kState].abortController.abort(reason); + // Materialize lazily so construction stays cheap, but abort() must + // still abort the same signal later observed via controller.signal. + (controller[kState].abortController ??= new AbortController()).abort(reason); state = stream[kState].state; if (state === 'closed' || state === 'errored') @@ -1169,6 +1181,21 @@ function writableStreamDefaultControllerWrite(controller, chunk, chunkSize) { writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); } +function writableStreamDefaultControllerCompleteWrite(controller) { + const stream = controller[kState].stream; + writableStreamFinishInFlightWrite(stream); + const streamState = stream[kState]; + const { + state, + } = streamState; + assert(state === 'writable' || state === 'erroring'); + dequeueValue(controller); + if (!streamState.closeQueuedOrInFlight && + state === 'writable') { + writableStreamUpdateBackpressure(controller, streamState); + } +} + function writableStreamDefaultControllerProcessWrite(controller, chunk) { const { stream, @@ -1181,17 +1208,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // so they are created once on the first write and reused for every // subsequent write instead of allocating two fresh closures per chunk. controller[kState].writeFulfilled = () => { - writableStreamFinishInFlightWrite(stream); - const streamState = stream[kState]; - const { - state, - } = streamState; - assert(state === 'writable' || state === 'erroring'); - dequeueValue(controller); - if (!streamState.closeQueuedOrInFlight && - state === 'writable') { - writableStreamUpdateBackpressure(controller, streamState); - } + writableStreamDefaultControllerCompleteWrite(controller); writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); }; controller[kState].writeRejected = (error) => { @@ -1204,7 +1221,8 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // The write algorithm may be a raw callback (a wrapped user sink.write // returns its result uncoerced; a synchronous throw surfaces here) or an // internal algorithm that always returns a promise; thenAlgorithmResult - // handles both. + // handles both. Non-thenable results react on kResolvedPromise so each + // write completion is still separated by a microtask. let result; try { result = writeAlgorithm(chunk, controller); @@ -1373,7 +1391,7 @@ function setupWritableStreamDefaultController( highWaterMark, queue: kEmptyQueue, queueTotalSize: 0, - abortController: new AbortController(), + abortController: undefined, sizeAlgorithm, started: false, stream, @@ -1387,8 +1405,7 @@ function setupWritableStreamDefaultController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // Non-thenable start result: fulfillment is guaranteed and no .then // lookup on the result is observable, so run the post-start step // directly at the exact microtask position the promise reaction diff --git a/test/parallel/test-whatwg-webstreams-hotpath.js b/test/parallel/test-whatwg-webstreams-hotpath.js new file mode 100644 index 00000000000..3cff7f555e9 --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-hotpath.js @@ -0,0 +1,306 @@ +// Flags: --expose-internals --no-warnings +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + ReadableStream, + WritableStream, +} = require('node:stream/web'); +const { + cloneAsUint8Array, + isNonThenable, + kState, +} = require('internal/webstreams/util'); +const { + kControllerErrorFunction, +} = require('internal/streams/utils'); + +assert.strictEqual(typeof isNonThenable, 'function'); +assert.strictEqual(typeof cloneAsUint8Array, 'function'); + +assert.strictEqual(isNonThenable(undefined), true); +assert.strictEqual(isNonThenable(null), true); +assert.strictEqual(isNonThenable(1), true); +assert.strictEqual(isNonThenable('x'), true); +assert.strictEqual(isNonThenable(true), true); +assert.strictEqual(isNonThenable({}), false); +assert.strictEqual(isNonThenable(() => {}), false); +assert.strictEqual(isNonThenable(Promise.resolve()), false); +assert.strictEqual(isNonThenable(new Proxy({}, {})), false); +assert.strictEqual(isNonThenable(new Proxy(Object(1), {})), false); +assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); + +{ + const src = new Uint8Array([1, 2, 3, 4]); + const cloned = cloneAsUint8Array(src); + assert.ok(cloned instanceof Uint8Array); + assert.deepStrictEqual([...cloned], [1, 2, 3, 4]); + src[0] = 9; + assert.strictEqual(cloned[0], 1); +} + +{ + const view = new DataView(new ArrayBuffer(4)); + new Uint8Array(view.buffer).set([9, 8, 7, 6]); + const cloned = cloneAsUint8Array(view); + assert.ok(cloned instanceof Uint8Array); + assert.deepStrictEqual([...cloned], [9, 8, 7, 6]); +} + +{ + const cloned = cloneAsUint8Array(new Uint8Array()); + assert.ok(cloned instanceof Uint8Array); + assert.strictEqual(cloned.byteLength, 0); +} + +{ + const buf = Buffer.alloc(16); + buf[4] = 7; + buf[5] = 8; + const sliced = buf.subarray(4, 8); + const cloned = cloneAsUint8Array(sliced); + assert.deepStrictEqual([...cloned], [7, 8, 0, 0]); + assert.strictEqual(cloned.buffer.byteLength, 4); +} + +{ + const ab = new ArrayBuffer(8); + const view = new Uint8Array(ab); + ab.transfer(); + assert.throws(() => cloneAsUint8Array(view), { + name: 'TypeError', + }); +} + +// Public API: pull-driven ReadableStream + read(). +(async () => { + const rs = new ReadableStream({ + start(controller) { + controller.enqueue('a'); + controller.enqueue('b'); + controller.close(); + }, + }); + const reader = rs.getReader(); + { + const { value, done } = await reader.read(); + assert.strictEqual(value, 'a'); + assert.strictEqual(done, false); + } + { + const { value, done } = await reader.read(); + assert.strictEqual(value, 'b'); + assert.strictEqual(done, false); + } + { + const { value, done } = await reader.read(); + assert.strictEqual(value, undefined); + assert.strictEqual(done, true); + } +})().then(common.mustCall(), common.mustNotCall()); + +// Public API: pipeTo with a sync sink. +(async () => { + const expected = []; + const received = []; + const rs = new ReadableStream({ + start(controller) { + for (let i = 0; i < 32; i++) { + expected.push(i); + controller.enqueue(i); + } + controller.close(); + }, + }); + await rs.pipeTo(new WritableStream({ + write(chunk) { + received.push(chunk); + }, + })); + assert.deepStrictEqual(received, expected); +})().then(common.mustCall(), common.mustNotCall()); + +// Spec path: each pull is separated by a microtask. Start schedules one +// pull; further pulls wait for that fulfillment and do not run in the +// same turn. Nested queueMicrotask checks encode that exact schedule +// and will fail if an unrelated change shifts pull timing. +{ + let calls = 0; + new ReadableStream({ + pull(controller) { + controller.enqueue(++calls); + }, + }, { + highWaterMark: 4, + }); + queueMicrotask(common.mustCall(() => { + assert.strictEqual(calls, 1); + // The next pull is queued only after fulfillment, so it is not + // invoked in this same turn. + queueMicrotask(common.mustCall(() => { + assert.strictEqual(calls, 2); + })); + })); +} + +// pipeTo of a pull-driven source must deliver every chunk. +(async () => { + const n = 64; + let i = 0; + const received = []; + const rs = new ReadableStream({ + pull(controller) { + if (i < n) + controller.enqueue(i++); + else + controller.close(); + }, + }, { highWaterMark: 8 }); + await rs.pipeTo(new WritableStream({ + write(chunk) { + received.push(chunk); + }, + }, { highWaterMark: 8 })); + assert.strictEqual(received.length, n); + assert.deepStrictEqual(received, Array.from({ length: n }, (_, k) => k)); +})().then(common.mustCall(), common.mustNotCall()); + +{ + // Empty-argument construction defers the controller. cancel() and + // getReader() must still work on the public API. + const rs = new ReadableStream(); + rs.cancel().then(common.mustCall(), common.mustNotCall()); +} + +{ + // Subclass that calls getReader() before the subclass constructor + // finishes: the deferred controller is materialized then. + class Sub extends ReadableStream { + constructor() { + super(); + this.reader = this.getReader(); + } + } + const rs = new Sub(); + assert.ok(rs.locked); + rs.reader.cancel().then(common.mustCall(), common.mustNotCall()); +} + +{ + // Subclass that calls cancel() before the subclass constructor finishes. + class Sub extends ReadableStream { + constructor() { + super(); + this.closed = this.cancel(); + } + } + const rs = new Sub(); + rs.closed.then(common.mustCall(), common.mustNotCall()); +} + +{ + // Passing a source leaves the empty-argument path, so start() receives + // a controller during super() even if the subclass constructor later + // calls getReader(). + let sawController = false; + class Sub extends ReadableStream { + constructor() { + super({ + start(controller) { + sawController = controller != null; + }, + }); + this.reader = this.getReader(); + } + } + const rs = new Sub(); + assert.ok(rs.locked); + queueMicrotask(common.mustCall(() => { + assert.strictEqual(sawController, true); + rs.reader.cancel().then(common.mustCall(), common.mustNotCall()); + })); +} + +{ + const rs = new ReadableStream(); + const reader = rs.getReader(); + reader.cancel().then(common.mustCall(), common.mustNotCall()); +} + +{ + // A Proxy around a thenable must not take the non-thenable shortcut. + let pulled = false; + const thenable = new Proxy({ + then(resolve) { + resolve(); + }, + }, {}); + const rs = new ReadableStream({ + pull(controller) { + if (pulled) { + controller.close(); + return thenable; + } + pulled = true; + controller.enqueue('proxied'); + return thenable; + }, + }); + rs.getReader().read().then(common.mustCall(({ value, done }) => { + assert.strictEqual(value, 'proxied'); + assert.strictEqual(done, false); + })); +} + +{ + // Do not read controller.signal before abort(): the lazy AbortController + // must still report the abort reason on first access. + let ctrl; + const err = new Error('hotpath-abort-before-signal'); + const ws = new WritableStream({ + start(c) { ctrl = c; }, + }); + ws.abort(err); + assert.strictEqual(ctrl.signal.aborted, true); + assert.strictEqual(ctrl.signal.reason, err); +} + +{ + // writer.ready and writer.closed are distinct spec slots, and two + // writers must not share a process-wide resolved promise. + const a = new WritableStream().getWriter(); + const b = new WritableStream().getWriter(); + assert.notStrictEqual(a.ready, b.ready); + assert.notStrictEqual(a.closed, b.closed); + assert.notStrictEqual(a.ready, a.closed); +} + +(async () => { + const ws = new WritableStream(); + await ws.close(); + const writer = ws.getWriter(); + assert.notStrictEqual(writer.ready, writer.closed); + await Promise.all([writer.ready, writer.closed]); +})().then(common.mustCall(), common.mustNotCall()); + +{ + // stream.cancel() must not return a process-shared resolved promise. + const a = new ReadableStream(); + const b = new ReadableStream(); + const pa = a.cancel(); + const pb = b.cancel(); + assert.notStrictEqual(pa, pb); + pa.then(common.mustCall(), common.mustNotCall()); + pb.then(common.mustCall(), common.mustNotCall()); +} + +{ + // Erroring an empty stream via interop must not allocate a controller, + // and a later cancel() must not allocate one either. + const rs = new ReadableStream(); + rs[kControllerErrorFunction](new Error('empty-error')); + assert.strictEqual(rs[kState].controller, undefined); + rs.cancel().then(common.mustNotCall(), common.mustCall()); + assert.strictEqual(rs[kState].controller, undefined); +} diff --git a/test/parallel/test-whatwg-writablestream.js b/test/parallel/test-whatwg-writablestream.js index 88d9c57b9de..66b181c8eb5 100644 --- a/test/parallel/test-whatwg-writablestream.js +++ b/test/parallel/test-whatwg-writablestream.js @@ -248,6 +248,20 @@ class Sink { }); } +{ + // abort() must abort the controller signal even if .signal was never + // observed before the abort (lazy AbortController materialization). + let ctrl; + const err = new Error('abort-before-signal'); + const ws = new WritableStream({ + start(c) { ctrl = c; }, + }); + assert.ok(ctrl); + ws.abort(err); + assert.strictEqual(ctrl.signal.aborted, true); + assert.strictEqual(ctrl.signal.reason, err); +} + { let controller; const writable = new WritableStream({