stream, quic: update iterable streams backpressure

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode/Opus

Author: jasnell

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 const bench = common.createBenchmark(main, {
   consumers: [2, 8, 32],
   batches: [1e4],
-  backpressure: ['block'],
+  backpressure: ['unbounded'],
   n: [5],
 }, {
   flags: ['--experimental-stream-iter'],
@@ -24,7 +24,7 @@ async function main({ consumers, batches, backpressure, n }) {
 
   bench.start();
   for (let i = 0; i < n; i++) {
-    const shared = share(source(), { highWaterMark: 64, backpressure });
+    const shared = share(source(), { budget: 65536, backpressure });
     const readers = Array.from({ length: consumers }, () => array(shared.pull()));
     await Promise.all(readers);
   }

doc/api/quic.md

@@ -1296,7 +1296,7 @@ added: v23.8.0
     interleaved with data from other streams of the same priority level.
     When `false`, the stream should be completed before same-priority peers.
     **Default:** `false`.
-  * `highWaterMark` {number} The maximum number of bytes that the writer
+  * `budget` {number} The maximum number of bytes that the writer
     will buffer before `writeSync()` returns `false`. When the buffered
     data exceeds this limit, the caller should wait for drain before
     writing more. **Default:** `65536` (64 KB).
@@ -1337,7 +1337,7 @@ added: v23.8.0
     interleaved with data from other streams of the same priority level.
     When `false`, the stream should be completed before same-priority peers.
     **Default:** `false`.
-  * `highWaterMark` {number} The maximum number of bytes that the writer
+  * `budget` {number} The maximum number of bytes that the writer
     will buffer before `writeSync()` returns `false`. When the buffered
     data exceeds this limit, the caller should wait for drain before
     writing more. **Default:** `65536` (64 KB).
@@ -1944,7 +1944,7 @@ added: v23.8.0
 The directionality of the stream, or `null` if the stream has been destroyed
 or is still pending. Read only.
 
-### `stream.highWaterMark`
+### `stream.budget`
 
 <!-- YAML
 added: v26.2.0
@@ -2256,7 +2256,8 @@ The Writer has the following methods:
   the QUIC transport-layer `INTERNAL_ERROR` (`0x1`) for raw QUIC).
   See [`stream.destroy()`][] for a full-stream abort that also resets
   the readable side via `STOP_SENDING`.
-* `desiredSize` — Available capacity in bytes, or `null` if closed/errored.
+* `canWrite` — `true` if writes will be accepted, `false` if at capacity,
+  or `null` if closed/errored.
 
 The bytes from each `writeSync()` / `writevSync()` / `write()` / `writev()`
 input chunk are copied into an internal buffer, so the caller's source

doc/api/stream_iter.md

@@ -226,7 +226,7 @@ How each policy uses these buffers:
 
 | Policy          | Buffer limit | Pending writes limit |
 | --------------- | ------------ | -------------------- |
-| `'strict'`      | `budget`     | `budget`             |
+| `'strict'`      | `budget`     | 1                    |
 | `'unbounded'`   | `budget`     | Unbounded            |
 | `'drop-oldest'` | `budget`     | N/A (never waits)    |
 | `'drop-newest'` | `budget`     | N/A (never waits)    |
@@ -235,8 +235,8 @@ How each policy uses these buffers:
 
 Strict mode catches "fire-and-forget" patterns where the producer calls
 `write()` without awaiting, which would cause unbounded memory growth.
-It limits both the buffer and the pending writes queue to
-`budget` bytes.
+It limits the buffer to `budget` bytes and the pending writes queue
+to a single entry.
 
 If you properly await each write, you can only ever have one pending
 write at a time (yours), so you never hit the pending writes limit.

lib/internal/quic/quic.js

@@ -297,8 +297,8 @@ const endpointRegistry = new SafeSet();
  *   (e.g. HTTP/3).
  * @property {'high'|'default'|'low'} [priority] The priority level of the stream.
  * @property {boolean} [incremental] Whether to interleave data with same-priority streams.
- * @property {number} [highWaterMark] The high water mark for write
- *   backpressure, in bytes. **Default:** `65536`.
+ * @property {number} [budget] The byte budget for write backpressure.
+ *   **Default:** `65536`.
  * @property {OnHeadersCallback} [onheaders] Callback for incoming initial headers
  * @property {OnTrailersCallback} [ontrailers] Callback for incoming trailing headers
  * @property {OnInfoCallback} [oninfo] Callback for informational (1xx) headers
@@ -1325,7 +1325,7 @@ function applyCallbacks(session, cbs) {
  * @param {QuicStream} stream The JS stream object
  * @param {any} body The body source
  */
-const kDefaultHighWaterMark = 65536;
+const kDefaultBudget = 65536;
 const kDefaultMaxPendingDatagrams = 128;
 
 function configureOutbound(handle, stream, body) {
@@ -1405,20 +1405,20 @@ function configureOutbound(handle, stream, body) {
   );
 }
 
-// Sets the high water mark and initial writeDesiredSize for a streaming
+// Sets the budget and initial writeDesiredSize for a streaming
 // outbound source. Called after handle.initStreamingSource() for both
 // body-source and writer paths. One-shot body sources (string, Uint8Array,
 // Blob, FileHandle, etc.) do not use this -- they go through attachSource
 // and are not subject to backpressure.
 function initStreamingBackpressure(stream) {
   const state = getQuicStreamState(stream);
   // Only set defaults if the user hasn't already configured them
-  // (e.g., via createBidirectionalStream({ highWaterMark: N })).
-  if (state.highWaterMark === 0) {
-    state.highWaterMark = kDefaultHighWaterMark;
+  // (e.g., via createBidirectionalStream({ budget: N })).
+  if (state.budget === 0) {
+    state.budget = kDefaultBudget;
   }
   if (state.writeDesiredSize === 0) {
-    state.writeDesiredSize = state.highWaterMark;
+    state.writeDesiredSize = state.budget;
   }
 }
 
@@ -1699,23 +1699,23 @@ class QuicStream {
   }
 
   /**
-   * The high water mark for write backpressure. When the total queued
+   * The byte budget for write backpressure. When the total queued
    * outbound bytes exceeds this value, writeSync returns false and
-   * desiredSize drops to 0. Default is 65536 (64KB).
+   * canWrite returns false. Default is 65536 (64KB).
    * @type {number}
    */
-  get highWaterMark() {
+  get budget() {
     assertIsQuicStream(this);
-    return this.#inner.state.highWaterMark;
+    return this.#inner.state.budget;
   }
 
-  set highWaterMark(val) {
+  set budget(val) {
     assertIsQuicStream(this);
-    validateInteger(val, 'highWaterMark', 0, 0xFFFFFFFF);
+    validateInteger(val, 'budget', 0, 0xFFFFFFFF);
     const inner = this.#inner;
-    inner.state.highWaterMark = val;
+    inner.state.budget = val;
     // If writeDesiredSize hasn't been set yet (still 0 from initialization),
-    // initialize it to the highWaterMark so the first write can proceed.
+    // initialize it to the budget so the first write can proceed.
     if (inner.state.writeDesiredSize === 0 && val > 0) {
       inner.state.writeDesiredSize = val;
     }
@@ -2163,8 +2163,8 @@ class QuicStream {
       // will accept the data into the DataQueue and
       // UpdateWriteDesiredSize() will drop writeDesiredSize toward 0,
       // at which point the standard drain mechanism takes over.
-      // This follows the Web Streams model where writes beyond the HWM
-      // succeed and backpressure applies to *subsequent* writes.
+      // This follows the iter-streams model where writes beyond the
+      // budget succeed and backpressure applies to *subsequent* writes.
       if (stream.#inner.state.writeDesiredSize === 0) return false;
       const result = handle.write([chunk]);
       if (result === undefined) return false;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
     const writer = {
       __proto__: null,
-      get desiredSize() {
+      get canWrite() {
         if (closed || errored || stream.#inner.state.writeEnded) return null;
-        return stream.#inner.state.writeDesiredSize;
+        return stream.#inner.state.writeDesiredSize > 0;
       },
       writeSync,
       write,
@@ -3254,7 +3254,7 @@ class QuicSession {
       body,
       priority = 'default',
       incremental = false,
-      highWaterMark = kDefaultHighWaterMark,
+      budget = kDefaultBudget,
       headers,
       onheaders,
       ontrailers,
@@ -3290,8 +3290,8 @@ class QuicSession {
       stream[kAttachFileHandle](body);
     }
 
-    // Set the high water mark for backpressure.
-    stream.highWaterMark = highWaterMark;
+    // Set the byte budget for backpressure.
+    stream.budget = budget;
 
     // Set stream callbacks before sending headers to avoid missing events.
     if (onheaders) stream.onheaders = onheaders;
@@ -4047,8 +4047,8 @@ class QuicSession {
     const stream = new QuicStream(kPrivateConstructor, handle, this, direction,
                                   false /* isLocal */);
 
-    // Set the default high water mark for received streams.
-    stream.highWaterMark = kDefaultHighWaterMark;
+    // Set the default byte budget for received streams.
+    stream.budget = kDefaultBudget;
 
     // A new stream was received. If we don't have an onstream callback, then
     // there's nothing we can do about it. Destroy the stream in this case.

lib/internal/quic/state.js

@@ -104,7 +104,7 @@ const {
   IDX_STATE_STREAM_WANTS_TRAILERS,
   IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
   IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
-  IDX_STATE_STREAM_HIGH_WATER_MARK,
+  IDX_STATE_STREAM_BUDGET,
   IDX_STATE_STREAM_RESET_CODE,
 } = internalBinding('quic');
 
@@ -871,18 +871,18 @@ class QuicStreamState {
   }
 
   /** @type {number} */
-  get highWaterMark() {
+  get budget() {
     const handle = this.#handle;
     if (handle === undefined) return undefined;
     return DataViewPrototypeGetUint32(
-      handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, kIsLittleEndian);
+      handle, this.#offset + IDX_STATE_STREAM_BUDGET, kIsLittleEndian);
   }
 
-  set highWaterMark(val) {
+  set budget(val) {
     const handle = this.#handle;
     if (handle === undefined) return;
     DataViewPrototypeSetUint32(
-      handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, val, kIsLittleEndian);
+      handle, this.#offset + IDX_STATE_STREAM_BUDGET, val, kIsLittleEndian);
   }
 
   toString() {
@@ -908,7 +908,7 @@ class QuicStreamState {
       early,
       resetCode,
       writeDesiredSize,
-      highWaterMark,
+      budget,
     } = this;
     return {
       __proto__: null,
@@ -928,7 +928,7 @@ class QuicStreamState {
       early,
       resetCode,
       writeDesiredSize,
-      highWaterMark,
+      budget,
     };
   }
 
@@ -965,7 +965,7 @@ class QuicStreamState {
       early,
       resetCode,
       writeDesiredSize,
-      highWaterMark,
+      budget,
     } = this;
 
     return `QuicStreamState ${inspect({
@@ -985,7 +985,7 @@ class QuicStreamState {
       early,
       resetCode,
       writeDesiredSize,
-      highWaterMark,
+      budget,
     }, opts)}`;
   }
 

lib/internal/streams/iter/broadcast.js

@@ -55,7 +55,6 @@ const {
 const {
   kMultiConsumerDefaultBudget,
   kResolvedPromise,
-  clampBudget,
   convertChunks,
   getWriterSignal,
   getMinCursor,
@@ -779,7 +778,7 @@ function broadcast(options = { __proto__: null }) {
 
   const opts = {
     __proto__: null,
-    budget: clampBudget(budget),
+    budget,
     backpressure,
     signal,
   };

lib/internal/streams/iter/push.js

@@ -37,7 +37,6 @@ const {
 const {
   kPushDefaultBudget,
   kResolvedPromise,
-  clampBudget,
   onSignalAbort,
   toUint8Array,
   convertChunks,
@@ -99,7 +98,7 @@ class PushQueue {
     if (signal !== undefined) {
       validateAbortSignal(signal, 'options.signal');
     }
-    this.#budget = clampBudget(budget);
+    this.#budget = budget;
     this.#backpressure = backpressure;
     this.#signal = signal;
     this.#abortHandler = undefined;
@@ -127,7 +126,12 @@ class PushQueue {
     if (this.#writerState !== 'open' || this.#consumerState !== 'active') {
       return null;
     }
-    return this.#bufferedBytes < this.#budget;
+    if ((this.#backpressure === 'strict' ||
+          this.#backpressure === 'unbounded') &&
+        this.#bufferedBytes >= this.#budget) {
+      return false;
+    }
+    return true;
   }
 
   /**
@@ -156,6 +160,10 @@ class PushQueue {
 
     const batchSize = this.#batchByteSize(chunks);
 
+    // Skip empty chunks -- zero-byte writes would accumulate infinitely
+    // without ever triggering backpressure under a byte-budget model.
+    if (batchSize === 0) return true;
+
     if (this.#bufferedBytes >= this.#budget) {
       switch (this.#backpressure) {
         case 'strict':
@@ -181,6 +189,11 @@ class PushQueue {
     this.#bytesWritten += batchSize;
 
     this.#resolvePendingReads();
+    // After drop-oldest, evicting a large chunk may bring us under budget.
+    // Resolve pending drains so writers waiting on backpressure can proceed.
+    if (this.#bufferedBytes < this.#budget) {
+      this.#resolvePendingDrains(true);
+    }
     return true;
   }
 

lib/internal/streams/iter/share.js

@@ -36,7 +36,6 @@ const {
 
 const {
   kMultiConsumerDefaultBudget,
-  clampBudget,
   getMinCursor,
   hasProtocol,
   onSignalAbort,
@@ -696,7 +695,7 @@ function share(source, options = { __proto__: null }) {
 
   const opts = {
     __proto__: null,
-    budget: clampBudget(budget),
+    budget,
     backpressure,
     signal,
   };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
   const opts = {
     __proto__: null,
-    budget: clampBudget(budget),
+    budget,
     backpressure,
   };