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

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);
  }
}