stream/iter: BroadcastWriter.end() strands backpressured writes and truncates the broadcast

[trivikr]

Version

main

Platform

N/A

Subsystem

stream

What steps will reproduce the bug?

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

const { writer, broadcast: bc } = broadcast({
 budget: 16384,
 backpressure: 'unbounded',
});
const reader = bc.push()[Symbol.asyncIterator]();

await writer.write(new Uint8Array(16384));

const pendingWrite = writer.write(Uint8Array.of(1));
const endPromise = writer.end();

console.log('first read done:', (await reader.next()).done);

const writeState = await Promise.race([
 pendingWrite.then(
 () => 'resolved',
 () => 'rejected',
 ),
 setTimeout(100, 'still pending'),
]);
console.log('blocked write:', writeState);

console.log('second read done:', (await reader.next()).done);
console.log('final read done:', (await reader.next()).done);
console.log('end():', await endPromise);

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

Always

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

first read done: false
blocked write: resolved
second read done: false
final read done: true
end(): 16385

Writes queued before end() should remain ahead of end-of-stream. The pending write should be promoted when the first read frees capacity, and end() should resolve only after the consumer reaches the final end sentinel.

As per §7.2.2, Writer.write():

“return a promise that resolves when the batch is transferred to the slots buffer.”

And §7.2.5, Writer.end() / Writer.endSync():

“signals end-of-stream and waits for buffered data to drain.”

What do you see instead?

first read done: false
blocked write: still pending
second read done: true
final read done: true
end(): 16384

After end() marks the broadcast ended, freeing buffer capacity cannot promote the queued write. Its promise remains pending, the byte is never delivered, and end() reports only the first write.

Additional information

No response