stream/iter: Broadcast and Share leak consumers for invalid or pre-aborted signals

[trivikr]

Version

main

Platform

N/A

Subsystem

stream

What steps will reproduce the bug?

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

const cases = [
 ['Broadcast', () => {
 const { broadcast: bc } = broadcast();
 return [bc, (options) => bc.push(options)];
 }],
 ['Share', () => {
 const shared = share(from('x'));
 return [shared, (options) => shared.pull(options)];
 }],
];

for (const [name, make] of cases) {
 let [owner, consume] = make();

 try {
 consume({ signal: {} });
 } catch (error) {
 console.log(name, 'invalid signal:', error.code,
 'consumerCount:', owner.consumerCount);
 }

 [owner, consume] = make();

 const controller = new AbortController();
 controller.abort(new Error('already aborted'));

 const iterator =
 consume({ signal: controller.signal })[Symbol.asyncIterator]();

 try {
 await iterator.next();
 } catch (error) {
 console.log(name, 'pre-aborted signal:', error.message,
 'consumerCount:', owner.consumerCount);
 }
}

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

Always

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

Signal validation or pre-abort handling should not register a consumer - or should immediately detach it - so consumerCount remains 0.

As per §13.1.3, Broadcast.push() and §13.2.3, Share.pull()

On break/error, abort pipelineController and detach the cursor.

What do you see instead?

Broadcast invalid signal: ERR_INVALID_ARG_TYPE consumerCount: 1
Broadcast pre-aborted signal: already aborted consumerCount: 1
Share invalid signal: ERR_INVALID_ARG_TYPE consumerCount: 1
Share pre-aborted signal: already aborted consumerCount: 1

Passing an invalid or already-aborted signal throws or rejects, but leaves an inaccessible consumer registered with consumerCount === 1, potentially causing permanent backpressure.

Additional information

No response