stream: fix pipeToSync byte accounting

Stop pipeToSync() from counting chunks when writeSync() or writevSync()
returns false without accepting the data. Preserve writers that use
false as an accepted backpressure signal.
Fixes: #63562
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63564
Backport-PR-URL: #64675Fixes: #63562
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: trivikr

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),

lib/internal/streams/iter/pull.js

@@ -863,6 +863,18 @@ 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()
 // =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
 consthasEndSync=typeofwriter.endSync==='function';
 
 try{
+letcanContinue=true;
 for(constbatchofpipeline){
+if(!canContinue){
+break;
+}
 if(hasWritevSync&&batch.length>1){
-writer.writevSync(batch);
+constresult=writer.writevSync(batch);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+break;
+}
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 }else{
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-writer.writeSync(chunk);
+constresult=writer.writeSync(chunk);
+if(result===false&&
+!isAcceptedSyncWriteBackpressure(writer,result)){
+canContinue=false;
+break;
+}
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 }
 }
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
 consthasWritev=typeofwriter.writev==='function';
 consthasWritevSync=typeofwriter.writevSync==='function';
 consthasEndSync=typeofwriter.endSync==='function';
-constsyncFalseCanBeAccepted=writer[kSyncWriteAcceptedOnFalse]===true;
-
-functionsyncFalseWasAccepted(){
-returnsyncFalseCanBeAccepted&&writer.desiredSize===0;
-}
 
 functionwaitForSyncBackpressure(){
 constondrain=writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
 asyncfunctionwriteBatchAsyncFallback(batch,startIndex){
 for(leti=startIndex;i<batch.length;i++){
 constchunk=batch[i];
-if(hasWriteSync&&writer.writeSync(chunk)){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(result){
 // Sync retry succeeded
-}elseif(syncFalseWasAccepted()){
+}elseif(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 awaitwaitForSyncBackpressure();
 continue;
@@ -1054,22 +1075,23 @@ 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){
-if(!hasWritevSync||!writer.writevSync(batch)){
-if(hasWritevSync&&syncFalseWasAccepted()){
+constresult=hasWritevSync&&writer.writevSync(batch);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 returnwaitForSyncBackpressure();
 }
 constopts=signal ? {__proto__: null, signal } : undefined;
-constresult=writer.writev(batch,opts);
-if(result===undefined){
+constwritevResult=writer.writev(batch,opts);
+if(writevResult===undefined){
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
 return;
 }
-returnPromisePrototypeThen(PromiseResolve(result),()=>{
+returnPromisePrototypeThen(PromiseResolve(writevResult),()=>{
 for(leti=0;i<batch.length;i++){
 totalBytes+=TypedArrayPrototypeGetByteLength(batch[i]);
 }
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
 }
 for(leti=0;i<batch.length;i++){
 constchunk=batch[i];
-if(!hasWriteSync||!writer.writeSync(chunk)){
-if(hasWriteSync&&syncFalseWasAccepted()){
+constresult=hasWriteSync&&writer.writeSync(chunk);
+if(!result){
+if(isAcceptedSyncWriteBackpressure(writer,result)){
 totalBytes+=TypedArrayPrototypeGetByteLength(chunk);
 returnwriteBatchAfterAcceptedBackpressure(batch,i+1);
 }

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

@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
 })(),'abcd',4);
 }
 
+asyncfunctiontestPipeToSyncPushWriterStrictFalseRejected(){
+constdecoder=newTextDecoder();
+const{ writer, readable }=push({highWaterMark: 1});
+
+consttotal=pipeToSync(['a','b'],writer,{preventClose: true});
+assert.strictEqual(total,1);
+
+constiter=readable[Symbol.asyncIterator]();
+constfirst=awaititer.next();
+assert.strictEqual(first.done,false);
+assert.strictEqual(decoder.decode(first.value[0]),'a');
+
+constsecond=awaitPromise.race([
+iter.next().then((result)=>{
+returnresult.done ? '<done>' : decoder.decode(result.value[0]);
+}),
+setImmediatePromise().then(()=>'<no second chunk>'),
+]);
+assert.strictEqual(second,'<no second chunk>');
+
+awaititer.return?.();
+}
+
+asyncfunctiontestPipeToSyncWritevFalseNotCounted(){
+constwriter={
+writevSync(){returnfalse;},
+writeSync: common.mustNotCall(),
+endSync(){return0;},
+};
+function*source(){
+yield[newUint8Array([1]),newUint8Array([2])];
+}
+
+consttotal=pipeToSync(source(),writer);
+assert.strictEqual(total,0);
+}
+
 // pipeToSync with writevSync
 asyncfunctiontestPipeToSyncWritev(){
 constbatches=[];
@@ -215,6 +252,8 @@ Promise.all([
 testWriteSyncFailsMidBatch(),
 testWriteSyncAlwaysFails(),
 testPushWriterBlockSyncFalseAccepted(),
+testPipeToSyncPushWriterStrictFalseRejected(),
+testPipeToSyncWritevFalseNotCounted(),
 testPipeToSyncWritev(),
 testPipeToSyncPlainChunksWritev(),
 testPipeToSyncWriteFallback(),