Version
main
Platform
Subsystem
stream
What steps will reproduce the bug?
import { share } from 'node:stream/iter';
async function* source() {
for (let i = 0; i < 4; i++)
yield [new TextEncoder().encode(`${i}`)];
}
const shared = share(source(), {
highWaterMark: 2,
backpressure: 'drop-newest',
});
const fast = shared.pull();
const slow = shared.pull();
const read = async (stream) => (await Array.fromAsync(stream))
.flat()
.map((chunk) => new TextDecoder().decode(chunk));
console.log('fast:', await read(fast));
console.log('bufferSize:', shared.bufferSize);
console.log('slow:', await read(slow));
How often does it reproduce? Is there a required condition?
Always
What is the expected behavior? Why is that the expected behavior?
fast: [ '0', '1' ]
bufferSize: 2
slow: [ '0', '1' ]
Once the two-slot buffer is full, upstream results '2' and '3' are discarded under "drop-newest".
Neither consumer receives them, and the buffer never exceeds highWaterMark.
From §13.2.2 — “Share buffering and backpressure”
With "drop-newest", the upstream pull result is discarded.
What do you see instead?
fast: [ '0', '1', '2', '3' ]
bufferSize: 4
slow: [ '0', '1', '2', '3' ]
The buffer grows to four entries despite highWaterMark: 2, and the stalled consumer eventually receives every batch
Additional information
No response
Version
main
Platform
Subsystem
stream
What steps will reproduce the bug?
How often does it reproduce? Is there a required condition?
Always
What is the expected behavior? Why is that the expected behavior?
Once the two-slot buffer is full, upstream results
'2'and'3'are discarded under"drop-newest".Neither consumer receives them, and the buffer never exceeds
highWaterMark.From §13.2.2 — “Share buffering and backpressure”
What do you see instead?
The buffer grows to four entries despite
highWaterMark: 2, and the stalled consumer eventually receives every batchAdditional information
No response