From fdbfc173bf615dd5c6d9781382aaf8270f8694eb Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 14:29:12 +0000 Subject: [PATCH] stream: improve handling of falsy errors in stream/iter Signed-off-by: James M Snell Assisted-by: Opencode --- doc/api/stream_iter.md | 35 ++- lib/internal/abort_controller.js | 1 + lib/internal/fs/promises.js | 237 +++++++++------ lib/internal/quic/quic.js | 14 +- lib/internal/streams/iter/broadcast.js | 41 +-- lib/internal/streams/iter/classic.js | 206 +++++++++---- lib/internal/streams/iter/pull.js | 13 +- lib/internal/streams/iter/push.js | 53 ++-- lib/internal/streams/iter/share.js | 66 +++-- lib/internal/streams/iter/transform.js | 4 +- lib/internal/streams/iter/utils.js | 13 - .../test-fs-promises-file-handle-pull.js | 4 +- .../test-fs-promises-file-handle-writer.js | 55 +++- test/parallel/test-quic-stream-writer-api.mjs | 12 +- ...test-stream-iter-broadcast-backpressure.js | 17 +- .../test-stream-iter-reason-propagation.js | 271 ++++++++++++++++++ .../test-stream-iter-share-coverage.js | 105 ++++++- test/parallel/test-stream-iter-to-readable.js | 159 ++++++++-- .../test-stream-iter-writable-from.js | 241 +++++++++++----- .../test-stream-iter-writable-interop.js | 65 ++++- 20 files changed, 1230 insertions(+), 382 deletions(-) create mode 100644 test/parallel/test-stream-iter-reason-propagation.js diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index b5ed8e06f6e1..f2fbba2286e0 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -463,14 +463,15 @@ if (result < 0) { } ``` -#### `writer.fail(reason)` +#### `writer.fail([reason])` * `reason` {any} Put the writer into a terminal error state. If the writer is already closed 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. +transition with no async work to perform. The reason is stored and propagated +without modification. If omitted, the reason is `undefined`. #### `writer[Symbol.asyncDispose]()` @@ -1337,9 +1338,10 @@ run().catch(console.error); #### `broadcast.cancel([reason])` -* `reason` {Error} +* `reason` {any} -Cancel the broadcast. All consumers receive an error. +Cancel the broadcast. If `reason` is provided, all consumers reject with that +exact reason. If it is omitted, consumers complete normally. #### `broadcast.consumerCount` @@ -1447,9 +1449,10 @@ Create a {Share} from an existing source. #### `share.cancel([reason])` -* `reason` {Error} +* `reason` {any} -Cancel the share. All consumers receive an error. +Cancel the share. If `reason` is provided, all consumers reject with that exact +reason. If it is omitted, consumers complete normally. #### `share.consumerCount` @@ -1521,9 +1524,10 @@ The number of chunks currently buffered. #### `share.cancel([reason])` -* `reason` {Error} +* `reason` {any} -Cancel the share. All consumers receive an error. +Cancel the share. If `reason` is provided, all consumers throw that exact +reason. If it is omitted, consumers complete normally. #### `share.consumerCount` @@ -1647,6 +1651,11 @@ the synchronous Writer methods (`writeSync`, `writevSync`, `endSync`) always return `false` or `-1`, deferring to the async path. The per-write `options.signal` parameter from the Writer interface is also ignored. +If `writer.fail(reason)` receives a non-Error reason, the classic Writable is +destroyed with an `ERR_FALSY_VALUE_REJECTION` or `ERR_OPERATION_FAILED` error. +Its `reason` property contains the original value, which remains the Writer's +stored failure reason. + The result is cached per instance and backpressure policy -- calling `fromWritable()` twice with the same stream and `backpressure` option returns the same Writer. @@ -1705,6 +1714,11 @@ Creates a byte-mode [`stream.Readable`][] from the `source` (the native batch format used by the stream/iter API). Each `Uint8Array` in a yielded batch is pushed as a separate chunk into the Readable. +Classic streams cannot represent arbitrary values as emitted errors. A +non-Error reason is wrapped in an `ERR_FALSY_VALUE_REJECTION` or +`ERR_OPERATION_FAILED` error whose `reason` property contains the original +value. + ```mjs import { createWriteStream } from 'node:fs'; import { from, pull, toReadable } from 'node:stream/iter'; @@ -1789,6 +1803,11 @@ sync path returns `false`. Similarly, `_final()` tries `endSync()` before `end()`. When the sync path succeeds, the callback is deferred via `queueMicrotask` to preserve the async resolution contract. +Classic stream callbacks cannot represent arbitrary values as errors. A +non-Error reason is wrapped in an `ERR_FALSY_VALUE_REJECTION` or +`ERR_OPERATION_FAILED` error before it is passed to the callback. The error's +`reason` property contains the original value. + The Writable's `highWaterMark` is set to `Number.MAX_SAFE_INTEGER` to effectively disable its internal buffering, allowing the underlying Writer to manage backpressure directly. diff --git a/lib/internal/abort_controller.js b/lib/internal/abort_controller.js index 09b160e9fe5a..e05463d581d0 100644 --- a/lib/internal/abort_controller.js +++ b/lib/internal/abort_controller.js @@ -639,6 +639,7 @@ module.exports = { AbortController, AbortSignal, ClonedAbortSignal, + abortSignal, aborted, transferableAbortSignal, transferableAbortController, diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index d5a6b9a2c853..8b511f572f0e 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -15,6 +15,7 @@ const { SafeArrayIterator, SafePromisePrototypeFinally, SafePromiseRace, + SafeSet, Symbol, SymbolAsyncDispose, SymbolAsyncIterator, @@ -537,9 +538,7 @@ if (getOptionValue('--experimental-stream-iter')) { // Signal-aware path while (remaining !== 0) { if (signal.aborted) { - throw signal.reason ?? - lazyDOMException('The operation was aborted', - 'AbortError'); + throw signal.reason; } const toRead = remaining > 0 ? MathMin(readSize, remaining) : readSize; @@ -745,9 +744,13 @@ if (getOptionValue('--experimental-stream-iter')) { let totalBytesWritten = 0; let closed = false; let closing = false; - let pendingEndPromise = null; - let error = null; - let asyncPending = false; + let pendingEnd = null; + let endingCleanup = false; + let errored = false; + let error; + let asyncPending = 0; + let released = false; + const pendingWrites = new SafeSet(); validateBoolean(autoClose, 'options.autoClose'); @@ -766,85 +769,124 @@ if (getOptionValue('--experimental-stream-iter')) { // Write a single buffer with EAGAIN retry (up to 5 retries). async function writeAll(buf, offset, length, position, signal) { - asyncPending = true; - try { - let retries = 0; - while (length > 0) { - const bytesWritten = (await PromisePrototypeThen( - binding.writeBuffer(fd, buf, offset, length, position, - kUsePromises), - undefined, - handleErrorFromBinding, - )) || 0; - - signal?.throwIfAborted(); - - if (bytesWritten === 0) { - if (++retries > 5) { - throw new ERR_OPERATION_FAILED('write failed after retries'); - } - } else { - retries = 0; - } + let retries = 0; + while (length > 0) { + const bytesWritten = (await PromisePrototypeThen( + binding.writeBuffer(fd, buf, offset, length, position, + kUsePromises), + undefined, + handleErrorFromBinding, + )) || 0; + + signal?.throwIfAborted(); - totalBytesWritten += bytesWritten; - offset += bytesWritten; - length -= bytesWritten; - if (position >= 0) position += bytesWritten; + if (bytesWritten === 0) { + if (++retries > 5) { + throw new ERR_OPERATION_FAILED('write failed after retries'); + } + } else { + retries = 0; } - } finally { - asyncPending = false; + + totalBytesWritten += bytesWritten; + offset += bytesWritten; + length -= bytesWritten; + if (position >= 0) position += bytesWritten; } } // Writev with EAGAIN retry. On partial write, concatenates remaining // buffers and falls back to writeAll (same approach as WriteStream). async function writevAll(buffers, position, signal) { - asyncPending = true; - try { - let totalSize = 0; - for (let i = 0; i < buffers.length; i++) { - totalSize += buffers[i].byteLength; - } + let totalSize = 0; + for (let i = 0; i < buffers.length; i++) { + totalSize += buffers[i].byteLength; + } - let retries = 0; - while (totalSize > 0) { - const bytesWritten = (await PromisePrototypeThen( - binding.writeBuffers(fd, buffers, position, kUsePromises), - undefined, - handleErrorFromBinding, - )) || 0; + let retries = 0; + while (totalSize > 0) { + const bytesWritten = (await PromisePrototypeThen( + binding.writeBuffers(fd, buffers, position, kUsePromises), + undefined, + handleErrorFromBinding, + )) || 0; - signal?.throwIfAborted(); + signal?.throwIfAborted(); - if (bytesWritten === 0) { - if (++retries > 5) { - throw new ERR_OPERATION_FAILED('writev failed after retries'); - } - } else { - retries = 0; + if (bytesWritten === 0) { + if (++retries > 5) { + throw new ERR_OPERATION_FAILED('writev failed after retries'); } + } else { + retries = 0; + } - totalBytesWritten += bytesWritten; - totalSize -= bytesWritten; - if (position >= 0) position += bytesWritten; - - if (totalSize > 0) { - // Partial write - concatenate remaining and use writeAll. - const remaining = Buffer.concat(buffers); - const wrote = bytesWritten; - // writeAll is already inside asyncPending = true, but - // writeAll sets it again - that's fine (idempotent). - await writeAll(remaining, wrote, remaining.length - wrote, - position, signal); - return; - } + totalBytesWritten += bytesWritten; + totalSize -= bytesWritten; + if (position >= 0) position += bytesWritten; + + if (totalSize > 0) { + // Partial write - concatenate remaining and use writeAll. + const remaining = Buffer.concat(buffers); + const wrote = bytesWritten; + await writeAll(remaining, wrote, remaining.length - wrote, + position, signal); + return; } - } finally { - asyncPending = false; } } + function releaseAfterFailure() { + if (!errored || asyncPending !== 0 || released) return; + released = true; + handle[kLocked] = false; + handle[kUnref](); + if (autoClose) { + handle[kCloseSync](); + } + } + + function finishEnd() { + if (!closing || errored || asyncPending !== 0 || endingCleanup) return; + endingCleanup = true; + PromisePrototypeThen(cleanup(), () => { + endingCleanup = false; + if (errored) return; + closing = false; + closed = true; + pendingEnd.resolve(totalBytesWritten); + pendingEnd = null; + }, (error) => { + endingCleanup = false; + if (errored) return; + closing = false; + closed = true; + pendingEnd.reject(error); + pendingEnd = null; + }); + } + + function trackOperation(operation) { + const { promise, resolve, reject } = PromiseWithResolvers(); + const pending = { __proto__: null, reject }; + pendingWrites.add(pending); + asyncPending++; + PromisePrototypeThen(operation, (value) => { + const active = pendingWrites.delete(pending); + asyncPending--; + releaseAfterFailure(); + if (active) resolve(value); + finishEnd(); + }, (error) => { + const active = pendingWrites.delete(pending); + asyncPending--; + releaseAfterFailure(); + if (active) reject(error); + finishEnd(); + }); + return promise; + } + // Synchronous write with EAGAIN retry. Throws on I/O error. // Used by writeSync for the full write, and by writevSync for // completing a partial writev. @@ -872,8 +914,8 @@ if (getOptionValue('--experimental-stream-iter')) { } async function cleanup() { - if (closed) return; - closed = true; + if (released) return; + released = true; handle[kLocked] = false; handle[kUnref](); if (autoClose) { @@ -886,13 +928,17 @@ if (getOptionValue('--experimental-stream-iter')) { write(chunk, options = kNullPrototo) { chunk = newStreamsToWriterUint8Array(chunk); const signal = newStreamsGetWriterSignal(options); - if (error) { + if (errored) { return PromiseReject(error); } if (closed) { return PromiseReject( new ERR_INVALID_STATE.TypeError('The writer is closed')); } + if (closing) { + return PromiseReject( + new ERR_INVALID_STATE.TypeError('The writer is closing')); + } if (signal?.aborted) { return PromiseReject(signal.reason); } @@ -904,19 +950,24 @@ if (getOptionValue('--experimental-stream-iter')) { if (bytesRemaining > 0) bytesRemaining -= chunk.byteLength; const position = pos; if (pos >= 0) pos += chunk.byteLength; - return writeAll(chunk, 0, chunk.byteLength, position, signal); + return trackOperation( + writeAll(chunk, 0, chunk.byteLength, position, signal)); }, writev(chunks, options = kNullPrototo) { chunks = newStreamsConvertChunks(chunks); const signal = newStreamsGetWriterSignal(options); - if (error) { + if (errored) { return PromiseReject(error); } if (closed) { return PromiseReject( new ERR_INVALID_STATE.TypeError('The writer is closed')); } + if (closing) { + return PromiseReject( + new ERR_INVALID_STATE.TypeError('The writer is closing')); + } if (signal?.aborted) { return PromiseReject(signal.reason); } @@ -932,12 +983,12 @@ if (getOptionValue('--experimental-stream-iter')) { if (bytesRemaining > 0) bytesRemaining -= totalSize; const position = pos; if (pos >= 0) pos += totalSize; - return writevAll(chunks, position, signal); + return trackOperation(writevAll(chunks, position, signal)); }, writeSync(chunk) { chunk = newStreamsToWriterUint8Array(chunk); - if (error || closed || asyncPending) return false; + if (errored || closed || closing || asyncPending) return false; const length = chunk.byteLength; if (length > syncWriteThreshold) return false; if (length === 0) return true; @@ -968,7 +1019,7 @@ if (getOptionValue('--experimental-stream-iter')) { writevSync(chunks) { chunks = newStreamsConvertChunks(chunks); - if (error || closed || asyncPending) return false; + if (errored || closed || closing || asyncPending) return false; let totalSize = 0; for (let i = 0; i < chunks.length; i++) { totalSize += chunks[i].byteLength; @@ -1004,29 +1055,31 @@ if (getOptionValue('--experimental-stream-iter')) { end(options = kNullPrototo) { const signal = newStreamsGetWriterSignal(options); - if (error) { + if (errored) { return PromiseReject(error); } if (closed) { return PromiseResolve(totalBytesWritten); } if (closing) { - return pendingEndPromise; + return pendingEnd.promise; } if (signal?.aborted) { return PromiseReject(signal.reason); } closing = true; - pendingEndPromise = PromisePrototypeThen( - cleanup(), () => totalBytesWritten); - return pendingEndPromise; + pendingEnd = PromiseWithResolvers(); + finishEnd(); + return pendingEnd.promise; }, endSync() { - if (error) return -1; + if (errored) return -1; if (closed) return totalBytesWritten; + if (closing) return -1; if (asyncPending) return -1; closed = true; + released = true; handle[kLocked] = false; handle[kUnref](); if (autoClose) { @@ -1036,21 +1089,25 @@ if (getOptionValue('--experimental-stream-iter')) { }, fail(reason) { - if (closed || error) return; - error = reason ?? new ERR_INVALID_STATE('Failed'); + if (closed || errored) return; + errored = true; + error = reason; + closing = false; closed = true; - handle[kLocked] = false; - handle[kUnref](); - if (autoClose) { - handle[kCloseSync](); + pendingEnd?.reject(reason); + pendingEnd = null; + for (const pending of pendingWrites) { + pending.reject(reason); } + pendingWrites.clear(); + releaseAfterFailure(); }, [SymbolAsyncDispose]() { if (closing) { - return pendingEndPromise ?? PromiseResolve(); + return pendingEnd?.promise ?? PromiseResolve(); } - if (!closed && !error) { + if (!closed && !errored) { this.fail(); } return PromiseResolve(); diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 8ecd5fc776a3..cd80d809fd63 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -2233,11 +2233,11 @@ class QuicStream { } 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'); } + signal?.throwIfAborted(); // If a drain is already pending, another operation is waiting // for capacity. Under strict policy, reject immediately. // Later, if we add support for other backpressure policies, @@ -2276,12 +2276,11 @@ class QuicStream { } async function writevAsync(chunks, signal) { - signal?.throwIfAborted(); - if (errored) throw error; if (closed || stream.#inner.state.writeEnded) { throw new ERR_INVALID_STATE('Writer is closed'); } + signal?.throwIfAborted(); // If a drain is already pending, another operation is waiting // for capacity. Under strict policy, reject immediately. @@ -2322,6 +2321,8 @@ class QuicStream { } async function endAsync(signal) { + if (errored) throw error; + if (closed) return totalBytesWritten; if (signal !== undefined) { signal.throwIfAborted(); // TODO(@jasnell): The stream/iter spec allows individual sync end @@ -2353,13 +2354,16 @@ class QuicStream { } finally { drainWakeup = null; } - return endSync(); + if (errored) throw error; + const result = endSync(); + if (errored) throw error; + return result; } function fail(reason) { if (closed || errored) return; errored = true; - error = reason ?? new ERR_INVALID_STATE('Failed'); + error = reason; // `writer.fail()` is always an error path, so the wire code on // RESET_STREAM must never be `0n` (which means "no error" in // most application protocols). Resolve the code in priority diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 16cb4c06533c..ddce0d12d42c 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -61,7 +61,6 @@ const { hasProtocol, onSignalAbort, parsePullArgs, - wrapError, toWriterUint8Array, validateBatchEntry, } = require('internal/streams/iter/utils'); @@ -81,6 +80,7 @@ const kCanWrite = Symbol('kCanWrite'); const kOnBufferDrained = Symbol('kOnBufferDrained'); const kOnEndDrained = Symbol('kOnEndDrained'); const kPendingWriteRemoved = Symbol('kPendingWriteRemoved'); +const kNoBroadcastError = Symbol('kNoBroadcastError'); function raceEndWithSignal(promise, signal) { if (!signal) return promise; @@ -107,6 +107,7 @@ class BroadcastImpl { #waiters = []; // Consumers with pending resolve (subset of #consumers) #ended = false; #error; + #errored = false; #cancelled = false; #options; #writer = null; @@ -176,6 +177,7 @@ class BroadcastImpl { reject: null, pending: [], detached: false, + error: kNoBroadcastError, }; this.#consumers.add(state); @@ -211,8 +213,8 @@ class BroadcastImpl { __proto__: null, next() { if (state.detached) { - if (self.#error !== undefined) { - return PromiseReject(self.#error); + if (state.error !== kNoBroadcastError) { + return PromiseReject(state.error); } return kDone; } @@ -231,8 +233,9 @@ class BroadcastImpl { { __proto__: null, done: false, value: chunk }); } - if (self.#error !== undefined) { + if (self.#errored) { state.detached = true; + state.error = self.#error; self.#deleteConsumer(state); return PromiseReject(self.#error); } @@ -272,11 +275,13 @@ class BroadcastImpl { cancel(reason) { if (this.#cancelled) return; + const hasReason = arguments.length > 0; this.#cancelled = true; this.#ended = true; // Prevents [kAbort]() from redundantly iterating consumers - if (reason !== undefined) { + if (hasReason) { this.#error = reason; + this.#errored = true; } // Reject pending writes on the writer so the pump doesn't hang @@ -284,7 +289,7 @@ class BroadcastImpl { for (const consumer of this.#consumers) { if (consumer.resolve) { - if (reason !== undefined) { + if (hasReason) { consumer.reject?.(reason); } else { consumer.resolve({ __proto__: null, done: true, value: undefined }); @@ -292,7 +297,8 @@ class BroadcastImpl { consumer.resolve = null; consumer.reject = null; } - if (reason !== undefined) { + if (hasReason) { + consumer.error = reason; this.#rejectPending(consumer, reason); } else { this.#resolvePendingDone(consumer); @@ -389,7 +395,8 @@ class BroadcastImpl { } [kAbort](reason) { - if (this.#error !== undefined) return; + if (this.#errored) return; + this.#errored = true; this.#error = reason; this.#ended = true; @@ -401,6 +408,7 @@ class BroadcastImpl { consumer.reject = null; } this.#rejectPending(consumer, reason); + consumer.error = reason; consumer.detached = true; } this.#consumers.clear(); @@ -460,7 +468,7 @@ class BroadcastImpl { return validateBatchEntry(entry); } catch (error) { this.#writer.fail(error); - if (this.#error === undefined) this[kAbort](error); + if (!this.#errored) this[kAbort](error); this.#buffer.clear(); this.#bufferedBytes = 0; return null; @@ -706,12 +714,11 @@ class BroadcastWriter { fail(reason) { if (this.#state === 'errored' || this.#state === 'closed') return; this.#state = 'errored'; - const error = reason ?? new ERR_INVALID_STATE.TypeError('Failed'); - this.#error = error; - this.#rejectPendingWrites(error); - this.#rejectPendingDrains(error); - this.#pendingEnd?.reject(error); - this.#broadcast[kAbort](error); + this.#error = reason; + this.#rejectPendingWrites(reason); + this.#rejectPendingDrains(reason); + this.#pendingEnd?.reject(reason); + this.#broadcast[kAbort](reason); } [SymbolAsyncDispose]() { @@ -818,7 +825,7 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) { const idx = pendingWrites.indexOf(entry); if (idx !== -1) pendingWrites.removeAt(idx); entry.batch = null; - reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError')); + reject(signal.reason); if (idx !== -1) self[kPendingWriteRemoved](); }; entry.resolve = function() { @@ -937,7 +944,7 @@ const Broadcast = { await w.end(signal ? { signal } : undefined); } } catch (error) { - w.fail(wrapError(error)); + w.fail(error); } }; PromisePrototypeThen(pump(), undefined, () => {}); diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js index 28f6079e6fe2..4ca384cae9ab 100644 --- a/lib/internal/streams/iter/classic.js +++ b/lib/internal/streams/iter/classic.js @@ -13,6 +13,7 @@ const { ArrayPrototypePush, + FunctionPrototypeCall, NumberMAX_SAFE_INTEGER, Promise, PromisePrototypeThen, @@ -32,9 +33,11 @@ const { AbortError, aggregateTwoErrors, codes: { + ERR_FALSY_VALUE_REJECTION, ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, ERR_INVALID_STATE, + ERR_OPERATION_FAILED, ERR_STREAM_WRITE_AFTER_END, }, } = require('internal/errors'); @@ -67,6 +70,25 @@ const { const { Buffer } = require('buffer'); const destroyImpl = require('internal/streams/destroy'); +const { isError } = require('internal/util'); + +// Classic stream error channels require a truthy Error object. +function toClassicError(reason, reasonMap) { + try { + if (isError(reason)) return reason; + } catch { + // Wrap values whose proxy traps make the Error check fail. + } + let error; + if (!reason) { + error = new ERR_FALSY_VALUE_REJECTION.HideStackFramesError(reason); + } else { + error = new ERR_OPERATION_FAILED('Non-Error value'); + error.reason = reason; + } + reasonMap?.set(error, { __proto__: null, reason }); + return error; +} // Lazy-loaded to avoid circular dependencies. Readable and Writable // both require this module's parent, so we defer the require. @@ -299,11 +321,17 @@ function toReadable(source, options = kNullPrototype) { backpressure.resolve(); backpressure = null; } - if (typeof iterator.return === 'function') { - PromisePrototypeThen(iterator.return(), - () => cb(err), (e) => cb(e || err)); - } else { - cb(err); + try { + const returnMethod = iterator.return; + if (typeof returnMethod !== 'function') { + cb(err); + return; + } + const returned = FunctionPrototypeCall(returnMethod, iterator); + PromisePrototypeThen(PromiseResolve(returned), () => cb(err), + (error) => cb(err || toClassicError(error))); + } catch (error) { + cb(err || toClassicError(error)); } }, }); @@ -331,7 +359,7 @@ function toReadable(source, options = kNullPrototype) { } } catch (err) { done = true; - readable.destroy(err); + readable.destroy(toClassicError(err)); } } @@ -372,30 +400,43 @@ function toReadableSync(source, options = kNullPrototype) { __proto__: null, highWaterMark, read() { - for (;;) { - if (hasBatch) { - while (batchIndex < batch.length) { - if (!this.push(batch[batchIndex++])) return; + try { + for (;;) { + if (hasBatch) { + while (batchIndex < batch.length) { + if (!this.push(batch[batchIndex++])) return; + } + batch = undefined; + hasBatch = false; + batchIndex = 0; } - batch = undefined; - hasBatch = false; - batchIndex = 0; - } - const result = iterator.next(); - const { done } = result; - if (done) { - this.push(null); - return; + const result = iterator.next(); + const { done } = result; + if (done) { + this.push(null); + return; + } + batch = result.value; + hasBatch = true; } - batch = result.value; - hasBatch = true; + } catch (error) { + const classicError = toClassicError(error); + throw classicError; } }, destroy(err, cb) { batch = undefined; hasBatch = false; - if (typeof iterator.return === 'function') iterator.return(); + try { + const returnMethod = iterator.return; + if (typeof returnMethod === 'function') { + FunctionPrototypeCall(returnMethod, iterator); + } + } catch (error) { + cb(err || toClassicError(error)); + return; + } cb(err); }, }); @@ -471,6 +512,9 @@ function fromWritable(writable, options = kNullPrototype) { // expose the full stream.Writable property set. const hwm = writable.writableHighWaterMark ?? 16384; let totalBytes = 0; + let errored = false; + let error; + let pendingEnd; // Waiters pending on backpressure resolution (block policy only). // Multiple un-awaited writes can each add a waiter, so this must be @@ -504,11 +548,17 @@ function fromWritable(writable, options = kNullPrototype) { } // Reject all pending waiters and remove the drain/error listeners. - function cleanup(err) { + function cleanup(err, preserveReason = false) { const pending = waiters; waiters = []; for (let i = 0; i < pending.length; i++) { - pending[i].reject(err ?? new AbortError()); + if (!preserveReason && + (err === undefined || err === null) && + pending[i].close !== undefined) { + pending[i].close(); + } else { + pending[i].reject(preserveReason ? err : err ?? new AbortError()); + } } if (!listenersInstalled) return; listenersInstalled = false; @@ -526,7 +576,8 @@ function fromWritable(writable, options = kNullPrototype) { function isWritable() { // Duck-typed streams may not have these properties -- treat missing // as false (i.e., writable is still open). - return !(writable.destroyed ?? false) && + return !errored && + !(writable.destroyed ?? false) && !(writable.writableFinished ?? false) && !(writable.writableEnded ?? false); } @@ -580,6 +631,7 @@ function fromWritable(writable, options = kNullPrototype) { write(chunk, options) { const bytes = toWriterUint8Array(chunk); getWriterSignal(options); + if (errored) return PromiseReject(error); if (!isWritable()) { return PromiseReject(new ERR_STREAM_WRITE_AFTER_END()); } @@ -617,6 +669,7 @@ function fromWritable(writable, options = kNullPrototype) { writev(chunks, options) { chunks = convertChunks(chunks); getWriterSignal(options); + if (errored) return PromiseReject(error); if (!isWritable()) { return PromiseReject(new ERR_STREAM_WRITE_AFTER_END()); } @@ -665,51 +718,71 @@ function fromWritable(writable, options = kNullPrototype) { // write(). end(options) { getWriterSignal(options); + if (errored) return PromiseReject(error); + if (pendingEnd) return pendingEnd.promise; if ((writable.writableFinished ?? false) || (writable.destroyed ?? false)) { cleanup(); return PromiseResolve(totalBytes); } - const { promise, resolve, reject } = PromiseWithResolvers(); + pendingEnd = PromiseWithResolvers(); + const { promise, resolve, reject } = pendingEnd; - if (!(writable.writableEnded ?? false)) { - writable.end(); - } + try { + if (!(writable.writableEnded ?? false)) { + writable.end(); + } - eos(writable, { writable: true, readable: false }, (err) => { - cleanup(err); - if (err) reject(err); - else resolve(totalBytes); - }); + eos(writable, { writable: true, readable: false }, (err) => { + if (errored) return; + pendingEnd = undefined; + cleanup(err); + if (err) reject(err); + else resolve(totalBytes); + }); + } catch (reason) { + pendingEnd = undefined; + errored = true; + error = reason; + cleanup(reason, true); + reject(reason); + try { + writable.destroy?.(toClassicError(reason)); + } catch { + // Preserve the original terminal reason. + } + } return promise; }, fail(reason) { - cleanup(reason); + if (errored || + (writable.writableFinished ?? false) || + (writable.destroyed ?? false)) { + return; + } + errored = true; + error = reason; + pendingEnd?.reject(reason); + pendingEnd = undefined; + cleanup(reason, true); if (typeof writable.destroy === 'function') { - writable.destroy(reason); + writable.destroy(toClassicError(reason)); } }, [SymbolAsyncDispose]() { + if (pendingEnd) return pendingEnd.promise; if (isWritable()) { - cleanup(); - if (typeof writable.destroy === 'function') { - writable.destroy(); - } + this.fail(); } return PromiseResolve(); }, [SymbolDispose]() { - if (isWritable()) { - cleanup(); - if (typeof writable.destroy === 'function') { - writable.destroy(); - } - } + this.fail(); }, }; @@ -719,11 +792,12 @@ function fromWritable(writable, options = kNullPrototype) { if ((writable.writableLength ?? 0) < hwm) { return PromiseResolve(true); } - const { promise, resolve } = PromiseWithResolvers(); + const { promise, resolve, reject } = PromiseWithResolvers(); ArrayPrototypePush(waiters, { __proto__: null, resolve() { resolve(true); }, - reject() { resolve(false); }, + reject, + close() { resolve(false); }, }); installListeners(); return promise; @@ -761,6 +835,7 @@ function toWritable(writer) { const hasEndSync = hasEnd && typeof writer.endSync === 'function'; const hasFail = typeof writer.fail === 'function'; + const classicErrorReasons = new SafeWeakMap(); // Try-sync-first pattern: attempt the synchronous method and fall back to the // async method if it returns false (data not accepted synchronously). // When the sync path succeeds, the callback is deferred via queueMicrotask @@ -778,14 +853,16 @@ function toWritable(writer) { } // WriteSync returned false: not accepted, fall through to async. } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); return; } } try { - PromisePrototypeThen(writer.write(bytes), () => cb(), cb); + PromisePrototypeThen( + writer.write(bytes), () => cb(), + (err) => cb(toClassicError(err, classicErrorReasons))); } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); } } @@ -804,14 +881,16 @@ function toWritable(writer) { } // WritevSync returned false: not accepted, fall through to async. } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); return; } } try { - PromisePrototypeThen(writer.writev(chunks), () => cb(), cb); + PromisePrototypeThen( + writer.writev(chunks), () => cb(), + (err) => cb(toClassicError(err, classicErrorReasons))); } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); } } @@ -829,22 +908,31 @@ function toWritable(writer) { } // Result < 0: can't end synchronously, fall through to async. } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); return; } } try { - PromisePrototypeThen(writer.end(), () => cb(), cb); + PromisePrototypeThen( + writer.end(), () => cb(), + (err) => cb(toClassicError(err, classicErrorReasons))); } catch (err) { - cb(err); + cb(toClassicError(err, classicErrorReasons)); } } function _destroy(err, cb) { if (err && hasFail) { - writer.fail(err); + const wrapped = classicErrorReasons.get(err); + classicErrorReasons.delete(err); + try { + writer.fail(wrapped === undefined ? err : wrapped.reason); + } catch (error) { + cb(err || toClassicError(error, classicErrorReasons)); + return; + } } - cb(); + cb(err); } const writableOptions = { diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index 6b44f5c5e431..01a9504dc871 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -35,6 +35,7 @@ const { const { AbortController, AbortSignal, + abortSignal, } = require('internal/abort_controller'); const { @@ -54,7 +55,6 @@ const { toUint8Array, validateBatchEntry, validateByteView, - wrapError, yieldAbortable, } = require('internal/streams/iter/utils'); const { @@ -723,8 +723,7 @@ async function* createAsyncPipeline(source, transforms, signal) { let abortHandler; if (signal) { abortHandler = () => { - controller.abort(signal.reason ?? - lazyDOMException('Aborted', 'AbortError')); + abortSignal(controller.signal, signal.reason); }; signal.addEventListener('abort', abortHandler, { __proto__: null, once: true }); } @@ -785,7 +784,7 @@ async function* createAsyncPipeline(source, transforms, signal) { completed = true; } catch (error) { if (!controller.signal.aborted) { - controller.abort(wrapError(error)); + abortSignal(controller.signal, error); } throw error; } finally { @@ -868,7 +867,7 @@ function pull(source, ...args) { return iterator.return(value); }, throw(error) { - controller.abort(error); + abortSignal(controller.signal, error); return iterator.throw(error); }, [SymbolAsyncIterator]() { @@ -1028,7 +1027,7 @@ function pipeToSync(source, ...args) { } } catch (error) { if (!options.preventFail) { - writer.fail?.(wrapError(error)); + writer.fail?.(error); } throw error; } @@ -1053,7 +1052,7 @@ async function pipeTo(source, ...args) { function failWriter(error) { if (!options.preventFail) { - writer.fail?.(wrapError(error)); + writer.fail?.(error); } } diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 0536d8212c2a..31b437c85b75 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -12,7 +12,6 @@ const { PromiseResolve, PromiseWithResolvers, SafeWeakSet, - Symbol, SymbolAsyncDispose, SymbolAsyncIterator, SymbolDispose, @@ -23,7 +22,6 @@ const { ERR_INVALID_STATE, }, } = require('internal/errors'); -const { lazyDOMException } = require('internal/util'); const { validateInteger, } = require('internal/validators'); @@ -55,7 +53,6 @@ const { RingBuffer, } = require('internal/streams/iter/ringbuffer'); -const kNoFailReason = Symbol('kNoFailReason'); const consumerReturnErrors = new SafeWeakSet(); function isConsumerReturnError(error) { @@ -108,8 +105,10 @@ class PushQueue { #writerState = 'open'; /** Consumer state: 'active' | 'returned' | 'thrown' */ #consumerState = 'active'; - /** Error that closed the stream */ - #error = null; + /** Error that closed the writer */ + #writerError; + /** Error supplied by the consumer */ + #consumerError; /** Total bytes written */ #bytesWritten = 0; /** Pending end promise (resolves when consumer drains past end sentinel) */ @@ -255,12 +254,11 @@ class PushQueue { throw new ERR_INVALID_STATE.TypeError('Writer is closing'); } if (this.#writerState === 'errored') { - throw this.#error; + throw this.#writerError; } if (this.#consumerState !== 'active') { - throw this.#consumerState === 'thrown' && this.#error ? - this.#error : - new ERR_INVALID_STATE.TypeError('Stream closed by consumer'); + if (this.#consumerState === 'thrown') throw this.#consumerError; + throw new ERR_INVALID_STATE.TypeError('Stream closed by consumer'); } // Check for pre-aborted signal (after state checks per spec) @@ -307,7 +305,7 @@ class PushQueue { this.#pendingWrites.removeAt(idx); this.#resolvePendingReads(); } - reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError')); + reject(signal.reason); }; // Wrap resolve/reject to clean up signal listener @@ -379,25 +377,23 @@ class PushQueue { * No-op if errored or closed (fully drained). * If closing (draining), short-circuits the drain. */ - fail(reason = kNoFailReason) { + fail(reason) { if (this.#writerState === 'errored' || this.#writerState === 'closed') { return; } const wasClosing = this.#writerState === 'closing'; this.#writerState = 'errored'; - this.#error = reason === kNoFailReason ? - new ERR_INVALID_STATE('Failed') : - reason; + this.#writerError = reason; this.#cleanup(); - this.#rejectPendingReads(this.#error); - this.#rejectPendingDrains(this.#error); - this.#rejectPendingWrites(this.#error); + this.#rejectPendingReads(this.#writerError); + this.#rejectPendingDrains(this.#writerError); + this.#rejectPendingWrites(this.#writerError); if (wasClosing) { // Short-circuit the graceful drain: reject the pending end promise if (this.#pendingEnd) { - this.#pendingEnd.reject(this.#error); + this.#pendingEnd.reject(this.#writerError); this.#pendingEnd = null; } } @@ -408,7 +404,7 @@ class PushQueue { } get error() { - return this.#error; + return this.#writerError; } get backpressurePolicy() { @@ -446,7 +442,7 @@ class PushQueue { return { __proto__: null, done: true, value: undefined }; } if (this.#consumerState === 'thrown') { - throw this.#error; + throw this.#consumerError; } // If there's data in the buffer, return it immediately @@ -467,7 +463,7 @@ class PushQueue { } if (this.#writerState === 'errored') { - throw this.#error; + throw this.#writerError; } const { promise, resolve, reject } = PromiseWithResolvers(); @@ -489,7 +485,7 @@ class PushQueue { consumerThrow(error) { if (this.#consumerState !== 'active') return; this.#consumerState = 'thrown'; - this.#error = error; + this.#consumerError = error; this.#terminateWriterFromConsumer(error); this.#rejectPendingReads(error); // Reject pending drains - the consumer errored @@ -531,7 +527,7 @@ class PushQueue { this.#bufferedBytes = 0; if (this.#writerState === 'open' || this.#writerState === 'closing') { this.#writerState = 'errored'; - this.#error = error; + this.#writerError = error; } this.#cleanup(); this.#rejectPendingWrites(error); @@ -548,7 +544,7 @@ class PushQueue { pending.resolve({ __proto__: null, done: true, value: undefined }); } else if (this.#consumerState === 'thrown') { const pending = this.#pendingReads.shift(); - pending.reject(this.#error); + pending.reject(this.#consumerError); } else if (this.#slots.length > 0) { const pending = this.#pendingReads.shift(); try { @@ -569,7 +565,7 @@ class PushQueue { pending.resolve({ __proto__: null, done: true, value: undefined }); } else if (this.#writerState === 'errored') { const pending = this.#pendingReads.shift(); - pending.reject(this.#error); + pending.reject(this.#writerError); } else { break; } @@ -685,6 +681,11 @@ class PushWriter { end(options) { const signal = getWriterSignal(options); + const state = this.#queue.writerState; + if (state === 'errored') return PromiseReject(this.#queue.error); + if (state === 'closed') { + return PromiseResolve(this.#queue.totalBytesWritten); + } if (signal?.aborted) return PromiseReject(signal.reason); const result = this.#queue.end(); @@ -707,7 +708,7 @@ class PushWriter { } fail(reason) { - this.#queue.fail(arguments.length === 0 ? kNoFailReason : reason); + this.#queue.fail(reason); } [SymbolAsyncDispose]() { diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index 70179dffd3c1..6154509a5b64 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -7,6 +7,7 @@ const { ArrayPrototypePush, + FunctionPrototypeCall, PromisePrototypeThen, PromiseResolve, PromiseWithResolvers, @@ -41,7 +42,6 @@ const { getMinCursor, hasProtocol, onSignalAbort, - wrapError, parsePullArgs, validateBatchEntry, } = require('internal/streams/iter/utils'); @@ -79,7 +79,7 @@ class ShareImpl { #consumers = new SafeSet(); #sourceIterator = null; #sourceExhausted = false; - #sourceError; + #sourceError = kNoShareError; #cancelled = false; #pulling = false; #pullWaiters = []; @@ -192,7 +192,7 @@ class ShareImpl { if (self.#sourceExhausted) { state.detached = true; self.#deleteConsumer(state); - if (self.#sourceError !== undefined) { + if (self.#sourceError !== kNoShareError) { state.error = self.#sourceError; throw state.error; } @@ -254,28 +254,33 @@ class ShareImpl { cancel(reason) { if (this.#cancelled) return; + const hasReason = arguments.length > 0; this.#cancelled = true; - if (reason !== undefined) { + if (hasReason) { this.#cancelError = reason; } this.#resolveCancel(kShareCancelled); this.#resolveCancel = null; - if (this.#sourceIterator?.return) { - try { + try { + const returnMethod = this.#sourceIterator?.return; + if (typeof returnMethod === 'function') { PromisePrototypeThen( - PromiseResolve(this.#sourceIterator.return()), undefined, () => {}); - } catch { - // Cancellation has precedence over source cleanup errors. + PromiseResolve(FunctionPrototypeCall( + returnMethod, this.#sourceIterator)), + undefined, + () => {}); } + } catch { + // Cancellation has precedence over source cleanup errors. } for (const consumer of this.#consumers) { consumer.error = this.#cancelError; if (consumer.resolve) { - if (reason !== undefined) { + if (hasReason) { consumer.reject?.(reason); } else { consumer.resolve({ __proto__: null, done: true, value: undefined }); @@ -304,7 +309,7 @@ class ShareImpl { async #waitForBufferSpace() { while (this.#bufferedBytes >= this.#options.budget) { if (this.#cancelled || - this.#sourceError !== undefined || + this.#sourceError !== kNoShareError || this.#sourceExhausted) { return this.#cancelled ? null : true; } @@ -345,7 +350,7 @@ class ShareImpl { async #waitForBufferSpaceAfterDrop() { while (this.#bufferedBytes >= this.#options.budget && !this.#cancelled && - this.#sourceError === undefined && + this.#sourceError === kNoShareError && !this.#sourceExhausted) { const { promise, resolve } = PromiseWithResolvers(); ArrayPrototypePush(this.#pullWaiters, resolve); @@ -406,7 +411,7 @@ class ShareImpl { this.#bufferedBytes += entry.byteLength; } } catch (error) { - this.#sourceError = wrapError(error); + this.#sourceError = error; this.#sourceExhausted = true; } finally { this.#pulling = false; @@ -483,7 +488,7 @@ class SyncShareImpl { #consumers = new SafeSet(); #sourceIterator = null; #sourceExhausted = false; - #sourceError; + #sourceError = kNoShareError; #cancelled = false; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; @@ -513,6 +518,7 @@ class SyncShareImpl { __proto__: null, cursor: this.#bufferStart, detached: false, + error: kNoShareError, }; this.#consumers.add(state); @@ -532,14 +538,16 @@ class SyncShareImpl { return { __proto__: null, next() { - if (self.#sourceError !== undefined) { - state.detached = true; - self.#deleteConsumer(state); - throw self.#sourceError; - } if (state.detached) { + if (state.error !== kNoShareError) throw state.error; return { __proto__: null, done: true, value: undefined }; } + if (self.#sourceError !== kNoShareError) { + state.detached = true; + state.error = self.#sourceError; + self.#deleteConsumer(state); + throw state.error; + } if (self.#cancelled) { state.detached = true; self.#deleteConsumer(state); @@ -600,10 +608,11 @@ class SyncShareImpl { self.#pullFromSource(); - if (self.#sourceError !== undefined) { + if (self.#sourceError !== kNoShareError) { state.detached = true; + state.error = self.#sourceError; self.#deleteConsumer(state); - throw self.#sourceError; + throw state.error; } const newBufferIndex = state.cursor - self.#bufferStart; @@ -649,17 +658,24 @@ class SyncShareImpl { cancel(reason) { if (this.#cancelled) return; + const hasReason = arguments.length > 0; this.#cancelled = true; - if (reason !== undefined) { + if (hasReason) { this.#sourceError = reason; } - if (this.#sourceIterator?.return) { - this.#sourceIterator.return(); + try { + const returnMethod = this.#sourceIterator?.return; + if (typeof returnMethod === 'function') { + FunctionPrototypeCall(returnMethod, this.#sourceIterator); + } + } catch { + // Cancellation has precedence over source cleanup errors. } for (const consumer of this.#consumers) { + if (hasReason) consumer.error = reason; consumer.detached = true; } this.#consumers.clear(); @@ -685,7 +701,7 @@ class SyncShareImpl { this.#bufferedBytes += entry.byteLength; } } catch (error) { - this.#sourceError = wrapError(error); + this.#sourceError = error; this.#sourceExhausted = true; } } diff --git a/lib/internal/streams/iter/transform.js b/lib/internal/streams/iter/transform.js index cb35906ded02..c4f08b8cd32b 100644 --- a/lib/internal/streams/iter/transform.js +++ b/lib/internal/streams/iter/transform.js @@ -35,7 +35,6 @@ const { }, genericNodeError, } = require('internal/errors'); -const { lazyDOMException } = require('internal/util'); const { isArrayBufferView, isAnyArrayBuffer } = require('internal/util/types'); const { kValidatedTransform } = require('internal/streams/iter/types'); const { @@ -399,8 +398,7 @@ function makeZlibTransform(createHandleFn, processFlag, finishFlag) { resolveWrite = undefined; rejectWrite = undefined; if (reject) { - reject(signal.reason ?? - lazyDOMException('The operation was aborted', 'AbortError')); + reject(signal.reason); } }; signal.addEventListener('abort', onAbort, { __proto__: null, once: true }); diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index 08337a5536f2..95d2d914eb1a 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -9,7 +9,6 @@ const { PromiseWithResolvers, SafePromisePrototypeFinally, SafePromiseRace, - String, SymbolAsyncIterator, TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, @@ -27,10 +26,8 @@ const { codes: { ERR_INVALID_ARG_TYPE, ERR_INVALID_STATE, - ERR_OPERATION_FAILED, }, } = require('internal/errors'); -const { isError } = require('internal/util'); const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types'); @@ -337,15 +334,6 @@ function toWriterUint8Array(chunk) { })); } -/** - * Wrap a caught value as an Error, converting non-Error values. - * @param {unknown} error - * @returns {Error} - */ -function wrapError(error) { - return isError(error) ? error : new ERR_OPERATION_FAILED(String(error)); -} - /** * Check if a value implements a Symbol-keyed protocol (has a function * at the given symbol key). @@ -445,6 +433,5 @@ module.exports = { validateBackpressure, validateBatchEntry, validateByteView, - wrapError, yieldAbortable, }; diff --git a/test/parallel/test-fs-promises-file-handle-pull.js b/test/parallel/test-fs-promises-file-handle-pull.js index 3fc531baf713..cdfaf273f933 100644 --- a/test/parallel/test-fs-promises-file-handle-pull.js +++ b/test/parallel/test-fs-promises-file-handle-pull.js @@ -198,7 +198,7 @@ async function testPullAbortSignal() { const ac = new AbortController(); const fh = await open(filePath, 'r'); try { - ac.abort(); + ac.abort(null); const readable = fh.pull({ signal: ac.signal }); await assert.rejects( @@ -208,7 +208,7 @@ async function testPullAbortSignal() { assert.fail('Should not reach here'); } }, - (err) => err.name === 'AbortError', + (err) => err === null, ); } finally { await fh.close(); diff --git a/test/parallel/test-fs-promises-file-handle-writer.js b/test/parallel/test-fs-promises-file-handle-writer.js index d43bababad1e..a636844f8e57 100644 --- a/test/parallel/test-fs-promises-file-handle-writer.js +++ b/test/parallel/test-fs-promises-file-handle-writer.js @@ -764,8 +764,12 @@ async function testEndSyncReturnsFalseDuringAsync() { const p = w.write(Buffer.from('data')); assert.strictEqual(w.endSync(), -1); + const ending = w.end(); + assert.strictEqual(w.writeSync(Buffer.from('more')), false); + assert.strictEqual(w.endSync(), -1); + await assert.rejects(w.write('more'), { code: 'ERR_INVALID_STATE' }); await p; - const totalBytes = await w.end(); + const totalBytes = await ending; await fh.close(); assert.strictEqual(totalBytes, 4); @@ -850,6 +854,50 @@ async function testEndRejectsOnErrored() { await fh.close(); } +async function testFailPreservesReason() { + for (const reason of [undefined, null, false, 0, '', 'failure']) { + const suffix = String(reason).replaceAll(' ', '-'); + const filePath = path.join(tmpDir, `writer-fail-${suffix}.txt`); + const fh = await open(filePath, 'w'); + const w = fh.writer(); + + w.fail(reason); + + await assert.rejects(w.write('data'), (error) => error === reason); + await assert.rejects(w.end(), (error) => error === reason); + await fh.close(); + } +} + +async function testFailRejectsPendingWriteWithReason() { + const filePath = path.join(tmpDir, 'writer-fail-pending.txt'); + const fh = await open(filePath, 'w'); + const w = fh.writer(); + const reason = null; + const pending = w.write(Buffer.alloc(1024 * 1024)); + + w.fail(reason); + + await assert.rejects(pending, (error) => error === reason); + await fh.close(); +} + +async function testFailWhileClosingPreservesReason() { + const filePath = path.join(tmpDir, 'writer-fail-closing.txt'); + const fh = await open(filePath, 'w'); + const w = fh.writer(); + const reason = false; + const pendingWrite = w.write(Buffer.alloc(1024 * 1024)); + const pendingEnd = w.end(); + + w.fail(reason); + + for (const promise of [pendingWrite, pendingEnd]) { + await assert.rejects(promise, (error) => error === reason); + } + await fh.close(); +} + // ============================================================================= // end() is idempotent when closing/closed // ============================================================================= @@ -914,7 +962,7 @@ async function testAsyncDisposeCallsFail() { // Writer should be in errored state - write should reject await assert.rejects( w.write(Buffer.from('more')), - (err) => err instanceof Error, + (reason) => reason === undefined, ); // Handle should be unlocked and reusable @@ -1125,6 +1173,9 @@ Promise.all([ testEndSyncAutoClose(), testFullSyncPipeline(), testEndRejectsOnErrored(), + testFailPreservesReason(), + testFailRejectsPendingWriteWithReason(), + testFailWhileClosingPreservesReason(), testEndIdempotent(), testAsyncDisposeWhileClosing(), testAsyncDisposeCallsFail(), diff --git a/test/parallel/test-quic-stream-writer-api.mjs b/test/parallel/test-quic-stream-writer-api.mjs index 4f19502d8579..6ccd52b046a4 100644 --- a/test/parallel/test-quic-stream-writer-api.mjs +++ b/test/parallel/test-quic-stream-writer-api.mjs @@ -130,8 +130,8 @@ await clientSession.opened; { const stream = await clientSession.createBidirectionalStream(); const w = stream.writer; - const testError = new Error('writer fail test'); - w.fail(testError); + const reason = null; + w.fail(reason); // After fail, canWrite is null. assert.strictEqual(w.canWrite, null); // drainableProtocol returns null when errored. @@ -141,8 +141,12 @@ await clientSession.opened; assert.strictEqual(w.endSync(), -1); // WriteSync after fail returns false. assert.strictEqual(w.writeSync(encoder.encode('x')), false); - // Write after fail throws with the original error. - await assert.rejects(w.write(encoder.encode('x')), testError); + // Stored failure takes precedence over per-operation cancellation. + const signal = AbortSignal.abort('operation cancelled'); + await assert.rejects( + w.write(encoder.encode('x'), { signal }), + (error) => error === reason); + await assert.rejects(w.end({ signal }), (error) => error === reason); // Don't await stream.closed here — the reset stream may not trigger // server onstream (no data was sent before fail), so the server // won't count it. The stream is cleaned up when the session closes. diff --git a/test/parallel/test-stream-iter-broadcast-backpressure.js b/test/parallel/test-stream-iter-broadcast-backpressure.js index 6efa7eb54c73..e42bb93d5f1f 100644 --- a/test/parallel/test-stream-iter-broadcast-backpressure.js +++ b/test/parallel/test-stream-iter-broadcast-backpressure.js @@ -4,6 +4,7 @@ const common = require('../common'); const assert = require('assert'); const { broadcast, ondrain, text } = require('stream/iter'); +const { setImmediate } = require('timers/promises'); // ============================================================================= // Backpressure policies @@ -67,7 +68,7 @@ async function testDropPoliciesReportPhysicalCapacity() { // 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); + await setImmediate(); assert.strictEqual(drained, false); assert.strictEqual((await iterator.next()).done, false); @@ -93,14 +94,14 @@ async function testBlockBackpressure() { // Next write should block let writeResolved = false; const writePromise = writer.write(kChunk).then(() => { writeResolved = true; }); - await new Promise(setImmediate); + await setImmediate(); assert.strictEqual(writeResolved, false); // Drain consumer to unblock the pending write const iter = consumer[Symbol.asyncIterator](); const first = await iter.next(); assert.strictEqual(first.done, false); - await new Promise(setImmediate); + await setImmediate(); assert.strictEqual(writeResolved, true); writer.endSync(); @@ -122,7 +123,7 @@ async function testBlockBackpressureContent() { writer.writeSync(chunk1); const writePromise = writer.write(chunk2); - await new Promise(setImmediate); + await setImmediate(); // Read all and verify content const iter = consumer[Symbol.asyncIterator](); @@ -158,11 +159,7 @@ async function testStrictBackpressureOverflow() { }); writer.fail(); - await assert.rejects(pending, { - name: 'TypeError', - code: 'ERR_INVALID_STATE', - message: 'Invalid state: Failed', - }); + await assert.rejects(pending, (reason) => reason === undefined); } async function testEndDrainsPendingWrite() { @@ -195,7 +192,7 @@ async function testEndDrainsPendingWrite() { let endResolved = false; endPromise.then(common.mustCall(() => { endResolved = true; })); - await new Promise(setImmediate); + await setImmediate(); assert.strictEqual(endResolved, false); assert.strictEqual((await iter.next()).done, true); diff --git a/test/parallel/test-stream-iter-reason-propagation.js b/test/parallel/test-stream-iter-reason-propagation.js new file mode 100644 index 000000000000..7511dbb2c410 --- /dev/null +++ b/test/parallel/test-stream-iter-reason-propagation.js @@ -0,0 +1,271 @@ +// Flags: --experimental-stream-iter +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + Broadcast, + broadcast, + from, + fromSync, + pipeTo, + pipeToSync, + pull, + push, + share, + shareSync, +} = require('stream/iter'); + +const reasons = [undefined, null, false, 0, '', 'failure']; + +async function rejectsWith(promise, expected) { + await assert.rejects(promise, (reason) => reason === expected); +} + +function throwsWith(fn, expected) { + assert.throws(fn, (reason) => reason === expected); +} + +function asyncThrowingSource(reason) { + return { + __proto__: null, + [Symbol.asyncIterator]() { + return { + __proto__: null, + next() { return Promise.reject(reason); }, + }; + }, + }; +} + +function syncThrowingSource(reason) { + return { + __proto__: null, + [Symbol.iterator]() { + return { + __proto__: null, + next() { throw reason; }, + }; + }, + }; +} + +async function testPipeReasons() { + for (const reason of reasons) { + let asyncFailCalled = false; + let asyncFailReason; + const asyncWriter = { + __proto__: null, + write() {}, + fail(error) { asyncFailCalled = true; asyncFailReason = error; }, + }; + const asyncSource = asyncThrowingSource(reason); + + await rejectsWith(pipeTo(asyncSource, asyncWriter), reason); + assert.strictEqual(asyncFailCalled, true); + assert.strictEqual(asyncFailReason, reason); + + let syncFailCalled = false; + let syncFailReason; + const syncWriter = { + __proto__: null, + writeSync() { return true; }, + endSync() { return 0; }, + fail(error) { syncFailCalled = true; syncFailReason = error; }, + }; + const syncSource = syncThrowingSource(reason); + + throwsWith(() => pipeToSync(syncSource, syncWriter), reason); + assert.strictEqual(syncFailCalled, true); + assert.strictEqual(syncFailReason, reason); + } +} + +async function testSharedSourceReasons() { + for (const reason of reasons) { + const asyncSource = asyncThrowingSource(reason); + const asyncIterator = share(asyncSource).pull()[Symbol.asyncIterator](); + await rejectsWith(asyncIterator.next(), reason); + + const syncSource = syncThrowingSource(reason); + const syncIterator = shareSync(syncSource).pull()[Symbol.iterator](); + throwsWith(() => syncIterator.next(), reason); + } +} + +async function testBroadcastFromReason() { + const reason = 'source failure'; + const source = asyncThrowingSource(reason); + const { broadcast: channel } = Broadcast.from(source); + await rejectsWith(channel.push()[Symbol.asyncIterator]().next(), reason); +} + +async function testWriterFailReasons() { + for (const reason of reasons) { + const pushed = push(); + pushed.writer.fail(reason); + await rejectsWith( + pushed.readable[Symbol.asyncIterator]().next(), reason); + await rejectsWith(pushed.writer.write('data'), reason); + await rejectsWith(pushed.writer.end(), reason); + + const broadcasted = broadcast(); + const iterator = broadcasted.broadcast.push()[Symbol.asyncIterator](); + const pendingRead = iterator.next(); + broadcasted.writer.fail(reason); + await rejectsWith(pendingRead, reason); + await rejectsWith(broadcasted.writer.write('data'), reason); + await rejectsWith(broadcasted.writer.end(), reason); + await rejectsWith( + broadcasted.broadcast.push()[Symbol.asyncIterator]().next(), reason); + } +} + +async function testDisposeFailsWithUndefined() { + const pushed = push(); + pushed.writer[Symbol.dispose](); + await rejectsWith( + pushed.readable[Symbol.asyncIterator]().next(), undefined); + + const broadcasted = broadcast(); + const next = broadcasted.broadcast + .push()[Symbol.asyncIterator]().next(); + broadcasted.writer[Symbol.dispose](); + await rejectsWith(next, undefined); +} + +async function testExplicitUndefinedCancellation() { + const broadcasted = broadcast(); + const broadcastNext = broadcasted.broadcast + .push()[Symbol.asyncIterator]().next(); + broadcasted.broadcast.cancel(undefined); + await rejectsWith(broadcastNext, undefined); + + const shared = share(from('data')); + const shareIterator = shared.pull()[Symbol.asyncIterator](); + shared.cancel(undefined); + await rejectsWith(shareIterator.next(), undefined); + + const syncShared = shareSync(fromSync('data')); + const syncIterator = syncShared.pull()[Symbol.iterator](); + syncShared.cancel(undefined); + throwsWith(() => syncIterator.next(), undefined); +} + +async function testPendingWriteAbortReasons() { + const chunk = new Uint8Array(16384); + + const pushed = push({ budget: chunk.byteLength }); + pushed.writer.writeSync(chunk); + const pushController = new AbortController(); + const pushWrite = pushed.writer.write('data', { + signal: pushController.signal, + }); + pushController.abort(null); + await rejectsWith(pushWrite, null); + + const broadcasted = broadcast({ budget: chunk.byteLength }); + broadcasted.broadcast.push(); + broadcasted.writer.writeSync(chunk); + const broadcastController = new AbortController(); + const broadcastWrite = broadcasted.writer.write('data', { + signal: broadcastController.signal, + }); + broadcastController.abort(null); + await rejectsWith(broadcastWrite, null); + broadcasted.broadcast.cancel(); +} + +async function testTransformReasons() { + for (const thrownReason of [undefined, 'transform failure']) { + let observedReason; + const watchSignal = (batch, { signal }) => { + signal.addEventListener('abort', () => { + observedReason = signal.reason; + }, { once: true }); + return batch; + }; + const throwReason = () => { throw thrownReason; }; + const transformed = pull(from('data'), watchSignal, throwReason); + const iterator = transformed[Symbol.asyncIterator](); + await rejectsWith(iterator.next(), thrownReason); + assert.strictEqual(observedReason, thrownReason); + } + + const controller = new AbortController(); + const started = Promise.withResolvers(); + const waitForAbort = (batch, { signal }) => { + started.resolve(); + const { promise, reject } = Promise.withResolvers(); + signal.addEventListener('abort', () => reject(signal.reason), { + once: true, + }); + return promise; + }; + const abortedIterator = pull(from('data'), waitForAbort, { + signal: controller.signal, + })[Symbol.asyncIterator](); + const next = abortedIterator.next(); + await started.promise; + controller.abort(null); + await rejectsWith(next, null); +} + +async function testIteratorThrowReasonReachesTransforms() { + const reason = undefined; + let observedReason; + const transformed = pull(from('data'), (batch, { signal }) => { + signal.addEventListener('abort', () => { + observedReason = signal.reason; + }, { once: true }); + return batch; + }); + const iterator = transformed[Symbol.asyncIterator](); + + await iterator.next(); + await rejectsWith(iterator.throw(reason), reason); + assert.strictEqual(observedReason, reason); +} + +async function testErroredWriterPrecedesOperationSignal() { + const failure = 0; + const consumerReason = 'consumer failure'; + const signal = AbortSignal.abort(null); + const { writer, readable } = push(); + const iterator = readable[Symbol.asyncIterator](); + + writer.fail(failure); + + await rejectsWith(writer.write('data', { signal }), failure); + await rejectsWith(writer.end({ signal }), failure); + await rejectsWith(iterator.throw(consumerReason), consumerReason); + await rejectsWith(writer.write('data'), failure); +} + +async function testCompletedBroadcastConsumerStaysCompleted() { + const { writer, broadcast: channel } = broadcast(); + const iterator = channel.push()[Symbol.asyncIterator](); + + await iterator.return(); + writer.fail(undefined); + + assert.deepStrictEqual(await iterator.next(), { + __proto__: null, + done: true, + value: undefined, + }); +} + +Promise.all([ + testPipeReasons(), + testSharedSourceReasons(), + testBroadcastFromReason(), + testWriterFailReasons(), + testDisposeFailsWithUndefined(), + testExplicitUndefinedCancellation(), + testPendingWriteAbortReasons(), + testTransformReasons(), + testIteratorThrowReasonReachesTransforms(), + testErroredWriterPrecedesOperationSignal(), + testCompletedBroadcastConsumerStaysCompleted(), +]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-share-coverage.js b/test/parallel/test-stream-iter-share-coverage.js index 48866e61deab..8b0cc1967f2f 100644 --- a/test/parallel/test-stream-iter-share-coverage.js +++ b/test/parallel/test-stream-iter-share-coverage.js @@ -88,32 +88,122 @@ async function testSyncIteratorThrow() { assert.strictEqual(shared.consumerCount, 0); } -// Async source throws non-Error value → wrapError +async function testCompletedSyncConsumerStaysCompleted() { + const reason = undefined; + const source = { + __proto__: null, + [Symbol.iterator]() { + return { + __proto__: null, + next() { throw reason; }, + }; + }, + }; + const shared = shareSync(source); + const completed = shared.pull()[Symbol.iterator](); + const active = shared.pull()[Symbol.iterator](); + + completed.return(); + let caught = false; + try { + active.next(); + } catch (error) { + caught = true; + assert.strictEqual(error, reason); + } + assert.strictEqual(caught, true); + assert.deepStrictEqual(completed.next(), { + __proto__: null, + done: true, + value: undefined, + }); +} + +async function testSyncCancelIgnoresCleanupError() { + const reason = null; + const source = { + __proto__: null, + [Symbol.iterator]() { + let done = false; + return { + __proto__: null, + next() { + if (done) return { done: true, value: undefined }; + done = true; + return { done: false, value: [Buffer.from('data')] }; + }, + get return() { throw new Error('cleanup failed'); }, + }; + }, + }; + const shared = shareSync(source); + const iterator = shared.pull()[Symbol.iterator](); + + iterator.next(); + shared.cancel(reason); + + assert.strictEqual(shared.consumerCount, 0); + assert.throws(() => iterator.next(), (error) => error === reason); +} + +async function testAsyncCancelIgnoresCleanupGetterError() { + const reason = null; + const source = { + __proto__: null, + [Symbol.asyncIterator]() { + let done = false; + return { + __proto__: null, + next() { + if (done) return Promise.resolve({ done: true, value: undefined }); + done = true; + return Promise.resolve({ + done: false, + value: [Buffer.from('data')], + }); + }, + get return() { throw new Error('cleanup failed'); }, + }; + }, + }; + const shared = share(source); + const iterator = shared.pull()[Symbol.asyncIterator](); + + await iterator.next(); + shared.cancel(reason); + + assert.strictEqual(shared.consumerCount, 0); + await assert.rejects(iterator.next(), (error) => error === reason); +} + +// Async source preserves a non-Error thrown value. async function testShareSourceThrowsNonError() { + const reason = 'not an error'; async function* source() { yield [new TextEncoder().encode('ok')]; - throw 'not an error'; // eslint-disable-line no-throw-literal + throw reason; } const shared = share(source()); const consumer = shared.pull(); await assert.rejects(async () => { // eslint-disable-next-line no-unused-vars for await (const batch of consumer) { /* consume */ } - }, { code: 'ERR_OPERATION_FAILED' }); + }, (error) => error === reason); } -// Sync source throws non-Error value → wrapError +// Sync source preserves a non-Error thrown value. async function testSyncShareSourceThrowsNonError() { + const reason = 42; function* source() { yield [new TextEncoder().encode('ok')]; - throw 42; // eslint-disable-line no-throw-literal + throw reason; } const shared = shareSync(source()); const consumer = shared.pull(); assert.throws(() => { // eslint-disable-next-line no-unused-vars for (const batch of consumer) { /* consume */ } - }, { code: 'ERR_OPERATION_FAILED' }); + }, (error) => error === reason); } Promise.all([ @@ -123,6 +213,9 @@ Promise.all([ testSyncShareDispose(), testAsyncIteratorThrow(), testSyncIteratorThrow(), + testCompletedSyncConsumerStaysCompleted(), + testSyncCancelIgnoresCleanupError(), + testAsyncCancelIgnoresCleanupGetterError(), testShareSourceThrowsNonError(), testSyncShareSourceThrowsNonError(), ]).then(common.mustCall()); diff --git a/test/parallel/test-stream-iter-to-readable.js b/test/parallel/test-stream-iter-to-readable.js index d8287036e7e7..a58bf0087df5 100644 --- a/test/parallel/test-stream-iter-to-readable.js +++ b/test/parallel/test-stream-iter-to-readable.js @@ -16,13 +16,15 @@ const { toReadableSync, } = require('stream/iter'); +const kNeverResolves = new Promise(() => { }); + function collect(readable) { - return new Promise((resolve, reject) => { - const chunks = []; - readable.on('data', (chunk) => chunks.push(chunk)); - readable.on('end', () => resolve(Buffer.concat(chunks))); - readable.on('error', reject); - }); + const { promise, resolve, reject } = Promise.withResolvers(); + const chunks = []; + readable.on('data', (chunk) => chunks.push(chunk)); + readable.on('end', () => resolve(Buffer.concat(chunks))); + readable.on('error', reject); + return promise; } // ============================================================================= @@ -105,6 +107,84 @@ async function testErrorAsync() { }, { message: 'source failed' }); } +async function testFalsyErrorAsync() { + for (const [reason, code] of [ + [null, 'ERR_FALSY_VALUE_REJECTION'], + [{ __proto__: null }, 'ERR_OPERATION_FAILED'], + ]) { + const source = { + __proto__: null, + [Symbol.asyncIterator]() { + return { + __proto__: null, + next() { return Promise.reject(reason); }, + return() { return Promise.resolve({ done: true }); }, + }; + }, + }; + + await assert.rejects(collect(toReadable(source)), (error) => { + return error.code === code && error.reason === reason; + }); + } +} + +async function testFalsyThenableCleanupError() { + const reason = null; + const source = { + __proto__: null, + [Symbol.asyncIterator]() { + return { + __proto__: null, + next() { return kNeverResolves; }, + return() { + return { + __proto__: null, + then(resolve, reject) { reject(reason); }, + }; + }, + }; + }, + }; + const readable = toReadable(source); + const { promise, resolve } = Promise.withResolvers(); + readable.once('error', resolve); + + readable.destroy(); + + const result = await promise; + assert.strictEqual(result.code, 'ERR_FALSY_VALUE_REJECTION'); + assert.strictEqual(result.reason, reason); +} + +async function testFalsyCleanupGetterErrors() { + for (const [symbol, create] of [ + [Symbol.asyncIterator, toReadable], + [Symbol.iterator, toReadableSync], + ]) { + const reason = false; + const source = { + __proto__: null, + [symbol]() { + return { + __proto__: null, + next() { return kNeverResolves; }, + get return() { throw reason; }, + }; + }, + }; + const readable = create(source); + const { promise, resolve } = Promise.withResolvers(); + readable.once('error', resolve); + + readable.destroy(); + + const result = await promise; + assert.strictEqual(result.code, 'ERR_FALSY_VALUE_REJECTION'); + assert.strictEqual(result.reason, reason); + } +} + // ============================================================================= // fromStreamIter: empty source // ============================================================================= @@ -155,16 +235,16 @@ async function testDestroyAsync() { // Read a couple chunks then destroy const chunks = []; - await new Promise((resolve, reject) => { - readable.on('data', (chunk) => { - chunks.push(chunk); - if (chunks.length >= 3) { - readable.destroy(); - } - }); - readable.on('close', resolve); - readable.on('error', reject); + const { promise, resolve, reject } = Promise.withResolvers(); + readable.on('data', (chunk) => { + chunks.push(chunk); + if (chunks.length >= 3) { + readable.destroy(); + } }); + readable.on('close', resolve); + readable.on('error', reject); + await promise; assert.ok(chunks.length >= 3); assert.ok(returnCalled, 'iterator.return() should have been called'); @@ -190,15 +270,20 @@ async function testDestroyDuringBackpressure() { const readable = toReadable(gen(), { highWaterMark: 1 }); // Read one chunk to start the pump, then destroy while it's waiting - const chunk = await new Promise((resolve) => { + { + const { promise, resolve } = Promise.withResolvers(); readable.once('readable', () => resolve(readable.read())); - }); - assert.ok(chunk); + assert.ok(await promise); + } // The pump should be waiting on backpressure now. Destroy the stream. readable.destroy(); - await new Promise((resolve) => readable.on('close', resolve)); + { + const { promise, resolve } = Promise.withResolvers(); + readable.on('close', resolve); + await promise; + } assert.ok(readable.destroyed); assert.ok(returnCalled, 'iterator.return() should have been called'); } @@ -243,11 +328,11 @@ async function testPipeAsync() { }, }); - await new Promise((resolve, reject) => { - readable.pipe(writable); - writable.on('finish', resolve); - writable.on('error', reject); - }); + const { promise, resolve, reject } = Promise.withResolvers(); + readable.pipe(writable); + writable.on('finish', resolve); + writable.on('error', reject); + await promise; assert.strictEqual(Buffer.concat(chunks).toString(), 'pipe test data'); } @@ -472,6 +557,24 @@ async function testErrorSync() { }, { message: 'sync source failed' }); } +async function testFalsyErrorSync() { + const reason = false; + const source = { + __proto__: null, + [Symbol.iterator]() { + return { + __proto__: null, + next() { throw reason; }, + }; + }, + }; + + await assert.rejects(collect(toReadableSync(source)), (error) => { + return error.code === 'ERR_FALSY_VALUE_REJECTION' && + error.reason === reason; + }); +} + // ============================================================================= // fromStreamIterSync: empty source // ============================================================================= @@ -506,7 +609,9 @@ async function testDestroySync() { readable.read(); // Start iteration readable.destroy(); - await new Promise((resolve) => readable.on('close', resolve)); + const { promise, resolve } = Promise.withResolvers(); + readable.on('close', resolve); + await promise; assert.ok(returnCalled, 'iterator.return() should have been called'); } @@ -615,6 +720,9 @@ Promise.all([ testMultiBatchAsync(), testBackpressureAsync(), testErrorAsync(), + testFalsyErrorAsync(), + testFalsyThenableCleanupError(), + testFalsyCleanupGetterErrors(), testEmptyAsync(), testEmptyBatchAsync(), testDestroyAsync(), @@ -628,6 +736,7 @@ Promise.all([ testBackpressureSync(), testBackpressureSyncMultiChunkBatch(), testErrorSync(), + testFalsyErrorSync(), testDestroySync(), testRoundTrip(), testRoundTripWithCompression(), diff --git a/test/parallel/test-stream-iter-writable-from.js b/test/parallel/test-stream-iter-writable-from.js index 46cfb627cb4a..5cf7371caa5a 100644 --- a/test/parallel/test-stream-iter-writable-from.js +++ b/test/parallel/test-stream-iter-writable-from.js @@ -6,6 +6,7 @@ const common = require('../common'); const assert = require('assert'); +const { setImmediate, setTimeout } = require('timers/promises'); const { push, text, @@ -28,6 +29,79 @@ async function testBasicWrite() { assert.strictEqual(result, 'hello world'); } +async function testFalsyWriterRejectionBecomesClassicError() { + const nonCoercible = { __proto__: null }; + const trapped = new Proxy({}, { + getPrototypeOf() { throw new Error('unexpected coercion'); }, + }); + for (const [reason, code] of [ + [null, 'ERR_FALSY_VALUE_REJECTION'], + [nonCoercible, 'ERR_OPERATION_FAILED'], + [trapped, 'ERR_OPERATION_FAILED'], + ]) { + let failed = false; + let failReason; + const writable = toWritable({ + __proto__: null, + write() { return Promise.reject(reason); }, + fail(error) { failed = true; failReason = error; }, + }); + writable.on('error', common.mustCall()); + + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('data', common.mustCall((error) => { + for (const symbol of Object.getOwnPropertySymbols(error)) { + delete error[symbol]; + } + if (error) reject(error); + else resolve(); + })); + + await assert.rejects(promise, (error) => { + return error.code === code && error.reason === reason; + }); + await setImmediate(); + assert.strictEqual(failed, true); + assert.strictEqual(failReason, reason); + } +} + +async function testClassicWrapperReusePreservesErrorIdentity() { + const first = toWritable({ + __proto__: null, + write() { return Promise.reject(null); }, + fail() {}, + }); + first.on('error', common.mustCall()); + let wrapper; + { + const { promise, resolve } = Promise.withResolvers(); + first.write('first', common.mustCall((error) => { + wrapper = error; + resolve(); + })); + await promise; + } + + let failReason; + const second = toWritable({ + __proto__: null, + write() { return Promise.reject(wrapper); }, + fail(reason) { failReason = reason; }, + }); + second.on('error', common.mustCall()); + { + const { promise, resolve } = Promise.withResolvers(); + second.write('second', common.mustCall((error) => { + assert.strictEqual(error, wrapper); + resolve(); + })); + await promise; + } + await setImmediate(); + assert.strictEqual(failReason, wrapper); +} + // ============================================================================= // _write delegates to writer.write() // ============================================================================= @@ -46,12 +120,12 @@ async function testWriteDelegatesToWriter() { const writable = toWritable(writer); - await new Promise((resolve, reject) => { - writable.write('hello', (err) => { - if (err) reject(err); - else resolve(); - }); - }); + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('hello', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + await promise; assert.strictEqual(Buffer.concat(chunks).toString(), 'hello'); } @@ -86,7 +160,9 @@ async function testWritevDelegation() { writable.write('c'); writable.uncork(); - await new Promise((resolve) => writable.end(resolve)); + const { promise, resolve } = Promise.withResolvers(); + writable.end(resolve); + await promise; // Writev should have been called with the batched chunks assert.ok(batches.length > 0, 'writev should have been called'); @@ -129,9 +205,9 @@ async function testWriteSyncFirst() { const writable = toWritable(writer); - await new Promise((resolve) => { - writable.write('test', resolve); - }); + const { promise, resolve } = Promise.withResolvers(); + writable.write('test', resolve); + await promise; assert.ok(syncCalled, 'writeSync should have been called'); assert.ok(!asyncCalled, 'write should not have been called'); @@ -160,9 +236,9 @@ async function testWriteSyncFallback() { const writable = toWritable(writer); - await new Promise((resolve) => { - writable.write('test', resolve); - }); + const { promise, resolve } = Promise.withResolvers(); + writable.write('test', resolve); + await promise; assert.ok(syncCalled, 'writeSync should have been called'); assert.ok(asyncCalled, 'write should have been called as fallback'); @@ -191,7 +267,9 @@ async function testEndSyncFirst() { const writable = toWritable(writer); - await new Promise((resolve) => writable.end(resolve)); + const { promise, resolve } = Promise.withResolvers(); + writable.end(resolve); + await promise; assert.ok(endSyncCalled, 'endSync should have been called'); assert.ok(!endAsyncCalled, 'end should not have been called'); @@ -220,7 +298,9 @@ async function testEndSyncFallback() { const writable = toWritable(writer); - await new Promise((resolve) => writable.end(resolve)); + const { promise, resolve } = Promise.withResolvers(); + writable.end(resolve); + await promise; assert.ok(endSyncCalled, 'endSync should have been called'); assert.ok(endAsyncCalled, 'end should have been called as fallback'); @@ -243,7 +323,9 @@ async function testFinalDelegatesToEnd() { const writable = toWritable(writer); - await new Promise((resolve) => writable.end(resolve)); + const { promise, resolve } = Promise.withResolvers(); + writable.end(resolve); + await promise; assert.ok(endCalled, 'writer.end() should have been called'); } @@ -261,13 +343,13 @@ async function testDestroyDelegatesToFail() { }; const writable = toWritable(writer); - writable.on('error', () => {}); // Prevent unhandled + writable.on('error', common.mustCall()); const testErr = new Error('destroy test'); writable.destroy(testErr); // Give a tick for destroy to propagate - await new Promise((resolve) => setTimeout(resolve, 10)); + await setTimeout(10); assert.strictEqual(failReason, testErr); } @@ -286,13 +368,14 @@ async function testWriteErrorPropagation() { }; const writable = toWritable(writer); - - await assert.rejects(new Promise((resolve, reject) => { - writable.write('data', (err) => { - if (err) reject(err); - else resolve(); - }); - }), { message: 'write failed' }); + writable.on('error', common.mustCall()); + + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('data', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + await assert.rejects(promise, { message: 'write failed' }); } // ============================================================================= @@ -343,12 +426,12 @@ async function testPushWriterBlockBackpressureNoDuplicate() { const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); const writable = toWritable(writer); - await new Promise((resolve, reject) => { - writable.write('a', (err) => { - if (err) reject(err); - else resolve(); - }); - }); + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('a', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + await promise; writable.write('b'); writable.end(); @@ -365,12 +448,12 @@ async function testPushWriterBlockBackpressureWritevNoDuplicate() { const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); const writable = toWritable(writer); - await new Promise((resolve, reject) => { - writable.write('a', (err) => { - if (err) reject(err); - else resolve(); - }); - }); + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('a', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + await promise; writable.cork(); writable.write('b'); @@ -423,16 +506,14 @@ async function testSyncCallbackDeferred() { const writable = toWritable(writer); - const p = new Promise((resolve) => { - writable.write('test', () => { - callbackTick = true; - resolve(); - }); - // Callback should NOT have fired synchronously - assert.strictEqual(callbackTick, false); - }); - - await p; + const { promise, resolve } = Promise.withResolvers(); + writable.write('test', common.mustCall(() => { + callbackTick = true; + resolve(); + })); + // Callback should NOT have fired synchronously + assert.strictEqual(callbackTick, false); + await promise; assert.strictEqual(callbackTick, true); } @@ -452,10 +533,10 @@ async function testMinimalWriter() { const writable = toWritable(writer); - await new Promise((resolve) => { - writable.write('minimal'); - writable.end(resolve); - }); + const { promise, resolve } = Promise.withResolvers(); + writable.write('minimal'); + writable.end(resolve); + await promise; assert.strictEqual(Buffer.concat(chunks).toString(), 'minimal'); } @@ -474,7 +555,7 @@ async function testDestroyWithoutError() { const writable = toWritable(writer); writable.destroy(); - await new Promise((resolve) => setTimeout(resolve, 10)); + await setTimeout(10); assert.ok(!failCalled, 'fail should not be called on clean destroy'); } @@ -491,12 +572,12 @@ async function testDestroyWithError() { }; const writable = toWritable(writer); - writable.on('error', () => {}); + writable.on('error', common.mustCall()); const err = new Error('test'); writable.destroy(err); - await new Promise((resolve) => setTimeout(resolve, 10)); + await setTimeout(10); assert.strictEqual(failReason, err); } @@ -512,12 +593,12 @@ async function testDestroyWithoutFail() { }; const writable = toWritable(writer); - writable.on('error', () => {}); + writable.on('error', common.mustCall()); // Should not throw even though writer has no fail() writable.destroy(new Error('test')); - await new Promise((resolve) => setTimeout(resolve, 10)); + await setTimeout(10); assert.ok(writable.destroyed); } @@ -553,13 +634,14 @@ async function testWriteSyncThrowsPropagation() { }; const writable = toWritable(writer); - - await assert.rejects(new Promise((resolve, reject) => { - writable.write('test', (err) => { - if (err) reject(err); - else resolve(); - }); - }), { message: 'sync broken' }); + writable.on('error', common.mustCall()); + + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('test', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + await assert.rejects(promise, { message: 'sync broken' }); } // ============================================================================= @@ -575,13 +657,15 @@ async function testWriteThrowsSyncPropagation() { }; const writable = toWritable(writer); + writable.on('error', common.mustCall()); - await assert.rejects(new Promise((resolve, reject) => { - writable.write('data', (err) => { - if (err) reject(err); - else resolve(); - }); - }), { message: 'sync throw from write' }); + const { promise, resolve, reject } = Promise.withResolvers(); + writable.write('data', common.mustCall((err) => { + if (err) reject(err); + else resolve(); + })); + + await assert.rejects(promise, { message: 'sync throw from write' }); } // ============================================================================= @@ -598,15 +682,16 @@ async function testEndThrowsSyncPropagation() { }; const writable = toWritable(writer); - writable.on('error', () => {}); + writable.on('error', common.mustCall()); - await new Promise((resolve) => { - writable.end(common.mustCall((err) => { - assert.ok(err); - assert.strictEqual(err.message, 'sync throw from end'); - resolve(); - })); - }); + const { promise, resolve } = Promise.withResolvers(); + writable.end(common.mustCall((err) => { + assert.ok(err); + assert.strictEqual(err.message, 'sync throw from end'); + resolve(); + })); + + await promise; } // ============================================================================= @@ -619,6 +704,8 @@ testHighWaterMarkIsMaxSafeInt(); Promise.all([ testBasicWrite(), + testFalsyWriterRejectionBecomesClassicError(), + testClassicWrapperReusePreservesErrorIdentity(), testWriteDelegatesToWriter(), testWritevDelegation(), testWriteSyncFirst(), diff --git a/test/parallel/test-stream-iter-writable-interop.js b/test/parallel/test-stream-iter-writable-interop.js index a8ae0a99cad7..9940ffc34c93 100644 --- a/test/parallel/test-stream-iter-writable-interop.js +++ b/test/parallel/test-stream-iter-writable-interop.js @@ -7,6 +7,7 @@ const common = require('../common'); const assert = require('assert'); const { Writable } = require('stream'); +const { setImmediate } = require('timers/promises'); const { from, fromWritable, @@ -356,7 +357,7 @@ async function testEndReturnsByteCount() { async function testFail() { const writable = new Writable({ write(chunk, enc, cb) { cb(); } }); - writable.on('error', () => {}); // Prevent unhandled error + writable.on('error', common.mustCall()); const writer = fromWritable(writable); writer.fail(new Error('test fail')); @@ -496,6 +497,7 @@ async function testPipeToWithTransform() { async function testDispose() { const writable = new Writable({ write(chunk, enc, cb) { cb(); } }); + writable.on('error', common.mustCall()); const writer = fromWritable(writable); writer[Symbol.dispose](); @@ -504,6 +506,7 @@ async function testDispose() { async function testAsyncDispose() { const writable = new Writable({ write(chunk, enc, cb) { cb(); } }); + writable.on('error', common.mustCall()); const writer = fromWritable(writable); await writer[Symbol.asyncDispose](); @@ -581,7 +584,7 @@ async function testFailRejectsPendingWaiters() { // Never call cb -- stuck }, }); - writable.on('error', () => {}); // Prevent unhandled error + writable.on('error', common.mustCall()); const writer = fromWritable(writable, { backpressure: 'unbounded' }); @@ -594,6 +597,58 @@ async function testFailRejectsPendingWaiters() { await assert.rejects(writePromise, { message: 'fail reason' }); } +async function testFailPreservesReason() { + let classicError; + const writable = new Writable({ + highWaterMark: 1, + write() {}, + }); + writable.on('error', common.mustCall((error) => { classicError = error; })); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); + const pending = writer.write('blocked data'); + const draining = ondrain(writer); + + writer.fail(null); + + await assert.rejects(pending, (reason) => reason === null); + await assert.rejects(draining, (reason) => reason === null); + await assert.rejects(writer.write('more'), (reason) => reason === null); + await assert.rejects(writer.end(), (reason) => reason === null); + await setImmediate(); + assert.strictEqual(classicError.code, 'ERR_FALSY_VALUE_REJECTION'); + assert.strictEqual(classicError.reason, null); +} + +async function testEndThrowPreservesReason() { + const reason = undefined; + const writable = new Writable({ + write(chunk, encoding, callback) { callback(); }, + }); + writable.on('error', common.mustCall()); + writable.end = () => { throw reason; }; + const writer = fromWritable(writable); + + await assert.rejects(writer.end(), (error) => error === reason); + await assert.rejects(writer.write('more'), (error) => error === reason); +} + +async function testFailWhileClosingPreservesReason() { + let finish; + const writable = new Writable({ + write(chunk, encoding, callback) { callback(); }, + final(callback) { finish = callback; }, + }); + writable.on('error', common.mustCall()); + const writer = fromWritable(writable); + const ending = writer.end(); + + writer.fail(false); + + await assert.rejects(ending, (reason) => reason === false); + await assert.rejects(writer.write('more'), (reason) => reason === false); + finish(); +} + // ============================================================================= // dispose rejects pending block waiters // ============================================================================= @@ -605,6 +660,7 @@ async function testDisposeRejectsPendingWaiters() { // Never call cb -- stuck }, }); + writable.on('error', common.mustCall()); const writer = fromWritable(writable, { backpressure: 'unbounded' }); @@ -613,7 +669,7 @@ async function testDisposeRejectsPendingWaiters() { writer[Symbol.dispose](); - await assert.rejects(writePromise, { name: 'AbortError' }); + await assert.rejects(writePromise, (reason) => reason === undefined); } // ============================================================================= @@ -672,5 +728,8 @@ Promise.all([ testAsyncDispose(), testWriteInvalidChunkType(), testFailRejectsPendingWaiters(), + testFailPreservesReason(), + testFailWhileClosingPreservesReason(), + testEndThrowPreservesReason(), testDisposeRejectsPendingWaiters(), ]).then(common.mustCall());