Skip to content

Commit f8d77be

Browse files
committed
stream: skip zero-byte broadcast writes
Zero-byte writes append entries to the broadcast buffer without increasing bufferedBytes. This allows write('') and writev([]) to bypass backpressure and grow the buffer without bound. Treat zero-byte batches as successful no-ops, matching push streams, and add coverage for the synchronous and asynchronous writer methods. Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: codex:gpt-5.6-sol
1 parent 54a5095 commit f8d77be

2 files changed

Lines changed: 27 additions & 1 deletion

File tree

lib/internal/streams/iter/broadcast.js

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -274,6 +274,12 @@ class BroadcastImpl {
274274
[kWrite](chunk) {
275275
if (this.#ended || this.#cancelled) return false;
276276

277+
const batchSize = this.#batchByteSize(chunk);
278+
279+
// Skip empty chunks -- zero-byte writes would accumulate infinitely
280+
// without ever triggering backpressure under a byte-budget model.
281+
if (batchSize === 0) return true;
282+
277283
if (this.#bufferedBytes >= this.#options.budget) {
278284
switch (this.#options.backpressure) {
279285
case 'strict':
@@ -299,7 +305,6 @@ class BroadcastImpl {
299305
}
300306
}
301307

302-
const batchSize = this.#batchByteSize(chunk);
303308
this.#buffer.push(chunk);
304309
this.#bufferedBytes += batchSize;
305310
this.#notifyConsumers();

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,26 @@ async function testWritevAsync() {
147147
assert.strictEqual(data, 'hello world');
148148
}
149149

150+
// Zero-byte writes do not consume buffer entries.
151+
async function testZeroByteWrites() {
152+
const { writer, broadcast: bc } = broadcast({ budget: 16384 });
153+
const consumer = bc.push();
154+
155+
for (let i = 0; i < 1000; i++) {
156+
assert.strictEqual(writer.writeSync(''), true);
157+
assert.strictEqual(writer.writevSync([]), true);
158+
}
159+
await writer.write('');
160+
await writer.writev([]);
161+
assert.strictEqual(writer.canWrite, true);
162+
writer.endSync();
163+
164+
let entries = 0;
165+
const iterator = consumer[Symbol.asyncIterator]();
166+
while (!(await iterator.next()).done) entries++;
167+
assert.strictEqual(entries, 0);
168+
}
169+
150170
// endSync returns the total byte count
151171
async function testEndSyncReturnValue() {
152172
const { writer, broadcast: bc } = broadcast({ budget: 16384 });
@@ -165,5 +185,6 @@ Promise.all([
165185
testBlockBackpressureContent(),
166186
testStrictBackpressureOverflow(),
167187
testWritevAsync(),
188+
testZeroByteWrites(),
168189
testEndSyncReturnValue(),
169190
]).then(common.mustCall());

0 commit comments

Comments
 (0)