stream: refine the stream/iter backpressure

In `Writer`, there are two queues that matter: the slot queue
and the pending queue. The sync `writeSync`/`writevSync` only
use the slot queue. If writes cannot be accepted directly into
slots, they return `false`. The sync methods *never* enqueue
into the pending queue. The async `write`/`writev` will first
attempt to add to the slot queue; if it is full, then it will
attempt to add to the pending queue; if that is also full, the
backpressure policy kicks in.

Signed-off-by: James M Snell <jasnell@gmail.com>

Author: jasnell

lib/internal/streams/iter/classic.js

@@ -57,7 +57,6 @@ const {
 const {
   toAsyncStreamable: kToAsyncStreamable,
   kValidatedSource,
-  kSyncWriteAccepted,
   drainableProtocol,
 } = require('internal/streams/iter/types');
 
@@ -765,41 +764,11 @@ function toWritable(writer) {
   const hasEndSync = hasEnd &&
                      typeof writer.endSync === 'function';
   const hasFail = typeof writer.fail === 'function';
-  const hasSyncWriteAccepted =
-    typeof writer[kSyncWriteAccepted] === 'function';
-
-  function syncWriteAccepted() {
-    return hasSyncWriteAccepted && writer[kSyncWriteAccepted]();
-  }
-
-  function finishAfterSyncBackpressure(cb) {
-    let ondrain;
-    try {
-      if (typeof writer[drainableProtocol] === 'function') {
-        ondrain = writer[drainableProtocol]();
-      }
-    } catch (err) {
-      cb(err);
-      return;
-    }
-    if (ondrain !== null && ondrain !== undefined) {
-      PromisePrototypeThen(ondrain, (drained) => {
-        if (drained === false) {
-          cb(new ERR_INVALID_STATE.TypeError('Stream closed by consumer'));
-          return;
-        }
-        cb();
-      }, cb);
-      return;
-    }
-    queueMicrotask(cb);
-  }
-
   // Try-sync-first pattern: attempt the synchronous method and fall back to the
-  // async method if it returns false without accepting the data, or if it
-  // throws. When the sync path succeeds, the callback is deferred via
-  // queueMicrotask to preserve the async resolution contract that Writable
-  // internals expect from _write/_writev/_final callbacks.
+  // async method if it returns false (data not accepted synchronously).
+  // When the sync path succeeds, the callback is deferred via queueMicrotask
+  // to preserve the async resolution contract that Writable internals expect
+  // from _write/_writev/_final callbacks.
 
   function _write(chunk, encoding, cb) {
     const bytes = typeof chunk === 'string' ?
@@ -810,11 +779,6 @@ 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.
       }
@@ -839,11 +803,6 @@ 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/pull.js

@@ -56,8 +56,6 @@ const {
 } = require('internal/streams/iter/utils');
 
 const {
-  drainableProtocol,
-  kSyncWriteAcceptedOnFalse,
   kValidatedSource,
   kValidatedTransform,
   toAsyncStreamable,
@@ -872,21 +870,6 @@ async function pipeTo(source, ...args) {
   const hasWritev = typeof writer.writev === 'function';
   const hasWritevSync = typeof writer.writevSync === 'function';
   const hasEndSync = typeof writer.endSync === 'function';
-  const syncFalseCanBeAccepted = writer[kSyncWriteAcceptedOnFalse] === true;
-
-  function syncFalseWasAccepted() {
-    return syncFalseCanBeAccepted && writer.desiredSize === 0;
-  }
-
-  function waitForSyncBackpressure() {
-    const ondrain = writer[drainableProtocol];
-    return ondrain?.call(writer);
-  }
-
-  async function writeBatchAfterAcceptedBackpressure(batch, startIndex) {
-    await waitForSyncBackpressure();
-    await writeBatchAsyncFallback(batch, startIndex);
-  }
 
   // Async fallback for writeBatch when sync write fails partway through.
   // Continues writing from batch[startIndex] using async write().
@@ -895,10 +878,6 @@ async function pipeTo(source, ...args) {
       const chunk = batch[i];
       if (hasWriteSync && writer.writeSync(chunk)) {
         // Sync retry succeeded
-      } else if (syncFalseWasAccepted()) {
-        totalBytes += TypedArrayPrototypeGetByteLength(chunk);
-        await waitForSyncBackpressure();
-        continue;
       } else {
         const result = writer.write(
           chunk, signal ? { __proto__: null, signal } : undefined);
@@ -916,12 +895,6 @@ async function pipeTo(source, ...args) {
   function writeBatch(batch) {
     if (hasWritev && batch.length > 1) {
       if (!hasWritevSync || !writer.writevSync(batch)) {
-        if (hasWritevSync && syncFalseWasAccepted()) {
-          for (let i = 0; i < batch.length; i++) {
-            totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
-          }
-          return waitForSyncBackpressure();
-        }
         const opts = signal ? { __proto__: null, signal } : undefined;
         const result = writer.writev(batch, opts);
         if (result === undefined) {
@@ -944,12 +917,7 @@ async function pipeTo(source, ...args) {
     for (let i = 0; i < batch.length; i++) {
       const chunk = batch[i];
       if (!hasWriteSync || !writer.writeSync(chunk)) {
-        if (hasWriteSync && syncFalseWasAccepted()) {
-          totalBytes += TypedArrayPrototypeGetByteLength(chunk);
-          return writeBatchAfterAcceptedBackpressure(batch, i + 1);
-        }
         // Sync path failed at index i - fall back to async for the rest.
-        // Count bytes for chunks already written synchronously (0..i-1).
         return writeBatchAsyncFallback(batch, i);
       }
       totalBytes += TypedArrayPrototypeGetByteLength(chunk);

lib/internal/streams/iter/push.js

@@ -33,8 +33,6 @@ const {
 
 const {
   drainableProtocol,
-  kSyncWriteAccepted,
-  kSyncWriteAcceptedOnFalse,
 } = require('internal/streams/iter/types');
 
 const {
@@ -369,19 +367,6 @@ class PushQueue {
     this.#pendingEnd = pending;
   }
 
-  /**
-   * Force-enqueue chunks into the slots buffer, bypassing capacity checks.
-   * Used by PushWriter.writeSync() for 'block' policy where the data is
-   * accepted but false is returned as a backpressure signal.
-   */
-  forceEnqueue(chunks) {
-    this.#slots.push(chunks);
-    for (let i = 0; i < chunks.length; i++) {
-      this.#bytesWritten += TypedArrayPrototypeGetByteLength(chunks[i]);
-    }
-    this.#resolvePendingReads();
-  }
-
   /**
    * Wait for backpressure to clear (desiredSize > 0).
    * @returns {Promise<void>}
@@ -563,16 +548,11 @@ class PushQueue {
 
 class PushWriter {
   #queue;
-  #syncWriteAccepted = false;
 
   constructor(queue) {
     this.#queue = queue;
   }
 
-  [kSyncWriteAccepted]() {
-    return this.#syncWriteAccepted;
-  }
-
   [drainableProtocol]() {
     const desired = this.desiredSize;
     if (desired === null) return null;
@@ -584,10 +564,6 @@ class PushWriter {
     return this.#queue.desiredSize;
   }
 
-  get [kSyncWriteAcceptedOnFalse]() {
-    return this.#queue.backpressurePolicy === 'block';
-  }
-
   write(chunk, options) {
     if (!options?.signal && this.#queue.canWriteSync()) {
       const bytes = toUint8Array(chunk);
@@ -612,36 +588,16 @@ class PushWriter {
   }
 
   writeSync(chunk) {
-    this.#syncWriteAccepted = false;
     const bytes = toUint8Array(chunk);
-    const result = 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;
-      return false;
-    }
-    this.#syncWriteAccepted = result;
-    return result;
+    return this.#queue.writeSync([bytes]);
   }
 
   writevSync(chunks) {
-    this.#syncWriteAccepted = false;
     if (!ArrayIsArray(chunks)) {
       throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
     }
     const bytes = convertChunks(chunks);
-    const result = this.#queue.writeSync(bytes);
-    if (!result && this.#queue.backpressurePolicy === 'block' &&
-        this.#queue.desiredSize === 0) {
-      this.#queue.forceEnqueue(bytes);
-      this.#syncWriteAccepted = true;
-      return false;
-    }
-    this.#syncWriteAccepted = result;
-    return result;
+    return this.#queue.writeSync(bytes);
   }
 
   end(options) {

lib/internal/streams/iter/types.js

@@ -64,25 +64,9 @@ const kValidatedTransform = Symbol('kValidatedTransform');
  */
 const kValidatedSource = Symbol('kValidatedSource');
 
-/**
- * Internal sentinel for writers whose sync write methods can return false
- * after accepting data as a backpressure signal.
- */
-const kSyncWriteAccepted = 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.
- */
-const kSyncWriteAcceptedOnFalse = Symbol('kSyncWriteAcceptedOnFalse');
-
 module.exports = {
   broadcastProtocol,
   drainableProtocol,
-  kSyncWriteAccepted,
-  kSyncWriteAcceptedOnFalse,
   kValidatedSource,
   kValidatedTransform,
   shareProtocol,

test/parallel/test-stream-iter-push-backpressure.js

@@ -95,21 +95,25 @@ async function testBlockBackpressure() {
   await writePromise;
 }
 
-async function testBlockWriteSyncEnqueues() {
-  // With block policy, writeSync should enqueue the data even when the buffer
-  // is full, returning false as a backpressure signal. The data IS accepted.
+async function testBlockWriteSyncDoesNotEnqueue() {
+  // With block policy, writeSync returns false when the buffer is full.
+  // The data is NOT accepted — writeSync only operates on the slots buffer.
+  // The caller should fall back to write() which uses the pending queue.
   const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' });
 
   // Fill the buffer
   assert.strictEqual(writer.writeSync('a'), true);
 
-  // Buffer full: writeSync should enqueue and return false (data accepted)
+  // Buffer full: writeSync returns false, data NOT enqueued
   assert.strictEqual(writer.writeSync('b'), false);
 
-  writer.endSync();
+  // Use the async write (try-fallback pattern) — this goes to pending queue
+  const consuming = text(readable);
+  await writer.write('b');
+  await writer.end();
 
-  // Both chunks should be delivered (drain flushes all slots into one batch)
-  const result = await text(readable);
+  // Both chunks should be delivered
+  const result = await consuming;
   assert.strictEqual(result, 'ab');
 }
 
@@ -151,6 +155,6 @@ Promise.all([
   testDropOldest(),
   testDropNewest(),
   testBlockBackpressure(),
-  testBlockWriteSyncEnqueues(),
+  testBlockWriteSyncDoesNotEnqueue(),
   testStrictPendingQueueOverflow(),
 ]).then(common.mustCall());