pipeline causes memory leak when used with async generator #37084

const { PassThrough, pipeline } = require('stream')
const { promisify } = require('util')
const nextTick = promisify(process.nextTick)

const consume = async (it) => {
  for await (const _item of it) continue
}

async function* genItems(count) {
  const source = async function* () {
    // Generate way too many items
    while (true) {
      await nextTick() // simulate async
      yield { foo: 'bar' }
    }
  }

  const duplex = new PassThrough({ objectMode: true })

  const stream = pipeline(source, duplex, (err) => {
    console.log('done', err) // err.code === ERR_STREAM_PREMATURE_CLOSE
  })

  for await (const item of stream) {
    yield item
    if (--count <= 0) break
  }
}

const test = async () => {
  const stream = genItems(2)
  await consume(stream)
}

const main = async () => {
  debugger // <== Start memory allocation timeline profiling now
  for (let i = 0; i < 10; i++) await test()
  global.gc()
  debugger // <== Stop memory profiling now
}

main()