stream, quic: update iterable streams backpressure

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode/Opus
PR-URL: #64464
Reviewed-By: Tim Perry <pimterry@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: jasnell

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };
 

benchmark/streams/iter-throughput-share.js

@@ -5,7 +5,7 @@ const common = require('../common.js');
 constbench=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(leti=0;i<n;i++){
-constshared=share(source(),{highWaterMark: 64, backpressure });
+constshared=share(source(),{budget: 65536, backpressure });
 constreaders=Array.from({length: consumers},()=>array(shared.pull()));
 awaitPromise.all(readers);
 }

doc/api/quic.md

@@ -1276,7 +1276,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).
@@ -1317,7 +1317,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).
@@ -1924,7 +1924,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: REPLACEME
@@ -2236,7 +2236,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

@@ -223,7 +223,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) |
@@ -232,8 +232,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
  */
-constkDefaultHighWaterMark=65536;
+constkDefaultBudget=65536;
 constkDefaultMaxPendingDatagrams=128;
 
 functionconfigureOutbound(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.
 functioninitStreamingBackpressure(stream){
 conststate=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}
  */
-gethighWaterMark(){
+getbudget(){
 assertIsQuicStream(this);
-returnthis.#inner.state.highWaterMark;
+returnthis.#inner.state.budget;
 }
 
-sethighWaterMark(val){
+setbudget(val){
 assertIsQuicStream(this);
-validateInteger(val,'highWaterMark',0,0xFFFFFFFF);
+validateInteger(val,'budget',0,0xFFFFFFFF);
 constinner=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)returnfalse;
 constresult=handle.write([chunk]);
 if(result===undefined)returnfalse;
@@ -2323,9 +2323,9 @@ class QuicStream {
 
 constwriter={
 __proto__: null,
-getdesiredSize(){
+getcanWrite(){
 if(closed||errored||stream.#inner.state.writeEnded)returnnull;
-returnstream.#inner.state.writeDesiredSize;
+returnstream.#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 {
 conststream=newQuicStream(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} */
-gethighWaterMark(){
+getbudget(){
 consthandle=this.#handle;
 if(handle===undefined)returnundefined;
 returnDataViewPrototypeGetUint32(
-handle,this.#offset +IDX_STATE_STREAM_HIGH_WATER_MARK,kIsLittleEndian);
+handle,this.#offset +IDX_STATE_STREAM_BUDGET,kIsLittleEndian);
 }
 
-sethighWaterMark(val){
+setbudget(val){
 consthandle=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 }) {
 
 constopts={
 __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'){
 returnnull;
 }
-returnthis.#bufferedBytes <this.#budget;
+if((this.#backpressure ==='strict'||
+this.#backpressure ==='unbounded')&&
+this.#bufferedBytes >=this.#budget){
+returnfalse;
+}
+returntrue;
 }
 
 /**
@@ -156,6 +160,10 @@ class PushQueue {
 
 constbatchSize=this.#batchByteSize(chunks);
 
+// Skip empty chunks -- zero-byte writes would accumulate infinitely
+// without ever triggering backpressure under a byte-budget model.
+if(batchSize===0)returntrue;
+
 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);
+}
 returntrue;
 }
 

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 }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
  signal,
 };
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
 
 constopts={
 __proto__: null,
-budget: clampBudget(budget),
+ budget,
  backpressure,
 };