`stream.Duplex.from` with rejecting promise leads to uncaught exception

[SimenB]

Version

v16.19.0, v19.3.0

Platform

Darwin Simens-MacBook-Pro.local 22.2.0 Darwin Kernel Version 22.2.0: Fri Nov 11 02:03:51 PST 2022; root:xnu-8792.61.2~4/RELEASE_ARM64_T6000 arm64

Subsystem

stream

What steps will reproduce the bug?

If passing an array of Promises to stream.Duplex.from and one rejects before all others are settled, an uncaught error is emitted. If all promises except for the rejected one are settled, then it correctly emits an error event on the stream instead.

import { Duplex } from 'node:stream';
import getStream from 'get-stream';

process.on('uncaughtException', error => {
  console.error('Got uncaught exception', error);

  process.exit(1);
});

try {
  const dup = Duplex.from([
    'hello',
    Promise.resolve('1'),
    Promise.resolve('2'),
    // Setting timeout to 500 correctly errors instead of emitting uncaught 
    new Promise(resolve => setTimeout(() => resolve('3'), 1500)),
    'end',
    new Promise((resolve, reject) =>
      setTimeout(() => reject(new Error('booo')), 1000),
    ),
  ]);

  const res = await getStream(dup);
  console.log(res);
} catch (error) {
  process.exitCode = 1;
  console.error('Got error', error);
}

get-stream: https://www.npmjs.com/package/get-stream (I don't know a good way with core stream to get a promise of a stream).

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

Always.

What is the expected behavior?

There should never be an uncaught error, it should always be emitted as an error event on the stream.

What do you see instead?

Uncaught error emitted.

Additional information

No response

[marco-ippolito]

It doesn't require an external library, this is a reproducible example:

import { Duplex } from 'node:stream';
import assert from "node:assert";

process.on('uncaughtException', () => {
 assert.fail('uncaught exception')
});

assert.doesNotThrow(() => {
    Duplex.from([
      new Promise((_, reject) => setTimeout(() => reject(new Error('1')), 1000)),
      new Promise(resolve => setTimeout(() => resolve('2'), 1500)),
    ]);
})

I'll try to fix it

[marco-ippolito]

The problem originates from using an Iterable that contains Promises.
Note that the documentation doesn’t strictly mention Arrays, which are supported because Duplex.from in this case is internally converted to Readable.from which only supports (async) Iterables.

Duplex.from([ new Promise(resolve => setTimeout(() => resolve('3'), 1500)), 'end', new Promise((resolve, reject) => setTimeout(() => reject(new Error('booo')), 1000), )]);

When executing from the code inside both promises is executed, the timeouts are set.

node/lib/internal/streams/from.js

Lines 83 to 110 in 6668c4d

	try {
	const { value, done } = isAsync ?
	await iterator.next() :
	iterator.next();
	if (done) {
	readable.push(null);
	} else {
	const res = (value &&
	typeof value.then === 'function') ?
	await value :
	value;
	if (res === null) {
	reading = false;
	throw new ERR_STREAM_NULL_VALUES();
	} else if (readable.push(res)) {
	continue;
	} else {
	reading = false;
	}
	}
	} catch (err) {
	readable.destroy(err);
	}
	break;
	}
	}
	return readable;

At line 93 while the loop is blocked awaiting the first promise, the timeout of the second promise is over and the callback is executed. Since the loop is still blocked on the first promise, the second promise rejects without a handler and results in uncaught exception.
It's not a bug, it's normal behaviour of Node.
@ShogunPanda

[ShogunPanda]

@nodejs/streams Do you think we should document this somewhere? Seems to be a "Broken Promises" problem rather than a specific stream problem.

[mcollina]

I think we should document this behavior as a caveat. The current behavior is correct, even if it's surprising.

[marco-ippolito]

@mcollina I could add a note in the documentation of Duplex.from and Readable.from warning that the use of array with promises leds to uncaught exceptions

[mcollina]

Go for it.