Skip to content

Commit 4b83c6c

Browse files
jasnellrichardlau
jasnell
authored and
richardlau
committed
stream: accept explicit stream/iter budgets below 16384
push(), duplex(), broadcast(), share() and shareSync() rejected an explicit budget below 16384 bytes with ERR_OUT_OF_RANGE. The spec (push() step 3, broadcast() step 1 and share() step 2) only requires the implementation-defined default to be at least 16384 bytes; an explicit budget is used as given. Small budgets are also useful in tests and in memory-constrained code. Accept any explicit budget of at least 1 byte, and validate it in one place. Defaults are unchanged. The validation tests are updated to the new lower bound; note that WebIDL conversion truncates fractions, so 1.5 is now a valid budget of 1 and 0.5 is used to exercise the rejection instead. Assisted-by: OpenCode Signed-off-by: James M Snell <[email protected]> PR-URL: #66483 Reviewed-By: Trivikram Kamat <[email protected]>
1 parent f302fbd commit 4b83c6c

7 files changed

Lines changed: 62 additions & 38 deletions

File tree

‎doc/api/stream_iter.md‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -801,7 +801,7 @@ added: v25.9.0
801801
readable side.
802802
* `options` {Object}
803803
* `budget` {number} Maximum number of buffered bytes before
804-
backpressure is applied. Must be >= 16384.
804+
backpressure is applied. Must be a positive integer.
805805
**Default:** `16384`.
806806
* `backpressure` {string} Backpressure policy: `'strict'`, `'unbounded'`,
807807
`'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`.
@@ -865,7 +865,7 @@ added: v25.9.0
865865

866866
* `options` {Object}
867867
* `budget` {number} Buffer size in bytes for both directions.
868-
Must be >= 16384. **Default:** `16384`.
868+
Must be a positive integer. **Default:** `16384`.
869869
* `backpressure` {string} Policy for both directions.
870870
**Default:** `'strict'`.
871871
* `signal` {AbortSignal} Cancellation signal for both channels.
@@ -1323,7 +1323,7 @@ added: v25.9.0
13231323
-->
13241324

13251325
* `options` {Object}
1326-
* `budget` {number} Buffer size in bytes. Must be >= 16384.
1326+
* `budget` {number} Buffer size in bytes. Must be a positive integer.
13271327
**Default:** `65536`.
13281328
* `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or
13291329
`'drop-newest'`. **Default:** `'strict'`.
@@ -1433,7 +1433,7 @@ added: v25.9.0
14331433

14341434
* `source` {AsyncIterable} The source to share.
14351435
* `options` {Object}
1436-
* `budget` {number} Buffer size in bytes. Must be >= 16384.
1436+
* `budget` {number} Buffer size in bytes. Must be a positive integer.
14371437
**Default:** `65536`.
14381438
* `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or
14391439
`'drop-newest'`. **Default:** `'strict'`.
@@ -1559,7 +1559,7 @@ added: v25.9.0
15591559

15601560
* `source` {Iterable} The sync source to share.
15611561
* `options` {Object}
1562-
* `budget` {number} Must be >= 16384.
1562+
* `budget` {number} Must be a positive integer.
15631563
**Default:** `65536`.
15641564
* `backpressure` {string} `'strict'` or `'drop-oldest'`.
15651565
**Default:** `'strict'`.

‎lib/internal/streams/iter/broadcast.js‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -40,9 +40,6 @@ const {
4040
ERR_INVALID_STATE,
4141
},
4242
} = require('internal/errors');
43-
const {
44-
validateInteger,
45-
} = require('internal/validators');
4643

4744
const {
4845
broadcastProtocol,
@@ -72,6 +69,7 @@ const {
7269
toWriterUint8Array,
7370
validateBatchEntry,
7471
yieldAbortable,
72+
validateBudget,
7573
} = require('internal/streams/iter/utils');
7674
const {
7775
converters,
@@ -895,7 +893,7 @@ function broadcast(options = { __proto__: null }) {
895893
backpressure = 'strict',
896894
signal,
897895
} = options;
898-
validateInteger(budget, 'options.budget', 16384);
896+
validateBudget(budget);
899897

900898
const opts = {
901899
__proto__: null,

‎lib/internal/streams/iter/push.js‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,6 @@ const {
2222
ERR_INVALID_STATE,
2323
},
2424
} = require('internal/errors');
25-
const {
26-
validateInteger,
27-
} = require('internal/validators');
2825

2926
const {
3027
drainableProtocol,
@@ -40,6 +37,7 @@ const {
4037
getWriterSignal,
4138
parsePullArgs,
4239
validateBatchEntry,
40+
validateBudget,
4341
} = require('internal/streams/iter/utils');
4442
const {
4543
converters,
@@ -132,7 +130,7 @@ class PushQueue {
132130
backpressure = 'strict',
133131
signal,
134132
} = options;
135-
validateInteger(budget, 'options.budget', 16384);
133+
validateBudget(budget);
136134
this.#budget = budget;
137135
this.#backpressure = backpressure;
138136
this.#signal = signal;

‎lib/internal/streams/iter/share.js‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ const {
4646
parsePullArgs,
4747
splitBatchEntry,
4848
validateBatchEntry,
49+
validateBudget,
4950
} = require('internal/streams/iter/utils');
5051
const {
5152
converters,
@@ -64,9 +65,6 @@ const {
6465
ERR_OUT_OF_RANGE,
6566
},
6667
} = require('internal/errors');
67-
const {
68-
validateInteger,
69-
} = require('internal/validators');
7068

7169
const { markPromiseAsHandled } = internalBinding('util');
7270

@@ -865,7 +863,7 @@ function share(source, options = { __proto__: null }) {
865863
backpressure = 'strict',
866864
signal,
867865
} = options;
868-
validateInteger(budget, 'options.budget', 16384);
866+
validateBudget(budget);
869867

870868
const opts = {
871869
__proto__: null,
@@ -894,7 +892,7 @@ function shareSync(source, options = { __proto__: null }) {
894892
budget = kMultiConsumerDefaultBudget,
895893
backpressure = 'strict',
896894
} = options;
897-
validateInteger(budget, 'options.budget', 16384);
895+
validateBudget(budget);
898896
// A synchronous consumer can neither wait for the slowest consumer to
899897
// release budget ('unbounded') nor keep pulling and discarding until it
900898
// does ('drop-newest'): the slowest consumer cannot advance while the

‎lib/internal/streams/iter/utils.js‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ const {
3535
const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types');
3636

3737
const {
38+
validateInteger,
3839
validateOneOf,
3940
} = require('internal/validators');
4041
const {
@@ -236,6 +237,15 @@ function validateByteView(snapshot) {
236237
return value;
237238
}
238239

240+
/**
241+
* Validate an explicit `budget` option. The spec only requires the default
242+
* budget to be at least 16384 bytes; any positive explicit budget is valid.
243+
* @param {unknown} budget
244+
*/
245+
function validateBudget(budget) {
246+
validateInteger(budget, 'options.budget', 1);
247+
}
248+
239249
/**
240250
* Append a chunk to `chunks` for later concatenation, and the information
241251
* needed to detect that it was resized or detached in the meantime to
@@ -550,6 +560,7 @@ module.exports = {
550560
toWriterUint8Array,
551561
validateBackpressure,
552562
validateBatchEntry,
563+
validateBudget,
553564
validateRecordedChunks,
554565
validateByteView,
555566
yieldAbortable,

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

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,11 +150,28 @@ async function testStrictPendingQueueOverflow() {
150150
await iter.return();
151151
}
152152

153+
async function testSmallBudget() {
154+
// An explicit budget below the 16384-byte default is honored.
155+
const { writer, readable } = push({ budget: 4 });
156+
assert.strictEqual(writer.writeSync('abc'), true);
157+
assert.strictEqual(writer.canWrite, true);
158+
// The buffer may overshoot the budget by one write.
159+
assert.strictEqual(writer.writeSync('defgh'), true);
160+
assert.strictEqual(writer.canWrite, false);
161+
assert.strictEqual(writer.writeSync('i'), false);
162+
const write = writer.write('ij');
163+
const result = text(readable);
164+
await write;
165+
writer.endSync();
166+
assert.strictEqual(await result, 'abcdefghij');
167+
}
168+
153169
Promise.all([
154170
testStrictBackpressure(),
155171
testDropOldest(),
156172
testDropNewest(),
157173
testBlockBackpressure(),
158174
testBlockWriteSyncDoesNotEnqueue(),
159175
testStrictPendingQueueOverflow(),
176+
testSmallBudget(),
160177
]).then(common.mustCall());

‎test/parallel/test-stream-iter-validation.js‎

Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,18 @@ const {
2121
// push() validation
2222
// =============================================================================
2323

24-
// Budget must be integer >= 16384
24+
// Budget must be an integer >= 1
2525
assert.throws(() => push({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' });
26-
assert.throws(() => push({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' });
27-
// Values < 16384 are rejected
26+
// WebIDL conversion truncates fractions: 0.5 becomes 0, 1.5 becomes 1.
27+
assert.throws(() => push({ budget: 0.5 }), { code: 'ERR_OUT_OF_RANGE' });
28+
assert.strictEqual(push({ budget: 1.5 }).writer.canWrite, true);
29+
// Values < 1 are rejected
2830
assert.throws(() => push({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });
2931
assert.throws(() => push({ budget: -1 }), { code: 'ERR_OUT_OF_RANGE' });
3032
assert.throws(() => push({ budget: -100 }), { code: 'ERR_OUT_OF_RANGE' });
31-
assert.throws(() => push({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
32-
// 16384 is the minimum accepted value
33+
// Only the default must be at least 16384; smaller explicit budgets are valid
34+
assert.strictEqual(push({ budget: 1 }).writer.canWrite, true);
35+
assert.strictEqual(push({ budget: 16383 }).writer.canWrite, true);
3336
assert.strictEqual(push({ budget: 16384 }).writer.canWrite, true);
3437
// MAX_SAFE_INTEGER is accepted
3538
assert.strictEqual(push({ budget: Number.MAX_SAFE_INTEGER }).writer.canWrite,
@@ -87,11 +90,11 @@ assert.throws(() => duplex({ b: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' });
8790

8891
// Budget validation (cascades through to push())
8992
assert.throws(() => duplex({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' });
90-
assert.throws(() => duplex({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' });
93+
assert.throws(() => duplex({ budget: 0.5 }), { code: 'ERR_OUT_OF_RANGE' });
9194
assert.throws(() => duplex({ budget: Number.MAX_SAFE_INTEGER + 1 }),
9295
{ code: 'ERR_OUT_OF_RANGE' });
9396

94-
// Values < 16384 are rejected (both directions)
97+
// Values < 1 are rejected (both directions)
9598
assert.throws(() => duplex({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });
9699
// MAX_SAFE_INTEGER is accepted
97100
{
@@ -129,17 +132,16 @@ assert.throws(() => pullSync(fromSync('a'), 42), { code: 'ERR_INVALID_ARG_TYPE'
129132
// =============================================================================
130133

131134
assert.throws(() => broadcast({ budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' });
132-
assert.throws(() => broadcast({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' });
135+
assert.throws(() => broadcast({ budget: 0.5 }), { code: 'ERR_OUT_OF_RANGE' });
133136
assert.throws(() => broadcast({ budget: Number.MAX_SAFE_INTEGER + 1 }),
134137
{ code: 'ERR_OUT_OF_RANGE' });
135138

136-
// Values < 16384 are rejected
139+
// Values < 1 are rejected
137140
assert.throws(() => broadcast({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });
138141
assert.throws(() => broadcast({ budget: -1 }), { code: 'ERR_OUT_OF_RANGE' });
139-
assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' });
140-
// 16384 is the minimum accepted value
142+
// Small explicit budgets are accepted
141143
{
142-
const bc = broadcast({ budget: 16384 });
144+
const bc = broadcast({ budget: 1 });
143145
bc.broadcast.push();
144146
assert.strictEqual(bc.writer.canWrite, true);
145147
bc.writer.endSync();
@@ -235,7 +237,7 @@ assert.throws(
235237

236238
assert.throws(() => share(42), { code: 'ERR_INVALID_ARG_TYPE' });
237239
assert.throws(() => share(from('a'), { budget: 'bad' }), { code: 'ERR_OUT_OF_RANGE' });
238-
assert.throws(() => share(from('a'), { budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' });
240+
assert.throws(() => share(from('a'), { budget: 0.5 }), { code: 'ERR_OUT_OF_RANGE' });
239241
assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
240242
{ code: 'ERR_OUT_OF_RANGE' });
241243
assert.throws(() => share(from('a'), { signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
@@ -250,28 +252,28 @@ assert.throws(() => share(from('a'), { backpressure: 'bad' }), { code: 'ERR_INVA
250252
assert.strictEqual(shared.consumerCount, 0);
251253
}
252254

253-
// share() values < 16384 are rejected
255+
// share() values < 1 are rejected
254256
assert.throws(() => share(from('a'), { budget: 0 }), { code: 'ERR_OUT_OF_RANGE' });
255257
assert.throws(() => share(from('a'), { budget: -1 }), { code: 'ERR_OUT_OF_RANGE' });
256-
// 16384 is the minimum, MAX_SAFE_INTEGER is accepted
257-
share(from('a'), { budget: 16384 }).cancel();
258+
// 1 is the minimum, MAX_SAFE_INTEGER is accepted
259+
share(from('a'), { budget: 1 }).cancel();
258260
share(from('a'), { budget: Number.MAX_SAFE_INTEGER }).cancel();
259261

260262
assert.throws(() => shareSync(42), { code: 'ERR_INVALID_ARG_TYPE' });
261263
assert.throws(() => shareSync(fromSync('a'), { budget: 'bad' }),
262264
{ code: 'ERR_OUT_OF_RANGE' });
263-
assert.throws(() => shareSync(fromSync('a'), { budget: 1.5 }),
265+
assert.throws(() => shareSync(fromSync('a'), { budget: 0.5 }),
264266
{ code: 'ERR_OUT_OF_RANGE' });
265267
assert.throws(() => shareSync(fromSync('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }),
266268
{ code: 'ERR_OUT_OF_RANGE' });
267269

268-
// shareSync() values < 16384 are rejected
270+
// shareSync() values < 1 are rejected
269271
assert.throws(() => shareSync(fromSync('a'), { budget: 0 }),
270272
{ code: 'ERR_OUT_OF_RANGE' });
271273
assert.throws(() => shareSync(fromSync('a'), { budget: -1 }),
272274
{ code: 'ERR_OUT_OF_RANGE' });
273-
// 16384 is the minimum, MAX_SAFE_INTEGER is accepted
274-
shareSync(fromSync('a'), { budget: 16384 }).cancel();
275+
// 1 is the minimum, MAX_SAFE_INTEGER is accepted
276+
shareSync(fromSync('a'), { budget: 1 }).cancel();
275277
shareSync(fromSync('a'), { budget: Number.MAX_SAFE_INTEGER }).cancel();
276278

277279
// Share.from / SyncShare.fromSync reject non-iterable

0 commit comments

Comments
 (0)