stream: avoid retrying accepted pipeTo writes

PushWriter in block backpressure mode can return false from writeSync()
and writevSync() after accepting data. Treat that false return as
backpressure and wait for drain instead of retrying the same chunks
asynchronously.
Fixes: #63296
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63297
Backport-PR-URL: #64675Fixes: #63296
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: trivikr

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());

lib/internal/streams/iter/pull.js

@@ -51,6 +51,8 @@ const {
 }=require('internal/streams/iter/utils');
 
 const{
+ drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedTransform,
 }=require('internal/streams/iter/types');
 
@@ -828,13 +830,33 @@ async function pipeTo(source, ...args) {
 consthasWriteSync=typeofwriter.writeSync==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
+constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
+
+functionsyncFalseWasAccepted(){
+returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
+}
+
+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];
 if(hasWriteSync&&writer.writeSync(chunk)){
 // Sync retry succeeded
+}elseif(syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+awaitwaitForSyncBackpressure();
+continue;
 }else{
 constresult=writer.write(
 chunk,signal ? {__proto__: null, signal } : undefined);
@@ -852,6 +874,12 @@ async function pipeTo(source, ...args) {
 functionwriteBatch(batch){
 if(hasWritev&&batch.length>1){
 if(!hasWritevSync||!writer.writevSync(batch)){
+if(hasWritevSync&&syncFalseWasAccepted()){
+for(leti=0;i<batch.length;i++){
+totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
+}
+returnwaitForSyncBackpressure();
+}
 constopts=signal ? {__proto__: null, signal } : undefined;
 returnPromisePrototypeThen(writer.writev(batch,opts),()=>{
 for(leti=0;i<batch.length;i++){
@@ -867,6 +895,10 @@ async function pipeTo(source, ...args) {
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
 if(!hasWriteSync||!writer.writeSync(chunk)){
+if(hasWriteSync&&syncFalseWasAccepted()){
+totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
+returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
+}
 // 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);

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
 const{
@@ -560,6 +561,10 @@ class PushWriter {
 returnthis.#queue.desiredSize;
 }
 
+get[kSyncWriteAcceptedOnFalse](){
+returnthis.#queue.backpressurePolicy==='block';
+}
+
 write(chunk,options){
 if(!options?.signal&&this.#queue.canWriteSync()){
 constbytes=toUint8Array(chunk);
@@ -586,7 +591,8 @@ class PushWriter {
 writeSync(chunk){
 constbytes=toUint8Array(chunk);
 constresult=this.#queue.writeSync([bytes]);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+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]);
@@ -601,7 +607,8 @@ class PushWriter {
 }
 constbytes=convertChunks(chunks);
 constresult=this.#queue.writeSync(bytes);
-if(!result&&this.#queue.backpressurePolicy==='block'){
+if(!result&&this.#queue.backpressurePolicy==='block'&&
+this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
 returnfalse;
 }

lib/internal/streams/iter/types.js

@@ -64,9 +64,12 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
+
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,
  shareProtocol,

test/parallel/test-stream-iter-pipeto-writev.js

@@ -5,7 +5,8 @@
 
 constcommon=require('../common');
 constassert=require('assert');
-const{ pipeTo, pipeToSync }=require('stream/iter');
+const{setImmediate: setImmediatePromise}=require('timers/promises');
+const{ pipeTo, pipeToSync, push, text }=require('stream/iter');
 
 // Multi-chunk batch with writevSync (sync success path)
 asyncfunctiontestWritevSyncSuccess(){
@@ -104,6 +105,35 @@ async function testWriteSyncAlwaysFails() {
 assert.strictEqual(total,2);
 }
 
+// PushWriter block mode accepts sync writes even when returning false for
+// backpressure. pipeTo must wait for drain, not retry the same write.
+asyncfunctionassertPushWriterBlockPipeTo(source,expected,expectedTotal){
+const{ writer, readable }=push({
+highWaterMark: 1,
+backpressure: 'block',
+});
+
+constpipe=pipeTo(source,writer);
+awaitsetImmediatePromise();
+constdata=awaittext(readable);
+consttotal=awaitpipe;
+
+assert.strictEqual(data,expected);
+assert.strictEqual(total,expectedTotal);
+}
+
+asyncfunctiontestPushWriterBlockSyncFalseAccepted(){
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97])];
+yield[newUint8Array([98])];
+})(),'ab',2);
+
+awaitassertPushWriterBlockPipeTo((asyncfunction*(){
+yield[newUint8Array([97,98])];
+yield[newUint8Array([99]),newUint8Array([100])];
+})(),'abcd',4);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -142,6 +172,7 @@ Promise.all([
 testWritevSyncFails(),
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
+testPushWriterBlockSyncFalseAccepted(),
 testPipeToSyncWritev(),
 testPipeToSyncWriteFallback(),
 ]).then(common.mustCall());