stream: using an async iterator on a pipeline works on v18 but in v19 needs Readable.from(asyncItFn)

[ErickWendel]

Version

19.3

Platform

darwin

Subsystem

No response

What steps will reproduce the bug?

The code below works on v18.13.0 but crashes on v19.3.0

import { pipeline } from 'node:stream/promises'
async function* myCustomReadable() {
    yield Buffer.from(`tick: ${new Date().toISOString()}`)
}

async function* myCustomWritable(stream) {
  for await (const chunk of stream) {
    console.log(chunk.toString())
  }
}

await pipeline(
  myCustomReadable,
  myCustomWritable,
)

node:internal/streams/readable:192
  const isDuplex = this instanceof Stream.Duplex;
                        ^

TypeError: Right-hand side of 'instanceof' is not an object
    at PassThrough.Readable (node:internal/streams/readable:192:25)
    at PassThrough.Duplex (node:internal/streams/duplex:58:12)
    at PassThrough.Transform (node:internal/streams/transform:106:10)
    at new PassThrough (node:internal/streams/passthrough:42:13)
    at pipelineImpl (node:internal/streams/pipeline:280:20)
    at node:stream/promises:28:5
    at new Promise (<anonymous>)
    at pipeline (node:stream/promises:17:10)
    at file:///Users/erickwendel/Downloads/projetos/cursos/mastering-streams/mastering-streams-code/example.mjs:12:7
    at ModuleJob.run (node:internal/modules/esm/module_job:194:25)

Node.js v19.3.0

but it works if explicitly convert it to a Readable

await pipeline(
  Readable.from(myCustomReadable()),
  myCustomWritable,
)

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

No response

What is the expected behavior?

No response

What do you see instead?

TypeError: Right-hand side of 'instanceof' is not an object

Additional information

@nodejs/streams @nodejs/tooling

[ronag]

Adding import 'node:stream' at the top fixes it :/

[ronag]

@mcollina the problem here is all the global stream init logic lives in 'stream', so at the moment you must first import 'stream' before anything else under 'stream/...'

[ErickWendel]

OMG 😂

I'd like to help working on it. Do you know where to look at?

[ronag]

node/lib/stream.js
node/lib/stream/promises.js

[ErickWendel]

@mcollina the problem here is all the global stream init logic lives in 'stream', so at the moment you must first import 'stream' before anything else under 'stream/...'

do other bugs are happening because of it?

[ErickWendel]

Analyzing here it works without going on 'node:stream/promises':

import { promises } from 'node:stream'
const { pipeline } = promises
async function* myCustomReadable() {
  yield Buffer.from(`tick: ${new Date().toISOString()}`)
}

async function* myCustomWritable(stream) {
  for await (const chunk of stream) {
    console.log(chunk.toString())
  }
}

await pipeline(
  myCustomReadable,
  myCustomWritable,
)

But looking at the code, here is the problem:

if I change from require('internal/streams/passthrough') to require('stream').Passthrough it works. it seems to be something on the loader side

Gonna take a look at it tonight