stream: validate fromWritable() options before cache

Validate options before returning a cached fromWritable() adapter so
invalid later options still throw.

Cache adapters by backpressure policy as well as Writable instance,
since the policy changes write behavior.

Fixes: https://github.com/nodejs/node/issues/63277

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: https://github.com/nodejs/node/pull/63278
Fixes: https://github.com/nodejs/node/issues/63277
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>
This commit is contained in:
Trivikram Kamat authored and GitHub committed 2026-05-15 20:03:31 +00:00
1 parent be7ded1693
commit 2d6cbea73a
3 files changed
+61 -7

No files matched your search

+3 -2
View File
@@ -1543,8 +1543,9 @@ 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.
The result is cached per instance -- calling `fromWritable()` twice with the
same stream returns the same Writer.
The result is cached per instance and backpressure policy -- calling
`fromWritable()` twice with the same stream and `backpressure` option returns
the same Writer.
For duck-typed streams that do not expose `writableHighWaterMark`,
`writableLength`, or similar properties, sensible defaults are used.
+13 -5
View File
@@ -21,6 +21,7 @@ const {
PromiseReject,
PromiseResolve,
PromiseWithResolvers,
SafeMap,
SafeWeakMap,
SymbolAsyncDispose,
SymbolAsyncIterator,
@@ -427,10 +428,6 @@ function fromWritable(writable, options = kNullPrototype) {
throw new ERR_INVALID_ARG_TYPE('writable', 'Writable', writable);
}
// Return cached adapter if available.
const cached = fromWritableCache.get(writable);
if (cached !== undefined) return cached;
validateObject(options, 'options');
const {
backpressure = 'strict',
@@ -459,6 +456,17 @@ function fromWritable(writable, options = kNullPrototype) {
'drop-oldest is not supported for classic stream.Writable');
}
// Return cached adapter if available. Backpressure policy changes writer
// behavior, so cache one adapter per policy.
let cachedByBackpressure = fromWritableCache.get(writable);
if (cachedByBackpressure !== undefined) {
const cached = cachedByBackpressure.get(backpressure);
if (cached !== undefined) return cached;
} else {
cachedByBackpressure = new SafeMap();
fromWritableCache.set(writable, cachedByBackpressure);
}
// Fall back to sensible defaults for duck-typed streams that may not
// expose the full stream.Writable property set.
const hwm = writable.writableHighWaterMark ?? 16384;
@@ -710,7 +718,7 @@ function fromWritable(writable, options = kNullPrototype) {
return promise;
};
fromWritableCache.set(writable, writer);
cachedByBackpressure.set(backpressure, writer);
return writer;
}
@@ -0,0 +1,45 @@
// Flags: --experimental-stream-iter
'use strict';
const common = require('../common');
const assert = require('assert');
const { Writable } = require('stream');
const { fromWritable } = require('stream/iter');
{
const writable = new Writable({ write() {} });
fromWritable(writable);
assert.throws(
() => fromWritable(writable, { backpressure: 'invalid' }),
{ code: 'ERR_INVALID_ARG_VALUE' },
);
writable.destroy();
}
async function testCachedWritableUsesLaterBackpressureOptions() {
const chunks = [];
const writable = new Writable({
highWaterMark: 1,
write(chunk, encoding, callback) {
chunks.push(Buffer.from(chunk));
},
});
fromWritable(writable);
const writer = fromWritable(writable, { backpressure: 'drop-newest' });
await writer.write('a');
await writer.write('b');
assert.deepStrictEqual(
chunks.map((chunk) => chunk.toString()),
['a'],
);
writable.destroy();
}
testCachedWritableUsesLaterBackpressureOptions().then(common.mustCall());