stream: avoid duplicate writes in toWritable

PushWriter can return false after accepting a chunk when block
backpressure is active. Teach the classic Writable adapter to treat
that case as accepted backpressure instead of retrying through the
async write path.
Fixes: #63359
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63360
Backport-PR-URL: #64675Fixes: #63359
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),

lib/internal/streams/iter/classic.js

@@ -57,6 +57,7 @@ const {
 const{
 toAsyncStreamable: kToAsyncStreamable,
  kValidatedSource,
+ kSyncWriteAccepted,
  drainableProtocol,
 }=require('internal/streams/iter/types');
 
@@ -764,13 +765,41 @@ function toWritable(writer) {
 consthasEndSync=hasEnd&&
 typeofwriter.endSync==='function';
 consthasFail=typeofwriter.fail==='function';
+consthasSyncWriteAccepted=
+typeofwriter[kSyncWriteAccepted]==='function';
 
-// Try-sync-first pattern: attempt the synchronous method and
-// fall back to the async method if it returns false (indicating
-// the sync path was not accepted) or 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.
+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.
 
 function_write(chunk,encoding,cb){
 constbytes=typeofchunk==='string' ?
@@ -781,6 +810,11 @@ 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.
 }
@@ -805,6 +839,11 @@ 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.
 }

lib/internal/streams/iter/push.js

@@ -32,6 +32,7 @@ const {
 
 const{
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
 }=require('internal/streams/iter/types');
 
@@ -545,11 +546,16 @@ class PushQueue {
 
 classPushWriter{
  #queue;
+ #syncWriteAccepted =false;
 
 constructor(queue){
 this.#queue =queue;
 }
 
+[kSyncWriteAccepted](){
+returnthis.#syncWriteAccepted;
+}
+
 [drainableProtocol](){
 constdesired=this.desiredSize;
 if(desired===null)returnnull;
@@ -589,19 +595,23 @@ 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;
 }
 
 writevSync(chunks){
+this.#syncWriteAccepted =false;
 if(!ArrayIsArray(chunks)){
 thrownewERR_INVALID_ARG_TYPE('chunks','Array',chunks);
 }
@@ -610,8 +620,10 @@ class PushWriter {
 if(!result&&this.#queue.backpressurePolicy==='block'&&
 this.#queue.desiredSize===0){
 this.#queue.forceEnqueue(bytes);
+this.#syncWriteAccepted =true;
 returnfalse;
 }
+this.#syncWriteAccepted =result;
 returnresult;
 }
 

lib/internal/streams/iter/types.js

@@ -64,11 +64,24 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 constkValidatedSource=Symbol('kValidatedSource');
 
+/**
+ * Internal sentinel for writers whose sync write methods can return false
+ * after accepting data as a backpressure signal.
+ */
+constkSyncWriteAccepted=Symbol('kSyncWriteAccepted');
+
+/**
+ * Internal sentinel for writers whose sync write methods may return false
+ * after accepting data when backpressure is applied. Such writers must expose
+ * desiredSize so callers can distinguish accepted backpressure from a sync
+ * write that was not performed.
+ */
 constkSyncWriteAcceptedOnFalse=Symbol('kSyncWriteAcceptedOnFalse');
 
 module.exports={
  broadcastProtocol,
  drainableProtocol,
+ kSyncWriteAccepted,
  kSyncWriteAcceptedOnFalse,
  kValidatedSource,
  kValidatedTransform,

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

@@ -335,6 +335,53 @@ async function testRoundTrip() {
 assert.strictEqual(result,data);
 }
 
+// =============================================================================
+// PushWriter writeSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.write('b');
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'ab');
+}
+
+// =============================================================================
+// PushWriter writevSync false accepted as backpressure is not retried
+// =============================================================================
+
+asyncfunctiontestPushWriterBlockBackpressureWritevNoDuplicate(){
+const{ writer, readable }=push({highWaterMark: 1,backpressure: 'block'});
+constwritable=toWritable(writer);
+
+awaitnewPromise((resolve,reject)=>{
+writable.write('a',(err)=>{
+if(err)reject(err);
+elseresolve();
+});
+});
+
+writable.cork();
+writable.write('b');
+writable.write('c');
+writable.uncork();
+writable.end();
+
+constresult=awaittext(readable);
+assert.strictEqual(result,'abc');
+}
+
 // =============================================================================
 // Multiple sequential writes
 // =============================================================================
@@ -590,6 +637,8 @@ Promise.all([
 testWriteThrowsSyncPropagation(),
 testEndThrowsSyncPropagation(),
 testRoundTrip(),
+testPushWriterBlockBackpressureNoDuplicate(),
+testPushWriterBlockBackpressureWritevNoDuplicate(),
 testSequentialWrites(),
 testSyncCallbackDeferred(),
 testMinimalWriter(),