Version
main
Platform
Subsystem
stream
What steps will reproduce the bug?
import{share}from'node:stream/iter';asyncfunction*source(){for(leti=0;i<4;i++)yield[newTextEncoder().encode(`${i}`)];}constshared=share(source(),{highWaterMark: 2,backpressure: 'drop-newest',});constfast=shared.pull();constslow=shared.pull();constread=async(stream)=>(awaitArray.fromAsync(stream)).flat().map((chunk)=>newTextDecoder().decode(chunk));console.log('fast:',awaitread(fast));console.log('bufferSize:',shared.bufferSize);console.log('slow:',awaitread(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: 2slow: [ '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: 4slow: [ '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