mirrors/node
0
0 Fork 0
mirror of https://github.com/nodejs/node.git synced 2026-10-28 10:15:07 +08:00
Code Activity

stream: improve handling of falsy errors in stream/iter

Browse Source 
Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: https://github.com/nodejs/node/pull/65864
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
This commit is contained in:
James M Snell authored and GitHub committed 2026-09-08 20:56:03 +00:00
1 parent 23a05a6e98
commit 60e363046f
20 files changed
+1232 -384
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/api/stream_iter.md
+27 -8
View File
@@ -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.
lib/internal/abort_controller.js
+1
View File
@@ -642,6 +642,7 @@ module.exports = {
AbortController,
AbortSignal,
ClonedAbortSignal,
abortSignal,
aborted,
transferableAbortSignal,
transferableAbortController,
lib/internal/fs/promises.js
+151 -94
View File
@@ -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,83 +769,122 @@ 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;
let retries = 0;
while (length > 0) {
const bytesWritten = (await PromisePrototypeThen(
binding.writeBuffer(fd, buf, offset, length, position,
kUsePromises),
undefined,
handleErrorFromBinding,
)) || 0;
signal?.throwIfAborted();
signal?.throwIfAborted();
if (bytesWritten === 0) {
if (++retries > 5) {
throw new ERR_OPERATION_FAILED('write failed after retries');
}
} else {
retries = 0;
if (bytesWritten === 0) {
if (++retries > 5) {
throw new ERR_OPERATION_FAILED('write failed after retries');
}
totalBytesWritten += bytesWritten;
offset += bytesWritten;
length -= bytesWritten;
if (position >= 0) position += bytesWritten;
} 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 retries = 0;
while (totalSize > 0) {
const bytesWritten = (await PromisePrototypeThen(
binding.writeBuffers(fd, buffers, position, kUsePromises),
undefined,
handleErrorFromBinding,
)) || 0;
signal?.throwIfAborted();
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;
}
}
} finally {
asyncPending = false;
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;
signal?.throwIfAborted();
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;
await writeAll(remaining, wrote, remaining.length - wrote,
position, signal);
return;
}
}
}
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.
@@ -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();
lib/internal/quic/quic.js
+9 -5
View File
@@ -2250,11 +2250,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,
@@ -2293,12 +2293,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.
@@ -2339,6 +2338,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
@@ -2370,13 +2371,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
lib/internal/streams/iter/broadcast.js
+24 -17
View File
@@ -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, () => {});
lib/internal/streams/iter/classic.js
+147 -59
View File
@@ -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) => {
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.
}
}
eos(writable, { writable: true, readable: false }, (err) => {
cleanup(err);
if (err) reject(err);
else resolve(totalBytes);
});
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 = {
lib/internal/streams/iter/pull.js
+6 -7
View File
@@ -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);
}
}
lib/internal/streams/iter/push.js
+27 -26
View File
@@ -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]() {
lib/internal/streams/iter/share.js
+41 -25
View File
@@ -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;
}
}
lib/internal/streams/iter/transform.js
+1 -3
View File
@@ -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 });
lib/internal/streams/iter/utils.js
-13
View File
@@ -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,
};
test/parallel/test-fs-promises-file-handle-pull.js
+2 -2
View File
@@ -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();
test/parallel/test-fs-promises-file-handle-writer.js
+53 -2
View File
@@ -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(),
test/parallel/test-quic-stream-writer-api.mjs
+8 -4
View File
@@ -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.
test/parallel/test-stream-iter-broadcast-backpressure.js
+7 -10
View File
@@ -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);
test/parallel/test-stream-iter-reason-propagation.js
+271
View File
@@ -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());
test/parallel/test-stream-iter-share-coverage.js
+99 -6
View File
@@ -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());
test/parallel/test-stream-iter-to-readable.js
+134 -25
View File
@@ -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(),
test/parallel/test-stream-iter-writable-from.js
+162 -75
View File
@@ -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);
writable.on('error', common.mustCall());
await assert.rejects(new Promise((resolve, reject) => {
writable.write('data', (err) => {
if (err) reject(err);
else resolve();
});
}), { message: 'write failed' });
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);
writable.on('error', common.mustCall());
await assert.rejects(new Promise((resolve, reject) => {
writable.write('test', (err) => {
if (err) reject(err);
else resolve();
});
}), { message: 'sync broken' });
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(),
test/parallel/test-stream-iter-writable-interop.js
+62 -3
View File
@@ -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());