stream: refine the stream/iter backpressure

In `Writer`, there are two queues that matter: the slot queue
and the pending queue. The sync `writeSync`/`writevSync` only
use the slot queue. If writes cannot be accepted directly into
slots, they return `false`. The sync methods *never* enqueue
into the pending queue. The async `write`/`writev` will first
attempt to add to the slot queue; if it is full, then it will
attempt to add to the pending queue; if that is also full, the
backpressure policy kicks in.
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: #63697
Backport-PR-URL: #64675
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: jasnell

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){

doc/api/stream_iter.md

@@ -1649,7 +1649,7 @@ Creates a classic [`stream.Writable`][] backed by a stream/iter Writer.
 
 Each `_write()` / `_writev()` call attempts the Writer's synchronous method
 first (`writeSync` / `writevSync`), falling back to the async method if the
-sync path returns `false` or throws. Similarly, `_final()` tries `endSync()`
+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.
 

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
- kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
-consthasSyncWriteAccepted=
-typeofwriter[kSyncWriteAccepted]==='function';
-
-functionsyncWriteAccepted(){
-returnhasSyncWriteAccepted&&writer[kSyncWriteAccepted]();
-}
-
-functionfinishAfterSyncBackpressure(cb){
-letondrain;
-try{
-if(typeofwriter[drainableProtocol]==='function'){
-ondrain=writer[drainableProtocol]();
-}
-}catch(err){
-cb(err);
-return;
-}
-if(ondrain!==null&&ondrain!==undefined){
-PromisePrototypeThen(ondrain,(drained)=>{
-if(drained===false){
-cb(newERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-return;
-}
-cb();
-},cb);
-return;
-}
-queueMicrotask(cb);
-}
-
 // Try-sync-first pattern: attempt the synchronous method and fall back to the
-// async method if it returns false without accepting the data, or if it
-// throws. When the sync path succeeds, the callback is deferred via
-// queueMicrotask to preserve the async resolution contract that Writable
-// internals expect from _write/_writev/_final callbacks.
+// async method if it returns false (data not accepted synchronously).
+// When the sync path succeeds, the callback is deferred via queueMicrotask
+// to preserve the async resolution contract that Writable internals expect
+// from _write/_writev/_final callbacks.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -810,13 +779,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunk was accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WriteSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -839,13 +805,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-if(syncWriteAccepted()){
-// The chunks were accepted; false only signaled backpressure.
-finishAfterSyncBackpressure(cb);
-return;
-}
-}catch{
-// Sync path threw -- fall through to async.
+// WritevSync returned false: not accepted, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{
@@ -867,8 +830,10 @@ function toWritable(writer) {
 queueMicrotask(cb);
 return;
 }
-}catch{
-// Sync path threw -- fall through to async.
+// Result < 0: can't end synchronously, fall through to async.
+}catch(err){
+cb(err);
+return;
 }
 }
 try{

lib/internal/streams/iter/consumers.js

@@ -480,6 +480,7 @@ function merge(...args) {
 );
 }
 
+letprimaryError;
 try{
 while(activeCount>0||ready.length>0){
 signal?.throwIfAborted();
@@ -500,22 +501,46 @@ function merge(...args) {
 });
 }
 }
+}catch(err){
+primaryError=err;
 }finally{
-// Clean up: return all iterators
-awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
-if(iterator.return){
-try{
-awaititerator.return();
-}catch{
-// Ignore return errors
-}
-}
-});
+// Clean up: return all iterators. Cleanup errors are not
+// swallowed - a broken iterator.return() (e.g., failing to
+// release a resource) should be visible to the caller.
+awaitcleanupIterators(iterators,primaryError);
 }
 },
 };
 }
 
+asyncfunctioncleanupIterators(iterators,primaryError){
+letcleanupError;
+awaitSafePromiseAllReturnVoid(iterators,async(iterator)=>{
+if(iterator.return){
+try{
+awaititerator.return();
+}catch(err){
+// Keep the first cleanup error encountered.
+cleanupError??=err;
+}
+}
+});
+if(cleanupError!==undefined){
+if(primaryError!==undefined){
+// Both a primary error and a cleanup error occurred.
+// Wrap in SuppressedError so neither is lost:
+// .error = primaryError, .suppressed = cleanupError.
+// eslint-disable-next-line no-restricted-syntax
+thrownewSuppressedError(primaryError,cleanupError);
+}
+// No primary error - the cleanup error is the only error.
+throwcleanupError;
+}
+if(primaryError!==undefined){
+throwprimaryError;
+}
+}
+
 module.exports={
  array,
  arrayBuffer,

lib/internal/streams/iter/pull.js

@@ -63,8 +63,6 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
- drainableProtocol,
- kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  toAsyncStreamable,
@@ -863,18 +861,6 @@ async function* createAsyncPipeline(source, transforms, signal) {
 }
 }
 
-/**
- * Check if a false sync write result means accepted backpressure.
- * @param {object} writer - The writer whose sync method returned.
- * @param {*} result - The return value from writeSync() or writevSync().
- * @returns {boolean}
- */
-functionisAcceptedSyncWriteBackpressure(writer,result){
-returnresult===false&&
-writer[kSyncWriteAcceptedOnFalse]===true&&
-writer.desiredSize===0;
-}
-
 // =============================================================================
 // Public API: pull() and pullSync()
 // =============================================================================
@@ -963,9 +949,7 @@ function pipeToSync(source, ...args) {
 break;
 }
 if(hasWritevSync&&batch.length>1){
-constresult=writer.writevSync(batch);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writevSync(batch)===false){
 break;
 }
 for(leti=0;i<batch.length;i++){
@@ -974,9 +958,7 @@ function pipeToSync(source, ...args) {
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=writer.writeSync(chunk);
-if(result===false&&
-!isAcceptedSyncWriteBackpressure(writer,result)){
+if(writer.writeSync(chunk)===false){
 canContinue=false;
 break;
 }
@@ -1027,28 +1009,13 @@ async function pipeTo(source, ...args) {
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
 
-functionwaitForSyncBackpressure(){
-constondrain=writer[drainableProtocol];
-returnondrain?.call(writer);
-}
-
-asyncfunctionwriteBatchAfterAcceptedBackpressure(batch,startIndex){
-awaitwaitForSyncBackpressure();
-awaitwriteBatchAsyncFallback(batch,startIndex);
-}
-
 // Async fallback for writeBatch when sync write fails partway through.
 // Continues writing from batch[startIndex] using async write().
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(result){
+if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
-}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-awaitwaitForSyncBackpressure();
-continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -1065,14 +1032,7 @@ async function pipeTo(source, ...args) {
 // is required. Callers must check: const p = writeBatch(b); if (p) await p;
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
-constresult=hasWritevSync&&writer.writevSync(batch);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-for(leti=0;i<batch.length;i++){
-totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
-}
-returnwaitForSyncBackpressure();
-}
+if(!hasWritevSync||!writer.writevSync(batch)){
 constopts=signal ? {__proto__: null, signal } : undefined;
 constwritevResult=writer.writev(batch,opts);
 if(writevResult===undefined){
@@ -1094,14 +1054,8 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-constresult=hasWriteSync&&writer.writeSync(chunk);
-if(!result){
-if(isAcceptedSyncWriteBackpressure(writer,result)){
-totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
-returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
-}
+if(!hasWriteSync||!writer.writeSync(chunk)){
 // Sync path failed at index i - fall back to async for the rest.
-// Count bytes for chunks already written synchronously (0..i-1).
 returnwriteBatchAsyncFallback(batch,i);
 }
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -32,8 +32,6 @@ const {
 
 const{
  drainableProtocol,
- kSyncWriteAccepted,
- kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -364,19 +362,6 @@ class PushQueue {
 this.#pendingEnd =pending;
 }
 
-/**
- * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
- * Used by PushWriter.writeSync() for 'block' policy where the data is
- * accepted but false is returned as a backpressure signal.
- */
-forceEnqueue(chunks){
-this.#slots.push(chunks);
-for(leti=0;i<chunks.length;i++){
-this.#bytesWritten +=TypedArrayPrototypeGetByteLength(chunks[i]);
-}
-this.#resolvePendingReads();
-}
-
 /**
  * Wait for backpressure to clear (desiredSize > 0).
  * @returns {Promise<void>}
@@ -558,16 +543,11 @@ class PushQueue {
 
 classPushWriter{
  #queue;
- #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
-[kSyncWriteAccepted](){
-returnthis.#syncWriteAccepted;
-}
-
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -579,10 +559,6 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
-get[kSyncWriteAcceptedOnFalse](){
-returnthis.#queue.backpressurePolicy==='block';
-}
-
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -607,36 +583,16 @@ class PushWriter {
 }
 
 writeSync(chunk){
-this.#syncWriteAccepted =false;
 constbytes=toUint8Array(chunk);
-constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-// Block policy: force-enqueue and return false as backpressure signal.
-// Data IS accepted; false tells caller to slow down.
-this.#queue.forceEnqueue([bytes]);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync([bytes]);
 }
 
 writevSync(chunks){
-this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
 constbytes=convertChunks(chunks);
-constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'&&
-this.#queue.desiredSize===0){
-this.#queue.forceEnqueue(bytes);
-this.#syncWriteAccepted =true;
-returnfalse;
-}
-this.#syncWriteAccepted =result;
-returnresult;
+returnthis.#queue.writeSync(bytes);
 }
 
 end(options){