stream/iter: pipeTo retries accepted PushWriter writes under block backpressure #63296

import { pipeTo, push, text } from "node:stream/iter";

const enc = new TextEncoder();

async function repro(source) {
  const { writer, readable } = push({
    highWaterMark: 1,
    backpressure: "block",
  });

  const piping = pipeTo(source(), writer);

  // Let pipeTo() fill the PushWriter queue before the reader drains it.
  await new Promise(setImmediate);

  const output = await text(readable);
  await piping;
  return output;
}

async function* writevSource() {
  yield [enc.encode("ab")];
  yield [enc.encode("c"), enc.encode("d")];
}

async function* writeSource() {
  yield [enc.encode("a")];
  yield [enc.encode("b")];
}

console.log(await repro(writevSource)); // actual: abcdcd, expected: abcd
console.log(await repro(writeSource)); // actual: abbb, expected: ab