stream/iter: broadcast consumer can leave earlier next() pending after overlapping next() calls

[trivikr]

Version

26.2.0

Platform

macOS 26.5.0

Subsystem

stream

What steps will reproduce the bug?

import { broadcast } from 'node:stream/iter';

const { writer, broadcast: bc } = broadcast();
const it = bc.push()[Symbol.asyncIterator]();

const first = it.next();
const second = it.next();

await writer.write('x');

console.log('second next():', await second);

const result = await Promise.race([
 first.then((value) => ['settled', value]),
 new Promise((resolve) => setTimeout(resolve, 50, ['pending'])),
]);

console.log('first next():', ...result);

How often does it reproduce? Is there a required condition?

Always

What is the expected behavior? Why is that the expected behavior?

second next(): { done: true, value: undefined }
first next(): settled { done: false, value: [ Uint8Array(1) [ 120 ] ] }

What do you see instead?

second next(): { done: false, value: [ Uint8Array(1) [ 120 ] ] }
first next(): pending

Additional information

No response

[trivikr]

As per spec 13.1.3, the second next should be pending and not resolve.

Broadcast.push() creates a consumer cursor that “advances the cursor on each step, waiting for new data if needed” in the spec. One written chunk should satisfy the earliest pending next() for that iterator. The second next() has no second chunk to read, and the broadcast has not been ended/cancelled, so it should wait for new data. It should only resolve { done: true, value: undefined } after clean completion, for example writer.end() or broadcast.cancel() with no reason.

New repro

import { broadcast } from 'node:stream/iter';
import { setTimeout } from 'node:timers/promises';

const { writer, broadcast: bc } = broadcast();
const it = bc.push()[Symbol.asyncIterator]();

const first = it.next();
const second = it.next();

await writer.write('x');

const secondResult = await Promise.race([
 second.then((value) => ['settled', value]),
 setTimeout(50, ['pending']),
]);

console.log('second next():', ...secondResult);

const result = await Promise.race([
 first.then((value) => ['settled', value]),
 setTimeout(50, ['pending']),
]);

console.log('first next():', ...result);

Observed

second next(): settled [Object: null prototype] {
 done: false,
 value: [ Uint8Array(1) [ 120 ] ]
}
first next(): pending

Expected as previous ask

second next(): settled [Object: null prototype] { done: true, value: undefined }
first next(): settled [Object: null prototype] {
 done: false,
 value: [ Uint8Array(1) [ 120 ] ]
}

Expected (actual)

second next(): pending
first next(): settled [Object: null prototype] {
 done: false,
 value: [ Uint8Array(1) [ 120 ] ]
}