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: #63277
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63278Fixes: #63277
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: trivikr

doc/api/stream_iter.md

@@ -1517,8 +1517,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.

lib/internal/streams/iter/classic.js

@@ -21,6 +21,7 @@ const {
  PromiseReject,
  PromiseResolve,
  PromiseWithResolvers,
+ SafeMap,
  SafeWeakMap,
  SymbolAsyncDispose,
  SymbolAsyncIterator,
@@ -427,10 +428,6 @@ function fromWritable(writable, options = kNullPrototype) {
 thrownewERR_INVALID_ARG_TYPE('writable','Writable',writable);
 }
 
-// Return cached adapter if available.
-constcached=fromWritableCache.get(writable);
-if(cached!==undefined)returncached;
-
 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.
+letcachedByBackpressure=fromWritableCache.get(writable);
+if(cachedByBackpressure!==undefined){
+constcached=cachedByBackpressure.get(backpressure);
+if(cached!==undefined)returncached;
+}else{
+cachedByBackpressure=newSafeMap();
+fromWritableCache.set(writable,cachedByBackpressure);
+}
+
 // Fall back to sensible defaults for duck-typed streams that may not
 // expose the full stream.Writable property set.
 consthwm=writable.writableHighWaterMark??16384;
@@ -710,7 +718,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnpromise;
 };
 
-fromWritableCache.set(writable,writer);
+cachedByBackpressure.set(backpressure,writer);
 returnwriter;
 }
 

test/parallel/test-stream-iter-from-writable-cache-options.js

@@ -0,0 +1,45 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+constcommon=require('../common');
+constassert=require('assert');
+const{ Writable }=require('stream');
+const{ fromWritable }=require('stream/iter');
+
+{
+constwritable=newWritable({write(){}});
+
+fromWritable(writable);
+
+assert.throws(
+()=>fromWritable(writable,{backpressure: 'invalid'}),
+{code: 'ERR_INVALID_ARG_VALUE'},
+);
+
+writable.destroy();
+}
+
+asyncfunctiontestCachedWritableUsesLaterBackpressureOptions(){
+constchunks=[];
+constwritable=newWritable({
+highWaterMark: 1,
+write(chunk,encoding,callback){
+chunks.push(Buffer.from(chunk));
+},
+});
+
+fromWritable(writable);
+constwriter=fromWritable(writable,{backpressure: 'drop-newest'});
+
+awaitwriter.write('a');
+awaitwriter.write('b');
+
+assert.deepStrictEqual(
+chunks.map((chunk)=>chunk.toString()),
+['a'],
+);
+
+writable.destroy();
+}
+
+testCachedWritableUsesLaterBackpressureOptions().then(common.mustCall());