.pipe() and .pipeTo() fail when the source stream is constructed from a Buffer

[gabrielschulhof]

Version

v22.9.0, v23.4.0

Platform

Darwin XXXXXXXXXX 24.2.0 Darwin Kernel Version 24.2.0: Fri Dec  6 19:01:59 PST 2024; root:xnu-11215.61.5~2/RELEASE_ARM64_T6000 arm64

Subsystem

stream

What steps will reproduce the bug?

const path = require('node:path');
const fs = require('node:fs');
const { Readable, Writable } = require('node:stream');

const fileBuffer = fs.readFileSync(__filename);
const readable = fs.createReadStream(path.resolve(__filename));
const writable = fs.createWriteStream(`${__filename}_copy`);
const webReadable = Readable.toWeb(readable);
const webWritable = Writable.toWeb(writable);
const webReadableFromBuffer = ReadableStream.from(fileBuffer);
const readableFromBuffer = Readable.from(fileBuffer);
const readableFromWebReadableFromBuffer = Readable.from(webReadableFromBuffer);
const readableFromWebReadableFromReadable = Readable.from(webReadable);

// readable.pipe(writable);
// OK

// readableFromBuffer.pipe(writable);
// OK

// readableFromWebReadableFromReadable.pipe(writable);
// OK

// webReadable.pipeTo(webWritable);
// TypeError [ERR_INVALID_STATE]: Invalid state: The ReadableStream is locked

// webReadableFromBuffer.pipeTo(webWritable);
// TypeError [ERR_INVALID_STATE]: Invalid state: The ReadableStream is locked

// readableFromWebReadableFromBuffer.pipe(writable);
// TypeError [ERR_INVALID_ARG_TYPE]: The "chunk" argument must be of type string or an instance of Buffer, TypedArray, or DataView. Received type number (60)

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

Every time. No preconditions.

What is the expected behavior? Why is that the expected behavior?

All the above combinations should succeed.

What do you see instead?

Errors in the cases where the Readable or ReadableStream is constructed from a Buffer.

Additional information

No response

[gabrielschulhof]

Potentially related: #56139

[gabrielschulhof]

Also potentially related: #44866

[gabrielschulhof]

Creating a new stream each time solves the problem of the ReadableStream being locked:

const path = require('node:path');
const fs = require('node:fs');
const { Readable, Writable } = require('node:stream');

const fileBuffer = () => fs.readFileSync(__filename);
const readable = () => fs.createReadStream(path.resolve(__filename));
const writable = () => fs.createWriteStream(`${__filename}_copy`);
const webReadable = () => Readable.toWeb(readable());
const webWritable = () => Writable.toWeb(writable());
const webReadableFromBuffer = () => ReadableStream.from(fileBuffer());
const readableFromBuffer = () => Readable.from(fileBuffer());
const readableFromWebReadableFromBuffer = () => Readable.from(webReadableFromBuffer());

// If we don't return a new webReadable every time, the one we create gets
// locked at this point, because it is used exclusively by the Readable that
// wraps it.
const readableFromWebReadableFromReadable = () => Readable.from(webReadable());

// readable().pipe(writable());
// OK

// readableFromBuffer().pipe(writable());
// OK

// readableFromWebReadableFromReadable().pipe(writable());
// OK

// webReadable().pipeTo(webWritable());
// OK

// webReadableFromBuffer().pipeTo(webWritable());
// TypeError [ERR_INVALID_ARG_TYPE]: The "chunk" argument must be of type string or an instance of Buffer, TypedArray, or DataView. Received type number (99)

// readableFromWebReadableFromBuffer().pipe(writable());
// TypeError [ERR_INVALID_ARG_TYPE]: The "chunk" argument must be of type string or an instance of Buffer, TypedArray, or DataView. Received type number (99)

[gabrielschulhof]

As best I can tell,

const wr = webReadableFromBuffer()
;(async () => {
for await (const chunk of wr) {
  console.log(`|${JSON.stringify(chunk, null, 2)}|`)
}
})()

outputs a sequence of numbers, whereas doing the same loop over a ReadableStream in a browser outputs a sequence of Uint8Array objects like

|{
  "0": 87,
  "1": 119,
}|
|{
  "0": 32,
  "1": 48,
  "2": 51,
}|

This may be the discrepancy.

[gabrielschulhof]

OK. Need to pass a sequence of buffers to ReadableStream.from().