From 9b3419115f76aa5d5e60f6d7b8623ba969682eb8 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:12:34 +0000 Subject: [PATCH 01/12] stream: canWrite and ondrain now reflect physical capacity Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 20 +++++++---- lib/internal/streams/iter/broadcast.js | 14 ++++---- lib/internal/streams/iter/push.js | 6 ++-- ...test-stream-iter-broadcast-backpressure.js | 33 ++++++++++++++++++- test/parallel/test-stream-iter-push-writer.js | 31 +++++++++++++++++ 5 files changed, 87 insertions(+), 17 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index c1ebacd3dfad..6b430dd1d8e6 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -417,9 +417,13 @@ writer.fail(err); // Always synchronous, no fallback needed * {boolean|null} -Returns `true` if the next write is likely to be accepted (buffered data is -below capacity), `false` if backpressure is active, or `null` if the writer -is closed or the consumer has disconnected. +Returns `true` if the slots buffer has physical capacity (buffered data is +below the configured byte budget), `false` if the budget is exhausted, or +`null` if the writer is closed or the consumer has disconnected. + +This reports physical capacity independently of the backpressure policy. With +`'drop-oldest'` or `'drop-newest'`, writes still complete when this is `false` +by evicting buffered data or discarding the incoming data, respectively. This is a hint, not a guarantee: the state can change between the check and the write. Use [`ondrain()`][] to wait for capacity rather than polling. @@ -1089,9 +1093,13 @@ added: * `drainable` {Object} An object implementing the drainable protocol. * Returns: {Promise|null} -Wait for a drainable writer's backpressure to clear. Returns `null` if -the object does not implement the drainable protocol, or a promise that -fulfills with `true` when the writer can accept more data. +Wait for a drainable writer to regain physical buffer capacity. Returns `null` +if the object does not implement the drainable protocol, or a promise that +fulfills with `true` when buffered data falls below the byte budget. + +For writers using `'drop-oldest'` or `'drop-newest'`, this waits for physical +capacity even though writes do not block. This allows producers to avoid data +loss by waiting before writing. ```mjs import { push, ondrain, text } from 'node:stream/iter'; diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 0dc6364768d5..4131fbf2b8ec 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -310,6 +310,7 @@ class BroadcastImpl { if (this.#ended || this.#cancelled) return false; const batchSize = entry.byteLength; + let droppedOldest = false; // Skip empty chunks -- zero-byte writes would accumulate infinitely // without ever triggering backpressure under a byte-budget model. @@ -321,6 +322,7 @@ class BroadcastImpl { case 'unbounded': return false; case 'drop-oldest': + droppedOldest = true; while (this.#bufferedBytes >= this.#options.budget && this.#buffer.length > 0) { const evicted = this.#buffer.shift(); @@ -343,6 +345,10 @@ class BroadcastImpl { this.#buffer.push(entry); this.#bufferedBytes += batchSize; this.#notifyConsumers(); + if (droppedOldest && + this.#bufferedBytes < this.#options.budget) { + this[kOnBufferDrained]?.(); + } return true; } @@ -399,15 +405,13 @@ class BroadcastImpl { } /** - * Check if the next write is likely to be accepted. + * Check whether the slots buffer has capacity. * Returns null if ended/cancelled, true/false otherwise. * @returns {boolean | null} */ [kCanWrite]() { if (this.#ended || this.#cancelled) return null; - if ((this.#options.backpressure === 'strict' || - this.#options.backpressure === 'unbounded') && - this.#bufferedBytes >= this.#options.budget) { + if (this.#bufferedBytes >= this.#options.budget) { return false; } return true; @@ -666,7 +670,6 @@ class BroadcastWriter { writeSync(chunk) { if (this.#state !== 'open') return false; - if (!this.#broadcast[kCanWrite]()) return false; const converted = toUint8Array(chunk); const batch = createBatchEntry([converted]); @@ -680,7 +683,6 @@ class BroadcastWriter { writevSync(chunks) { validateArray(chunks, 'chunks'); if (this.#state !== 'open') return false; - if (!this.#broadcast[kCanWrite]()) return false; const converted = convertChunks(chunks); const batch = createBatchEntry(converted); if (this.#broadcast[kWrite](batch)) { diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index fa84cafdfde7..c53fc7e902c6 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -152,7 +152,7 @@ class PushQueue { // =========================================================================== /** - * Check if the next write is likely to be accepted. + * Check whether the slots buffer has capacity. * Returns null if writer is closed/errored or consumer has terminated. * @returns {boolean | null} */ @@ -160,9 +160,7 @@ class PushQueue { if (this.#writerState !== 'open' || this.#consumerState !== 'active') { return null; } - if ((this.#backpressure === 'strict' || - this.#backpressure === 'unbounded') && - this.#bufferedBytes >= this.#budget) { + if (this.#bufferedBytes >= this.#budget) { return false; } return true; diff --git a/test/parallel/test-stream-iter-broadcast-backpressure.js b/test/parallel/test-stream-iter-broadcast-backpressure.js index 698f1821aeea..6efa7eb54c73 100644 --- a/test/parallel/test-stream-iter-broadcast-backpressure.js +++ b/test/parallel/test-stream-iter-broadcast-backpressure.js @@ -3,7 +3,7 @@ const common = require('../common'); const assert = require('assert'); -const { broadcast, text } = require('stream/iter'); +const { broadcast, ondrain, text } = require('stream/iter'); // ============================================================================= // Backpressure policies @@ -47,6 +47,36 @@ async function testDropNewest() { assert.strictEqual(data, 'K'.repeat(16384)); } +async function testDropPoliciesReportPhysicalCapacity() { + const chunk = new Uint8Array(16384); + + for (const backpressure of ['drop-oldest', 'drop-newest']) { + const { writer, broadcast: bc } = broadcast({ + budget: chunk.byteLength, + backpressure, + }); + const iterator = bc.push()[Symbol.asyncIterator](); + + assert.strictEqual(writer.writeSync(chunk), true); + assert.strictEqual(writer.canWrite, false); + + let drained = false; + const drain = ondrain(writer); + drain.then(common.mustCall(() => { drained = true; })); + + // Drop policies still accept writes despite having no physical capacity. + assert.strictEqual(writer.writeSync(chunk), true); + assert.strictEqual(writer.canWrite, false); + await new Promise(setImmediate); + assert.strictEqual(drained, false); + + assert.strictEqual((await iterator.next()).done, false); + assert.strictEqual(await drain, true); + assert.strictEqual(writer.canWrite, true); + bc.cancel(); + } +} + // ============================================================================= // Block backpressure // ============================================================================= @@ -269,6 +299,7 @@ async function testEndSyncReturnValue() { Promise.all([ testDropOldest(), testDropNewest(), + testDropPoliciesReportPhysicalCapacity(), testBlockBackpressure(), testBlockBackpressureContent(), testStrictBackpressureOverflow(), diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index dd6cf7d494b2..4f47675da146 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -19,6 +19,36 @@ async function testOndrain() { assert.strictEqual(ondrain(writer), null); } +async function testDropPoliciesReportPhysicalCapacity() { + const chunk = new Uint8Array(16384); + + for (const backpressure of ['drop-oldest', 'drop-newest']) { + const { writer, readable } = push({ + budget: chunk.byteLength, + backpressure, + }); + const iterator = readable[Symbol.asyncIterator](); + + assert.strictEqual(writer.writeSync(chunk), true); + assert.strictEqual(writer.canWrite, false); + + let drained = false; + const drain = ondrain(writer); + drain.then(common.mustCall(() => { drained = true; })); + + // Drop policies still accept writes despite having no physical capacity. + assert.strictEqual(writer.writeSync(chunk), true); + assert.strictEqual(writer.canWrite, false); + await new Promise(setImmediate); + assert.strictEqual(drained, false); + + assert.strictEqual((await iterator.next()).done, false); + assert.strictEqual(await drain, true); + assert.strictEqual(writer.canWrite, true); + await iterator.return(); + } +} + async function testOndrainNonDrainable() { // Non-drainable objects return null assert.strictEqual(ondrain(null), null); @@ -586,6 +616,7 @@ async function testFailRejectsPendingReadWithFalsyReason() { Promise.all([ testOndrain(), + testDropPoliciesReportPhysicalCapacity(), testOndrainNonDrainable(), testWriteWithSignalRejects(), testWriteWithPreAbortedSignal(), From 9f50e02216a0f08f3032abcb61c7dc385365e86a Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:23:12 +0000 Subject: [PATCH 02/12] stream: ensure factory signals remain active through closing Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 4 +++- lib/internal/streams/iter/push.js | 3 ++- test/parallel/test-stream-iter-push-writer.js | 16 ++++++++++++++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index 6b430dd1d8e6..6d6361472194 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -783,7 +783,9 @@ added: **Default:** `16384`. * `backpressure` {string} Backpressure policy: `'strict'`, `'unbounded'`, `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. - * `signal` {AbortSignal} Abort the stream. + * `signal` {AbortSignal} Abort the stream. The signal remains active while + buffered data drains after `writer.end()`; aborting during that time fails + the writer and rejects the pending `end()` promise. * Returns: {Object} * `writer` {Writable} The writer side. * `readable` {AsyncIterable} whose chunks fulfill with {Uint8Array\[]} diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index c53fc7e902c6..8af326418b33 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -334,7 +334,6 @@ class PushQueue { return this.#bytesWritten; // Idempotent } - this.#cleanup(); this.#rejectPendingWrites( new ERR_INVALID_STATE.TypeError('Writer closed')); this.#resolvePendingDrains(false); @@ -342,6 +341,7 @@ class PushQueue { // If buffer is empty, close immediately if (this.#slots.length === 0) { this.#writerState = 'closed'; + this.#cleanup(); this.#resolvePendingReads(); return this.#bytesWritten; } @@ -359,6 +359,7 @@ class PushQueue { endDrained() { if (this.#writerState !== 'closing') return; this.#writerState = 'closed'; + this.#cleanup(); if (this.#pendingEnd) { this.#pendingEnd.resolve(this.#bytesWritten); this.#pendingEnd = null; diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index 4f47675da146..8d9109d9eb3e 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -324,6 +324,21 @@ async function testEndSignalAbortWhileDraining() { assert.strictEqual(await completedEnd, 5); } +async function testFactorySignalAbortWhileDraining() { + const controller = new AbortController(); + const reason = new Error('stream aborted while draining'); + const { writer, readable } = push({ signal: controller.signal }); + + writer.writeSync('hello'); + const end = writer.end(); + const endRejected = assert.rejects(end, (error) => error === reason); + controller.abort(reason); + + await endRejected; + await assert.rejects(text(readable), (error) => error === reason); + await assert.rejects(writer.end(), (error) => error === reason); +} + async function testEndAfterEndSyncWaitsForDrain() { const { writer, readable } = push(); writer.writeSync('hello'); @@ -634,6 +649,7 @@ Promise.all([ testEndAsyncReturnValue(), testEndWithPreAbortedSignal(), testEndSignalAbortWhileDraining(), + testFactorySignalAbortWhileDraining(), testEndAfterEndSyncWaitsForDrain(), testWriteUint8Array(), testOndrainWaitsForDrain(), From 14ba0d5925aadd85cf4af35074604dc0d8d66248 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:27:42 +0000 Subject: [PATCH 03/12] stream: ensure async dispoal after endSync awaits for drain Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 6 +++++ lib/internal/streams/iter/push.js | 25 ++++++------------- test/parallel/test-stream-iter-push-writer.js | 18 +++++++++++++ 3 files changed, 32 insertions(+), 17 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index 6d6361472194..f8e27c3200e9 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -462,6 +462,12 @@ or errored, this is a no-op. Unlike `write()` and `end()`, `fail()` is unconditionally synchronous because failing a writer is a pure state transition with no async work to perform. +#### `writer[Symbol.asyncDispose]()` + +If the writer is open, calls `writer.fail()`. If the writer is closing after +`end()` or `endSync()`, waits for buffered data to drain. If the writer is +already closed or errored, resolves immediately. + #### `writer.write(chunk[, options])` * `chunk` {Uint8Array|string} diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 8af326418b33..767a54681064 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -412,12 +412,12 @@ class PushQueue { return this.#writerState; } - get pendingEndPromise() { - return this.#pendingEnd?.promise ?? null; - } - - setPendingEnd(pending) { - this.#pendingEnd = pending; + getPendingEndPromise() { + if (this.#pendingEnd === null) { + const { promise, resolve, reject } = PromiseWithResolvers(); + this.#pendingEnd = { __proto__: null, promise, resolve, reject }; + } + return this.#pendingEnd.promise; } /** @@ -691,15 +691,7 @@ class PushWriter { return PromiseReject(this.#queue.error); } if (result === -3) { - // Closing: buffer has data, create deferred promise that resolves - // when consumer drains past the end sentinel - const pendingEndPromise = this.#queue.pendingEndPromise; - if (pendingEndPromise !== null) { - return raceEndWithSignal(pendingEndPromise, signal); - } - const { promise, resolve, reject } = PromiseWithResolvers(); - this.#queue.setPendingEnd({ __proto__: null, promise, resolve, reject }); - return raceEndWithSignal(promise, signal); + return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal); } // >= 0: byte count (immediate close or idempotent) return PromiseResolve(result); @@ -719,8 +711,7 @@ class PushWriter { [SymbolAsyncDispose]() { const state = this.#queue.writerState; if (state === 'closing') { - // Wait for graceful drain - return this.#queue.pendingEndPromise ?? PromiseResolve(); + return this.#queue.getPendingEndPromise(); } if (state === 'open') { this.fail(); diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index 8d9109d9eb3e..2e4eeafbda50 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -561,6 +561,23 @@ async function testAsyncDispose() { } } +async function testAsyncDisposeWaitsAfterEndSync() { + const { writer, readable } = push({ budget: 16384 }); + writer.writeSync('hello'); + assert.strictEqual(writer.endSync(), -1); + + let disposed = false; + const disposal = writer[Symbol.asyncDispose]().then(() => { + disposed = true; + }); + await Promise.resolve(); + assert.strictEqual(disposed, false); + + assert.strictEqual(await text(readable), 'hello'); + await disposal; + assert.strictEqual(disposed, true); +} + async function testSyncDispose() { const { writer, readable } = push({ budget: 16384 }); writer.writeSync('hello'); @@ -666,5 +683,6 @@ Promise.all([ testEndIdempotentWhenClosed(), testEndRejectsWhenErrored(), testAsyncDispose(), + testAsyncDisposeWaitsAfterEndSync(), testSyncDispose(), ]).then(common.mustCall()); From 8f679be4f2d29620c14b0e8f0797bfd803e69b17 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:38:22 +0000 Subject: [PATCH 04/12] stream: ensure pre-existing writes drain before EOF and end() waits Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 6 +- lib/internal/streams/iter/push.js | 35 +++++------ test/parallel/test-stream-iter-push-writer.js | 61 +++++++++++++------ 3 files changed, 65 insertions(+), 37 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index f8e27c3200e9..b46057f60371 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -435,7 +435,11 @@ the write. Use [`ondrain()`][] to wait for capacity rather than polling. the pending `end()` call; it does not fail the writer itself. * Returns: {Promise} Fulfills with the total number of bytes written. -Signals that no more data will be written and waits for buffered data to drain. +Signals that no more data will be written. Writes already waiting for buffer +space remain ordered before the end of the stream, while later writes fail. If +data is outstanding, the returned promise fulfills after the consumer pulls +`done: true` beyond the final batch. If no data is buffered or pending, the +writer closes immediately. #### `writer.endSync()` diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 767a54681064..8d680cd2ad01 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -299,7 +299,10 @@ class PushQueue { const onAbort = () => { // Remove from queue so it doesn't occupy a slot const idx = this.#pendingWrites.indexOf(entry); - if (idx !== -1) this.#pendingWrites.removeAt(idx); + if (idx !== -1) { + this.#pendingWrites.removeAt(idx); + this.#resolvePendingReads(); + } reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError')); }; @@ -334,30 +337,31 @@ class PushQueue { return this.#bytesWritten; // Idempotent } - this.#rejectPendingWrites( - new ERR_INVALID_STATE.TypeError('Writer closed')); this.#resolvePendingDrains(false); - // If buffer is empty, close immediately - if (this.#slots.length === 0) { + // If there is no accepted work to drain, close immediately. + if (this.#slots.length === 0 && this.#pendingWrites.length === 0) { this.#writerState = 'closed'; this.#cleanup(); this.#resolvePendingReads(); return this.#bytesWritten; } - // Buffer has data: transition to closing, defer completion until drained + // Accepted work remains: close after it drains and the consumer pulls EOF. this.#writerState = 'closing'; return -3; // Signal to PushWriter: create deferred end promise } /** - * Called by the read path when the consumer has drained all data while - * the writer is in the 'closing' state. Transitions to 'closed' and - * resolves the pending end promise. + * Called when the consumer pulls past all accepted data while the writer is + * closing. Transitions to 'closed' and resolves the pending end promise. */ endDrained() { - if (this.#writerState !== 'closing') return; + if (this.#writerState !== 'closing' || + this.#slots.length > 0 || + this.#pendingWrites.length > 0) { + return; + } this.#writerState = 'closed'; this.#cleanup(); if (this.#pendingEnd) { @@ -384,6 +388,7 @@ class PushQueue { this.#cleanup(); this.#rejectPendingReads(this.#error); this.#rejectPendingDrains(this.#error); + this.#rejectPendingWrites(this.#error); if (wasClosing) { // Short-circuit the graceful drain: reject the pending end promise @@ -391,8 +396,6 @@ class PushQueue { this.#pendingEnd.reject(this.#error); this.#pendingEnd = null; } - } else { - this.#rejectPendingWrites(this.#error); } } @@ -446,10 +449,6 @@ class PushQueue { if (this.#slots.length > 0) { const result = this.#drain(); this.#resolvePendingWrites(); - // After draining, check if writer was closing and buffer is now empty - if (this.#writerState === 'closing' && this.#slots.length === 0) { - this.endDrained(); - } return { __proto__: null, done: false, value: result }; } @@ -555,7 +554,9 @@ class PushQueue { } catch (error) { pending.reject(error); } - } else if (this.#writerState === 'closing' && this.#slots.length === 0) { + } else if (this.#writerState === 'closing' && + this.#slots.length === 0 && + this.#pendingWrites.length === 0) { this.endDrained(); const pending = this.#pendingReads.shift(); pending.resolve({ __proto__: null, done: true, value: undefined }); diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index 2e4eeafbda50..98f084ab01a6 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -84,9 +84,9 @@ async function testWriteWithSignalRejects() { await assert.rejects(writePromise, { name: 'AbortError' }); // Clean up - writer.end(); - // eslint-disable-next-line no-unused-vars - for await (const _ of readable) { break; } + const end = writer.end(); + await text(readable); + await end; } async function testWriteWithPreAbortedSignal() { @@ -100,9 +100,10 @@ async function testWriteWithPreAbortedSignal() { // Writer should still be usable for other writes writer.write('ok'); - writer.end(); + const end = writer.end(); const data = await text(readable); assert.strictEqual(data, 'ok'); + await end; } async function testCancelledWriteRemovedFromQueue() { @@ -127,7 +128,7 @@ async function testCancelledWriteRemovedFromQueue() { // The cancelled write should NOT occupy a pending slot. // A new write should succeed now that the buffer has room. await writer.write(kChunk); - writer.end(); + const end = writer.end(); const result = await iter.next(); assert.ok(!result.done); @@ -136,7 +137,8 @@ async function testCancelledWriteRemovedFromQueue() { totalBytes += chunk.byteLength; } assert.strictEqual(totalBytes, 16384); - await iter.return(); + assert.strictEqual((await iter.next()).done, true); + await end; } async function testOndrainResolvesFalseOnConsumerBreak() { @@ -505,28 +507,48 @@ async function testConsumerThrowRejectsPendingRead() { await readRejects; } -// end() while writes are pending rejects those writes -async function testEndRejectsPendingWrites() { +// end() drains writes that were already pending, then waits for EOF to be read. +async function testEndDrainsPendingWrites() { const kChunk = new Uint8Array(16384); const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); writer.writeSync(kChunk); // fill budget // This write blocks on backpressure const writePromise = writer.write(kChunk); + const endPromise = writer.end(); + await assert.rejects(writer.write(kChunk), { code: 'ERR_INVALID_STATE' }); - await new Promise(setImmediate); + let ended = false; + endPromise.then(common.mustCall(() => { ended = true; })); + const iterator = readable[Symbol.asyncIterator](); - // Ending should reject the pending write - writer.endSync(); + assert.strictEqual((await iterator.next()).done, false); + await writePromise; + assert.strictEqual((await iterator.next()).done, false); + await Promise.resolve(); + assert.strictEqual(ended, false); - await assert.rejects( - () => writePromise, - { code: 'ERR_INVALID_STATE' }, - ); + assert.strictEqual((await iterator.next()).done, true); + assert.strictEqual(await endPromise, kChunk.byteLength * 2); + assert.strictEqual(ended, true); +} - // Clean up: drain the readable - // eslint-disable-next-line no-unused-vars - for await (const _ of readable) { break; } +async function testEndWaitsForEofPull() { + const { writer, readable } = push(); + writer.writeSync('hello'); + const endPromise = writer.end(); + let ended = false; + endPromise.then(common.mustCall(() => { ended = true; })); + const iterator = readable[Symbol.asyncIterator](); + + const data = await iterator.next(); + assert.strictEqual(data.done, false); + await Promise.resolve(); + assert.strictEqual(ended, false); + + assert.strictEqual((await iterator.next()).done, true); + await endPromise; + assert.strictEqual(ended, true); } async function testEndIdempotentWhenClosed() { @@ -679,7 +701,8 @@ Promise.all([ testConsumerReturnResolvesPendingRead(), testEndRejectsAfterConsumerReturn(), testConsumerThrowRejectsPendingRead(), - testEndRejectsPendingWrites(), + testEndDrainsPendingWrites(), + testEndWaitsForEofPull(), testEndIdempotentWhenClosed(), testEndRejectsWhenErrored(), testAsyncDispose(), From dc181502d8b7ad3b042912d88ee34e16aa9ffcfa Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:44:08 +0000 Subject: [PATCH 05/12] stream: pre-aborted pipeTo now applies dest failure handling Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 3 +- lib/internal/streams/iter/pull.js | 18 +++++-- .../test-stream-iter-pipeto-signal.js | 47 +++++++++++++++++++ 3 files changed, 62 insertions(+), 6 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index b46057f60371..32120aaf4a3c 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -619,7 +619,8 @@ added: * `...transforms` {Function|Object} Zero or more transforms to apply. * `writer` {Object} Destination with `write(chunk)` method. * `options` {Object} - * `signal` {AbortSignal} Abort the pipeline. + * `signal` {AbortSignal} Abort the pipeline. Aborting fails the destination + writer unless `preventFail` is `true`. * `preventClose` {boolean} If `true`, do not call `writer.end()` when the source ends. **Default:** `false`. * `preventFail` {boolean} If `true`, do not call `writer.fail()` on diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index c1e42f03dc38..75a2f1a7f308 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -1054,8 +1054,18 @@ async function pipeTo(source, ...args) { const signal = options?.signal; - // Check for abort - signal?.throwIfAborted(); + function failWriter(error) { + if (!options?.preventFail) { + writer.fail?.(wrapError(error)); + } + } + + try { + signal?.throwIfAborted(); + } catch (error) { + failWriter(error); + throw error; + } const hasWriteSync = typeof writer.writeSync === 'function'; const useSyncIterableFastPath = @@ -1192,9 +1202,7 @@ async function pipeTo(source, ...args) { } } } catch (error) { - if (!options?.preventFail) { - writer.fail?.(wrapError(error)); - } + failWriter(error); throw error; } diff --git a/test/parallel/test-stream-iter-pipeto-signal.js b/test/parallel/test-stream-iter-pipeto-signal.js index ec1324a4e04b..be04c8c635df 100644 --- a/test/parallel/test-stream-iter-pipeto-signal.js +++ b/test/parallel/test-stream-iter-pipeto-signal.js @@ -9,6 +9,51 @@ const assert = require('assert'); const { setTimeout } = require('timers/promises'); const { pipeTo, from } = require('stream/iter'); +async function testPipeToPreAbortedSignalFailsWriter() { + const reason = new Error('already aborted'); + let sourceTouched = false; + const source = { + [Symbol.asyncIterator]() { + sourceTouched = true; + return {}; + }, + }; + const writer = { + write: common.mustNotCall(), + fail: common.mustCall((error) => assert.strictEqual(error, reason)), + }; + + await assert.rejects( + pipeTo(source, writer, { signal: AbortSignal.abort(reason) }), + (error) => error === reason, + ); + assert.strictEqual(sourceTouched, false); +} + +async function testPipeToPreAbortedSignalPreventFail() { + const reason = new Error('already aborted'); + let sourceTouched = false; + const source = { + [Symbol.asyncIterator]() { + sourceTouched = true; + return {}; + }, + }; + const writer = { + write: common.mustNotCall(), + fail: common.mustNotCall(), + }; + + await assert.rejects( + pipeTo(source, writer, { + signal: AbortSignal.abort(reason), + preventFail: true, + }), + (error) => error === reason, + ); + assert.strictEqual(sourceTouched, false); +} + // pipeTo with live signal, no transforms — abort mid-stream async function testPipeToLiveSignalNoTransforms() { const ac = new AbortController(); @@ -116,6 +161,8 @@ async function testPipeToLiveSignalWithTransformsCompletes() { } Promise.all([ + testPipeToPreAbortedSignalFailsWriter(), + testPipeToPreAbortedSignalPreventFail(), testPipeToLiveSignalNoTransforms(), testPipeToLiveSignalNoTransformsPendingNext(), testPipeToLiveSignalWithTransforms(), From 2a39d91146ba434d6315fba65278e0f56f370c81 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 01:55:01 +0000 Subject: [PATCH 06/12] stream: make pipeTo source normalization independent of Writer Signed-off-by: James M Snell Assisted-by: Opencode --- lib/internal/streams/iter/pull.js | 53 +----------------------- test/parallel/test-stream-iter-pipeto.js | 39 +++++++++++++---- 2 files changed, 34 insertions(+), 58 deletions(-) diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index 75a2f1a7f308..0be6a3975dbd 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -8,8 +8,6 @@ const { ArrayBufferIsView, - ArrayFromAsync, - ArrayIsArray, ArrayPrototypePush, ArrayPrototypeSlice, FunctionPrototypeCall, @@ -46,9 +44,7 @@ const { fromSync, isSyncIterable, isAsyncIterable, - isPrimitiveChunk, isUint8ArrayBatch, - normalizeAsyncValue, } = require('internal/streams/iter/from'); const { @@ -65,10 +61,7 @@ const { } = require('internal/streams/iter/utils'); const { - kValidatedSource, kValidatedTransform, - toAsyncStreamable, - toStreamable, } = require('internal/streams/iter/types'); // ============================================================================= @@ -131,22 +124,6 @@ function parsePipeToArgs(args, requiredMethod) { }; } -function canUseSyncIterablePipeToFastPath(source, transforms, signal) { - if (signal !== undefined || - transforms.length !== 0 || - isPrimitiveChunk(source) || - ArrayIsArray(source) || - source?.[kValidatedSource] || - !isSyncIterable(source) || - isAsyncIterable(source)) { - return false; - } - - // Preserve from()'s top-level protocol precedence for custom iterables. - return typeof source[toAsyncStreamable] !== 'function' && - typeof source[toStreamable] !== 'function'; -} - // ============================================================================= // Transform Output Flattening // ============================================================================= @@ -1068,9 +1045,7 @@ async function pipeTo(source, ...args) { } const hasWriteSync = typeof writer.writeSync === 'function'; - const useSyncIterableFastPath = - hasWriteSync && canUseSyncIterablePipeToFastPath(source, transforms, signal); - const normalized = useSyncIterableFastPath ? undefined : from(source); + const normalized = from(source); let totalBytes = 0; const hasWritev = typeof writer.writev === 'function'; @@ -1141,31 +1116,7 @@ async function pipeTo(source, ...args) { } try { - if (useSyncIterableFastPath) { - // Avoid from()'s async sync-iterable batching path. This keeps writes - // incremental for synchronous sources while preserving async - // normalization for non-primitive yielded values. - for (const value of source) { - if (isUint8ArrayBatch(value)) { - if (value.length > 0) { - const p = writeBatch(value); - if (p) await p; - } - continue; - } - if (isUint8Array(value)) { - const p = writeBatch([value]); - if (p) await p; - continue; - } - - const batch = await ArrayFromAsync(normalizeAsyncValue(value)); - if (batch.length > 0) { - const p = writeBatch(batch); - if (p) await p; - } - } - } else if (transforms.length === 0) { + if (transforms.length === 0) { // Fast path: no transforms - iterate normalized source directly if (signal) { for await (const batch of yieldAbortable(normalized, signal)) { diff --git a/test/parallel/test-stream-iter-pipeto.js b/test/parallel/test-stream-iter-pipeto.js index f16d7ae89972..4711b291a7af 100644 --- a/test/parallel/test-stream-iter-pipeto.js +++ b/test/parallel/test-stream-iter-pipeto.js @@ -247,7 +247,7 @@ async function testPipeToSyncMinimalWriter() { assert.strictEqual(chunks.length > 0, true); } -async function testPipeToSyncIterableFastPathWritesIncrementally() { +async function testPipeToSyncIterableUsesFromBatching() { let pulled = 0; let firstWritePulled = 0; const chunks = []; @@ -270,7 +270,7 @@ async function testPipeToSyncIterableFastPathWritesIncrementally() { const totalBytes = await pipeTo(source(), writer); assert.strictEqual(totalBytes, 3); - assert.strictEqual(firstWritePulled, 1); + assert.strictEqual(firstWritePulled, 3); assert.deepStrictEqual(chunks, [ new Uint8Array([0x61]), new Uint8Array([0x62]), @@ -278,7 +278,31 @@ async function testPipeToSyncIterableFastPathWritesIncrementally() { ]); } -async function testPipeToSyncIterableFastPathWriteFallback() { +async function testPipeToSourceNormalizationIndependentOfWriter() { + function source() { + return { + *[Symbol.iterator]() { + yield { + async *[Symbol.asyncIterator]() { + yield 'nested'; + }, + }; + }, + }; + } + + for (const hasWriteSync of [false, true]) { + const writer = { write: common.mustNotCall() }; + if (hasWriteSync) writer.writeSync = common.mustNotCall(); + + await assert.rejects( + pipeTo(source(), writer, { preventClose: true, preventFail: true }), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + } +} + +async function testPipeToSyncIterableWriteFallback() { const asyncWrites = []; const writer = { writeSync(chunk) { @@ -299,7 +323,7 @@ async function testPipeToSyncIterableFastPathWriteFallback() { assert.deepStrictEqual(asyncWrites, [new Uint8Array([0x62])]); } -async function testPipeToSyncIterableFastPathAsyncValue() { +async function testPipeToSyncIterableAsyncValue() { const chunks = []; const writer = { write: common.mustNotCall(), @@ -337,7 +361,8 @@ Promise.all([ testPipeToSyncPreventClose(), testPipeToMinimalWriter(), testPipeToSyncMinimalWriter(), - testPipeToSyncIterableFastPathWritesIncrementally(), - testPipeToSyncIterableFastPathWriteFallback(), - testPipeToSyncIterableFastPathAsyncValue(), + testPipeToSyncIterableUsesFromBatching(), + testPipeToSyncIterableWriteFallback(), + testPipeToSyncIterableAsyncValue(), + testPipeToSourceNormalizationIndependentOfWriter(), ]).then(common.mustCall()); From 248829333feb22f88248aed089c3f2fbefea59b3 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 02:04:08 +0000 Subject: [PATCH 07/12] stream: make consumer signals on longer alter source precedence Signed-off-by: James M Snell Assisted-by: Opencode --- lib/internal/streams/iter/consumers.js | 4 +-- .../test-stream-iter-consumers-bytes.js | 26 +++++++++++++++++++ 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js index 0a317939e8f4..95ab63a054b3 100644 --- a/lib/internal/streams/iter/consumers.js +++ b/lib/internal/streams/iter/consumers.js @@ -143,9 +143,7 @@ async function collectAsync(source, signal, limit) { signal?.throwIfAborted(); // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc. - const abortableSource = signal && isAsyncIterable(source) ? - yieldAbortable(source, signal) : source; - const normalized = from(abortableSource); + const normalized = from(source); const entries = []; // Fast path: no signal and no limit diff --git a/test/parallel/test-stream-iter-consumers-bytes.js b/test/parallel/test-stream-iter-consumers-bytes.js index 53d0a3858f87..531917eb9734 100644 --- a/test/parallel/test-stream-iter-consumers-bytes.js +++ b/test/parallel/test-stream-iter-consumers-bytes.js @@ -101,6 +101,31 @@ async function testAsyncConsumersAbortPendingNormalization() { } } +async function testAsyncConsumerSignalPreservesProtocolPrecedence() { + let protocolCalls = 0; + let iteratorCalls = 0; + const source = { + __proto__: null, + [toAsyncStreamable]() { + protocolCalls++; + return from('protocol'); + }, + async *[Symbol.asyncIterator]() { + iteratorCalls++; + yield 'iterator'; + }, + }; + + const result = await text(source, { + __proto__: null, + signal: new AbortController().signal, + }); + + assert.strictEqual(result, 'protocol'); + assert.strictEqual(protocolCalls, 1); + assert.strictEqual(iteratorCalls, 0); +} + async function testBytesEmpty() { const data = await bytes(from([])); assert.ok(data instanceof Uint8Array); @@ -255,6 +280,7 @@ Promise.all([ testBytesAsyncAbort(), testAsyncConsumersAbortPendingNext(), testAsyncConsumersAbortPendingNormalization(), + testAsyncConsumerSignalPreservesProtocolPrecedence(), testBytesEmpty(), testArrayBufferSyncBasic(), testArrayBufferAsync(), From 997df07c21d89fcc12750446ccc1aba87ab0b534 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 02:35:19 +0000 Subject: [PATCH 08/12] stream: apply source normalization once at call time Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 9 ++-- lib/internal/streams/iter/pull.js | 35 +++++++------ test/parallel/test-stream-iter-pull-async.js | 55 +++++++++++++++++--- test/parallel/test-stream-iter-pull-sync.js | 33 +++++++++++- 4 files changed, 105 insertions(+), 27 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index 32120aaf4a3c..45509c7f07e4 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -703,8 +703,10 @@ added: * `signal` {AbortSignal} Abort the pipeline. * Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array\[]} -Create a lazy async pipeline. Data is not read from `source` until the -returned iterable is consumed. Transforms are applied in order. +Create a lazy async pipeline. Source conversion and streamable protocol +dispatch occur when `pull()` is called, but data is not read from `source` +until the returned iterable is consumed. A signal that is already aborted is +thrown synchronously after source conversion. Transforms are applied in order. ```mjs import { from, pull, text } from 'node:stream/iter'; @@ -774,7 +776,8 @@ added: * `...transforms` {Function|Object} Zero or more sync transforms. * Returns: {Iterable} whose chunks return {Uint8Array\[]} -Synchronous version of [`pull()`][]. All transforms must be synchronous. +Synchronous version of [`pull()`][]. Source conversion and streamable protocol +dispatch occur when `pullSync()` is called. All transforms must be synchronous. ## Push streams diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index 0be6a3975dbd..6e00ab43f414 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -805,6 +805,7 @@ async function* createAsyncPipeline(source, transforms, signal) { * @returns {Iterable} */ function pullSync(source, ...transforms) { + const normalized = fromSync(source); for (let i = 0; i < transforms.length; i++) { if (!isTransform(transforms[i])) { throw new ERR_INVALID_ARG_TYPE( @@ -815,7 +816,7 @@ function pullSync(source, ...transforms) { return { __proto__: null, *[SymbolIterator]() { - yield* createSyncPipeline(fromSync(source), transforms); + yield* createSyncPipeline(normalized, transforms); }, }; } @@ -832,17 +833,9 @@ function pull(source, ...args) { const signal = options?.signal; if (signal !== undefined) { validateAbortSignal(signal, 'options.signal'); - // Eagerly check abort at call time per spec - if (signal.aborted) { - return { - __proto__: null, - // eslint-disable-next-line require-yield - async *[SymbolAsyncIterator]() { - throw signal.reason; - }, - }; - } } + const normalized = from(source); + signal?.throwIfAborted(); return { __proto__: null, @@ -852,7 +845,7 @@ function pull(source, ...args) { controller.signal : AbortSignal.any([signal, controller.signal]); async function* pipeline() { - yield* createAsyncPipeline(from(source), transforms, iteratorSignal); + yield* createAsyncPipeline(normalized, transforms, iteratorSignal); } const iterator = pipeline(); @@ -887,9 +880,6 @@ function pullWithConsumerCleanup(source, transforms, signal) { return sourceIterator; }, }; - const pipeline = signal === undefined ? - pull(pipelineSource, ...transforms) : - pull(pipelineSource, ...transforms, { __proto__: null, signal }); let sourceClosed = false; let abortHandler; @@ -906,6 +896,21 @@ function pullWithConsumerCleanup(source, transforms, signal) { } } + if (signal?.aborted) { + closeSource('throw', signal.reason); + return { + __proto__: null, + // eslint-disable-next-line require-yield + async *[SymbolAsyncIterator]() { + throw signal.reason; + }, + }; + } + + const pipeline = signal === undefined ? + pull(pipelineSource, ...transforms) : + pull(pipelineSource, ...transforms, { __proto__: null, signal }); + if (signal !== undefined) { abortHandler = () => closeSource('throw', signal.reason); signal.addEventListener('abort', abortHandler, diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js index 86c790c85cb2..6105d9250cad 100644 --- a/test/parallel/test-stream-iter-pull-async.js +++ b/test/parallel/test-stream-iter-pull-async.js @@ -11,6 +11,7 @@ const { share, tap, text, + toAsyncStreamable, } = require('stream/iter'); async function testPullIdentity() { @@ -53,18 +54,54 @@ async function testPullWithAbortSignal() { yield [new Uint8Array([1])]; } - const result = pull(gen(), { signal: AbortSignal.abort() }); - await assert.rejects( - async () => { - // eslint-disable-next-line no-unused-vars - for await (const _ of result) { - assert.fail('Should not reach here'); - } - }, + assert.throws( + () => pull(gen(), { signal: AbortSignal.abort() }), { name: 'AbortError' }, ); } +async function testPullNormalizesSourceAtCallTime() { + let protocolCalls = 0; + let iteratorCalls = 0; + const source = { + [toAsyncStreamable]() { + protocolCalls++; + return { + async *[Symbol.asyncIterator]() { + iteratorCalls++; + yield 'data'; + }, + }; + }, + }; + + const result = pull(source); + assert.strictEqual(protocolCalls, 1); + assert.strictEqual(iteratorCalls, 0); + assert.strictEqual(await text(result), 'data'); + assert.strictEqual(protocolCalls, 1); + assert.strictEqual(iteratorCalls, 1); +} + +function testPullPreAbortOrdering() { + const reason = new Error('already aborted'); + let protocolCalls = 0; + const source = { + [toAsyncStreamable]() { + protocolCalls++; + return from('data'); + }, + }; + const signal = AbortSignal.abort(reason); + + assert.throws(() => pull(source, { signal }), (error) => error === reason); + assert.strictEqual(protocolCalls, 1); + assert.throws( + () => pull(null, { signal }), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); +} + async function testPullChainedTransforms() { const enc = new TextEncoder(); const transforms = [ @@ -475,6 +512,8 @@ async function testTransformOptionsNotShared() { testPullStatelessTransform(), testPullStatefulTransform(), testPullWithAbortSignal(), + testPullNormalizesSourceAtCallTime(), + testPullPreAbortOrdering(), testPullChainedTransforms(), testPullSourceError(), testTapCallbackError(), diff --git a/test/parallel/test-stream-iter-pull-sync.js b/test/parallel/test-stream-iter-pull-sync.js index c47a6b3f9233..f14619cc1b42 100644 --- a/test/parallel/test-stream-iter-pull-sync.js +++ b/test/parallel/test-stream-iter-pull-sync.js @@ -3,7 +3,13 @@ const common = require('../common'); const assert = require('assert'); -const { pullSync, fromSync, bytesSync, tapSync } = require('stream/iter'); +const { + pullSync, + fromSync, + bytesSync, + tapSync, + toStreamable, +} = require('stream/iter'); function testPullSyncIdentity() { // No transforms - just pass through @@ -11,6 +17,30 @@ function testPullSyncIdentity() { assert.deepStrictEqual(data, new TextEncoder().encode('hello')); } +function testPullSyncNormalizesSourceAtCallTime() { + let protocolCalls = 0; + let iteratorCalls = 0; + const source = { + [toStreamable]() { + protocolCalls++; + return { + *[Symbol.iterator]() { + iteratorCalls++; + yield 'data'; + }, + }; + }, + }; + + const result = pullSync(source); + assert.strictEqual(protocolCalls, 1); + assert.strictEqual(iteratorCalls, 0); + assert.strictEqual(new TextDecoder().decode(bytesSync(result)), 'data'); + assert.strictEqual(protocolCalls, 1); + assert.strictEqual(iteratorCalls, 1); + assert.throws(() => pullSync(null), { code: 'ERR_INVALID_ARG_TYPE' }); +} + function testPullSyncStatelessTransform() { const upper = (chunks) => { if (chunks === null) return null; @@ -177,6 +207,7 @@ function testPullSyncInvalidTransform() { Promise.all([ testPullSyncIdentity(), + testPullSyncNormalizesSourceAtCallTime(), testPullSyncStatelessTransform(), testPullSyncStatefulTransform(), testPullSyncChainedTransforms(), From 4d0208472ff15d239cf6debbdb2761181cd6809e Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 02:42:36 +0000 Subject: [PATCH 09/12] stream: ensure from() observes returned rejecting promise correctly Signed-off-by: James M Snell Assisted-by: Opencode --- lib/internal/streams/iter/from.js | 9 ++++++++- test/parallel/test-stream-iter-from-async.js | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js index 59ac6a15775b..b33509ba90e6 100644 --- a/lib/internal/streams/iter/from.js +++ b/lib/internal/streams/iter/from.js @@ -15,6 +15,7 @@ const { DataViewPrototypeGetByteLength, DataViewPrototypeGetByteOffset, FunctionPrototypeCall, + PromisePrototypeThen, SymbolAsyncIterator, SymbolIterator, TypedArrayPrototypeGetBuffer, @@ -23,6 +24,8 @@ const { Uint8Array, } = primordials; +const { markPromiseAsHandled } = internalBinding('util'); + const { codes: { ERR_INVALID_ARG_TYPE, @@ -594,7 +597,11 @@ function from(input) { // Check toAsyncStreamable protocol (takes precedence over toStreamable and // iteration protocols) if (typeof input[toAsyncStreamable] === 'function') { - const result = input[toAsyncStreamable](); + let result = input[toAsyncStreamable](); + if (isPromise(result)) { + result = PromisePrototypeThen(result, undefined, undefined); + markPromiseAsHandled(result); + } // Synchronous validated source (e.g. Readable batched iterator) if (result?.[kValidatedSource]) { return result; diff --git a/test/parallel/test-stream-iter-from-async.js b/test/parallel/test-stream-iter-from-async.js index 5ef78088bfbd..cdc6cc23ea2f 100644 --- a/test/parallel/test-stream-iter-from-async.js +++ b/test/parallel/test-stream-iter-from-async.js @@ -235,6 +235,20 @@ async function testFromTopLevelProtocolOverIterator() { assert.strictEqual(result, 'from-protocol'); } +async function testFromHandlesProtocolRejectionUntilIteration() { + const reason = new Error('protocol failed'); + const iterable = from({ + [Symbol.for('Stream.toAsyncStreamable')]: common.mustCall( + () => Promise.reject(reason)), + }); + + await new Promise(setImmediate); + await assert.rejects( + iterable[Symbol.asyncIterator]().next(), + (error) => error === reason, + ); +} + // DataView input should be converted to Uint8Array (zero-copy) async function testFromDataView() { const buf = new ArrayBuffer(5); @@ -279,5 +293,6 @@ Promise.all([ testFromTopLevelToStreamable(), testFromTopLevelAsyncPrecedence(), testFromTopLevelProtocolOverIterator(), + testFromHandlesProtocolRejectionUntilIteration(), testFromDataView(), ]).then(common.mustCall()); From 091e3a1b0518b3622d7eac0e168d507828f5ef97 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 02:46:33 +0000 Subject: [PATCH 10/12] stream: fix nested async flushing with infinite sources Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 3 ++- lib/internal/streams/iter/from.js | 6 ++++- test/parallel/test-stream-iter-from-async.js | 25 ++++++++++++++++++++ 3 files changed, 32 insertions(+), 2 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index 45509c7f07e4..e146c95a4ff4 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -547,7 +547,8 @@ added: Create an async byte stream from the given input. Strings are UTF-8 encoded. `ArrayBuffer` and `ArrayBufferView` values are wrapped as `Uint8Array`. Arrays -and iterables in `input` are recursively flattened and normalized. +and iterables in `input` are recursively flattened and normalized. Flattened +values may be split across implementation-defined bounded batches. Objects implementing `Symbol.for('Stream.toAsyncStreamable')` or `Symbol.for('Stream.toStreamable')` are converted via those protocols. The diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js index b33509ba90e6..7fe03511861f 100644 --- a/lib/internal/streams/iter/from.js +++ b/lib/internal/streams/iter/from.js @@ -363,9 +363,13 @@ async function* normalizeAsyncSource(source) { continue; } // Slow path: normalize the value - const batch = []; + let batch = []; for await (const chunk of normalizeAsyncValue(value)) { ArrayPrototypePush(batch, chunk); + if (batch.length === FROM_BATCH_SIZE) { + yield batch; + batch = []; + } } if (batch.length > 0) { yield batch; diff --git a/test/parallel/test-stream-iter-from-async.js b/test/parallel/test-stream-iter-from-async.js index cdc6cc23ea2f..a4726d633ef0 100644 --- a/test/parallel/test-stream-iter-from-async.js +++ b/test/parallel/test-stream-iter-from-async.js @@ -31,6 +31,30 @@ async function testFromAsyncGenerator() { assert.deepStrictEqual(batches[1][0], new Uint8Array([30, 40])); } +async function testFromBoundsNestedAsyncIterable() { + let nestedClosed = false; + async function* nested() { + try { + let value = 0; + while (true) yield new Uint8Array([value++]); + } finally { + nestedClosed = true; + } + } + + async function* source() { + yield nested(); + } + + const iterator = from(source())[Symbol.asyncIterator](); + const first = await iterator.next(); + assert.strictEqual(first.done, false); + assert.strictEqual(first.value.length, 128); + + await iterator.return(); + assert.strictEqual(nestedClosed, true); +} + async function testFromSyncIterableAsAsync() { // Sync iterable passed to from() should work function* gen() { @@ -274,6 +298,7 @@ function testFromUndefinedThrows() { Promise.all([ testFromString(), testFromAsyncGenerator(), + testFromBoundsNestedAsyncIterable(), testFromSyncIterableAsAsync(), testFromSyncIterableAwaitsPromiseValues(), testFromSyncIterableRejectsNestedAsyncIterable(), From 3c3b4bbdd6feba2a143e2bce804041b8db28f949 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 02:52:01 +0000 Subject: [PATCH 11/12] stream: ensure that stateful transforms preserve this Signed-off-by: James M Snell Assisted-by: Opencode --- lib/internal/streams/iter/pull.js | 24 ++++++++++++------- test/parallel/test-stream-iter-pull-async.js | 14 +++++++++++ test/parallel/test-stream-iter-pull-sync.js | 15 ++++++++++++ .../test-stream-iter-transform-roundtrip.js | 13 ++++++++++ 4 files changed, 57 insertions(+), 9 deletions(-) diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index 6e00ab43f414..d1eb2e2c2409 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -505,8 +505,9 @@ function* withFlushSync(source) { yield null; } -function* applyStatefulSyncTransform(source, transform) { - const output = transform(withFlushSync(source)); +function* applyStatefulSyncTransform(source, transform, receiver) { + const output = FunctionPrototypeCall( + transform, receiver, withFlushSync(source)); for (const item of output) { if (item === null) continue; const batch = []; @@ -537,7 +538,8 @@ function* createSyncPipeline(source, transforms) { current = applyFusedStatelessSyncTransforms(current, statelessRun); statelessRun = []; } - current = applyStatefulSyncTransform(current, transform.transform); + current = applyStatefulSyncTransform( + current, transform.transform, transform); } else { ArrayPrototypePush(statelessRun, transform); } @@ -648,8 +650,10 @@ async function* withFlushAsync(source) { yield null; } -async function* applyStatefulAsyncTransform(source, transform, options) { - const output = transform(withFlushAsync(source), options); +async function* applyStatefulAsyncTransform( + source, transform, receiver, options) { + const output = FunctionPrototypeCall( + transform, receiver, withFlushAsync(source), options); for await (const item of output) { if (item === null) continue; // Fast path: item is already a Uint8Array[] batch (e.g. compression transforms) @@ -681,8 +685,10 @@ async function* applyStatefulAsyncTransform(source, transform, options) { * skips isUint8ArrayBatch validation (transform guarantees valid output). * @yields {Uint8Array[]} */ -async function* applyValidatedStatefulAsyncTransform(source, transform, options) { - const output = transform(source, options); +async function* applyValidatedStatefulAsyncTransform( + source, transform, receiver, options) { + const output = FunctionPrototypeCall( + transform, receiver, source, options); for await (const batch of output) { if (batch.length > 0) { yield batch; @@ -750,10 +756,10 @@ async function* createAsyncPipeline(source, transforms, signal) { const opts = { __proto__: null, signal: transformSignal }; if (transform[kValidatedTransform]) { current = applyValidatedStatefulAsyncTransform( - current, transform.transform, opts); + current, transform.transform, transform, opts); } else { current = applyStatefulAsyncTransform( - current, transform.transform, opts); + current, transform.transform, transform, opts); } } else { ArrayPrototypePush(statelessRun, transform); diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js index 6105d9250cad..c341cf296ddd 100644 --- a/test/parallel/test-stream-iter-pull-async.js +++ b/test/parallel/test-stream-iter-pull-async.js @@ -49,6 +49,19 @@ async function testPullStatefulTransform() { assert.strictEqual(data, 'data-ASYNC-END'); } +async function testPullStatefulTransformReceiver() { + const descriptor = {}; + descriptor.transform = common.mustCall( + async function*(source) { + assert.strictEqual(this, descriptor); + for await (const chunks of source) { + yield chunks; + } + }); + + assert.strictEqual(await text(pull(from('receiver'), descriptor)), 'receiver'); +} + async function testPullWithAbortSignal() { async function* gen() { yield [new Uint8Array([1])]; @@ -511,6 +524,7 @@ async function testTransformOptionsNotShared() { testPullIdentity(), testPullStatelessTransform(), testPullStatefulTransform(), + testPullStatefulTransformReceiver(), testPullWithAbortSignal(), testPullNormalizesSourceAtCallTime(), testPullPreAbortOrdering(), diff --git a/test/parallel/test-stream-iter-pull-sync.js b/test/parallel/test-stream-iter-pull-sync.js index f14619cc1b42..af481dbfa137 100644 --- a/test/parallel/test-stream-iter-pull-sync.js +++ b/test/parallel/test-stream-iter-pull-sync.js @@ -74,6 +74,20 @@ function testPullSyncStatefulTransform() { assert.strictEqual(data, 'data-END'); } +function testPullSyncStatefulTransformReceiver() { + const descriptor = {}; + descriptor.transform = common.mustCall( + function*(source) { + assert.strictEqual(this, descriptor); + yield* source; + }); + + assert.strictEqual( + new TextDecoder().decode(bytesSync(pullSync(fromSync('receiver'), descriptor))), + 'receiver', + ); +} + function testPullSyncChainedTransforms() { const addExcl = (chunks) => { if (chunks === null) return null; @@ -210,6 +224,7 @@ Promise.all([ testPullSyncNormalizesSourceAtCallTime(), testPullSyncStatelessTransform(), testPullSyncStatefulTransform(), + testPullSyncStatefulTransformReceiver(), testPullSyncChainedTransforms(), testPullSyncSourceError(), testPullSyncEmptySource(), diff --git a/test/parallel/test-stream-iter-transform-roundtrip.js b/test/parallel/test-stream-iter-transform-roundtrip.js index df63483d6c85..d9a745593c27 100644 --- a/test/parallel/test-stream-iter-transform-roundtrip.js +++ b/test/parallel/test-stream-iter-transform-roundtrip.js @@ -42,6 +42,18 @@ async function testGzipRoundTrip() { assert.strictEqual(result, input); } +async function testValidatedTransformReceiver() { + const descriptor = compressGzip(); + const transform = descriptor.transform; + descriptor.transform = common.mustCall(function(source, options) { + assert.strictEqual(this, descriptor); + return Reflect.apply(transform, this, [source, options]); + }); + + const result = await bytes(pull(from('receiver'), descriptor)); + assert.ok(result.byteLength > 0); +} + async function testGzipLargeData() { // 100KB of repeated text - exercises multi-chunk path const input = 'gzip large data test. '.repeat(5000); @@ -250,6 +262,7 @@ async function testGzipWithLevel() { (async () => { // Gzip await testGzipRoundTrip(); + await testValidatedTransformReceiver(); await testGzipLargeData(); await testGzipActuallyCompresses(); From 9fcfe317b8a2f2a716a8cb1e6bc17b6e5f4fabbf Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 03:49:45 +0000 Subject: [PATCH 12/12] stream: use webidl validation semantics for args Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 19 +- lib/internal/fs/promises.js | 50 ++--- lib/internal/quic/quic.js | 62 +++--- lib/internal/streams/iter/broadcast.js | 64 +++---- lib/internal/streams/iter/classic.js | 21 +-- lib/internal/streams/iter/consumers.js | 103 +++++----- lib/internal/streams/iter/duplex.js | 19 +- lib/internal/streams/iter/pull.js | 51 ++--- lib/internal/streams/iter/push.js | 31 +-- lib/internal/streams/iter/share.js | 31 +-- lib/internal/streams/iter/utils.js | 42 ++--- lib/internal/streams/iter/webidl.js | 177 ++++++++++++++++++ .../test-fs-promises-file-handle-writer.js | 22 +++ test/parallel/test-quic-stream-writer-api.mjs | 24 ++- .../test-stream-iter-consumers-text.js | 12 +- test/parallel/test-stream-iter-push-writer.js | 2 +- test/parallel/test-stream-iter-validation.js | 21 +-- test/parallel/test-stream-iter-webidl.js | 159 ++++++++++++++++ .../test-stream-iter-writable-interop.js | 19 +- 19 files changed, 651 insertions(+), 278 deletions(-) create mode 100644 lib/internal/streams/iter/webidl.js create mode 100644 test/parallel/test-stream-iter-webidl.js diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index e146c95a4ff4..b5ed8e06f6e1 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -401,6 +401,12 @@ const { writer, readable } = push({ A writer is any object conforming to the Writer interface. Only `write()` is required; all other methods are optional. +Writer arguments use Web IDL conversion semantics. A non-`Uint8Array` chunk is +converted to a `USVString` and then UTF-8 encoded. `writev()` and +`writevSync()` accept any iterable object whose values can be converted to +chunks. Writer option dictionaries treat `null` as an empty dictionary and +ignore unknown members. + Each async method has a synchronous `*Sync` counterpart designed for a try-fallback pattern: attempt the fast synchronous path first, and fall back to the async version only when the synchronous call indicates it could not @@ -492,7 +498,7 @@ Synchronous write. Does not block; returns `false` if backpressure is active. #### `writer.writev(chunks[, options])` -* `chunks` {Uint8Array\[]|string\[]} +* `chunks` {Iterable} of {Uint8Array|string} values * `options` {Object} * `signal` {AbortSignal} Cancel just this write operation. The signal cancels only the pending `writev()` call; it does not fail the writer itself. @@ -502,7 +508,7 @@ Write multiple chunks as a single batch. #### `writer.writevSync(chunks)` -* `chunks` {Uint8Array\[]|string\[]} +* `chunks` {Iterable} of {Uint8Array|string} values * Returns: {boolean} `true` if the write was accepted, `false` if the buffer is full. @@ -521,6 +527,12 @@ import { from, pull, bytes, Stream } from 'node:stream/iter'; Stream.from('hello'); ``` +Options dictionaries defined by the Iterable Streams API use Web IDL +conversion semantics. `null` is treated as an empty dictionary, unknown +members are ignored, and known members are converted to their declared types +before the operation runs. Conversion failures use Node.js error codes such as +`ERR_INVALID_ARG_TYPE`, `ERR_INVALID_ARG_VALUE`, and `ERR_OUT_OF_RANGE`. + ```cjs // Named exports const { from, pull, bytes, Stream } = require('node:stream/iter'); @@ -860,7 +872,7 @@ added: * `options` {Object} * `budget` {number} Buffer size in bytes for both directions. - **Default:** `16384`. + Must be >= 16384. **Default:** `16384`. * `backpressure` {string} Policy for both directions. **Default:** `'strict'`. * `signal` {AbortSignal} Cancellation signal for both channels. @@ -1379,6 +1391,7 @@ added: **Default:** `65536`. * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. + * `signal` {AbortSignal} * Returns: {Share} Create a pull-model multi-consumer shared stream. Unlike `broadcast()`, the diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index a9d03ebb1d18..c719f2d4b5e9 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -164,8 +164,9 @@ const lazyReadableStream = getLazy(() => let newStreamsPull; let newStreamsPullSync; let newStreamsParsePullArgs; -let newStreamsToUint8Array; +let newStreamsToWriterUint8Array; let newStreamsConvertChunks; +let newStreamsGetWriterSignal; function lazyNewStreams() { if (newStreamsPull === undefined) { const pullModule = require('internal/streams/iter/pull'); @@ -173,8 +174,9 @@ function lazyNewStreams() { newStreamsPullSync = pullModule.pullSync; const utils = require('internal/streams/iter/utils'); newStreamsParsePullArgs = utils.parsePullArgs; - newStreamsToUint8Array = utils.toUint8Array; + newStreamsToWriterUint8Array = utils.toWriterUint8Array; newStreamsConvertChunks = utils.convertChunks; + newStreamsGetWriterSignal = utils.getWriterSignal; } } @@ -885,6 +887,8 @@ if (getOptionValue('--experimental-stream-iter')) { return { __proto__: null, write(chunk, options = kNullPrototo) { + chunk = newStreamsToWriterUint8Array(chunk); + const signal = newStreamsGetWriterSignal(options); if (error) { return PromiseReject(error); } @@ -892,17 +896,9 @@ if (getOptionValue('--experimental-stream-iter')) { return PromiseReject( new ERR_INVALID_STATE.TypeError('The writer is closed')); } - validateObject(options, 'options'); - const { - signal, - } = options; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - if (signal.aborted) { - return PromiseReject(signal.reason); - } + if (signal?.aborted) { + return PromiseReject(signal.reason); } - chunk = newStreamsToUint8Array(chunk); if (bytesRemaining >= 0 && chunk.byteLength > bytesRemaining) { return PromiseReject( new ERR_OUT_OF_RANGE('write', `<= ${bytesRemaining} bytes`, @@ -915,6 +911,8 @@ if (getOptionValue('--experimental-stream-iter')) { }, writev(chunks, options = kNullPrototo) { + chunks = newStreamsConvertChunks(chunks); + const signal = newStreamsGetWriterSignal(options); if (error) { return PromiseReject(error); } @@ -922,17 +920,9 @@ if (getOptionValue('--experimental-stream-iter')) { return PromiseReject( new ERR_INVALID_STATE.TypeError('The writer is closed')); } - validateObject(options, 'options'); - const { - signal, - } = options; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - if (signal?.aborted) { - return PromiseReject(signal.reason); - } + if (signal?.aborted) { + return PromiseReject(signal.reason); } - chunks = newStreamsConvertChunks(chunks); let totalSize = 0; for (let i = 0; i < chunks.length; i++) { totalSize += chunks[i].byteLength; @@ -949,8 +939,8 @@ if (getOptionValue('--experimental-stream-iter')) { }, writeSync(chunk) { + chunk = newStreamsToWriterUint8Array(chunk); if (error || closed || asyncPending) return false; - chunk = newStreamsToUint8Array(chunk); const length = chunk.byteLength; if (length > syncWriteThreshold) return false; if (length === 0) return true; @@ -980,8 +970,8 @@ if (getOptionValue('--experimental-stream-iter')) { }, writevSync(chunks) { - if (error || closed || asyncPending) return false; chunks = newStreamsConvertChunks(chunks); + if (error || closed || asyncPending) return false; let totalSize = 0; for (let i = 0; i < chunks.length; i++) { totalSize += chunks[i].byteLength; @@ -1016,6 +1006,7 @@ if (getOptionValue('--experimental-stream-iter')) { }, end(options = kNullPrototo) { + const signal = newStreamsGetWriterSignal(options); if (error) { return PromiseReject(error); } @@ -1025,15 +1016,8 @@ if (getOptionValue('--experimental-stream-iter')) { if (closing) { return pendingEndPromise; } - validateObject(options, 'options'); - const { - signal, - } = options; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - if (signal.aborted) { - return PromiseReject(signal.reason); - } + if (signal?.aborted) { + return PromiseReject(signal.reason); } closing = true; pendingEndPromise = PromisePrototypeThen( diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index d55603daf1f4..8ecd5fc776a3 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -138,8 +138,9 @@ const { } = require('internal/streams/iter/types'); const { - toUint8Array, convertChunks, + getWriterSignal, + toWriterUint8Array, } = require('internal/streams/iter/utils'); const { @@ -164,7 +165,6 @@ const { } = require('internal/fs/promises'); const { - validateAbortSignal, validateBoolean, validateFunction, validateInteger, @@ -2197,7 +2197,7 @@ class QuicStream { // signals backpressure additional writes are rejected until the buffer has // capacity again. - function writeSync(chunk) { + function writeConvertedSync(chunk) { // If the stream is closed, errored, or write-ended, we cannot accept // more data. Refuse the sync write. // If a drain is already pending, another operation is waiting @@ -2205,7 +2205,6 @@ class QuicStream { if (closed || errored || stream.#inner.state.writeEnded || drainWakeup != null) { return false; } - chunk = toUint8Array(chunk); const len = TypedArrayPrototypeGetByteLength(chunk); if (len === 0) return true; // Refuse the write only when there is no available capacity at @@ -2223,13 +2222,18 @@ class QuicStream { return true; } - async function write(chunk, options = kEmptyObject) { - validateObject(options, 'options'); - const { signal } = options; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - signal.throwIfAborted(); - } + function writeSync(chunk) { + return writeConvertedSync(toWriterUint8Array(chunk)); + } + + function write(chunk, options = kEmptyObject) { + chunk = toWriterUint8Array(chunk); + const signal = getWriterSignal(options); + return writeAsync(chunk, signal); + } + + async function writeAsync(chunk, signal) { + signal?.throwIfAborted(); if (errored) throw error; if (closed || stream.#inner.state.writeEnded) { throw new ERR_INVALID_STATE('Writer is closed'); @@ -2242,16 +2246,15 @@ class QuicStream { throw new ERR_INVALID_STATE('Stream write buffer is full'); } - if (!writeSync(chunk)) { + if (!writeConvertedSync(chunk)) { throw new ERR_INVALID_STATE('Stream write buffer is full'); } } - function writevSync(chunks) { + function writevConvertedSync(chunks) { if (closed || errored || stream.#inner.state.writeEnded || drainWakeup != null) { return false; } - chunks = convertChunks(chunks); let len = 0; for (const c of chunks) len += TypedArrayPrototypeGetByteLength(c); if (len === 0) return true; @@ -2262,13 +2265,18 @@ class QuicStream { return true; } - async function writev(chunks, options = kEmptyObject) { - validateObject(options, 'options'); - const { signal } = options; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - signal.throwIfAborted(); - } + function writevSync(chunks) { + return writevConvertedSync(convertChunks(chunks)); + } + + function writev(chunks, options = kEmptyObject) { + chunks = convertChunks(chunks); + const signal = getWriterSignal(options); + return writevAsync(chunks, signal); + } + + async function writevAsync(chunks, signal) { + signal?.throwIfAborted(); if (errored) throw error; if (closed || stream.#inner.state.writeEnded) { @@ -2283,7 +2291,7 @@ class QuicStream { throw new ERR_INVALID_STATE('Stream write buffer is full'); } - if (!writevSync(chunks)) { + if (!writevConvertedSync(chunks)) { throw new ERR_INVALID_STATE('Stream write buffer is full'); } } @@ -2308,11 +2316,13 @@ class QuicStream { return totalBytesWritten; } - async function end(options = kEmptyObject) { - validateObject(options, 'options'); - const { signal } = options; + function end(options = kEmptyObject) { + const signal = getWriterSignal(options); + return endAsync(signal); + } + + async function endAsync(signal) { if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); signal.throwIfAborted(); // TODO(@jasnell): The stream/iter spec allows individual sync end // calls to be canceled via an AbortSignal. We currently do not support diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 4131fbf2b8ec..16cb4c06533c 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -33,10 +33,7 @@ const { }, } = require('internal/errors'); const { - validateAbortSignal, - validateArray, validateInteger, - validateObject, } = require('internal/validators'); const { @@ -65,10 +62,12 @@ const { onSignalAbort, parsePullArgs, wrapError, - toUint8Array, - validateBackpressure, + toWriterUint8Array, validateBatchEntry, } = require('internal/streams/iter/utils'); +const { + converters, +} = require('internal/streams/iter/webidl'); const { RingBuffer, @@ -135,9 +134,13 @@ class BroadcastImpl { } push(...args) { - const { transforms, options } = parsePullArgs(args); - const signal = options?.signal; - validateAbortSignal(signal, 'options.signal'); + const parsed = parsePullArgs(args); + const { transforms } = parsed; + const options = converters.PullOptions(parsed.options, { + __proto__: null, + context: 'options', + }); + const { signal } = options; // Avoid registering a consumer that the pre-aborted pipeline will never // read or detach. @@ -595,46 +598,30 @@ class BroadcastWriter { } write(chunk, options) { + const converted = toWriterUint8Array(chunk); const signal = getWriterSignal(options); // Fast path: no signal, writer open, buffer has space if (this.#canUseWriteFastPath(signal)) { - const converted = toUint8Array(chunk); const batch = createBatchEntry([converted]); this.#broadcast[kWrite](batch); this.#totalBytes += batch.byteLength; return kResolvedPromise; } - return this.#writevSlow([chunk], signal); + return this.#writeBatchSlow(createBatchEntry([converted]), signal); } writev(chunks, options) { - validateArray(chunks, 'chunks'); + const converted = convertChunks(chunks); const signal = getWriterSignal(options); + const batch = createBatchEntry(converted); // Fast path: no signal, writer open, buffer has space if (this.#canUseWriteFastPath(signal)) { - const converted = convertChunks(chunks); - const batch = createBatchEntry(converted); if (this.#state === 'open' && this.#broadcast[kWrite](batch)) { this.#totalBytes += batch.byteLength; return kResolvedPromise; } return this.#writeBatchSlow(batch, signal); } - return this.#writevSlow(chunks, signal); - } - - async #writevSlow(chunks, signal) { - if (this.#state === 'errored') { - throw this.#error; - } - if (this.#state !== 'open') { - throw new ERR_INVALID_STATE.TypeError('Writer is closed'); - } - - signal?.throwIfAborted(); - - const batch = createBatchEntry(convertChunks(chunks)); - return this.#writeBatchSlow(batch, signal); } @@ -669,9 +656,8 @@ class BroadcastWriter { } writeSync(chunk) { + const converted = toWriterUint8Array(chunk); if (this.#state !== 'open') return false; - const converted = - toUint8Array(chunk); const batch = createBatchEntry([converted]); if (this.#broadcast[kWrite](batch)) { this.#totalBytes += batch.byteLength; @@ -681,9 +667,8 @@ class BroadcastWriter { } writevSync(chunks) { - validateArray(chunks, 'chunks'); - if (this.#state !== 'open') return false; const converted = convertChunks(chunks); + if (this.#state !== 'open') return false; const batch = createBatchEntry(converted); if (this.#broadcast[kWrite](batch)) { this.#totalBytes += batch.byteLength; @@ -863,17 +848,16 @@ function onBroadcastCancel(broadcastImpl, signal) { * @returns {{ writer: Writer, broadcast: Broadcast }} */ function broadcast(options = { __proto__: null }) { - validateObject(options, 'options'); + options = converters.BroadcastOptions(options, { + __proto__: null, + context: 'options', + }); const { budget = kMultiConsumerDefaultBudget, backpressure = 'strict', signal, } = options; validateInteger(budget, 'options.budget', 16384); - validateBackpressure(backpressure); - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - } const opts = { __proto__: null, @@ -916,8 +900,12 @@ const Broadcast = { 'input', ['Broadcastable', 'AsyncIterable', 'Iterable'], input); } + options = converters.BroadcastOptions(options, { + __proto__: null, + context: 'options', + }); const result = broadcast(options); - const signal = options?.signal; + const { signal } = options; const pump = async () => { const w = result.writer; diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js index 0769348a3981..28f6079e6fe2 100644 --- a/lib/internal/streams/iter/classic.js +++ b/lib/internal/streams/iter/classic.js @@ -40,7 +40,6 @@ const { } = require('internal/errors'); const { - validateArray, validateInteger, validateObject, } = require('internal/validators'); @@ -60,9 +59,10 @@ const { } = require('internal/streams/iter/types'); const { + convertChunks, getWriterSignal, validateBackpressure, - toUint8Array, + toWriterUint8Array, } = require('internal/streams/iter/utils'); const { Buffer } = require('buffer'); @@ -538,7 +538,7 @@ function fromWritable(writable, options = kNullPrototype) { function writeChunks(chunks) { let ok = true; for (let i = 0; i < chunks.length; i++) { - const bytes = toUint8Array(chunks[i]); + const bytes = chunks[i]; totalBytes += TypedArrayPrototypeGetByteLength(bytes); ok = writable.write(bytes); } @@ -554,10 +554,12 @@ function fromWritable(writable, options = kNullPrototype) { }, writeSync(chunk) { + toWriterUint8Array(chunk); return false; }, writevSync(chunks) { + convertChunks(chunks); return false; }, @@ -576,18 +578,12 @@ function fromWritable(writable, options = kNullPrototype) { // otherwise ignored. Classic stream.Writable has no per-write abort signal // support; cancellation should be handled at the pipeline level instead. write(chunk, options) { + const bytes = toWriterUint8Array(chunk); getWriterSignal(options); if (!isWritable()) { return PromiseReject(new ERR_STREAM_WRITE_AFTER_END()); } - let bytes; - try { - bytes = toUint8Array(chunk); - } catch (err) { - return PromiseReject(err); - } - if (backpressure === 'strict' && isFull()) { return PromiseReject(new ERR_INVALID_STATE.RangeError( 'Backpressure violation: buffer is full. ' + @@ -619,7 +615,7 @@ function fromWritable(writable, options = kNullPrototype) { }, writev(chunks, options) { - validateArray(chunks, 'chunks'); + chunks = convertChunks(chunks); getWriterSignal(options); if (!isWritable()) { return PromiseReject(new ERR_STREAM_WRITE_AFTER_END()); @@ -634,8 +630,7 @@ function fromWritable(writable, options = kNullPrototype) { if (backpressure === 'drop-newest' && isFull()) { // Discard entire batch. for (let i = 0; i < chunks.length; i++) { - totalBytes += - TypedArrayPrototypeGetByteLength(toUint8Array(chunks[i])); + totalBytes += TypedArrayPrototypeGetByteLength(chunks[i]); } return PromiseResolve(); } diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js index 95ab63a054b3..75ad4026ad7c 100644 --- a/lib/internal/streams/iter/consumers.js +++ b/lib/internal/streams/iter/consumers.js @@ -36,11 +36,7 @@ const { } = require('internal/errors'); const { TextDecoder } = require('internal/encoding'); const { - validateAbortSignal, validateFunction, - validateInteger, - validateObject, - validateString, } = require('internal/validators'); const { @@ -66,6 +62,9 @@ const { toAsyncStreamable, toStreamable, } = require('internal/streams/iter/types'); +const { + converters, +} = require('internal/streams/iter/webidl'); const { isAnyArrayBuffer, @@ -197,37 +196,6 @@ function toArrayBuffer(data) { byteOffset + byteLength); } -// ============================================================================= -// Shared option validation -// ============================================================================= - -function validateBaseConsumerOptions(options) { - validateObject(options, 'options'); - if (options.limit !== undefined) { - validateInteger(options.limit, 'options.limit', 0); - } - if (options.encoding !== undefined) { - validateString(options.encoding, 'options.encoding'); - try { - new TextDecoder(options.encoding); - } catch { - throw new ERR_INVALID_ARG_VALUE.RangeError( - 'options.encoding', options.encoding); - } - } -} - -function validateConsumerOptions(options) { - validateBaseConsumerOptions(options); - if (options.signal !== undefined) { - validateAbortSignal(options.signal, 'options.signal'); - } -} - -function validateSyncConsumerOptions(options) { - validateBaseConsumerOptions(options); -} - // ============================================================================= // Sync Consumers // ============================================================================= @@ -241,7 +209,10 @@ const kNullPrototype = { __proto__: null }; * @returns {Uint8Array} */ function bytesSync(source, options = kNullPrototype) { - validateSyncConsumerOptions(options); + options = converters.ConsumeSyncOptions(options, { + __proto__: null, + context: 'options', + }); return concatBytes(collectSync(source, options.limit)); } @@ -252,9 +223,18 @@ function bytesSync(source, options = kNullPrototype) { * @returns {string} */ function textSync(source, options = kNullPrototype) { - validateSyncConsumerOptions(options); + options = converters.TextConsumeSyncOptions(options, { + __proto__: null, + context: 'options', + }); + try { + new TextDecoder(options.encoding); + } catch { + throw new ERR_INVALID_ARG_VALUE.RangeError( + 'options.encoding', options.encoding); + } const data = concatBytes(collectSync(source, options.limit)); - const decoder = new TextDecoder(options.encoding ?? 'utf-8', { + const decoder = new TextDecoder(options.encoding, { __proto__: null, fatal: true, }); @@ -268,7 +248,10 @@ function textSync(source, options = kNullPrototype) { * @returns {ArrayBuffer} */ function arrayBufferSync(source, options = kNullPrototype) { - validateSyncConsumerOptions(options); + options = converters.ConsumeSyncOptions(options, { + __proto__: null, + context: 'options', + }); return toArrayBuffer(concatBytes(collectSync(source, options.limit))); } @@ -279,7 +262,10 @@ function arrayBufferSync(source, options = kNullPrototype) { * @returns {Uint8Array[]} */ function arraySync(source, options = kNullPrototype) { - validateSyncConsumerOptions(options); + options = converters.ConsumeSyncOptions(options, { + __proto__: null, + context: 'options', + }); return collectSync(source, options.limit); } @@ -294,7 +280,10 @@ function arraySync(source, options = kNullPrototype) { * @returns {Promise} */ async function bytes(source, options = kNullPrototype) { - validateConsumerOptions(options); + options = converters.ConsumeOptions(options, { + __proto__: null, + context: 'options', + }); const chunks = await collectAsync(source, options.signal, options.limit); return concatBytes(chunks); } @@ -306,10 +295,19 @@ async function bytes(source, options = kNullPrototype) { * @returns {Promise} */ async function text(source, options = kNullPrototype) { - validateConsumerOptions(options); + options = converters.TextConsumeOptions(options, { + __proto__: null, + context: 'options', + }); + try { + new TextDecoder(options.encoding); + } catch { + throw new ERR_INVALID_ARG_VALUE.RangeError( + 'options.encoding', options.encoding); + } const chunks = await collectAsync(source, options.signal, options.limit); const data = concatBytes(chunks); - const decoder = new TextDecoder(options.encoding ?? 'utf-8', { + const decoder = new TextDecoder(options.encoding, { __proto__: null, fatal: true, }); @@ -323,7 +321,10 @@ async function text(source, options = kNullPrototype) { * @returns {Promise} */ async function arrayBuffer(source, options = kNullPrototype) { - validateConsumerOptions(options); + options = converters.ConsumeOptions(options, { + __proto__: null, + context: 'options', + }); const chunks = await collectAsync(source, options.signal, options.limit); return toArrayBuffer(concatBytes(chunks)); } @@ -335,7 +336,10 @@ async function arrayBuffer(source, options = kNullPrototype) { * @returns {Promise} */ async function array(source, options = kNullPrototype) { - validateConsumerOptions(options); + options = converters.ConsumeOptions(options, { + __proto__: null, + context: 'options', + }); return collectAsync(source, options.signal, options.limit); } @@ -419,9 +423,10 @@ function merge(...args) { sources = args; } - if (options?.signal !== undefined) { - validateAbortSignal(options.signal, 'options.signal'); - } + options = converters.MergeOptions(options, { + __proto__: null, + context: 'options', + }); // Normalize each source via from() const normalized = ArrayPrototypeMap(sources, (source) => from(source)); @@ -429,7 +434,7 @@ function merge(...args) { return { __proto__: null, async *[SymbolAsyncIterator]() { - const signal = options?.signal; + const { signal } = options; signal?.throwIfAborted(); diff --git a/lib/internal/streams/iter/duplex.js b/lib/internal/streams/iter/duplex.js index a60c510c4199..674ef81a53c9 100644 --- a/lib/internal/streams/iter/duplex.js +++ b/lib/internal/streams/iter/duplex.js @@ -16,9 +16,8 @@ const { push, } = require('internal/streams/iter/push'); const { - validateAbortSignal, - validateObject, -} = require('internal/validators'); + converters, +} = require('internal/streams/iter/webidl'); /** * Create a pair of connected duplex channels for bidirectional communication. @@ -27,17 +26,11 @@ const { * @returns {[DuplexChannel, DuplexChannel]} */ function duplex(options = { __proto__: null }) { - validateObject(options, 'options'); + options = converters.DuplexOptions(options, { + __proto__: null, + context: 'options', + }); const { budget, backpressure, signal, a, b } = options; - if (a !== undefined) { - validateObject(a, 'options.a'); - } - if (b !== undefined) { - validateObject(b, 'options.b'); - } - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - } // Channel A writes to B's readable (A->B direction). // Signal is NOT passed to push() -- we handle abort via close() below. diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index d1eb2e2c2409..6b44f5c5e431 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -27,7 +27,6 @@ const { }, } = require('internal/errors'); const { lazyDOMException } = require('internal/util'); -const { validateAbortSignal } = require('internal/validators'); const { isAnyArrayBuffer, isPromise, @@ -49,7 +48,6 @@ const { const { createBatchEntry, - isPullOptions, isTransform, isTransformObject, parsePullArgs, @@ -59,6 +57,9 @@ const { wrapError, yieldAbortable, } = require('internal/streams/iter/utils'); +const { + converters, +} = require('internal/streams/iter/webidl'); const { kValidatedTransform, @@ -80,7 +81,7 @@ function hasMethod(value, name) { * Parse pipeTo/pipeToSync arguments: [...transforms, writer, options?] * @param {Array} args * @param {string} requiredMethod - 'write' for pipeTo, 'writeSync' for pipeToSync - * @returns {{ transforms: Array, writer: object, options: object }} + * @returns {{ transforms: Array, writer: object, options: unknown }} */ function parsePipeToArgs(args, requiredMethod) { if (args.length === 0) { @@ -92,7 +93,7 @@ function parsePipeToArgs(args, requiredMethod) { // Check if last arg is options const last = args[args.length - 1]; - if (isPullOptions(last) && !hasMethod(last, requiredMethod)) { + if (!isTransform(last) && !hasMethod(last, requiredMethod)) { options = last; writerIndex = args.length - 2; } @@ -835,11 +836,13 @@ function pullSync(source, ...transforms) { * @returns {AsyncIterable} */ function pull(source, ...args) { - const { transforms, options } = parsePullArgs(args); - const signal = options?.signal; - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - } + const parsed = parsePullArgs(args); + const { transforms } = parsed; + const options = converters.PullOptions(parsed.options, { + __proto__: null, + context: 'options', + }); + const { signal } = options; const normalized = from(source); signal?.throwIfAborted(); @@ -969,11 +972,16 @@ function pullWithConsumerCleanup(source, transforms, signal) { * @returns {number} Total bytes written */ function pipeToSync(source, ...args) { - const { transforms, writer, options } = parsePipeToArgs(args, 'writeSync'); + const parsed = parsePipeToArgs(args, 'writeSync'); + const { transforms, writer } = parsed; + const options = converters.PipeToSyncOptions(parsed.options, { + __proto__: null, + context: 'options', + }); const hasWritevSync = typeof writer.writevSync === 'function'; const endSync = writer.endSync; - if (!options?.preventClose && typeof endSync !== 'function') { + if (!options.preventClose && typeof endSync !== 'function') { throw new ERR_INVALID_ARG_TYPE( 'writer.endSync', 'Function', endSync); } @@ -1012,14 +1020,14 @@ function pipeToSync(source, ...args) { } } - if (!options?.preventClose) { + if (!options.preventClose) { if (FunctionPrototypeCall(endSync, writer) < 0) { throw new ERR_INVALID_STATE( 'Writer could not be closed synchronously'); } } } catch (error) { - if (!options?.preventFail) { + if (!options.preventFail) { writer.fail?.(wrapError(error)); } throw error; @@ -1035,15 +1043,16 @@ function pipeToSync(source, ...args) { * @returns {Promise} Total bytes written */ async function pipeTo(source, ...args) { - const { transforms, writer, options } = parsePipeToArgs(args, 'write'); - if (options?.signal !== undefined) { - validateAbortSignal(options.signal, 'options.signal'); - } - - const signal = options?.signal; + const parsed = parsePipeToArgs(args, 'write'); + const { transforms, writer } = parsed; + const options = converters.PipeToOptions(parsed.options, { + __proto__: null, + context: 'options', + }); + const { signal } = options; function failWriter(error) { - if (!options?.preventFail) { + if (!options.preventFail) { writer.fail?.(wrapError(error)); } } @@ -1158,7 +1167,7 @@ async function pipeTo(source, ...args) { } } - if (!options?.preventClose) { + if (!options.preventClose) { if (!hasEndSync || writer.endSync() < 0) { await writer.end?.(signal ? { __proto__: null, signal } : undefined); } diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 8d680cd2ad01..0536d8212c2a 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -25,8 +25,6 @@ const { } = require('internal/errors'); const { lazyDOMException } = require('internal/util'); const { - validateAbortSignal, - validateArray, validateInteger, } = require('internal/validators'); @@ -39,13 +37,15 @@ const { kResolvedPromise, createBatchEntry, onSignalAbort, - toUint8Array, + toWriterUint8Array, convertChunks, getWriterSignal, parsePullArgs, - validateBackpressure, validateBatchEntry, } = require('internal/streams/iter/utils'); +const { + converters, +} = require('internal/streams/iter/webidl'); const { pullWithConsumerCleanup, @@ -124,16 +124,16 @@ class PushQueue { #bufferedBytes = 0; constructor(options = { __proto__: null }) { + options = converters.PushStreamOptions(options, { + __proto__: null, + context: 'options', + }); const { budget = kPushDefaultBudget, backpressure = 'strict', signal, } = options; validateInteger(budget, 'options.budget', 16384); - validateBackpressure(backpressure); - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - } this.#budget = budget; this.#backpressure = backpressure; this.#signal = signal; @@ -166,6 +166,10 @@ class PushQueue { return true; } + get signal() { + return this.#signal; + } + /** * Check if a sync write would be accepted. * @returns {boolean} @@ -651,20 +655,18 @@ class PushWriter { } write(chunk, options) { + const bytes = toWriterUint8Array(chunk); const signal = getWriterSignal(options); if (!signal && this.#queue.canWriteSync()) { - const bytes = toUint8Array(chunk); this.#queue.writeSync([bytes]); return kResolvedPromise; } - const bytes = toUint8Array(chunk); return this.#queue.writeAsync([bytes], signal); } writev(chunks, options) { - validateArray(chunks, 'chunks'); - const signal = getWriterSignal(options); const bytes = convertChunks(chunks); + const signal = getWriterSignal(options); if (!signal && this.#queue.writeSync(bytes)) { return kResolvedPromise; } @@ -672,12 +674,11 @@ class PushWriter { } writeSync(chunk) { - const bytes = toUint8Array(chunk); + const bytes = toWriterUint8Array(chunk); return this.#queue.writeSync([bytes]); } writevSync(chunks) { - validateArray(chunks, 'chunks'); const bytes = convertChunks(chunks); return this.#queue.writeSync(bytes); } @@ -780,7 +781,7 @@ function push(...args) { let readable; if (transforms.length > 0) { readable = pullWithConsumerCleanup( - rawReadable, transforms, options.signal); + rawReadable, transforms, queue.signal); } else { readable = rawReadable; } diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 00d9d387ce82..70179dffd3c1 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -43,9 +43,11 @@ const { onSignalAbort, wrapError, parsePullArgs, - validateBackpressure, validateBatchEntry, } = require('internal/streams/iter/utils'); +const { + converters, +} = require('internal/streams/iter/webidl'); const { RingBuffer, @@ -59,9 +61,7 @@ const { }, } = require('internal/errors'); const { - validateAbortSignal, validateInteger, - validateObject, } = require('internal/validators'); // ============================================================================= @@ -104,9 +104,13 @@ class ShareImpl { } pull(...args) { - const { transforms, options } = parsePullArgs(args); - const signal = options?.signal; - validateAbortSignal(signal, 'options.signal'); + const parsed = parsePullArgs(args); + const { transforms } = parsed; + const options = converters.PullOptions(parsed.options, { + __proto__: null, + context: 'options', + }); + const { signal } = options; // Avoid registering a consumer that the pre-aborted pipeline will never // read or detach. @@ -746,17 +750,16 @@ function onShareCancel(shareImpl, signal) { function share(source, options = { __proto__: null }) { // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc. const normalized = from(source); - validateObject(options, 'options'); + options = converters.ShareOptions(options, { + __proto__: null, + context: 'options', + }); const { budget = kMultiConsumerDefaultBudget, backpressure = 'strict', signal, } = options; validateInteger(budget, 'options.budget', 16384); - validateBackpressure(backpressure); - if (signal !== undefined) { - validateAbortSignal(signal, 'options.signal'); - } const opts = { __proto__: null, @@ -777,13 +780,15 @@ function share(source, options = { __proto__: null }) { function shareSync(source, options = { __proto__: null }) { // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc. const normalized = fromSync(source); - validateObject(options, 'options'); + options = converters.ShareSyncOptions(options, { + __proto__: null, + context: 'options', + }); const { budget = kMultiConsumerDefaultBudget, backpressure = 'strict', } = options; validateInteger(budget, 'options.budget', 16384); - validateBackpressure(backpressure); const opts = { __proto__: null, diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index 66e831a58523..08337a5536f2 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -35,9 +35,11 @@ const { isError } = require('internal/util'); const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types'); const { - validateAbortSignal, validateOneOf, } = require('internal/validators'); +const { + converters, +} = require('internal/streams/iter/webidl'); // Cached resolved promise to avoid allocating a new one on every sync fast-path. const kResolvedPromise = PromiseResolve(); @@ -304,6 +306,10 @@ function concatBytes(chunks) { * @returns {Uint8Array[]} */ function convertChunks(chunks) { + chunks = converters.WriterChunkSequence(chunks, { + __proto__: null, + context: 'chunks', + }); const len = chunks.length; const result = new Array(len); for (let i = 0; i < len; i++) { @@ -318,9 +324,17 @@ function convertChunks(chunks) { * @returns {AbortSignal|undefined} */ function getWriterSignal(options) { - const signal = options?.signal; - validateAbortSignal(signal, 'options.signal'); - return signal; + return converters.WriteOptions(options, { + __proto__: null, + context: 'options', + }).signal; +} + +function toWriterUint8Array(chunk) { + return toUint8Array(converters.WriterChunk(chunk, { + __proto__: null, + context: 'chunk', + })); } /** @@ -348,20 +362,6 @@ function hasProtocol(value, symbol) { ); } -/** - * Check if a value is PullOptions (object without transform or write property). - * @param {unknown} value - * @returns {boolean} - */ -function isPullOptions(value) { - return ( - value !== null && - typeof value === 'object' && - !('transform' in value) && - !('write' in value) - ); -} - /** * Check if a value is a stateful transform object (has a transform method). * @param {unknown} value @@ -384,7 +384,7 @@ function isTransform(value) { * Parse variadic arguments for pull/pullSync. * Returns { transforms, options } * @param {Array} args - * @returns {{ transforms: Array, options: object|undefined }} + * @returns {{ transforms: Array, options: unknown }} */ function parsePullArgs(args) { if (args.length === 0) { @@ -394,7 +394,7 @@ function parsePullArgs(args) { let transforms; let options; const last = args[args.length - 1]; - if (isPullOptions(last)) { + if (!isTransform(last)) { transforms = ArrayPrototypeSlice(args, 0, -1); options = last; } else { @@ -436,12 +436,12 @@ module.exports = { getWriterSignal, getMinCursor, hasProtocol, - isPullOptions, isTransform, isTransformObject, onSignalAbort, parsePullArgs, toUint8Array, + toWriterUint8Array, validateBackpressure, validateBatchEntry, validateByteView, diff --git a/lib/internal/streams/iter/webidl.js b/lib/internal/streams/iter/webidl.js new file mode 100644 index 000000000000..b971fa95071b --- /dev/null +++ b/lib/internal/streams/iter/webidl.js @@ -0,0 +1,177 @@ +'use strict'; + +const { + converters: baseConverters, + convertToInt, + createDictionaryConverter, + createEnumConverter, + createInterfaceConverter, + createSequenceConverter, +} = require('internal/webidl'); +const { AbortSignal } = require('internal/abort_controller'); +const { isUint8Array } = require('internal/util/types'); + +const converters = { __proto__: null }; + +function unsignedLongLong(value, options) { + return convertToInt(value, 64, 'unsigned', options); +} + +function enforceRangeUnsignedLongLong(value, options = { __proto__: null }) { + return convertToInt(value, 64, 'unsigned', { + __proto__: null, + prefix: options.prefix, + context: options.context, + code: options.code, + enforceRange: true, + }); +} + +function allowStreamBufferOptions(options) { + return { + __proto__: null, + prefix: options.prefix, + context: options.context, + code: options.code, + allowShared: true, + allowResizable: true, + }; +} + +converters.AbortSignal = createInterfaceConverter( + 'AbortSignal', AbortSignal.prototype); +converters.BackpressurePolicy = createEnumConverter('BackpressurePolicy', [ + 'strict', + 'unbounded', + 'drop-oldest', + 'drop-newest', +]); +converters.unsignedLongLong = unsignedLongLong; +converters.enforceRangeUnsignedLongLong = enforceRangeUnsignedLongLong; +converters.WriterChunk = (value, options = { __proto__: null }) => { + if (isUint8Array(value)) { + return baseConverters.Uint8Array( + value, allowStreamBufferOptions(options)); + } + return baseConverters.USVString(value, options); +}; +converters.WriterChunkSequence = createSequenceConverter( + converters.WriterChunk); + +const signalMember = { + __proto__: null, + key: 'signal', + converter: converters.AbortSignal, +}; +const budgetMember = { + __proto__: null, + key: 'budget', + converter: converters.unsignedLongLong, +}; +const backpressureMember = { + __proto__: null, + key: 'backpressure', + converter: converters.BackpressurePolicy, + defaultValue: () => 'strict', +}; +const limitMember = { + __proto__: null, + key: 'limit', + converter: converters.enforceRangeUnsignedLongLong, +}; + +converters.WriteOptions = createDictionaryConverter('WriteOptions', [ + signalMember, +]); +converters.PushStreamOptions = createDictionaryConverter( + 'PushStreamOptions', [budgetMember, backpressureMember, signalMember]); +converters.PullOptions = createDictionaryConverter('PullOptions', [ + signalMember, +]); +converters.PipeToOptions = createDictionaryConverter('PipeToOptions', [ + { + __proto__: null, + key: 'preventClose', + converter: baseConverters.boolean, + defaultValue: () => false, + }, + { + __proto__: null, + key: 'preventFail', + converter: baseConverters.boolean, + defaultValue: () => false, + }, + signalMember, +]); +converters.PipeToSyncOptions = createDictionaryConverter( + 'PipeToSyncOptions', [ + { + __proto__: null, + key: 'preventClose', + converter: baseConverters.boolean, + defaultValue: () => false, + }, + { + __proto__: null, + key: 'preventFail', + converter: baseConverters.boolean, + defaultValue: () => false, + }, + ]); +converters.ConsumeOptions = createDictionaryConverter('ConsumeOptions', [ + limitMember, + signalMember, +]); +converters.ConsumeSyncOptions = createDictionaryConverter( + 'ConsumeSyncOptions', [limitMember]); +const encodingMember = { + __proto__: null, + key: 'encoding', + converter: baseConverters.DOMString, + defaultValue: () => 'utf-8', +}; +converters.TextConsumeOptions = createDictionaryConverter( + 'TextConsumeOptions', [ + [limitMember, signalMember], + [encodingMember], + ]); +converters.TextConsumeSyncOptions = createDictionaryConverter( + 'TextConsumeSyncOptions', [ + [limitMember], + [encodingMember], + ]); +converters.MergeOptions = createDictionaryConverter('MergeOptions', [ + signalMember, +]); +converters.BroadcastOptions = createDictionaryConverter( + 'BroadcastOptions', [budgetMember, backpressureMember, signalMember]); +converters.ShareOptions = createDictionaryConverter( + 'ShareOptions', [budgetMember, backpressureMember, signalMember]); +converters.ShareSyncOptions = createDictionaryConverter( + 'ShareSyncOptions', [budgetMember, backpressureMember]); +converters.DuplexDirectionOptions = createDictionaryConverter( + 'DuplexDirectionOptions', [ + budgetMember, + { + __proto__: null, + key: 'backpressure', + converter: converters.BackpressurePolicy, + }, + ]); +converters.DuplexOptions = createDictionaryConverter('DuplexOptions', [ + { + __proto__: null, + key: 'a', + converter: converters.DuplexDirectionOptions, + }, + { + __proto__: null, + key: 'b', + converter: converters.DuplexDirectionOptions, + }, + budgetMember, + backpressureMember, + signalMember, +]); + +module.exports = { converters }; diff --git a/test/parallel/test-fs-promises-file-handle-writer.js b/test/parallel/test-fs-promises-file-handle-writer.js index ff90716400ef..d43bababad1e 100644 --- a/test/parallel/test-fs-promises-file-handle-writer.js +++ b/test/parallel/test-fs-promises-file-handle-writer.js @@ -1060,6 +1060,27 @@ async function testWriterArgumentValidation() { } } +async function testWriterWebIDLConversion() { + const filePath = path.join(tmpDir, 'writer-webidl.txt'); + const fh = await open(filePath, 'w'); + const w = fh.writer(); + + await w.write(42, null); + await w.writev(new Set([true, { toString: () => 'object' }])); + assert.throws( + () => w.write(Symbol('invalid')), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.throws( + () => w.write('invalid options', 1), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.strictEqual(await w.end(null), 12); + await fh.close(); + + assert.strictEqual(fs.readFileSync(filePath, 'utf8'), '42trueobject'); +} + // ============================================================================= // Run all tests // ============================================================================= @@ -1114,4 +1135,5 @@ Promise.all([ testWriterLimitWritevSync(), testWriterLimitAndStart(), testWriterArgumentValidation(), + testWriterWebIDLConversion(), ]).then(common.mustCall()); diff --git a/test/parallel/test-quic-stream-writer-api.mjs b/test/parallel/test-quic-stream-writer-api.mjs index 009cb4ff8a4a..4f19502d8579 100644 --- a/test/parallel/test-quic-stream-writer-api.mjs +++ b/test/parallel/test-quic-stream-writer-api.mjs @@ -16,7 +16,7 @@ const { bytes } = await import('stream/iter'); const encoder = new TextEncoder(); -const totalStreams = 5; +const totalStreams = 6; const serverResults = []; const allDone = Promise.withResolvers(); @@ -88,6 +88,25 @@ await clientSession.opened; await stream.closed; } +// Web IDL Writer argument conversion +{ + const stream = await clientSession.createBidirectionalStream(); + const w = stream.writer; + await w.write(42, null); + await w.writev(new Set([true, { toString: () => 'object' }])); + assert.throws( + () => w.write(Symbol('invalid')), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.throws( + () => w.write('invalid options', 1), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.strictEqual(w.endSync(), 12); + for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars + await stream.closed; +} + { const stream = await clientSession.createBidirectionalStream(); const w = stream.writer; @@ -139,4 +158,5 @@ assert.strictEqual(decoder.decode(serverResults[0]), 'async write'); assert.strictEqual(decoder.decode(serverResults[1]), 'hello writev'); assert.strictEqual(decoder.decode(serverResults[2]), 'async writev'); assert.strictEqual(decoder.decode(serverResults[3]), 'end async'); -assert.strictEqual(decoder.decode(serverResults[4]), 'capacity'); +assert.strictEqual(decoder.decode(serverResults[4]), '42trueobject'); +assert.strictEqual(decoder.decode(serverResults[5]), 'capacity'); diff --git a/test/parallel/test-stream-iter-consumers-text.js b/test/parallel/test-stream-iter-consumers-text.js index a8fb7a74367f..68be098ff8f6 100644 --- a/test/parallel/test-stream-iter-consumers-text.js +++ b/test/parallel/test-stream-iter-consumers-text.js @@ -143,17 +143,17 @@ function testTextSyncUnsupportedEncodingThrowsRangeError() { ); } -async function testTextNonStringEncodingThrowsTypeError() { +async function testTextConvertedEncodingThrowsRangeError() { await assert.rejects( () => text(from('hello'), { encoding: 1 }), - { code: 'ERR_INVALID_ARG_TYPE' }, + { code: 'ERR_INVALID_ARG_VALUE' }, ); } -function testTextSyncNonStringEncodingThrowsTypeError() { +function testTextSyncConvertedEncodingThrowsRangeError() { assert.throws( () => textSync(fromSync('hello'), { encoding: 1 }), - { code: 'ERR_INVALID_ARG_TYPE' }, + { code: 'ERR_INVALID_ARG_VALUE' }, ); } @@ -173,6 +173,6 @@ Promise.all([ testTextSyncBOMStripped(), testTextUnsupportedEncodingThrowsRangeError(), testTextSyncUnsupportedEncodingThrowsRangeError(), - testTextNonStringEncodingThrowsTypeError(), - testTextSyncNonStringEncodingThrowsTypeError(), + testTextConvertedEncodingThrowsRangeError(), + testTextSyncConvertedEncodingThrowsRangeError(), ]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index 98f084ab01a6..854bd69f0ac6 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -216,7 +216,7 @@ async function testWritevSyncInvalidChunkDoesNotQueue() { const { writer, readable } = push({ budget: 16384 }); assert.throws( - () => writer.writevSync([1]), + () => writer.writevSync([Symbol('invalid')]), { code: 'ERR_INVALID_ARG_TYPE' }, ); diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js index b93e6575490f..871c891a6539 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/test/parallel/test-stream-iter-validation.js @@ -22,7 +22,7 @@ const { // ============================================================================= // Budget must be integer >= 16384 -assert.throws(() => push({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => push({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => push({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); // Values < 16384 are rejected assert.throws(() => push({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); @@ -68,12 +68,11 @@ assert.throws(() => push('bad', {}), { code: 'ERR_INVALID_ARG_TYPE' }); writer.endSync(); } -// Writer.write rejects non-string/non-Uint8Array +// Writer chunks use the Web IDL (Uint8Array or USVString) conversion. { const { writer } = push(); - assert.throws(() => writer.writeSync(42), { code: 'ERR_INVALID_ARG_TYPE' }); - assert.throws(() => writer.writeSync({}), { code: 'ERR_INVALID_ARG_TYPE' }); - assert.throws(() => writer.writeSync(true), { code: 'ERR_INVALID_ARG_TYPE' }); + assert.throws(() => writer.writeSync(Symbol()), + { code: 'ERR_INVALID_ARG_TYPE' }); writer.endSync(); } @@ -87,7 +86,7 @@ assert.throws(() => duplex({ a: 42 }), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => duplex({ b: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); // Budget validation (cascades through to push()) -assert.throws(() => duplex({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => duplex({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => duplex({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => duplex({ budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); @@ -129,7 +128,7 @@ assert.throws(() => pullSync(fromSync('a'), 42), { code: 'ERR_INVALID_ARG_TYPE' // broadcast() validation // ============================================================================= -assert.throws(() => broadcast({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => broadcast({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => broadcast({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => broadcast({ budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); @@ -214,7 +213,7 @@ assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' }); // ============================================================================= assert.throws(() => share(42), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => share(from('a'), { budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => share(from('a'), { budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => share(from('a'), { budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); @@ -239,7 +238,7 @@ share(from('a'), { budget: Number.MAX_SAFE_INTEGER }).cancel(); assert.throws(() => shareSync(42), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => shareSync(fromSync('a'), { budget: 'bad' }), - { code: 'ERR_INVALID_ARG_TYPE' }); + { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => shareSync(fromSync('a'), { budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => shareSync(fromSync('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }), @@ -274,7 +273,7 @@ assert.throws(() => bytesSync(fromSync('a'), { limit: 'bad' }), assert.throws(() => bytesSync(fromSync('a'), { limit: -1 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => textSync(fromSync('a'), { encoding: 42 }), - { code: 'ERR_INVALID_ARG_TYPE' }); + { code: 'ERR_INVALID_ARG_VALUE' }); assert.throws(() => textSync(fromSync('a'), { encoding: 'bogus' }), { code: 'ERR_INVALID_ARG_VALUE' }); assert.throws(() => arrayBufferSync(fromSync('a'), { limit: 'bad' }), @@ -318,7 +317,7 @@ async function testAsyncValidation() { await assert.rejects( () => bytes(from('a'), { limit: -1 }), { code: 'ERR_OUT_OF_RANGE' }); await assert.rejects( - () => text(from('a'), { encoding: 42 }), { code: 'ERR_INVALID_ARG_TYPE' }); + () => text(from('a'), { encoding: 42 }), { code: 'ERR_INVALID_ARG_VALUE' }); await assert.rejects( () => text(from('a'), { encoding: 'not-a-real-encoding' }), { code: 'ERR_INVALID_ARG_VALUE' }); diff --git a/test/parallel/test-stream-iter-webidl.js b/test/parallel/test-stream-iter-webidl.js new file mode 100644 index 000000000000..904bf77a647f --- /dev/null +++ b/test/parallel/test-stream-iter-webidl.js @@ -0,0 +1,159 @@ +// Flags: --experimental-stream-iter +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { Writable } = require('stream'); +const { + Broadcast, + broadcast, + bytesSync, + from, + fromWritable, + fromSync, + pipeTo, + pull, + push, + share, + shareSync, + text, + textSync, +} = require('stream/iter'); + +function testDictionaryAndIntegerConversion() { + const pushed = push(null); + assert.strictEqual(pushed.writer.endSync(), 0); + const transformedWithNull = push((chunks) => chunks, null); + transformedWithNull.writer.endSync(); + + const broadcasted = broadcast(null); + assert.strictEqual(broadcasted.writer.endSync(), 0); + + share(from(''), null).cancel(); + shareSync(fromSync(''), null).cancel(); + + assert.deepStrictEqual(bytesSync(fromSync('data'), null), + new TextEncoder().encode('data')); + assert.deepStrictEqual( + bytesSync(fromSync('data'), { limit: '4.9' }), + new TextEncoder().encode('data'), + ); + assert.throws( + () => bytesSync(fromSync('data'), { limit: -1 }), + { name: 'TypeError', code: 'ERR_OUT_OF_RANGE' }, + ); + + const converted = push({ + budget: '16384.9', + backpressure: { toString: () => 'strict' }, + }); + assert.strictEqual(converted.writer.canWrite, true); + converted.writer.endSync(); + + let signalReads = 0; + const transformed = push((chunks) => chunks, { + get signal() { + signalReads++; + return undefined; + }, + }); + assert.strictEqual(signalReads, 1); + transformed.writer.endSync(); +} + +async function testUnknownDictionaryMembers() { + const source = pull(from('pull'), { + transform: 1, + write: 1, + unknown: true, + }); + assert.strictEqual(await text(source), 'pull'); + + let ended = false; + const writer = { + write() {}, + end() { ended = true; }, + }; + await pipeTo(from('pipe'), writer, { + transform: 1, + write: 1, + }); + assert.strictEqual(ended, true); + + const options = { + get encoding() { + throw new Error('unknown member was read'); + }, + }; + assert.deepStrictEqual(bytesSync(fromSync('bytes'), options), + new TextEncoder().encode('bytes')); + assert.strictEqual(textSync(fromSync('text'), { + encoding: { toString: () => 'utf-8' }, + }), 'text'); +} + +async function testWriterConversion() { + const { writer, readable } = push(); + + await writer.write(42, null); + assert.strictEqual(writer.writevSync(new Set([ + null, + true, + { toString: () => 'object' }, + ])), true); + + assert.throws( + () => writer.write(Symbol('invalid')), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.throws( + () => writer.writev('not a sequence object'), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.throws( + () => writer.write('invalid options', 1), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + + writer.endSync(); + assert.strictEqual(await text(readable), '42nulltrueobject'); +} + +async function testOtherWriterConversions() { + const output = []; + const writable = new Writable({ + write(chunk, encoding, callback) { + output.push(chunk.toString()); + callback(); + }, + }); + const classicWriter = fromWritable(writable); + await classicWriter.write(42, null); + await classicWriter.writev(new Set([false, '!'])); + await classicWriter.end(null); + assert.strictEqual(output.join(''), '42false!'); + + const result = broadcast(); + const source = result.broadcast.push(); + await result.writer.write(42, null); + await result.writer.writev(new Set([true])); + result.writer.endSync(); + assert.strictEqual(await text(source), '42true'); + + let signalReads = 0; + const fromResult = Broadcast.from(from(''), { + get signal() { + signalReads++; + return undefined; + }, + }); + assert.strictEqual(signalReads, 1); + fromResult.broadcast.cancel(); +} + +Promise.all([ + testDictionaryAndIntegerConversion(), + testUnknownDictionaryMembers(), + testWriterConversion(), + testOtherWriterConversions(), +]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-writable-interop.js b/test/parallel/test-stream-iter-writable-interop.js index 110e8e0a7d2e..a8ae0a99cad7 100644 --- a/test/parallel/test-stream-iter-writable-interop.js +++ b/test/parallel/test-stream-iter-writable-interop.js @@ -511,25 +511,18 @@ async function testAsyncDispose() { } // ============================================================================= -// write() validates chunk type +// write() rejects values that cannot be converted to USVString // ============================================================================= -async function testWriteInvalidChunkType() { +function testWriteInvalidChunkType() { const writable = new Writable({ write(chunk, enc, cb) { cb(); } }); const writer = fromWritable(writable); - await assert.rejects( - writer.write(42), - { code: 'ERR_INVALID_ARG_TYPE' }, - ); - await assert.rejects( - writer.write(null), - { code: 'ERR_INVALID_ARG_TYPE' }, - ); - await assert.rejects( - writer.write({}), + assert.throws( + () => writer.write(Symbol('invalid')), { code: 'ERR_INVALID_ARG_TYPE' }, ); + writable.destroy(); } // ============================================================================= @@ -559,7 +552,7 @@ function testWritevInvalidChunkUncorks() { const writer = fromWritable(writable); assert.throws( - () => writer.writev([new Uint8Array([1]), 42]), + () => writer.writev([new Uint8Array([1]), Symbol('invalid')]), { code: 'ERR_INVALID_ARG_TYPE' }, ); assert.strictEqual(writable.writableCorked, 0);