Commit 04f1b722420 for nodejs
commit 04f1b722420d1eaf7e98fc01368da3a774ac90fb
Author: James M Snell <jasnell@gmail.com>
Date: Sun Aug 30 18:05:28 2026 +0000
stream: correct strict pending-write behavior & cancellation
Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: https://github.com/nodejs/node/pull/66079
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md
index caf39715dca..522ae8db5ed 100644
--- a/doc/api/stream_iter.md
+++ b/doc/api/stream_iter.md
@@ -1583,8 +1583,9 @@ added:
> Stability: 1 - Experimental
-* `readable` {stream.Readable|Object} A classic Readable stream or any object
- with `read()`, `on()`, and `off()` methods.
+* `readable` {stream.Readable|Object} A classic Readable stream or a compatible
+ object with `read()`, `pipe()`, `destroy()`, `on()`, and `removeListener()`
+ methods.
* Returns: {AsyncIterable} whose chunks fulfill with {Uint8Array\[]}
Converts a classic Readable stream (or duck-typed equivalent) into a
@@ -1593,8 +1594,8 @@ stream/iter async iterable source that can be passed to [`from()`][],
If the object implements the [`toAsyncStreamable`][] protocol (as
`stream.Readable` does), that protocol is used. Otherwise, the function
-duck-types on `read()`, `on()`, and `off()` (EventEmitter) and wraps the
-stream with a batched async iterator.
+duck-types on `read()`, `pipe()`, `destroy()`, `on()`, and `removeListener()`
+(EventEmitter) and wraps the stream with a batched async iterator.
The result is cached per instance -- calling `fromReadable()` twice with the
same stream returns the same iterable.
@@ -1639,13 +1640,14 @@ added:
> Stability: 1 - Experimental
-* `writable` {stream.Writable|Object} A classic Writable stream or any object
- with `write()` and `on()` methods.
+* `writable` {stream.Writable|Object} A classic Writable stream or a compatible
+ object with `write()`, `end()`, `destroy()`, `on()`, and `removeListener()`
+ methods.
* `options` {Object}
* `backpressure` {string} Backpressure policy. **Default:** `'strict'`.
- * `'strict'` -- writes are rejected when the buffer is full. Catches
- callers that ignore backpressure.
- * `'unbounded'` -- writes wait for drain when the buffer is full. Recommended
+ * `'strict'` -- one write may wait while the buffer is full. Further writes
+ are rejected until it is accepted or canceled.
+ * `'unbounded'` -- writes are queued while the buffer is full. Recommended
for use with [`pipeTo()`][].
* `'drop-newest'` -- writes are silently discarded when the buffer is full.
* `'drop-oldest'` -- **not supported**. Throws `ERR_INVALID_ARG_VALUE`.
@@ -1657,8 +1659,9 @@ destination.
Since all writes on a classic Writable are fundamentally asynchronous,
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.
+return `false` or `-1`, deferring to the async path. A queued `write()` or
+`writev()` can be canceled with its `options.signal` before it reaches the
+classic Writable.
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.
@@ -1817,6 +1820,10 @@ 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.
+Destroying the Writable before successful completion calls `writer.fail()`.
+If `fail()` is unavailable, `Symbol.dispose` or `Symbol.asyncDispose` is used
+when implemented by the Writer.
+
The Writable uses the default classic stream `highWaterMark`. Classic stream
backpressure bounds writes waiting to reach the underlying Writer, while the
Writer controls completion of the active `_write()` or `_writev()` operation.
diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js
index d1faf2cdb88..6b5014291f3 100644
--- a/lib/internal/streams/iter/classic.js
+++ b/lib/internal/streams/iter/classic.js
@@ -63,6 +63,7 @@ const {
const {
convertChunks,
getWriterSignal,
+ onSignalAbort,
validateBackpressure,
toWriterUint8Array,
} = require('internal/streams/iter/utils');
@@ -70,6 +71,7 @@ const {
const { Buffer } = require('buffer');
const destroyImpl = require('internal/streams/destroy');
const { isError } = require('internal/util');
+const { RingBuffer } = require('internal/streams/iter/ringbuffer');
// Classic stream error channels require a truthy Error object.
function toClassicError(reason, reasonMap) {
@@ -89,6 +91,34 @@ function toClassicError(reason, reasonMap) {
return error;
}
+function raceWithSignal(promise, signal) {
+ if (signal === undefined) return promise;
+ if (signal.aborted) return PromiseReject(signal.reason);
+
+ const {
+ promise: signaledPromise,
+ resolve,
+ reject,
+ } = PromiseWithResolvers();
+ const onAbort = () => reject(signal.reason);
+ signal.addEventListener('abort', onAbort, {
+ __proto__: null,
+ once: true,
+ });
+ PromisePrototypeThen(
+ promise,
+ (value) => {
+ signal.removeEventListener('abort', onAbort);
+ resolve(value);
+ },
+ (reason) => {
+ signal.removeEventListener('abort', onAbort);
+ reject(reason);
+ },
+ );
+ return signaledPromise;
+}
+
// Lazy-loaded to avoid circular dependencies. Readable and Writable
// both require this module's parent, so we defer the require.
let Readable;
@@ -212,7 +242,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
(stream._readableState?.autoDestroy)) {
destroyImpl.destroyer(stream, null);
} else {
- stream.off('readable', next);
+ stream.removeListener('readable', next);
cleanup();
}
}
@@ -222,10 +252,11 @@ async function* createBatchedAsyncIterator(stream, normalize) {
* Convert a classic Readable (or duck-type) to a stream/iter async iterable.
*
* If the object implements the toAsyncStreamable protocol, delegates to it.
- * Otherwise, duck-type checks for read() + EventEmitter (on/off) and
+ * Otherwise, duck-type checks for read(), pipe(), destroy(), and EventEmitter
+ * on()/removeListener() methods and
* wraps with a batched async iterator.
* @param {object} readable - A classic Readable or duck-type with
- * read() and on()/off() methods.
+ * read(), pipe(), destroy(), on(), and removeListener() methods.
* @returns {AsyncIterable<Uint8Array[]>} A stream/iter async iterable source.
*/
function fromReadable(readable) {
@@ -246,7 +277,10 @@ function fromReadable(readable) {
// Duck-type path: object has read() and EventEmitter methods.
if (typeof readable.read !== 'function' ||
- typeof readable.on !== 'function') {
+ typeof readable.pipe !== 'function' ||
+ typeof readable.destroy !== 'function' ||
+ typeof readable.on !== 'function' ||
+ typeof readable.removeListener !== 'function') {
throw new ERR_INVALID_ARG_TYPE('readable', 'Readable', readable);
}
@@ -452,7 +486,8 @@ const fromWritableCache = new SafeWeakMap();
/**
* Create a stream/iter Writer adapter from a classic Writable (or duck-type).
*
- * Duck-type requirements: write() and on()/off() methods.
+ * Duck-type requirements: write(), end(), destroy(), on(), and
+ * removeListener() methods.
* Falls back to sensible defaults for missing properties like
* writableHighWaterMark, writableLength, writableObjectMode.
* @param {object} writable - A classic Writable or duck-type.
@@ -464,7 +499,10 @@ const fromWritableCache = new SafeWeakMap();
function fromWritable(writable, options = kNullPrototype) {
if (writable == null ||
typeof writable.write !== 'function' ||
- typeof writable.on !== 'function') {
+ typeof writable.end !== 'function' ||
+ typeof writable.destroy !== 'function' ||
+ typeof writable.on !== 'function' ||
+ typeof writable.removeListener !== 'function') {
throw new ERR_INVALID_ARG_TYPE('writable', 'Writable', writable);
}
@@ -513,94 +551,306 @@ function fromWritable(writable, options = kNullPrototype) {
let totalBytes = 0;
let errored = false;
let error;
+ let ending = false;
+ let endStarted = false;
let pendingEnd;
-
- // Waiters pending on backpressure resolution (block policy only).
- // Multiple un-awaited writes can each add a waiter, so this must be
- // a list. A single persistent 'drain' listener and 'error' listener
- // (installed once lazily) resolve or reject all waiters to avoid
- // accumulating per-write listeners on the stream.
- let waiters = [];
- let listenersInstalled = false;
- let onDrain;
- let onError;
-
- function installListeners() {
- if (listenersInstalled) return;
- listenersInstalled = true;
- onDrain = () => {
- const pending = waiters;
- waiters = [];
- for (let i = 0; i < pending.length; i++) {
- pending[i].resolve();
- }
- };
- onError = (err) => {
- const pending = waiters;
- waiters = [];
- for (let i = 0; i < pending.length; i++) {
- pending[i].reject(err);
- }
- };
+ let needsDrain = false;
+ let finished = false;
+ let pendingWrites = new RingBuffer();
+ let drainWaiters = [];
+ let drainListenerInstalled = false;
+ let terminalListenersInstalled = false;
+
+ function installDrainListener() {
+ if (drainListenerInstalled) return;
+ drainListenerInstalled = true;
writable.on('drain', onDrain);
- writable.on('error', onError);
}
- // Reject all pending waiters and remove the drain/error listeners.
- function cleanup(err, preserveReason = false) {
- const pending = waiters;
- waiters = [];
+ function removeDrainListenerIfIdle() {
+ if (!drainListenerInstalled ||
+ pendingWrites.length !== 0 ||
+ drainWaiters.length !== 0) {
+ return;
+ }
+ drainListenerInstalled = false;
+ writable.removeListener('drain', onDrain);
+ }
+
+ function removeTerminalListeners() {
+ if (!terminalListenersInstalled) return;
+ terminalListenersInstalled = false;
+ writable.removeListener('error', onError);
+ writable.removeListener('finish', onFinish);
+ writable.removeListener('close', onClose);
+ }
+
+ function cleanupPendingSignal(entry) {
+ if (entry.signal === undefined) return;
+ entry.signal.removeEventListener('abort', entry.onAbort);
+ entry.signal = undefined;
+ entry.onAbort = undefined;
+ }
+
+ function cleanup(
+ reason,
+ preserveReason = false,
+ keepTerminalListeners = false,
+ ) {
+ const pending = drainWaiters;
+ drainWaiters = [];
for (let i = 0; i < pending.length; i++) {
if (!preserveReason &&
- (err === undefined || err === null) &&
+ (reason === undefined || reason === null) &&
pending[i].close !== undefined) {
pending[i].close();
} else {
- pending[i].reject(preserveReason ? err : err ?? new AbortError());
+ pending[i].reject(
+ preserveReason ? reason : reason ?? new AbortError());
}
}
- if (!listenersInstalled) return;
- listenersInstalled = false;
- writable.removeListener('drain', onDrain);
- writable.removeListener('error', onError);
- }
- function waitForDrain() {
- const { promise, resolve, reject } = PromiseWithResolvers();
- ArrayPrototypePush(waiters, { __proto__: null, resolve, reject });
- installListeners();
- return promise;
+ const writes = pendingWrites;
+ pendingWrites = new RingBuffer();
+ while (writes.length !== 0) {
+ const entry = writes.shift();
+ cleanupPendingSignal(entry);
+ entry.reject(
+ preserveReason ? reason : reason ?? new AbortError());
+ }
+
+ if (drainListenerInstalled) {
+ drainListenerInstalled = false;
+ writable.removeListener('drain', onDrain);
+ }
+ if (!keepTerminalListeners) removeTerminalListeners();
}
- function isWritable() {
+ function isUnderlyingWritable() {
+ syncWritableError();
// Duck-typed streams may not have these properties -- treat missing
// as false (i.e., writable is still open).
- return !errored &&
+ return !errored && !finished &&
!(writable.destroyed ?? false) &&
!(writable.writableFinished ?? false) &&
!(writable.writableEnded ?? false);
}
+ function isWritable() {
+ return !ending && isUnderlyingWritable();
+ }
+
function isFull() {
- return (writable.writableLength ?? 0) >= hwm;
+ return needsDrain ||
+ (writable.writableNeedDrain ?? false) ||
+ (hwm > 0 && (writable.writableLength ?? 0) >= hwm);
}
function writeChunks(chunks) {
let ok = true;
for (let i = 0; i < chunks.length; i++) {
const bytes = chunks[i];
+ if (!writable.write(bytes)) {
+ needsDrain = true;
+ ok = false;
+ }
totalBytes += TypedArrayPrototypeGetByteLength(bytes);
- ok = writable.write(bytes);
}
return ok;
}
+ function writeBatch(chunks) {
+ if (typeof writable.cork !== 'function' ||
+ typeof writable.uncork !== 'function') {
+ return writeChunks(chunks);
+ }
+ writable.cork();
+ try {
+ return writeChunks(chunks);
+ } finally {
+ writable.uncork();
+ }
+ }
+
+ function maybeStartEnd() {
+ if (ending && pendingWrites.length === 0) startEnd();
+ }
+
+ function settleDrainWaiters(value) {
+ const pending = drainWaiters;
+ drainWaiters = [];
+ for (let i = 0; i < pending.length; i++) {
+ pending[i].resolve(value);
+ }
+ }
+
+ function flushPendingWrites() {
+ while (pendingWrites.length !== 0 &&
+ isUnderlyingWritable() &&
+ !isFull()) {
+ const entry = pendingWrites.shift();
+ cleanupPendingSignal(entry);
+ let ok;
+ try {
+ ok = writeBatch(entry.chunks);
+ } catch (reason) {
+ entry.reject(reason);
+ continue;
+ }
+ if (errored) {
+ entry.reject(error);
+ break;
+ }
+ entry.resolve();
+ if (!ok) break;
+ }
+
+ if (pendingWrites.length !== 0) installDrainListener();
+ if (pendingWrites.length === 0) {
+ if (ending) {
+ settleDrainWaiters(false);
+ } else if (isUnderlyingWritable() && !isFull()) {
+ settleDrainWaiters(true);
+ }
+ }
+ removeDrainListenerIfIdle();
+ maybeStartEnd();
+ }
+
+ function queueWrite(chunks, signal) {
+ const { promise, resolve, reject } = PromiseWithResolvers();
+ const entry = {
+ __proto__: null,
+ chunks,
+ resolve,
+ reject,
+ signal: undefined,
+ onAbort: undefined,
+ };
+ pendingWrites.push(entry);
+ installDrainListener();
+
+ if (signal !== undefined) {
+ entry.signal = signal;
+ entry.onAbort = () => {
+ const index = pendingWrites.indexOf(entry);
+ if (index === -1) return;
+ pendingWrites.removeAt(index);
+ cleanupPendingSignal(entry);
+ reject(signal.reason);
+ removeDrainListenerIfIdle();
+ flushPendingWrites();
+ };
+ onSignalAbort(signal, entry.onAbort);
+ }
+
+ return promise;
+ }
+
+ function onDrain() {
+ needsDrain = false;
+ flushPendingWrites();
+ }
+
+ function finishWithError(reason, keepTerminalListeners = false) {
+ if (errored) return;
+ errored = true;
+ error = reason;
+ ending = false;
+ const end = pendingEnd;
+ pendingEnd = undefined;
+ cleanup(reason, true, keepTerminalListeners);
+ end?.reject(reason);
+ }
+
+ function syncWritableError() {
+ if (errored) return;
+ const reason = writable.errored;
+ if (reason !== undefined && reason !== null) {
+ finishWithError(reason, true);
+ }
+ }
+
+ function onError(reason) {
+ if (errored) {
+ removeTerminalListeners();
+ return;
+ }
+ finishWithError(reason);
+ }
+
+ function onFinish() {
+ if (errored || finished) return;
+ finished = true;
+ ending = false;
+ const end = pendingEnd;
+ pendingEnd = undefined;
+ cleanup();
+ end?.resolve(totalBytes);
+ }
+
+ function onClose() {
+ syncWritableError();
+ if (errored) {
+ removeTerminalListeners();
+ return;
+ }
+ if (finished || (writable.writableFinished ?? false)) {
+ onFinish();
+ removeTerminalListeners();
+ return;
+ }
+
+ finished = true;
+ ending = false;
+ const end = pendingEnd;
+ pendingEnd = undefined;
+ const reason = new AbortError();
+ cleanup(reason, true);
+ end?.reject(reason);
+ }
+
+ function startEnd() {
+ if (endStarted || pendingEnd === undefined) return;
+ endStarted = true;
+ const end = pendingEnd;
+
+ try {
+ if (!(writable.writableEnded ?? false)) writable.end();
+ syncWritableError();
+ if (!errored && (writable.writableFinished ?? false)) onFinish();
+ } catch (reason) {
+ pendingEnd = undefined;
+ ending = false;
+ errored = true;
+ error = reason;
+ cleanup(reason, true, true);
+ end.reject(reason);
+ try {
+ writable.destroy(toClassicError(reason));
+ } catch {
+ removeTerminalListeners();
+ }
+ }
+ }
+
+ writable.on('error', onError);
+ writable.on('finish', onFinish);
+ writable.on('close', onClose);
+ terminalListenersInstalled = true;
+ syncWritableError();
+ if (!errored && (writable.writableFinished ?? false)) {
+ onFinish();
+ } else if (!errored && (writable.destroyed ?? false)) {
+ onClose();
+ }
+
const writer = {
__proto__: null,
get canWrite() {
if (!isWritable()) return null;
- return (writable.writableLength ?? 0) < hwm;
+ return pendingWrites.length === 0 && !isFull();
},
writeSync(chunk) {
@@ -613,99 +863,74 @@ function fromWritable(writable, options = kNullPrototype) {
return false;
},
- // Backpressure semantics: write() resolves when the data is accepted
- // into the Writable's internal buffer, NOT when _write() has flushed
- // it to the underlying resource. This matches the Writer spec -- the
- // PushWriter resolves on buffer acceptance too. Classic Writable flow
- // control works the same way: write rapidly until write() returns
- // false, then wait for 'drain'. The _write callback is involved in
- // backpressure indirectly -- 'drain' fires after callbacks drain the
- // buffer below highWaterMark. Per-write errors from _write surface
- // as 'error' events caught by our generic error handler, rejecting
- // the next pending operation rather than the already-resolved one.
- //
- // The options.signal parameter from the Writer interface is validated but
- // otherwise ignored. Classic stream.Writable has no per-write abort signal
- // support; cancellation should be handled at the pipeline level instead.
+ // Writes made with available capacity resolve when accepted by the classic
+ // Writable. Writes made while it is full stay in this adapter until drain,
+ // so operation signals can cancel them before they are committed.
write(chunk, options) {
const bytes = toWriterUint8Array(chunk);
- getWriterSignal(options);
+ const signal = getWriterSignal(options);
+ syncWritableError();
if (errored) return PromiseReject(error);
if (!isWritable()) {
return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
}
+ if (signal?.aborted) return PromiseReject(signal.reason);
- if (backpressure === 'strict' && isFull()) {
- return PromiseReject(new ERR_INVALID_STATE.RangeError(
- 'Backpressure violation: buffer is full. ' +
- 'Await each write() call to respect backpressure.'));
- }
-
- if (backpressure === 'drop-newest' && isFull()) {
- // Silently discard. Still count bytes for consistency with
- // PushWriter, which counts dropped bytes in totalBytes.
- totalBytes += TypedArrayPrototypeGetByteLength(bytes);
- return PromiseResolve();
+ if (pendingWrites.length !== 0 || isFull()) {
+ if (backpressure === 'drop-newest') {
+ totalBytes += TypedArrayPrototypeGetByteLength(bytes);
+ return PromiseResolve();
+ }
+ if (backpressure === 'strict' && pendingWrites.length !== 0) {
+ return PromiseReject(new ERR_INVALID_STATE.RangeError(
+ 'Backpressure violation: too many pending writes. ' +
+ 'Await each write() call to respect backpressure.'));
+ }
+ return queueWrite([bytes], signal);
}
- totalBytes += TypedArrayPrototypeGetByteLength(bytes);
- const ok = writable.write(bytes);
- if (ok) return PromiseResolve();
-
- // backpressure === 'unbounded' (or strict with room that filled on
- // this write -- writable.write() accepted the data but returned
- // false indicating the buffer is now at/over hwm).
- if (backpressure === 'unbounded') {
- return waitForDrain();
+ try {
+ writeChunks([bytes]);
+ } catch (reason) {
+ return PromiseReject(reason);
}
-
- // strict: the write was accepted (there was room before writing)
- // but the buffer is now full. Resolve -- the *next* write will
- // be rejected if the caller ignores backpressure.
+ if (errored) return PromiseReject(error);
return PromiseResolve();
},
writev(chunks, options) {
chunks = convertChunks(chunks);
- getWriterSignal(options);
+ const signal = getWriterSignal(options);
+ syncWritableError();
if (errored) return PromiseReject(error);
if (!isWritable()) {
return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
}
+ if (signal?.aborted) return PromiseReject(signal.reason);
+ if (chunks.length === 0) return PromiseResolve();
- if (backpressure === 'strict' && isFull()) {
- return PromiseReject(new ERR_INVALID_STATE.RangeError(
- 'Backpressure violation: buffer is full. ' +
- 'Await each write() call to respect backpressure.'));
- }
-
- if (backpressure === 'drop-newest' && isFull()) {
- // Discard entire batch.
- for (let i = 0; i < chunks.length; i++) {
- totalBytes += TypedArrayPrototypeGetByteLength(chunks[i]);
+ if (pendingWrites.length !== 0 || isFull()) {
+ if (backpressure === 'drop-newest') {
+ for (let i = 0; i < chunks.length; i++) {
+ totalBytes += TypedArrayPrototypeGetByteLength(chunks[i]);
+ }
+ return PromiseResolve();
}
- return PromiseResolve();
- }
-
- let ok = true;
- if (typeof writable.cork === 'function' &&
- typeof writable.uncork === 'function') {
- writable.cork();
- try {
- ok = writeChunks(chunks);
- } finally {
- writable.uncork();
+ if (backpressure === 'strict' && pendingWrites.length !== 0) {
+ return PromiseReject(new ERR_INVALID_STATE.RangeError(
+ 'Backpressure violation: too many pending writes. ' +
+ 'Await each write() call to respect backpressure.'));
}
- } else {
- ok = writeChunks(chunks);
+ return queueWrite(chunks, signal);
}
- if (ok) return PromiseResolve();
-
- if (backpressure === 'unbounded') {
- return waitForDrain();
+ try {
+ writeBatch(chunks);
+ } catch (reason) {
+ return PromiseReject(reason);
}
+ if (errored) return PromiseReject(error);
return PromiseResolve();
},
@@ -713,50 +938,27 @@ function fromWritable(writable, options = kNullPrototype) {
return -1;
},
- // options.signal is validated but otherwise ignored for the same reason as
- // write().
end(options) {
- getWriterSignal(options);
+ const signal = getWriterSignal(options);
+ syncWritableError();
if (errored) return PromiseReject(error);
- if (pendingEnd) return pendingEnd.promise;
+ if (signal?.aborted) return PromiseReject(signal.reason);
+ if (pendingEnd) return raceWithSignal(pendingEnd.promise, signal);
if ((writable.writableFinished ?? false) ||
(writable.destroyed ?? false)) {
cleanup();
- return PromiseResolve(totalBytes);
+ return raceWithSignal(PromiseResolve(totalBytes), signal);
}
pendingEnd = PromiseWithResolvers();
- const { promise, resolve, reject } = pendingEnd;
-
- 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.
- }
- }
-
- return promise;
+ const { promise } = pendingEnd;
+ ending = true;
+ maybeStartEnd();
+ return raceWithSignal(promise, signal);
},
fail(reason) {
+ syncWritableError();
if (errored ||
(writable.writableFinished ?? false) ||
(writable.destroyed ?? false)) {
@@ -764,11 +966,15 @@ function fromWritable(writable, options = kNullPrototype) {
}
errored = true;
error = reason;
+ ending = false;
pendingEnd?.reject(reason);
pendingEnd = undefined;
- cleanup(reason, true);
- if (typeof writable.destroy === 'function') {
+ cleanup(reason, true, true);
+ try {
writable.destroy(toClassicError(reason));
+ } catch (destroyError) {
+ removeTerminalListeners();
+ throw destroyError;
}
},
@@ -788,17 +994,17 @@ function fromWritable(writable, options = kNullPrototype) {
// drainableProtocol
writer[drainableProtocol] = function() {
if (!isWritable()) return null;
- if ((writable.writableLength ?? 0) < hwm) {
+ if (pendingWrites.length === 0 && !isFull()) {
return PromiseResolve(true);
}
const { promise, resolve, reject } = PromiseWithResolvers();
- ArrayPrototypePush(waiters, {
+ ArrayPrototypePush(drainWaiters, {
__proto__: null,
- resolve() { resolve(true); },
+ resolve,
reject,
close() { resolve(false); },
});
- installListeners();
+ installDrainListener();
return promise;
};
@@ -832,8 +1038,11 @@ function toWritable(writer) {
typeof writer.writevSync === 'function';
const hasEnd = typeof writer.end === 'function';
const hasEndSync = hasEnd &&
- typeof writer.endSync === 'function';
+ typeof writer.endSync === 'function';
const hasFail = typeof writer.fail === 'function';
+ const hasDispose = typeof writer[SymbolDispose] === 'function';
+ const hasAsyncDispose = typeof writer[SymbolAsyncDispose] === 'function';
+ let writerEnded = false;
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).
@@ -895,6 +1104,7 @@ function toWritable(writer) {
function _final(cb) {
if (!hasEnd) {
+ writerEnded = true;
queueMicrotask(cb);
return;
}
@@ -902,6 +1112,7 @@ function toWritable(writer) {
try {
const result = writer.endSync();
if (result >= 0) {
+ writerEnded = true;
queueMicrotask(cb);
return;
}
@@ -913,7 +1124,10 @@ function toWritable(writer) {
}
try {
PromisePrototypeThen(
- writer.end(), () => cb(),
+ writer.end(), () => {
+ writerEnded = true;
+ cb();
+ },
(err) => cb(toClassicError(err, classicErrorReasons)));
} catch (err) {
cb(toClassicError(err, classicErrorReasons));
@@ -921,15 +1135,37 @@ function toWritable(writer) {
}
function _destroy(err, cb) {
- if (err && hasFail) {
- const wrapped = classicErrorReasons.get(err);
- classicErrorReasons.delete(err);
- try {
- writer.fail(wrapped === undefined ? err : wrapped.reason);
- } catch (error) {
- cb(err || toClassicError(error, classicErrorReasons));
- return;
+ if (!err && writerEnded) {
+ cb();
+ return;
+ }
+
+ let result;
+ try {
+ if (hasFail) {
+ if (err) {
+ const wrapped = classicErrorReasons.get(err);
+ classicErrorReasons.delete(err);
+ writer.fail(wrapped === undefined ? err : wrapped.reason);
+ } else {
+ writer.fail();
+ }
+ } else if (hasDispose) {
+ writer[SymbolDispose]();
+ } else if (hasAsyncDispose) {
+ result = writer[SymbolAsyncDispose]();
}
+ } catch (error) {
+ cb(err || toClassicError(error, classicErrorReasons));
+ return;
+ }
+
+ if (result !== undefined) {
+ PromisePrototypeThen(
+ PromiseResolve(result),
+ () => cb(err),
+ (error) => cb(err || toClassicError(error, classicErrorReasons)));
+ return;
}
cb(err);
}
diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js
index 871c891a653..3f311ec8c46 100644
--- a/test/parallel/test-stream-iter-validation.js
+++ b/test/parallel/test-stream-iter-validation.js
@@ -5,7 +5,7 @@ const common = require('../common');
const assert = require('assert');
const { Writable } = require('stream');
const {
- from, fromSync, pull, pullSync, pipeTo, fromWritable,
+ from, fromSync, pull, pullSync, pipeTo, fromReadable, fromWritable,
push, duplex, broadcast, Broadcast, share, shareSync,
Share, SyncShare,
bytes, bytesSync, text, textSync,
@@ -190,6 +190,15 @@ assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG
// Broadcast.from rejects non-streamable input
assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' });
+assert.throws(
+ () => fromWritable({ write() {}, on() {} }),
+ { code: 'ERR_INVALID_ARG_TYPE' },
+);
+assert.throws(
+ () => fromReadable({ read() {}, on() {} }),
+ { code: 'ERR_INVALID_ARG_TYPE' },
+);
+
// fromWritable Writer options.signal must be AbortSignal
{
const writable = new Writable({
diff --git a/test/parallel/test-stream-iter-writable-from.js b/test/parallel/test-stream-iter-writable-from.js
index 5757f3126e7..12ea80df9fb 100644
--- a/test/parallel/test-stream-iter-writable-from.js
+++ b/test/parallel/test-stream-iter-writable-from.js
@@ -542,15 +542,30 @@ async function testMinimalWriter() {
assert.strictEqual(Buffer.concat(chunks).toString(), 'minimal');
}
+async function testNormalEndWithoutWriterEndDoesNotFail() {
+ const writable = toWritable({
+ write(chunk) { return Promise.resolve(); },
+ fail: common.mustNotCall(),
+ });
+ const closed = once(writable, 'close');
+
+ writable.end();
+ await closed;
+}
+
// =============================================================================
-// Destroy without error does not call fail()
+// Destroy without error calls fail()
// =============================================================================
async function testDestroyWithoutError() {
let failCalled = false;
const writer = {
write(chunk) { return Promise.resolve(); },
- fail() { failCalled = true; },
+ fail: common.mustCall(function(reason) {
+ assert.strictEqual(arguments.length, 0);
+ assert.strictEqual(reason, undefined);
+ failCalled = true;
+ }),
};
const writable = toWritable(writer);
@@ -558,7 +573,20 @@ async function testDestroyWithoutError() {
await setTimeout(10);
- assert.ok(!failCalled, 'fail should not be called on clean destroy');
+ assert.ok(failCalled, 'fail should be called on clean destroy');
+}
+
+async function testDestroyUsesDisposeFallback() {
+ let disposed = false;
+ const writable = toWritable({
+ write(chunk) { return Promise.resolve(); },
+ [Symbol.dispose]() { disposed = true; },
+ });
+
+ writable.destroy();
+ await setTimeout(10);
+
+ assert.strictEqual(disposed, true);
}
// =============================================================================
@@ -733,6 +761,7 @@ Promise.all([
testFinalDelegatesToEnd(),
testDestroyDelegatesToFail(),
testDestroyWithoutError(),
+ testDestroyUsesDisposeFallback(),
testDestroyWithError(),
testDestroyWithoutFail(),
testWriteErrorPropagation(),
@@ -744,4 +773,5 @@ Promise.all([
testSequentialWrites(),
testSyncCallbackDeferred(),
testMinimalWriter(),
+ testNormalEndWithoutWriterEndDoesNotFail(),
]).then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-writable-interop.js b/test/parallel/test-stream-iter-writable-interop.js
index 9940ffc34c9..52824b2168b 100644
--- a/test/parallel/test-stream-iter-writable-interop.js
+++ b/test/parallel/test-stream-iter-writable-interop.js
@@ -6,6 +6,7 @@
const common = require('../common');
const assert = require('assert');
+const { EventEmitter } = require('events');
const { Writable } = require('stream');
const { setImmediate } = require('timers/promises');
const {
@@ -120,7 +121,8 @@ async function testBlockErrorRejectsPendingWrite() {
const writer = fromWritable(writable, { backpressure: 'unbounded' });
- // First write fills the buffer, waits for drain
+ // The first write fills the buffer; the second waits for drain.
+ await writer.write('a');
const writePromise = writer.write('data that will block');
// Destroy with error while write is pending
@@ -129,15 +131,264 @@ async function testBlockErrorRejectsPendingWrite() {
await assert.rejects(writePromise, { message: 'stream broke' });
}
+async function testErrorBeforeBackpressureIsStored() {
+ const writable = new Writable({
+ write(chunk, enc, cb) { cb(); },
+ });
+ const writer = fromWritable(writable);
+ const reason = new Error('early stream error');
+ const closed = new Promise((resolve) => writable.once('close', resolve));
+
+ writable.destroy(reason);
+ await assert.rejects(writer.write('late'), (error) => error === reason);
+ await closed;
+
+ await assert.rejects(writer.end(), (error) => error === reason);
+}
+
+async function testAlreadyErroredWritablePreservesReason() {
+ const writable = new Writable({
+ write(chunk, enc, cb) { cb(); },
+ });
+ const reason = new Error('existing stream error');
+ const closed = new Promise((resolve) => writable.once('close', resolve));
+ writable.destroy(reason);
+
+ const writer = fromWritable(writable);
+ await assert.rejects(writer.write('late'), (error) => error === reason);
+ await closed;
+}
+
+async function testCleanDestroyRejectsQueuedOperations() {
+ const writable = new Writable({
+ highWaterMark: 1,
+ write(chunk, enc, cb) {},
+ });
+ const writer = fromWritable(writable);
+
+ await writer.write('a');
+ const pending = writer.write('b');
+ const ending = writer.end();
+ writable.destroy();
+
+ await assert.rejects(pending, { name: 'AbortError' });
+ await assert.rejects(ending, { name: 'AbortError' });
+}
+
+async function testPreAbortedWriteSignalsDoNotCommit() {
+ let writes = 0;
+ const writable = new Writable({
+ write(chunk, enc, cb) {
+ writes++;
+ cb();
+ },
+ });
+ const writer = fromWritable(writable);
+ const reason = { canceled: true };
+ const signal = AbortSignal.abort(reason);
+
+ await assert.rejects(
+ writer.write('a', { signal }),
+ (error) => error === reason,
+ );
+ await assert.rejects(
+ writer.writev([new Uint8Array([98])], { signal }),
+ (error) => error === reason,
+ );
+ assert.strictEqual(writes, 0);
+ await writer.end();
+}
+
+async function testPendingWriteSignalRemovesOperation() {
+ const callbacks = [];
+ const chunks = [];
+ const writable = new Writable({
+ highWaterMark: 1,
+ write(chunk, enc, cb) {
+ chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
+ },
+ });
+ const writer = fromWritable(writable);
+
+ await writer.write('a');
+ const controller = new AbortController();
+ const canceled = writer.write('b', { signal: controller.signal });
+ controller.abort('stop');
+ await assert.rejects(canceled, (reason) => reason === 'stop');
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a']);
+
+ // Cancellation frees the strict policy's single pending-operation slot.
+ const replacement = writer.write('c');
+ callbacks.shift()();
+ await replacement;
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a', 'c']);
+ const ending = writer.end();
+ callbacks.shift()();
+ await ending;
+}
+
+async function testZeroHighWaterMarkAcceptsFirstWrite() {
+ const callbacks = [];
+ const chunks = [];
+ const writable = new Writable({
+ highWaterMark: 0,
+ write(chunk, enc, cb) {
+ chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
+ },
+ });
+ const writer = fromWritable(writable);
+
+ await writer.write('a');
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a']);
+ const pending = writer.write('b');
+ callbacks.shift()();
+ await pending;
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a', 'b']);
+ const ending = writer.end();
+ callbacks.shift()();
+ await ending;
+}
+
+async function testDuckWritableLatchesWriteBackpressure() {
+ const writable = new EventEmitter();
+ const chunks = [];
+ writable.write = (chunk) => {
+ chunks.push(Buffer.from(chunk));
+ return false;
+ };
+ writable.end = () => {
+ writable.writableFinished = true;
+ writable.emit('finish');
+ };
+ writable.destroy = () => {
+ writable.destroyed = true;
+ writable.emit('close');
+ };
+
+ const writer = fromWritable(writable);
+ await writer.write('a');
+ const pending = writer.write('b');
+ await setImmediate();
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a']);
+ writable.emit('drain');
+ await pending;
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a', 'b']);
+ await writer.end();
+}
+
+async function testPendingWritesRemainFifoDuringDrain() {
+ const callbacks = [];
+ const chunks = [];
+ const writable = new Writable({
+ highWaterMark: 1,
+ write(chunk, enc, cb) {
+ chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
+ },
+ });
+ const writer = fromWritable(writable, { backpressure: 'unbounded' });
+
+ await writer.write('a');
+ let reentrant;
+ writable.once('drain', () => {
+ reentrant = writer.write('c');
+ });
+ const pending = writer.write('b');
+ callbacks.shift()();
+ await pending;
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['a', 'b']);
+ callbacks.shift()();
+ await reentrant;
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()),
+ ['a', 'b', 'c']);
+ const ending = writer.end();
+ callbacks.shift()();
+ await ending;
+}
+
+async function testOndrainWaitsForEntirePendingQueue() {
+ const callbacks = [];
+ const writable = new Writable({
+ highWaterMark: 1,
+ write(chunk, enc, cb) { callbacks.push(cb); },
+ });
+ const writer = fromWritable(writable, { backpressure: 'unbounded' });
+
+ await writer.write('a');
+ const second = writer.write('b');
+ const third = writer.write('c');
+ let drained = false;
+ const draining = ondrain(writer).then((value) => {
+ drained = value;
+ });
+
+ callbacks.shift()();
+ await second;
+ await setImmediate();
+ assert.strictEqual(drained, false);
+ callbacks.shift()();
+ await third;
+ await setImmediate();
+ assert.strictEqual(drained, false);
+ callbacks.shift()();
+ await draining;
+ assert.strictEqual(drained, true);
+ await writer.end();
+}
+
+async function testEndSignal() {
+ const callbacks = [];
+ const writable = new Writable({
+ highWaterMark: 1,
+ write(chunk, enc, cb) { callbacks.push(cb); },
+ });
+ const writer = fromWritable(writable);
+ const preAborted = AbortSignal.abort('before end');
+
+ await assert.rejects(
+ writer.end({ signal: preAborted }),
+ (reason) => reason === 'before end',
+ );
+ await writer.write('a');
+ const pending = writer.write('b');
+ const controller = new AbortController();
+ const signaledEnd = writer.end({ signal: controller.signal });
+ const completedEnd = writer.end();
+ controller.abort('during end');
+ await assert.rejects(signaledEnd, (reason) => reason === 'during end');
+
+ callbacks.shift()();
+ await pending;
+ callbacks.shift()();
+ await completedEnd;
+}
+
+async function testDirectWriteThrowRejects() {
+ const writable = new EventEmitter();
+ const reason = new Error('duck write failed');
+ writable.write = () => { throw reason; };
+ writable.end = () => {};
+ writable.destroy = () => {};
+ const writer = fromWritable(writable);
+
+ await assert.rejects(writer.write('a'), (error) => error === reason);
+ writer.fail();
+}
+
// =============================================================================
-// strict: rejects when buffer is full
+// strict: allows one pending write when the buffer is full
// =============================================================================
async function testStrictRejectsWhenFull() {
+ const callbacks = [];
+ const chunks = [];
const writable = new Writable({
highWaterMark: 5,
write(chunk, enc, cb) {
- // Never call cb -- data stays buffered
+ chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
},
});
@@ -146,22 +397,41 @@ async function testStrictRejectsWhenFull() {
// First write fills the buffer (5 bytes = hwm)
await writer.write('12345');
- // Second write should reject -- buffer is full
+ const pending = writer.write('more');
+ await setImmediate();
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['12345']);
+
+ // One operation is pending, so a further write violates strict policy.
await assert.rejects(
- writer.write('more'),
+ writer.write('overflow'),
{ code: 'ERR_INVALID_STATE' },
);
+
+ callbacks.shift()();
+ await pending;
+ assert.deepStrictEqual(
+ chunks.map((chunk) => chunk.toString()), ['12345', 'more']);
+ const ending = writer.end();
+ callbacks.shift()();
+ await ending;
}
// =============================================================================
-// strict: writev rejects when buffer is full
+// strict: allows one pending writev when the buffer is full
// =============================================================================
async function testStrictWritevRejectsWhenFull() {
+ const callbacks = [];
+ const chunks = [];
const writable = new Writable({
highWaterMark: 5,
write(chunk, enc, cb) {
- // Never call cb
+ chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
+ },
+ writev(entries, cb) {
+ for (const { chunk } of entries) chunks.push(Buffer.from(chunk));
+ callbacks.push(cb);
},
});
@@ -170,14 +440,27 @@ async function testStrictWritevRejectsWhenFull() {
// Fill buffer
await writer.write('12345');
- // Writev should reject entire batch
+ const pending = writer.writev([
+ new TextEncoder().encode('a'),
+ new TextEncoder().encode('b'),
+ ]);
+ await setImmediate();
+ assert.deepStrictEqual(chunks.map((chunk) => chunk.toString()), ['12345']);
+
await assert.rejects(
writer.writev([
- new TextEncoder().encode('a'),
- new TextEncoder().encode('b'),
+ new TextEncoder().encode('overflow'),
]),
{ code: 'ERR_INVALID_STATE' },
);
+
+ callbacks.shift()();
+ await pending;
+ assert.deepStrictEqual(
+ chunks.map((chunk) => chunk.toString()), ['12345', 'a', 'b']);
+ const ending = writer.end();
+ callbacks.shift()();
+ await ending;
}
// =============================================================================
@@ -551,7 +834,13 @@ function testWritevInvalidChunksType() {
// =============================================================================
function testWritevInvalidChunkUncorks() {
- const writable = new Writable({ write(chunk, enc, cb) { cb(); } });
+ let writes = 0;
+ const writable = new Writable({
+ write(chunk, enc, cb) {
+ writes++;
+ cb();
+ },
+ });
const writer = fromWritable(writable);
assert.throws(
@@ -559,6 +848,7 @@ function testWritevInvalidChunkUncorks() {
{ code: 'ERR_INVALID_ARG_TYPE' },
);
assert.strictEqual(writable.writableCorked, 0);
+ assert.strictEqual(writes, 0);
}
// =============================================================================
@@ -588,7 +878,7 @@ async function testFailRejectsPendingWaiters() {
const writer = fromWritable(writable, { backpressure: 'unbounded' });
- // This write will block on drain
+ await writer.write('a');
const writePromise = writer.write('blocked data');
// fail() should reject the pending waiter, not orphan it
@@ -605,6 +895,7 @@ async function testFailPreservesReason() {
});
writable.on('error', common.mustCall((error) => { classicError = error; }));
const writer = fromWritable(writable, { backpressure: 'unbounded' });
+ await writer.write('a');
const pending = writer.write('blocked data');
const draining = ondrain(writer);
@@ -664,7 +955,7 @@ async function testDisposeRejectsPendingWaiters() {
const writer = fromWritable(writable, { backpressure: 'unbounded' });
- // This write will block on drain
+ await writer.write('a');
const writePromise = writer.write('blocked data');
writer[Symbol.dispose]();
@@ -712,6 +1003,17 @@ Promise.all([
testWriteNoDrain(),
testBlockWaitsForDrain(),
testBlockErrorRejectsPendingWrite(),
+ testErrorBeforeBackpressureIsStored(),
+ testAlreadyErroredWritablePreservesReason(),
+ testCleanDestroyRejectsQueuedOperations(),
+ testPreAbortedWriteSignalsDoNotCommit(),
+ testPendingWriteSignalRemovesOperation(),
+ testZeroHighWaterMarkAcceptsFirstWrite(),
+ testDuckWritableLatchesWriteBackpressure(),
+ testPendingWritesRemainFifoDuringDrain(),
+ testOndrainWaitsForEntirePendingQueue(),
+ testEndSignal(),
+ testDirectWriteThrowRejects(),
testStrictRejectsWhenFull(),
testStrictWritevRejectsWhenFull(),
testDropNewestDiscards(),