diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 4f6bdc9f348a36..9b0ff9c99c4fc7 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -274,6 +274,12 @@ class BroadcastImpl { [kWrite](chunk) { if (this.#ended || this.#cancelled) return false; + const batchSize = this.#batchByteSize(chunk); + + // Skip empty chunks -- zero-byte writes would accumulate infinitely + // without ever triggering backpressure under a byte-budget model. + if (batchSize === 0) return true; + if (this.#bufferedBytes >= this.#options.budget) { switch (this.#options.backpressure) { case 'strict': @@ -299,7 +305,6 @@ class BroadcastImpl { } } - const batchSize = this.#batchByteSize(chunk); this.#buffer.push(chunk); this.#bufferedBytes += batchSize; this.#notifyConsumers(); diff --git a/test/parallel/test-stream-iter-broadcast-backpressure.js b/test/parallel/test-stream-iter-broadcast-backpressure.js index 0f2d8cc2cbc863..35a0cf0e238c60 100644 --- a/test/parallel/test-stream-iter-broadcast-backpressure.js +++ b/test/parallel/test-stream-iter-broadcast-backpressure.js @@ -147,6 +147,26 @@ async function testWritevAsync() { assert.strictEqual(data, 'hello world'); } +// Zero-byte writes do not consume buffer entries. +async function testZeroByteWrites() { + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); + const consumer = bc.push(); + + for (let i = 0; i < 1000; i++) { + assert.strictEqual(writer.writeSync(''), true); + assert.strictEqual(writer.writevSync([]), true); + } + await writer.write(''); + await writer.writev([]); + assert.strictEqual(writer.canWrite, true); + writer.endSync(); + + let entries = 0; + const iterator = consumer[Symbol.asyncIterator](); + while (!(await iterator.next()).done) entries++; + assert.strictEqual(entries, 0); +} + // endSync returns the total byte count async function testEndSyncReturnValue() { const { writer, broadcast: bc } = broadcast({ budget: 16384 }); @@ -165,5 +185,6 @@ Promise.all([ testBlockBackpressureContent(), testStrictBackpressureOverflow(), testWritevAsync(), + testZeroByteWrites(), testEndSyncReturnValue(), ]).then(common.mustCall());