stream/iter: per-consumer AbortSignal is ignored by broadcast.push() and share.pull() fast path #63302

import { setTimeout as delay } from "node:timers/promises";
import { broadcast, share } from "node:stream/iter";

async function check(label, next) {
  console.log(
    await Promise.race([
      next.then(
        (value) => `${label}: settled ${JSON.stringify(value)}`,
        (error) => `${label}: rejected ${error?.name ?? error}`,
      ),
      delay(50, `${label}: still pending`),
    ]),
  );
}

{
  const ac = new AbortController();
  const { broadcast: bc } = broadcast();
  const it = bc.push({ signal: ac.signal })[Symbol.asyncIterator]();

  const next = it.next();
  ac.abort(new Error("abort"));

  await check("broadcast.push({ signal }).next()", next);
}

{
  const ac = new AbortController();
  const shared = share(
    (async function* never() {
      await new Promise(() => {});
    })(),
  );
  const it = shared.pull({ signal: ac.signal })[Symbol.asyncIterator]();

  const next = it.next();
  ac.abort(new Error("abort"));

  await check("share.pull({ signal }).next()", next);
}